def test__read_text(e, s, a, b):
    with make_hdfs() as hdfs:
        with hdfs.open('/tmp/test/text.1.txt', 'wb') as f:
            f.write('Alice 100\nBob 200\nCharlie 300'.encode())

        with hdfs.open('/tmp/test/text.2.txt', 'wb') as f:
            f.write('Dan 400\nEdith 500\nFrank 600'.encode())

        with hdfs.open('/tmp/test/other.txt', 'wb') as f:
            f.write('a b\nc d'.encode())

        b = db.read_text('hdfs:///tmp/test/text.*.txt',
                         collection=True)
        yield gen.sleep(0.5)
        assert not s.tasks

        import dask
        b.compute(get=dask.get)

        future = e.compute(b.str.strip().str.split().map(len))
        yield gen.sleep(0.5)
        result = yield future._result()
        assert result == [2, 2, 2, 2, 2, 2]

        b = db.read_text('hdfs:///tmp/test/other.txt', collection=True)
        b = e.persist(b)
        future = e.compute(b.str.split().concat())
        result = yield future._result()
        assert result == ['a', 'b', 'c', 'd']

        L = db.read_text('hdfs:///tmp/test/text.*.txt', collection=False)
        assert all(isinstance(x, Delayed) for x in L)

    def test_pool_shrinking_with_shrink_delay(self):
        db = yield self.build_pool(auto_shrink=True, shrink_delay=datetime.timedelta(seconds=1),
                                   shrink_period=datetime.timedelta(milliseconds=500))
        f1 = db.execute("SELECT 1")
        f2 = db.execute("SELECT 2")
        f3 = db.execute("SELECT 3")
        f4 = db.execute("SELECT 4")
        f5 = db.execute("SELECT 5")
        cursors = yield [f1, f2, f3, f4, f5]
        yield gen.sleep(.7)

        self.assertEqual(db.conns.total, 5)
        self.assertEqual(cursors[0].fetchone()[0], 1)
        self.assertEqual(cursors[1].fetchone()[0], 2)
        self.assertEqual(cursors[2].fetchone()[0], 3)
        self.assertEqual(cursors[3].fetchone()[0], 4)
        self.assertEqual(cursors[4].fetchone()[0], 5)

        f1 = db.execute("SELECT 1")
        f2 = db.execute("SELECT 2")
        f3 = db.execute("SELECT 3")
        cursors = yield [f1, f2, f3]
        self.assertEqual(cursors[0].fetchone()[0], 1)
        self.assertEqual(cursors[1].fetchone()[0], 2)
        self.assertEqual(cursors[2].fetchone()[0], 3)

        yield gen.sleep(1)

        self.assertEqual(db.conns.total, 3)

def test_failed_worker_without_warning(c, s, a, b):
    L = c.map(inc, range(10))
    yield _wait(L)

    original_process = a.process
    a.process.terminate()
    start = time()
    while a.process is original_process and not isalive(a.process):
        yield gen.sleep(0.01)
        assert time() - start < 10

    yield gen.sleep(0.5)

    start = time()
    while len(s.ncores) < 2:
        yield gen.sleep(0.01)
        assert time() - start < 10

    yield _wait(L)

    L2 = c.map(inc, range(10, 20))
    yield _wait(L2)
    assert all(len(keys) > 0 for keys in s.has_what.values())
    ncores2 = s.ncores.copy()

    yield c._restart()

    L = c.map(inc, range(10))
    yield _wait(L)
    assert all(len(keys) > 0 for keys in s.has_what.values())

    assert not (set(ncores2) & set(s.ncores))  # no overlap

    def run(self):
        QueueItem.run(self)

        try:
            sickrage.srLogger.info("Beginning manual search for: [" + self.segment.prettyName() + "]")
            self.started = True

            searchResult = searchProviders(self.show, [self.segment], True, self.downCurQuality)

            if searchResult:
                # just use the first result for now
                sickrage.srLogger.info("Downloading " + searchResult[0].name + " from " + searchResult[0].provider.name)
                self.success = snatchEpisode(searchResult[0])

                # give the CPU a break
                gen.sleep(cpu_presets[sickrage.srConfig.CPU_PRESET])

            else:
                notifications.message('No downloads were found',
                                      "Couldn't find a download for <i>%s</i>" % self.segment.prettyName())

                sickrage.srLogger.info("Unable to find a download for: [" + self.segment.prettyName() + "]")

        except Exception:
            sickrage.srLogger.debug(traceback.format_exc())

        ### Keep a list with the 100 last executed searches
        fifo(MANUAL_SEARCH_HISTORY, self, MANUAL_SEARCH_HISTORY_SIZE)

        if self.success is None:
            self.success = False

        self.finish()

