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()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!