How to use test_server_exit method in avocado

Best Python code snippet using avocado_python

test_source_doc_client.py

Source:test_source_doc_client.py

...135    def setUp(self):136        self.env = TTestEnv(TestExit.port)137    def tearDown(self):138        self.env.tearDown(stop_server=False)139    def test_server_exit(self):140        file_data1 = b"12345_1"141        with open("test8484.txt", "wb") as outp:142            outp.write(file_data1)143        self.assertTrue(self.env.client.send_file("test8484.txt"))144        stats = self.env.client.get_stats(timeout=1)145        self.assertEqual(1, stats['source_doc_count'])146        with open(TSourceDocHTTPServer.stop_file, "w") as outp:147            outp.write(".")148        time.sleep(20)149        stats = self.env.client.get_stats(timeout=1)150        self.assertIsNone(stats)151class TestSourceDocArchive(TestCase):152    def setUp(self):153        self.env = TTestEnv(8496, 4, use_archives=True)...

test_services.py

Source:test_services.py

...23        self.s.main()24@pytest.fixture25def r():26    return redis.Redis('localhost')27def test_server_exit(r):28    s = Server('localhost', 6379, 0, None)29    ServerThread(s).start()30    ExitMessage(s.id).send(r)31def test_gatherer_exit(r):32    s = Gatherer()33    ServerThread(s).start()34    ExitMessage(s.id).send(r)35class ListenTestServer(Server):36    def __init__(self, r):37        super().__init__(redis_client=r)38        self.handler.register(EpisodeMessage, self.episode)39        self.handler.register(TrainCompleteMessage, self.handle_train_complete)40    def episode(self, msg):41        print(f'Episode {msg.steps}')...

test_servers.py

Source:test_servers.py

...27            servers.supervise_server(self.queue, server_tasks),28            timeout=0.01,29        )30    @kernels.with_kernel31    def test_server_exit(self):32        server_task = self.queue.spawn(noop)33        self.assert_state(False, 1, [])34        self.assertIsNone(self.run_supervisor([server_task]))35        self.assert_state(True, 0, [r'no op', r'server task exit: '])36        self.assertIsNone(server_task.get_result_nonblocking())37        self.assertFalse(tasks.get_all_tasks(), [])38    @kernels.with_kernel39    def test_server_error(self):40        server_task = self.queue.spawn(raises(ValueError('some error')))41        self.assert_state(False, 1, [])42        with self.assertRaisesRegex(43            servers.ServerError,44            r'server task error: ',45        ):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.