def test_spawner_poll(db, io_loop):
    first_spawner = new_spawner(db)
    user = first_spawner.user
    io_loop.run_sync(first_spawner.start)
    proc = first_spawner.proc
    status = io_loop.run_sync(first_spawner.poll)
    assert status is None
    user.state = first_spawner.get_state()
    assert "pid" in user.state

    # create a new Spawner, loading from state of previous
    spawner = new_spawner(db, user=first_spawner.user)
    spawner.start_polling()

    # wait for the process to get to the while True: loop
    io_loop.run_sync(lambda: gen.sleep(1))
    status = io_loop.run_sync(spawner.poll)
    assert status is None

    # kill the process
    proc.terminate()
    for i in range(10):
        if proc.poll() is None:
            time.sleep(1)
        else:
            break
    assert proc.poll() is not None

    io_loop.run_sync(lambda: gen.sleep(2))
    status = io_loop.run_sync(spawner.poll)
    assert status is not None

def wait_for_http_server(url, timeout=10):
    """Wait for an HTTP Server to respond at url
    
    Any non-5XX response code will do, even 404.
    """
    loop = ioloop.IOLoop.current()
    tic = loop.time()
    client = AsyncHTTPClient()
    while loop.time() - tic < timeout:
        try:
            r = yield client.fetch(url, follow_redirects=False)
        except HTTPError as e:
            if e.code >= 500:
                # failed to respond properly, wait and try again
                if e.code != 599:
                    # we expect 599 for no connection,
                    # but 502 or other proxy error is conceivable
                    app_log.warn("Server at %s responded with error: %s", url, e.code)
                yield gen.sleep(0.1)
            else:
                app_log.debug("Server at %s responded with %s", url, e.code)
                return
        except (OSError, socket.error) as e:
            if e.errno not in {errno.ECONNABORTED, errno.ECONNREFUSED, errno.ECONNRESET}:
                app_log.warn("Failed to connect to %s (%s)", url, e)
            yield gen.sleep(0.1)
        else:
            return
    
    raise TimeoutError("Server at {url} didn't respond in {timeout} seconds".format(
        **locals()
    ))

    def instantiate(self, stream=None, environment=None):
        """ Start a local worker process

        Blocks until the process is up and the scheduler is properly informed
        """
        if environment:
            if not os.path.isabs(environment):
                environment = os.path.join(self.local_dir, environment)
            self.environment = environment

        with log_errors():
            if self.process and isalive(self.process):
                raise ValueError("Existing process still alive. Please kill first")

            if self.environment != nanny_environment:
                with tmpfile() as fn:
                    self.process = run_worker_subprocess(self.environment, self.ip,
                            self.scheduler.ip, self.scheduler.port, self.ncores,
                            self.port, self._given_worker_port, self.name,
                            self.memory_limit, self.loop, fn, self.quiet)

                    while not os.path.exists(fn):
                        yield gen.sleep(0.01)

                    while True:
                        try:
                            with open(fn) as f:
                                msg = json.load(f)
                            self.worker_port = msg['port']
                            self.worker_dir = msg['local_directory']
                            break
                        except JSONDecodeError:
                            yield gen.sleep(0.01)
            else:
                q = Queue()
                self.process = Process(target=run_worker_fork,
                                       args=(q, self.ip, self.scheduler.ip,
                                             self.scheduler.port, self.ncores,
                                             self.port, self._given_worker_port,
                                             self.local_dir, self.services, self.name,
                                             self.memory_limit))
                self.process.daemon = True
                self.process.start()
                while True:
                    try:
                        msg = q.get_nowait()
                        if isinstance(msg, Exception):
                            raise msg
                        self.worker_port = msg['port']
                        self.worker_dir = msg['dir']
                        assert self.worker_port
                        break
                    except queues.Empty:
                        yield gen.sleep(0.1)



            logger.info("Nanny %s:%d starts worker process %s:%d",
                        self.ip, self.port, self.ip, self.worker_port)
            raise gen.Return('OK')

    def __wait(self):
        log.debug("Waiting for events")
        while not (yield sleep(0.001)):
            try:
                state = self.__connection.poll()
            except QueryCanceledError:
                yield sleep(0.1)
                continue

            f = Future()

            def resolve(fileno, io_op):
                if f.running():
                    f.set_result(True)
                self.__io_loop.remove_handler(fileno)

            if state == psycopg2.extensions.POLL_OK:
                raise Return(True)

            elif state == psycopg2.extensions.POLL_READ:
                self.__io_loop.add_handler(self.__connection.fileno(), resolve, IOLoop.READ)
                yield f

            elif state == psycopg2.extensions.POLL_WRITE:
                self.__io_loop.add_handler(self.__connection.fileno(), resolve, IOLoop.WRITE)
                yield f

	def test_Royal_Seal(self):
		tu.print_test_header("test Royal Seal")
		royal_seal = prosperity.Royal_Seal(self.game, self.player1)
		workers_village = prosperity.Workers_Village(self.game, self.player1)
		copper = supply_cards.Copper(self.game, self.player1)
		self.player1.hand.add(royal_seal)
		self.player1.hand.add(copper)
		tu.add_many_to_hand(self.player1, workers_village, 3)

		workers_village.play()
		workers_village.play()
		workers_village.play()

		royal_seal.play()
		yield tu.send_input(self.player1, "buyCard", "Curse")
		self.assertTrue(self.player1.last_mode["mode"] == "select")
		yield tu.send_input(self.player1, "post_selection", ["Yes"])
		self.assertTrue(self.player1.deck[-1].title == "Curse")
		yield gen.sleep(.2)
		self.assertTrue(self.player1.last_mode["mode"] == "buy")

		yield tu.send_input(self.player1, "buyCard", "Silver")
		self.assertTrue(self.player1.last_mode["mode"] == "select")

		yield tu.send_input(self.player1, "post_selection", ["No"])
		self.assertTrue(self.player1.discard_pile[-1].title == "Silver")
		yield gen.sleep(.2)
		self.assertTrue(self.player1.last_mode["mode"] == "buy")
		yield tu.send_input(self.player1, "buyCard", "Mint")
		self.assertTrue(self.player1.last_mode["mode"] == "buy")

