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

How to use check_db_socket method in autotest

Best Python code snippet using autotest_python

openvswitch.py

Source:openvswitch.py

...311        if not exists:312            logging.error("OpenVSwitch database file %s not exists.",313                          self.db_path)314        return exists315    def check_db_socket(self):316        """317        Check if db socket exists.318        """319        exists = os.path.exists(self.db_socket)320        if not exists:321            logging.error("OpenVSwitch database socket file %s not exists.",322                          self.db_socket)323        return exists324    def check(self):325        return (self.check_db_daemon() and self.check_switch_daemon() and326                self.check_db_file() and self.check_db_socket())327    def init_system(self):328        """329        Create new dbfile without any configuration.330        """331        sm = factory(ServiceManager)()332        try:333            if linux_modules.load_module("openvswitch"):334                sm.restart("openvswitch")335        except process.CmdError:336            logging.error("Service OpenVSwitch is probably not"337                          " installed in system.")338            raise339        self.pid_files_path = "/var/run/openvswitch/"340    def clean(self):...

test_simple_architecture.py

Source:test_simple_architecture.py

...208    check_output_file()209################210# SOCKET INPUT #211################212def check_db_socket():213    json_reports = extract_rapl_reports_with_2_sockets(10)214    mongo = pymongo.MongoClient(MONGO_URI)215    c_output = mongo[MONGO_DATABASE_NAME][MONGO_OUTPUT_COLLECTION_NAME]216    assert c_output.count_documents({}) == 20217    for report in json_reports:218        ts = datetime.strptime(report['timestamp'], "%Y-%m-%dT%H:%M:%S.%f")219        assert c_output.count_documents(220            {'timestamp': ts, 'sensor': report['sensor'],221             'target': report['target']}) == 2222def test_run_socket_to_mongo(mongo_database, unused_tcp_port, shutdown_system):223    config = {'verbose': True,224              'stream': False,225              'actor_system': SIMPLE_SYSTEM_IMP,226              'output': {'test_pusher': {'type': 'mongodb',227                                         'model': 'PowerReport',228                                         'uri': MONGO_URI,229                                         'db': MONGO_DATABASE_NAME,230                                         'collection': MONGO_OUTPUT_COLLECTION_NAME}},231              'input': {'test_puller': {'type': 'socket',232                                        'port': unused_tcp_port,233                                        'model': 'HWPCReport'}},234              }235    supervisor = Supervisor(verbose_mode=config['verbose'], system_imp=config['actor_system'])236    launch_simple_architecture(config, supervisor)237    client = ClientThread(extract_rapl_reports_with_2_sockets(10), unused_tcp_port)238    client.start()239    supervisor.monitor()240    check_db_socket()241def test_run_socket_with_delay_between_message_to_mongo(mongo_database, unused_tcp_port, shutdown_system):242    config = {'verbose': True,243              'stream': False,244              'actor_system': SIMPLE_SYSTEM_IMP,245              'output': {'test_pusher': {'type': 'mongodb',246                                         'model': 'PowerReport',247                                         'uri': MONGO_URI,248                                         'db': MONGO_DATABASE_NAME,249                                         'collection': MONGO_OUTPUT_COLLECTION_NAME}},250              'input': {'test_puller': {'type': 'socket',251                                        'port': unused_tcp_port,252                                        'model': 'HWPCReport'}},253              }254    supervisor = Supervisor(verbose_mode=config['verbose'], system_imp=config['actor_system'])255    launch_simple_architecture(config, supervisor)256    client = ClientThreadDelay(extract_rapl_reports_with_2_sockets(10), unused_tcp_port)257    client.start()258    supervisor.monitor()259    check_db_socket()260##############261# Socket to CSV #262##############263def check_output_file2():264    input_reports = extract_rapl_reports_with_2_sockets(10)265    output_file = open(OUTPUT_PATH + 'test_sensor-all/PowerReport.csv', 'r')266    # line count267    l_output = -1268    for _ in output_file:269        l_output += 1270    assert len(input_reports) * 2 == l_output271    output_file.seek(0)272    output_file.readline()273    # split socket0 report from socket1 report...

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.