KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use teardown_stream method in Airtest

Best Python code snippet using Airtest

resource.py

Source:resource.py

...44        return NOT_DONE_YET45    def setup_stream(self, request):46        return self.worker.register_client(self._rk, self.message_class,47                                           self._callback)48    def teardown_stream(self, err):49        if not (err is None or err.trap(ConnectionDone)):50            log.error(err)51        log.info('Unregistering: %s, %s' % (self._rk, err.getErrorMessage()))52        return self.worker.unregister_client(self._rk, self._callback)53    def publish(self, request, message):54        line = u'%s\n' % (message.to_json(),)55        request.write(line.encode(self.encoding))56class EventStream(BaseResource, StreamResourceMixin):57    message_class = TransportEvent58    routing_key = '%(transport_name)s.stream.event.%(conversation_key)s'59    def __init__(self, worker, conversation_key):60        BaseResource.__init__(self, worker, conversation_key)61        self.setup_stream_resource(worker, conversation_key)62class MessageStream(MessageResource, StreamResourceMixin):...

kafka_demo.py

Source:kafka_demo.py

...58            row["created"] = row["created"].strftime("%Y-%m-%d %H:%M:%S")59            producer.send(topic_name, json.dumps(row).encode())60            sleep(0.1)61        iteration += 162def teardown_stream(topic_name, servers=["localhost:9092"]):63    try:64        admin = KafkaAdminClient(bootstrap_servers=servers)65        print(admin.delete_topics([topic_name]))66        print(f"Topic {topic_name} deleted")67    except Exception as e:68        print(str(e))69        pass70if __name__ == "__main__":71    parsed_args = vars(args)72    mode = parsed_args["mode"]73    servers = parsed_args["bootstrap_servers"]74    teardown_stream("drivers", [servers])75    if mode == "setup":...

test_minicap.py

Source:test_minicap.py

...19                cnt += 120        return cnt21    @classmethod22    def tearDownClass(cls):23        cls.minicap.teardown_stream()24class TestMinicap(TestMinicapBase):25    def test_get_display_info(self):26        info = self.minicap.get_display_info()27        self.assertIsInstance(info, dict)28    def test_get_frame(self):29        frame = self.minicap.get_frame()30        frame = string_2_img(frame)31        self.assertIsInstance(frame, ndarray)32    def test_get_frames(self):33        frame = self.minicap.get_frame_from_stream()34        frame = string_2_img(frame)35        self.assertIsInstance(frame, ndarray)36    def test_rotation(self):37        self.dev.keyevent("HOME")38        time.sleep(1)39        frame_vertical = self.minicap.get_frame_from_stream()40        frame_vertical = string_2_img(frame_vertical)41        self.dev.start_app(PKG)42        time.sleep(1)43        frame_horizontal = self.minicap.get_frame_from_stream()44        frame_horizontal = string_2_img(frame_horizontal)45        self.assertEqual(frame_vertical.shape[0], frame_horizontal.shape[1])46        self.assertEqual(frame_vertical.shape[1], frame_horizontal.shape[0])47        self.assertEqual(self._count_server_proc(), 1)48class TestMinicapSetup(TestMinicapBase):49    def test_0_install(self):50        self.minicap.uninstall()51        self.minicap.install()52    def test_teardown(self):53        self.minicap.get_frame_from_stream()54        self.minicap.teardown_stream()55        self.assertEqual(self._count_server_proc(), 0)56if __name__ == '__main__':...

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.