def send():
    c = BufferedClient()
    c.send({'host': 'localhost', 'service': 'buffer test'})
    yield sleep(0.5)
    c.send({'host': 'localhost', 'service': 'buffer test'})
    c.send({'host': 'localhost', 'service': 'buffer test'})
    yield sleep(1.5)

def test_serializers():
    with echo_server() as e:
        comm = yield connect(e.address)

        b = BatchedSend(interval='10ms', serializers=['msgpack'])
        b.start(comm)

        b.send({'x': to_serialize(123)})
        b.send({'x': to_serialize('hello')})
        yield gen.sleep(0.100)

        b.send({'x': to_serialize(lambda x: x + 1)})

        with captured_logger('distributed.protocol') as sio:
            yield gen.sleep(0.100)

        value = sio.getvalue()
        assert 'serialize' in value
        assert 'type' in value
        assert 'function' in value

        msg = yield comm.read()
        assert list(msg) == [{'x': 123}, {'x': 'hello'}]

        with pytest.raises(gen.TimeoutError):
            msg = yield gen.with_timeout(timedelta(milliseconds=100), comm.read())

    def _call(self, api_function, *args, **kwargs):
        """Calls the provided api_function in a background thread.

        If the api function returns a response cleanly, this will return it.
        If the api function raises an exception, this raises it up.

        For as long as the api function returns a boto2 or boto3
        rate limiting exception, this will backoff and try again.
        """
        while True:
            try:
                result = yield self._thread(api_function, *args, **kwargs)
                self._decrease_delay()
                raise gen.Return(result)
            except boto_exception.BotoServerError as e:
                # Boto2 exception.
                if e.error_code in self.boto2_throttle_strings:
                    self._increase_delay()
                    yield gen.sleep(self.delay)
                else:
                    self._decrease_delay()
                    raise e
            except botocore_exceptions.ClientError as e:
                # Boto3 exception.
                if e.response['Error']['Code'] == 'Throttling':
                    self._increase_delay()
                    yield gen.sleep(self.delay)
                else:
                    self._decrease_delay()
                    raise e

