Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _configure_logging method in autotest

Best Python code snippet using autotest_python

test_worker.py

Source:test_worker.py

...451    def test_msg(self, m_logging_config, config, worker_name, exp_msg):452        worker = NowcastWorker(worker_name, "description")453        worker.config._dict = config454        worker._parsed_args = SimpleNamespace(debug=False)455        msg = worker._configure_logging()456        assert msg == exp_msg457    @pytest.mark.parametrize(458        "config, worker_name, exp_msg",459        [460            (461                filesystem_logging_config,462                "test_worker",463                "writing log messages to local file system",464            ),465            (466                zmq_logging_config_ports_list,467                "test_worker",468                "publishing log messages to tcp://*:4345",469            ),470            (471                zmq_logging_config_specific_port,472                "test_worker",473                "publishing log messages to tcp://*:4347",474            ),475            (476                zmq_logging_config_remote_worker,477                "remote_worker",478                "publishing log messages to tcp://*:4347",479            ),480            (481                zmq_logging_config_different_hosts_same_port,482                "remote_worker",483                "publishing log messages to tcp://*:4347",484            ),485            (486                zmq_logging_config_remote_worker_2_ports,487                "remote_worker",488                "publishing log messages to tcp://*:4347",489            ),490        ],491    )492    def test_msg_debug_mode(self, m_logging_config, config, worker_name, exp_msg):493        worker = NowcastWorker(worker_name, "description")494        worker.config._dict = config495        worker._parsed_args = SimpleNamespace(debug=True)496        msg = worker._configure_logging()497        assert msg == "**debug mode** writing log messages to console"498    @pytest.mark.parametrize(499        "config, worker_name",500        [501            (filesystem_logging_config, "test_worker"),502            (zmq_logging_config_ports_list, "test_worker"),503            (zmq_logging_config_specific_port, "test_worker"),504            (zmq_logging_config_remote_worker, "remote_worker"),505            (zmq_logging_config_different_hosts_same_port, "remote_worker"),506            (zmq_logging_config_remote_worker_2_ports, "remote_worker"),507        ],508    )509    def test_logger_name(self, m_logging_config, config, worker_name):510        worker = NowcastWorker(worker_name, "description")511        worker.config._dict = config512        worker._parsed_args = SimpleNamespace(debug=False)513        worker._configure_logging()514        assert worker.logger.name == worker_name515    @pytest.mark.parametrize(516        "config, worker_name",517        [518            (filesystem_logging_config, "test_worker"),519            (zmq_logging_config_ports_list, "test_worker"),520            (zmq_logging_config_specific_port, "test_worker"),521            (zmq_logging_config_remote_worker, "remote_worker"),522            (zmq_logging_config_different_hosts_same_port, "remote_worker"),523            (zmq_logging_config_remote_worker_2_ports, "remote_worker"),524        ],525    )526    def test_logging_dictConfig(self, m_logging_config, config, worker_name):527        worker = NowcastWorker(worker_name, "description")528        worker.config._dict = config529        worker._parsed_args = SimpleNamespace(debug=False)530        worker._configure_logging()531        if "publisher" in worker.config["logging"]:532            m_logging_config.dictConfig.assert_called_once_with(533                worker.config["logging"]["publisher"]534            )535        else:536            m_logging_config.dictConfig.assert_called_once_with(537                worker.config["logging"]538            )539    @patch("nemo_nowcast.worker.logging")540    def test_different_hosts_different_ports(self, m_logging, m_logging_config):541        worker = NowcastWorker("remote_worker", "description")542        worker.config._dict = self.zmq_logging_config_different_hosts_different_ports543        worker._parsed_args = SimpleNamespace(debug=False)544        m_handler = Mock(name="m_zmq_handler", spec=zmq.log.handlers.PUBHandler)545        m_logging.getLogger.return_value = Mock(root=Mock(handlers=[m_handler]))546        with pytest.raises(WorkerError):547            worker._configure_logging()548    @pytest.mark.parametrize(549        "config, worker_name, exception, exp_msg",550        [551            (552                zmq_logging_config_ports_list,553                "test_worker",554                zmq.ZMQError,555                "publishing log messages to tcp://*:4346",556            ),557            (558                zmq_logging_config_ports_list,559                "test_worker",560                ValueError,561                "publishing log messages to tcp://*:4346",562            ),563            (564                zmq_logging_config_remote_worker_2_ports,565                "remote_worker",566                zmq.ZMQError,567                "publishing log messages to tcp://*:4348",568            ),569            (570                zmq_logging_config_remote_worker_2_ports,571                "remote_worker",572                ValueError,573                "publishing log messages to tcp://*:4348",574            ),575        ],576    )577    def test_port_in_use(578        self, m_logging_config, config, worker_name, exception, exp_msg579    ):580        worker = NowcastWorker(worker_name, "description")581        worker._socket = Mock(name="zmq_socket")582        worker.config._dict = config583        worker._parsed_args = SimpleNamespace(debug=False)584        m_logging_config.dictConfig.side_effect = (exception, None)585        msg = worker._configure_logging()586        assert msg == exp_msg587    @pytest.mark.parametrize(588        "config, worker_name, exception",589        [590            (zmq_logging_config_ports_list, "test_worker", zmq.ZMQError),591            (zmq_logging_config_ports_list, "test_worker", ValueError),592            (zmq_logging_config_remote_worker_2_ports, "remote_worker", zmq.ZMQError),593            (zmq_logging_config_remote_worker_2_ports, "remote_worker", ValueError),594        ],595    )596    def test_all_ports_in_use(self, m_logging_config, config, worker_name, exception):597        worker = NowcastWorker(worker_name, "description")598        worker._socket = Mock(name="zmq_socket")599        worker.config._dict = config600        worker._parsed_args = SimpleNamespace(debug=False)601        m_logging_config.dictConfig.side_effect = exception602        with pytest.raises(WorkerError):603            worker._configure_logging()604    @patch("nemo_nowcast.worker.logging")605    def test_zmq_handler_root_topic(self, m_logging, m_logging_config):606        worker = NowcastWorker("test_worker", "description")607        worker.config._dict = self.zmq_logging_config_ports_list608        worker._parsed_args = SimpleNamespace(debug=False)609        m_handler = Mock(name="m_zmq_handler", spec=zmq.log.handlers.PUBHandler)610        m_logging.getLogger.return_value = Mock(root=Mock(handlers=[m_handler]))611        worker._configure_logging()612        assert m_handler.root_topic == "test_worker"613    @patch("nemo_nowcast.worker.logging")614    def test_zmq_handler_formatters(self, m_logging, m_logging_config):615        worker = NowcastWorker("test_worker", "description")616        worker.config._dict = self.zmq_logging_config_ports_list617        worker._parsed_args = SimpleNamespace(debug=False)618        m_handler = Mock(name="m_zmq_handler", spec=zmq.log.handlers.PUBHandler)619        m_logging.getLogger.return_value = Mock(root=Mock(handlers=[m_handler]))620        worker._configure_logging()621        expected = {622            m_logging.DEBUG: m_logging.Formatter("%(message)s\n"),623            m_logging.INFO: m_logging.Formatter("%(message)s\n"),624            m_logging.WARNING: m_logging.Formatter("%(message)s\n"),625            m_logging.ERROR: m_logging.Formatter("%(message)s\n"),626            m_logging.CRITICAL: m_logging.Formatter("%(message)s\n"),627        }628        assert m_handler.formatters == expected629    @pytest.mark.parametrize(630        "config, worker_name",631        [632            (filesystem_logging_config, "test_worker"),633            (zmq_logging_config_ports_list, "test_worker"),634            (zmq_logging_config_specific_port, "test_worker"),635            (zmq_logging_config_remote_worker, "remote_worker"),636        ],637    )638    @patch("nemo_nowcast.worker.logging")639    def test_debug_mode_console_logging_only(640        self, m_logging, m_logging_config, config, worker_name641    ):642        worker = NowcastWorker(worker_name, "description")643        if "publisher" in config["logging"]:644            p_config = patch.dict(645                config["logging"]["publisher"]["handlers"], {"console": {"level": 1000}}646            )647        else:648            p_config = patch.dict(649                config["logging"]["handlers"], {"console": {"level": 1000}}650            )651        worker._parsed_args = SimpleNamespace(debug=True)652        m_file_handler = Mock(name="m_file_handler")653        m_console_handler = Mock(name="m_console_handler")654        m_console_handler.name = "console"655        m_logging.getLogger.return_value = Mock(656            root=Mock(handlers=[m_file_handler, m_console_handler])657        )658        with p_config:659            worker.config._dict = config660            worker._configure_logging()661        m_console_handler.setLevel.assert_called_once_with(m_logging.DEBUG)662        m_file_handler.setLevel.assert_called_once_with(1000)663    @pytest.mark.parametrize(664        "config, worker_name",665        [666            (filesystem_logging_config, "test_worker"),667            (zmq_logging_config_ports_list, "test_worker"),668            (zmq_logging_config_specific_port, "test_worker"),669            (zmq_logging_config_remote_worker, "remote_worker"),670        ],671    )672    @patch("nemo_nowcast.worker.logging")673    def test_debug_mode_no_console_handler(674        self, m_logging, m_logging_config, config, worker_name675    ):676        worker = NowcastWorker(worker_name, "description")677        worker.config._dict = config678        worker._parsed_args = SimpleNamespace(debug=True)679        m_file_handler = Mock(name="m_file_handler")680        m_logging.getLogger.return_value = Mock(root=Mock(handlers=[m_file_handler]))681        worker._configure_logging()682        m_file_handler.setLevel.assert_called_once_with(100)683class TestInitZmqInterface:684    """Unit testss for NowcastWorker._init_zmq_interface method.685    """686    def test_debug_mode(self):687        worker = NowcastWorker("worker_name", "description")688        worker._parsed_args = Mock(debug=True)689        worker.logger = Mock(name="logger")690        worker._context = Mock(name="context")691        worker._init_zmq_interface()692        assert worker.logger.debug.call_count == 1693        assert not worker._context.socket.called694    def test_socket(self):695        worker = NowcastWorker("worker_name", "description")...

