Bài viết trước đã trình bày về cấu trúc Suite và quá trình chuyển đổi file sang định dạng dữ liệu Robot. Bài này sẽ tiếp tục phân tích sâu hơn.
Quay lại hàm entry point
Robotframework().main():
1 def main(self, datasources, **options):
2 settings = RobotSettings(options)
3 LOGGER.register_console_logger(**settings.console_logger_config)
4 LOGGER.info('Settings:\n%s' % unicode(settings))
5 suite = TestSuiteBuilder(settings['SuiteNames'],
6 settings['WarnOnSkipped'],
7 settings['RunEmptySuite']).build(*datasources)
8 suite.configure(**settings.suite_config)
9 result = suite.run(settings)
10 LOGGER.info("Tests execution ended. Statistics:\n%s"
11 % result.suite.stat_message)
12 if settings.log or settings.report or settings.xunit:
13 writer = ResultWriter(settings.output if settings.log else result)
14 writer.write_results(settings.get_rebot_settings())
15 return result.return_code
suite.run() là phương thức của lớp TestSuite trong robot/running/model.py. Cấu trúc cụ thể đã được đề cập trong bài trước.
1 def run(self, settings=None, **options):
2 with STOP_SIGNAL_MONITOR:
3 IMPORTER.reset()
4 pyloggingconf.initialize(settings['LogLevel'])
5 init_global_variables(settings)
6 output = Output(settings)
7 runner = Runner(output, settings)
8 self.visit(runner)
9 output.close(runner.result)
10 return runner.result
with STOP_SIGNAL_MONITOR. Đây là một cơ chế đặc biệt của Python. Nguyên lý hoạt động của with như sau:
1. Sau khi câu lệnh sau with được thực thi, phương thức __enter__() của đối tượng được gọi. Giá trị trả về của phương thức này sẽ được gán cho biến sau as.
2. Khi toàn bộ khối mã sau with kết thúc, phương thức __exit__() của đối tượng trả về sẽ được gọi.
Ví dụ minh họa:
1 class Demo:
2 def __enter__(self):
3 print "__enter__"
4 return self
5 def __exit__(self, *exc_info):
6 print "__exit__"
7 def show(self, msg):
8 print msg
9 with Demo() as d:
10 d.show('test')
1 __enter__
2 test
3 __exit__
_StopSignalMonitor:
1 class _StopSignalMonitor(object):
2
3 def __init__(self):
4 self._signal_count = 0
5 self._running_keyword = False
6 self._orig_sigint = None
7 self._orig_sigterm = None
8
9 def __call__(self, signum, frame):
10 self._signal_count += 1
11 LOGGER.info('Received signal: %s.' % signum)
12 if self._signal_count > 1:
13 sys.__stderr__.write('Execution forcefully stopped.\n')
14 raise SystemExit()
15 sys.__stderr__.write('Second signal will force exit.\n')
16 if self._running_keyword and not sys.platform.startswith('java'):
17 self._stop_execution_gracefully()
18
19 def _stop_execution_gracefully(self):
20 raise ExecutionFailed('Execution terminated by signal', exit=True)
21
22 def start(self):
23 self.__enter__()
24
25 def __enter__(self):
26 if signal:
27 self._orig_sigint = signal.getsignal(signal.SIGINT)
28 self._orig_sigterm = signal.getsignal(signal.SIGTERM)
29 for signum in signal.SIGINT, signal.SIGTERM:
30 self._register_signal_handler(signum)
31 return self
32
33 def __exit__(self, *exc_info):
34 if signal:
35 signal.signal(signal.SIGINT, self._orig_sigint)
36 signal.signal(signal.SIGTERM, self._orig_sigterm)
37
38 def _register_signal_handler(self, signum):
39 try:
40 signal.signal(signum, self)
41 except (ValueError, IllegalArgumentException), err:
42 if currentThread().getName() == 'MainThread':
43 self._warn_about_registeration_error(signum, err)
44
45 def _warn_about_registeration_error(self, signum, err):
46 name, ctrlc = {signal.SIGINT: ('INT', 'or with Ctrl-C '),
47 signal.SIGTERM: ('TERM', '')}[signum]
48 LOGGER.warn('Registering signal %s failed. Stopping execution '
49 'gracefully with this signal %sis not possible. '
50 'Original error was: %s' % (name, ctrlc, err))
51
52 def start_running_keyword(self, in_teardown):
53 self._running_keyword = True
54 if self._signal_count and not in_teardown:
55 self._stop_execution_gracefully()
56
57 def stop_running_keyword(self):
58 self._running_keyword = False
__enter__()**: Kiểm tra module signal đã được import thành công chưa. Nếu có, lấy trình xử lý tín hiệu hiện tại bằng signal.getsignal() cho SIGINT (kích hoạt bởi Ctrl+C) và SIGTERM (yêu cầu kết thúc từ tiến trình). Khi chương trình mới khởi tạo, getsignal trả về 0 (mặc định). Sau đó, gọi _register_signal_handler để đăng ký xử lý cho cả hai tín hiệu. Lưu ý: tham số thứ hai của signal.signal() là self, nghĩa là hàm xử lý chính là self() (tương ứng với _StopSignalMonitor.__call__). Cuối cùng, kiểm tra xem luồng hiện tại có phải là luồng chính không (vì tín hiệu chỉ có thể đặt trong luồng chính).
2. **Thoát bình thường**: Không có tín hiệu được gửi. Kết thúc khối with, gọi __exit__(). Phương thức này khôi phục lại trình xử lý tín hiệu ban đầu (lưu trong _orig_sigint và _orig_sigterm) thông qua signal.signal(). Các giá trị này ban đầu là 0 (signal.SIG_DFL – hành động mặc định). Kết thúc with.
3. **Khi nhận tín hiệu SIGINT hoặc SIGTERM**: Hàm __call__ được gọi. Logic như sau:
- Tăng biến đếm tín hiệu (_signal_count).
- Nếu nhận được tín hiệu lần thứ hai (>1), dừng ngay lập tức bằng SystemExit().
- Nếu lần đầu, in thông báo và kiểm tra xem có keyword nào đang chạy và không phải môi trường Java không. Nếu đúng, gọi _stop_execution_gracefully() để ném ngoại lệ ExecutionFailed (kết thúc gracefully sau khi chạy teardown). Sau đó, __exit__() được gọi để khôi phục tín hiệu mặc định.
Như vậy, để dừng quá trình chạy test, người dùng cần nhấn Ctrl+C hai lần: lần đầu kích hoạt kết thúc an toàn (sau teardown), lần thứ hai buộc dừng ngay.