def test_advertiser_intermediate_failure():

    @gen.coroutine
    def handle(request, response):
        body = yield request.get_body()
        if hb.count == 2:
            # fail the second request only
            raise Exception('great sadness')
        response.write_body(body)

    hb = Fakebahn(handle)
    try:
        hb.start()
        adv = hyperbahn.Advertiser(
            'foo', TChannel('foo', known_peers=[hb.hostport]),
            interval_secs=0.2,
            interval_max_jitter_secs=0.0,
        )

        yield adv.start()
        assert 1 == hb.count

        yield gen.sleep(0.25)
        assert 2 == hb.count

        yield gen.sleep(0.25)
        assert 3 == hb.count
    finally:
        hb.stop()

 def get(self, path):
     log.debug("Slow Sending file: %s (This may take several minutes)" % self.request.path)
     seenfiles.append(os.path.basename(str(self.request.path)))
     f = open(os.path.join('static', path), 'rb')
     f.seek(0, 2)
     length = f.tell()
     f.seek(0, 0)
     self.set_header('Content-Length', str(length))
     self.set_header('X-Powered-By', 'Express')
     self.set_header('Transfer-Encoding', '')
     self.set_header('Content-Disposition',
             'attachment; filename="{}"'.format(path))
     self.set_header('Accept-Ranges', 'bytes')
     self.set_header('Cache-Control', 'public, max-age=0')
     self.set_header('Content-Type', 'application/octet-stream')
     chunk = f.read(10)
     self.write(chunk)
     yield self.flush()
     yield gen.sleep(0.9)
     while True:
         chunk = f.read(1400)
         if not chunk:
             break
         self.write(chunk)
         yield self.flush()
         print("  {}%   ".format(int(f.tell()*100/length)), end="\r", flush=True)
         yield gen.sleep(0.3)

    def f():
        nn = rpc(ip=n.ip, port=n.port)
        yield n._start()
        first_dir = n.worker_dir

        assert os.path.exists(first_dir)

        ww = rpc(ip=n.ip, port=n.worker_port)
        yield ww.update_data(data={'x': 1, 'y': 2})
        with ignoring(StreamClosedError):
            yield ww.compute(function=sys.exit, args=(0,), key='z')

        start = time()
        while n.process.is_alive():  # wait while process dies
            yield gen.sleep(0.01)
            assert time() - start < 2

        start = time()
        while not n.process.is_alive():  # wait while process comes back
            yield gen.sleep(0.01)
            assert time() - start < 2

        start = time()
        while n.worker_address not in c.ncores or n.worker_dir is None:
            yield gen.sleep(0.01)
            assert time() - start < 2

        second_dir = n.worker_dir

        yield n._close()
        assert not os.path.exists(second_dir)
        assert not os.path.exists(first_dir)
        assert first_dir != n.worker_dir
        c.stop()

def test_upload_large_file(c, s, a, b):
    pytest.importorskip('crick')
    yield gen.sleep(0.05)
    with rpc(a.address) as aa:
        yield aa.upload_file(filename='myfile.dat', data=b'0' * 100000000)
        yield gen.sleep(0.05)
        assert a.digests['tick-duration'].components[0].max() < 0.050

    def test_here_now_multiple_channels(self):
        ch1 = 'test-here-now-channel1'
        ch2 = 'test-here-now-channel2'
        self.pubnub.config.uuid = 'test-here-now-uuid'
        # print("connecting to the first...")
        yield connect_to_channel(self.pubnub, ch1)
        # print("...connected to the first")
        yield gen.sleep(1)
        # print("connecting to the second...")
        self.pubnub.subscribe().channels(ch2).execute()
        # print("...connected to the second")
        yield gen.sleep(5)
        env = yield self.pubnub.here_now() \
            .channels([ch1, ch2]) \
            .future()

        assert env.result.total_channels == 2
        assert env.result.total_occupancy >= 1

        channels = env.result.channels

        assert len(channels) == 2
        assert channels[0].occupancy >= 1
        assert channels[0].occupants[0].uuid == self.pubnub.uuid
        assert channels[1].occupancy >= 1
        assert channels[1].occupants[0].uuid == self.pubnub.uuid

        yield disconnect_from_channel(self.pubnub, [ch1, ch2])
        self.pubnub.stop()
        self.stop()