cli.py

Source:cli.py

...29        '--debug', action='store_true', help="Show logging from DEBUG or above")30    # here you'd setup any other arguments/flags you cared about31    # return result of parsing cli args32    return parser.parse_args()33def _configure_logging(log_level=None):34    """Configure logging"""35    logging_format = '%(asctime)s %(levelname)s %(module)s %(lineno)d %(name)s - %(message)s'36    if log_level is None:37        log_level = logging.WARNING38    logging.basicConfig(stream=sys.stderr, level=log_level, format=logging_format)39def _update_logging_config(parsed_args):40    """Update the logging config based on the parsed cli args"""41    if parsed_args.verbose:42        _configure_logging(log_level=logging.INFO)43    elif parsed_args.debug:44        _configure_logging(log_level=logging.DEBUG)45def main():46    """Command line entry point"""47    _configure_logging()48    parsed_args = _parse_cli_args()49    _update_logging_config(parsed_args)50    _LOGGER.info("Begin running")51    _LOGGER.debug("Here is where you'd put calls to functions etc that did the work of your app")...

test.py

Source:test.py

...7import logging8import inotify.adapters9_DEFAULT_LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'10_LOGGER = logging.getLogger(__name__)11def _configure_logging():12    _LOGGER.setLevel(logging.DEBUG)13    ch = logging.StreamHandler()14    formatter = logging.Formatter(_DEFAULT_LOG_FORMAT)15    ch.setFormatter(formatter)16    _LOGGER.addHandler(ch)17def _main():18#    paths = [19#        '/tmp',20#    ]21#    22#    i = Inotify(paths=paths)23    i = inotify.adapters.Inotify()24    i.add_watch('/tmp')25    try:26        for event in i.event_gen():27            if event is not None:28                (header, type_names, path, filename) = event29                _LOGGER.info("WD=(%d) MASK=(%d) COOKIE=(%d) LEN=(%d) MASK->NAMES=%s "30                             "FILENAME=[%s]", 31                             header.wd, header.mask, header.cookie, header.len, type_names, 32                             filename)33    finally:34        i.remove_watch('/tmp')35if __name__ == '__main__':36    _configure_logging()...

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.