def test_worker_who_has_clears_after_failed_connection(c, s, a, b):
    n = Nanny(s.ip, s.port, ncores=2, loop=s.loop)
    n.start(0)

    start = time()
    while len(s.ncores) < 3:
        yield gen.sleep(0.01)
        assert time() < start + 5

    futures = c.map(slowinc, range(20), delay=0.01,
                    key=['f%d' % i for i in range(20)])
    yield wait(futures)

    result = yield c.submit(sum, futures, workers=a.address)
    for dep in set(a.dep_state) - set(a.task_state):
        a.release_dep(dep, report=True)

    n_worker_address = n.worker_address
    with ignoring(CommClosedError):
        yield c._run(os._exit, 1, workers=[n_worker_address])

    while len(s.workers) > 2:
        yield gen.sleep(0.01)

    total = c.submit(sum, futures, workers=a.address)
    yield total

    assert not a.has_what.get(n_worker_address)
    assert not any(n_worker_address in s for s in a.who_has.values())

    yield n._close()

    def do_connect(self, reconnect = False):
        # Create the socket and connect to the server
        if reconnect == True:
            logger.warning('Connection failed, retrying in %s seconds' % str(self._retrydelay))
            yield gen.sleep(self._retrydelay)

        while self._connection == None:
            logger.debug('Connecting to {}:{}'.format(config.ENVISALINKHOST, config.ENVISALINKPORT))
            try:
                self._connection = yield self.tcpclient.connect(config.ENVISALINKHOST, config.ENVISALINKPORT)
                self._connection.set_close_callback(self.handle_close)
            except StreamClosedError:
                #failed to connect, but got no connection object so we will loop here
                logger.warning('Connection failed, retrying in %s seconds' % str(self._retrydelay))
                yield gen.sleep(self._retrydelay)
                continue
            except gaierror:
                #could not resolve host provided, if this is a reconnect, will retry, if not, will fail
                if reconnect == True:
                    logger.warning('Connection failed, unable to resolve hostname %s, retrying in %s seconds' % (config.ENVISALINKHOST, str(self._retrydelay)))
                    yield gen.sleep(self._retrydelay)
                    continue
                else:
                    logger.warning('Connection failed, unable to resolve hostname %s.  Exiting due to incorrect hostname.' % config.ENVISALINKHOST)
                    sys.exit(0)

            try:
                line = yield self._connection.read_until(self._terminator)
            except StreamClosedError:
                #in this state, since the connection object isnt none, its going to throw the callback for handle_close so we just bomb out.
                #and let handle_close deal with this
                return

            logger.debug("Connected to %s:%i" % (config.ENVISALINKHOST, config.ENVISALINKPORT))
            self.handle_line(line)

def test_failed_worker_without_warning(c, s, a, b):
    L = c.map(inc, range(10))
    yield wait(L)

    original_pid = a.pid
    with ignoring(CommClosedError):
        yield c._run(os._exit, 1, workers=[a.worker_address])
    start = time()
    while a.pid == original_pid:
        yield gen.sleep(0.01)
        assert time() - start < 10

    yield gen.sleep(0.5)

    start = time()
    while len(s.ncores) < 2:
        yield gen.sleep(0.01)
        assert time() - start < 10

    yield wait(L)

    L2 = c.map(inc, range(10, 20))
    yield wait(L2)
    assert all(len(keys) > 0 for keys in s.has_what.values())
    ncores2 = dict(s.ncores)

    yield c._restart()

    L = c.map(inc, range(10))
    yield wait(L)
    assert all(len(keys) > 0 for keys in s.has_what.values())

    assert not (set(ncores2) & set(s.ncores))  # no overlap