/
githubmirror
/
oppia
Обзор
Документация
Войти
/
githubmirror
/
oppia
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
v3.2.3
scripts/servers_test.py
1 018 строк
41 KB
JG-11
Fix part of #13956: remove python_utils.string_io() and python_utils.url_open() (#14299)
02 дек 2021, 17:29
Не верифицирован
02 дек 2021, 17:29
014cdee
Код
Авторство
О чём код?
# Copyright 2019 The Oppia Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS-IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unit tests for scripts/servers.py.""" from __future__ import annotations import collections import contextlib import io import logging import os import re import shutil import signal import subprocess import sys import threading import time from unittest import mock from core import utils from core.tests import test_utils from scripts import common from scripts import scripts_test_utils from scripts import servers import psutil class ManagedProcessTests(test_utils.TestBase): # Helper class for improving the readability of tests. POPEN_CALL = ( collections.namedtuple('POPEN_CALL', ['program_args', 'kwargs'])) def setUp(self): super(ManagedProcessTests, self).setUp() self.exit_stack = contextlib.ExitStack() def tearDown(self): try: self.exit_stack.close() finally: super(ManagedProcessTests, self).tearDown() @contextlib.contextmanager def swap_popen(self, unresponsive=False, num_children=0, outputs=()): """Returns values for inspecting and mocking calls to psutil.Popen. Args: unresponsive: bool. Whether the processes created by the mock will stall when asked to terminate. num_children: int. The number of child processes the process created by the mock should create. Children inherit the same termination behavior. outputs: list(bytes). The outputs of the mock process. Returns: Context manager. A context manager in which calls to psutil.Popen() create a simple program that waits and then exits. Yields: list(POPEN_CALL). A list with the most up-to-date arguments passed to psutil.Popen from within the context manager returned. """ popen_calls = [] def mock_popen(program_args, **kwargs): """Mock of psutil.Popen that creates processes using os.fork(). The processes created will always terminate within ~1 minute. Args: program_args: list(*). Unused program arguments that would otherwise be passed to Popen. **kwargs: dict(str: *). Keyword arguments passed to Popen. Returns: PopenStub. The return value of psutil.Popen. """ popen_calls.append(self.POPEN_CALL(program_args, kwargs)) pid = 1 stdout = b''.join(b'%b\n' % o for o in outputs) child_procs = [ scripts_test_utils.PopenStub(pid=i, unresponsive=unresponsive) for i in range(pid + 1, pid + 1 + num_children) ] return scripts_test_utils.PopenStub( pid=pid, stdout=stdout, unresponsive=unresponsive, child_procs=child_procs) with self.swap(psutil, 'Popen', mock_popen): yield popen_calls @contextlib.contextmanager def swap_managed_cloud_datastore_emulator_io_operations( self, data_dir_exists): """Safely swaps IO operations used by managed_cloud_datastore_emulator. Args: data_dir_exists: bool. Return value of os.path.exists(DATA_DIR). Yields: tuple(CallCounter, CallCounter). CallCounter instances for rmtree and makedirs. """ old_exists = os.path.exists old_rmtree = shutil.rmtree old_makedirs = os.makedirs is_data_dir = lambda p: p == common.CLOUD_DATASTORE_EMULATOR_DATA_DIR new_exists = ( lambda p: data_dir_exists if is_data_dir(p) else old_exists(p)) new_rmtree = test_utils.CallCounter( lambda p, **kw: None if is_data_dir(p) else old_rmtree(p, **kw)) new_makedirs = test_utils.CallCounter( lambda p, **kw: None if is_data_dir(p) else old_makedirs(p, **kw)) with contextlib.ExitStack() as exit_stack: exit_stack.enter_context(self.swap(os.path, 'exists', new_exists)) exit_stack.enter_context(self.swap(shutil, 'rmtree', new_rmtree)) exit_stack.enter_context(self.swap(os, 'makedirs', new_makedirs)) yield new_rmtree, new_makedirs def assert_proc_was_managed_as_expected( self, logs, pid, manager_should_have_sent_terminate_signal=True, manager_should_have_sent_kill_signal=False): """Asserts that the process ended as expected. Args: logs: list(str). The logs emitted during the process's lifetime. pid: int. The process ID to inspect. manager_should_have_sent_terminate_signal: bool. Whether the manager should have sent a terminate signal to the process. manager_should_have_sent_kill_signal: bool. Whether the manager should have sent a kill signal to the process. """ proc_pattern = r'[A-Za-z ]+\((name="[A-Za-z]+", )?pid=%d\)' % (pid,) expected_patterns = [] if manager_should_have_sent_terminate_signal: expected_patterns.append(r'Terminating %s\.\.\.' % proc_pattern) if manager_should_have_sent_kill_signal: expected_patterns.append(r'Forced to kill %s!' % proc_pattern) else: expected_patterns.append(r'%s has already ended\.' % proc_pattern) logs_with_pid = [msg for msg in logs if re.search(proc_pattern, msg)] if expected_patterns and not logs_with_pid: self.fail(msg='%r has no match in logs=%r' % (proc_pattern, logs)) self.assert_matches_regexps(logs_with_pid, expected_patterns) def test_does_not_raise_when_psutil_not_in_path(self): self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(sys, 'path', [])) # Entering the context should not raise. self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) def test_concats_command_args_when_shell_is_true(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a', 1], timeout_secs=10, shell=True)) self.exit_stack.close() self.assert_proc_was_managed_as_expected(logs, proc.pid) self.assertEqual(popen_calls, [self.POPEN_CALL('a 1', {'shell': True})]) def test_passes_command_args_as_list_of_strings_when_shell_is_false(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a', 1], shell=False, timeout_secs=10)) self.exit_stack.close() self.assert_proc_was_managed_as_expected(logs, proc.pid) self.assertEqual( popen_calls, [self.POPEN_CALL(['a', '1'], {'shell': False})]) def test_filters_empty_strings_from_command_args_when_shell_is_true(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['', 'a', '', 1], timeout_secs=10, shell=True)) self.exit_stack.close() self.assert_proc_was_managed_as_expected(logs, proc.pid) self.assertEqual(popen_calls, [self.POPEN_CALL('a 1', {'shell': True})]) def test_filters_empty_strings_from_command_args_when_shell_is_false(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['', 'a', '', 1], shell=False, timeout_secs=10)) self.exit_stack.close() self.assert_proc_was_managed_as_expected(logs, proc.pid) self.assertEqual( popen_calls, [self.POPEN_CALL(['a', '1'], {'shell': False})]) def test_reports_killed_processes_as_warnings(self): self.exit_stack.enter_context(self.swap_popen( unresponsive=True)) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) self.exit_stack.close() self.assert_proc_was_managed_as_expected( logs, proc.pid, manager_should_have_sent_terminate_signal=True, manager_should_have_sent_kill_signal=True) def test_terminates_child_processes(self): self.exit_stack.enter_context(self.swap_popen(num_children=3)) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) pids = [c.pid for c in proc.children()] + [proc.pid] self.exit_stack.close() self.assertEqual(len(set(pids)), 4) for pid in pids: self.assert_proc_was_managed_as_expected(logs, pid) def test_kills_child_processes(self): self.exit_stack.enter_context(self.swap_popen( num_children=3, unresponsive=True)) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) pids = [c.pid for c in proc.children()] + [proc.pid] self.exit_stack.close() self.assertEqual(len(set(pids)), 4) for pid in pids: self.assert_proc_was_managed_as_expected( logs, pid, manager_should_have_sent_terminate_signal=True, manager_should_have_sent_kill_signal=True) def test_respects_processes_that_are_killed_early(self): self.exit_stack.enter_context(self.swap_popen()) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) time.sleep(1) proc.kill() proc.wait() self.exit_stack.close() self.assert_proc_was_managed_as_expected( logs, proc.pid, manager_should_have_sent_terminate_signal=False) def test_respects_processes_that_are_killed_after_delay(self): self.exit_stack.enter_context(self.swap_popen( unresponsive=True)) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_process( ['a'], timeout_secs=10)) def _kill_after_delay(): """Kills the targeted process after a short delay.""" time.sleep(5) proc.kill() assassin_thread = threading.Thread(target=_kill_after_delay) assassin_thread.start() self.exit_stack.close() assassin_thread.join() self.assert_proc_was_managed_as_expected( logs, proc.pid, manager_should_have_sent_terminate_signal=True, manager_should_have_sent_kill_signal=False) def test_does_not_raise_when_exit_fails(self): self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_raise( psutil, 'wait_procs', error=Exception('uh-oh'))) logs = self.exit_stack.enter_context(self.capture_logging( min_level=logging.ERROR)) self.exit_stack.enter_context(servers.managed_process(['a', 'bc'])) # Should not raise. self.exit_stack.close() self.assert_matches_regexps(logs, [ r'Failed to stop Process\(pid=1\) gracefully!\n' r'Traceback \(most recent call last\):\n' r'.*' r'Exception: uh-oh', ]) def test_managed_firebase_emulator(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(servers.managed_firebase_auth_emulator()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertIn('firebase', popen_calls[0].program_args) self.assertEqual(popen_calls[0].kwargs, {'shell': True}) def test_managed_cloud_datastore_emulator(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context( self.swap_managed_cloud_datastore_emulator_io_operations(True)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context( servers.managed_cloud_datastore_emulator()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertIn( 'beta emulators datastore start', popen_calls[0].program_args) self.assertNotIn('--no-store-on-disk', popen_calls[0].program_args) self.assertEqual(popen_calls[0].kwargs, {'shell': True}) def test_managed_cloud_datastore_emulator_creates_missing_data_dir(self): self.exit_stack.enter_context(self.swap_popen()) rmtree_counter, makedirs_counter = self.exit_stack.enter_context( self.swap_managed_cloud_datastore_emulator_io_operations(False)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context( servers.managed_cloud_datastore_emulator()) self.exit_stack.close() self.assertEqual(rmtree_counter.times_called, 0) self.assertEqual(makedirs_counter.times_called, 1) def test_managed_cloud_datastore_emulator_clears_data_dir(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) rmtree_counter, makedirs_counter = self.exit_stack.enter_context( self.swap_managed_cloud_datastore_emulator_io_operations(True)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(servers.managed_cloud_datastore_emulator( clear_datastore=True)) self.exit_stack.close() self.assertIn('--no-store-on-disk', popen_calls[0].program_args) self.assertEqual(rmtree_counter.times_called, 1) self.assertEqual(makedirs_counter.times_called, 1) def test_managed_cloud_datastore_emulator_acknowledges_data_dir(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) rmtree_counter, makedirs_counter = self.exit_stack.enter_context( self.swap_managed_cloud_datastore_emulator_io_operations(True)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(servers.managed_cloud_datastore_emulator( clear_datastore=False)) self.exit_stack.close() self.assertNotIn('--no-store-on-disk', popen_calls[0].program_args) self.assertEqual(rmtree_counter.times_called, 0) self.assertEqual(makedirs_counter.times_called, 0) def test_managed_dev_appserver(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(servers.managed_dev_appserver( 'app.yaml', env=None)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertIn('dev_appserver.py', popen_calls[0].program_args) self.assertEqual(popen_calls[0].kwargs, {'shell': True, 'env': None}) def test_managed_elasticsearch_dev_server(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context( servers.managed_elasticsearch_dev_server()) self.exit_stack.close() self.assertEqual( popen_calls[0].program_args, '%s/bin/elasticsearch -q' % common.ES_PATH) self.assertEqual(popen_calls[0].kwargs, { 'shell': True, 'env': {'ES_PATH_CONF': common.ES_PATH_CONFIG_DIR}, }) def test_start_server_removes_elasticsearch_data(self): check_function_calls = { 'shutil_rmtree_is_called': False } old_os_path_exists = os.path.exists def mock_os_remove_files(file_path): # pylint: disable=unused-argument check_function_calls['shutil_rmtree_is_called'] = True def mock_os_path_exists(file_path): # pylint: disable=unused-argument if file_path == common.ES_PATH_DATA_DIR: return True return old_os_path_exists(file_path) self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( subprocess, 'call', value=scripts_test_utils.PopenStub())) self.exit_stack.enter_context(self.swap( shutil, 'rmtree', mock_os_remove_files)) self.exit_stack.enter_context(self.swap( os.path, 'exists', mock_os_path_exists)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context( servers.managed_elasticsearch_dev_server()) self.exit_stack.close() self.assertTrue(check_function_calls['shutil_rmtree_is_called']) def test_managed_redis_server_throws_exception_when_on_windows_os(self): self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'is_windows_os', value=True)) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.assertRaisesRegexp( Exception, 'The redis command line interface is not installed because ' 'your machine is on the Windows operating system. The redis ' 'server cannot start.', lambda: self.exit_stack.enter_context( servers.managed_redis_server())) def test_managed_redis_server(self): original_os_remove = os.remove original_os_path_exists = os.path.exists @test_utils.CallCounter def mock_os_remove(path): if path == common.REDIS_DUMP_PATH: return original_os_remove(path) def mock_os_path_exists(path): if path == common.REDIS_DUMP_PATH: return original_os_path_exists(path) popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(self.swap_with_checks( os.path, 'exists', mock_os_path_exists)) self.exit_stack.enter_context(self.swap_with_checks( subprocess, 'check_call', lambda _: 0, expected_args=[([common.REDIS_CLI_PATH, 'shutdown', 'nosave'],)] )) self.exit_stack.enter_context(self.swap_with_checks( os, 'remove', mock_os_remove, called=False)) self.exit_stack.enter_context(servers.managed_redis_server()) self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s' % (common.REDIS_SERVER_PATH, common.REDIS_CONF_PATH)) self.assertEqual(popen_calls[0].kwargs, {'shell': True}) self.exit_stack.close() def test_managed_redis_server_deletes_redis_dump_when_it_exists(self): original_os_remove = os.remove original_os_path_exists = os.path.exists @test_utils.CallCounter def mock_os_remove(path): if path == common.REDIS_DUMP_PATH: return original_os_remove(path) def mock_os_path_exists(path): if path == common.REDIS_DUMP_PATH: return True original_os_path_exists(path) popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_to_always_return( common, 'wait_for_port_to_be_in_use')) self.exit_stack.enter_context(self.swap_with_checks( os.path, 'exists', mock_os_path_exists)) self.exit_stack.enter_context(self.swap_with_checks( os, 'remove', mock_os_remove)) self.exit_stack.enter_context(self.swap_with_checks( subprocess, 'check_call', lambda _: 0, expected_args=[([common.REDIS_CLI_PATH, 'shutdown', 'nosave'],)] )) self.exit_stack.enter_context(servers.managed_redis_server()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s' % (common.REDIS_SERVER_PATH, common.REDIS_CONF_PATH)) self.assertEqual(popen_calls[0].kwargs, {'shell': True}) self.assertEqual(mock_os_remove.times_called, 1) def test_managed_web_browser_on_linux_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Linux')) self.exit_stack.enter_context(self.swap_to_always_return( os, 'listdir', value=[])) managed_web_browser = servers.create_managed_web_browser(123) self.assertIsNotNone(managed_web_browser) self.exit_stack.enter_context(managed_web_browser) self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, ['xdg-open', 'http://localhost:123/']) def test_managed_web_browser_on_virtualbox_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Linux')) self.exit_stack.enter_context(self.swap_to_always_return( os, 'listdir', value=['VBOX-123'])) managed_web_browser = servers.create_managed_web_browser(123) self.assertIsNone(managed_web_browser) self.assertEqual(len(popen_calls), 0) def test_managed_web_browser_on_mac_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Darwin')) self.exit_stack.enter_context(self.swap_to_always_return( os, 'listdir', value=[])) managed_web_browser = servers.create_managed_web_browser(123) self.assertIsNotNone(managed_web_browser) self.exit_stack.enter_context(managed_web_browser) self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, ['open', 'http://localhost:123/']) def test_managed_web_browser_on_windows_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Windows')) self.exit_stack.enter_context(self.swap_to_always_return( os, 'listdir', value=[])) managed_web_browser = servers.create_managed_web_browser(123) self.assertIsNone(managed_web_browser) self.assertEqual(len(popen_calls), 0) def test_managed_portserver(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) proc = self.exit_stack.enter_context(servers.managed_portserver()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, 'python -m scripts.run_portserver ' '--portserver_unix_socket_address %s' % ( common.PORTSERVER_SOCKET_FILEPATH), ) self.assertEqual(proc.signals_received, [signal.SIGINT]) self.assertEqual(proc.terminate_count, 0) self.assertEqual(proc.kill_count, 0) def test_managed_portserver_removes_existing_socket(self): original_os_remove = os.remove original_os_path_exists = os.path.exists @test_utils.CallCounter def mock_os_remove(path): if path == common.PORTSERVER_SOCKET_FILEPATH: return original_os_remove(path) def mock_os_path_exists(path): if path == common.PORTSERVER_SOCKET_FILEPATH: return True original_os_path_exists(path) popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap_with_checks( os.path, 'exists', mock_os_path_exists)) self.exit_stack.enter_context(self.swap_with_checks( os, 'remove', mock_os_remove)) proc = self.exit_stack.enter_context(servers.managed_portserver()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, 'python -m scripts.run_portserver ' '--portserver_unix_socket_address %s' % ( common.PORTSERVER_SOCKET_FILEPATH), ) self.assertEqual(proc.signals_received, [signal.SIGINT]) self.assertEqual(mock_os_remove.times_called, 1) def test_managed_portserver_when_signals_are_rejected(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) proc = self.exit_stack.enter_context(servers.managed_portserver()) proc.reject_signal = True self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, 'python -m scripts.run_portserver ' '--portserver_unix_socket_address %s' % ( common.PORTSERVER_SOCKET_FILEPATH), ) self.assertEqual(proc.signals_received, [signal.SIGINT]) self.assertEqual(proc.terminate_count, 1) self.assertEqual(proc.kill_count, 0) def test_managed_portserver_when_unresponsive(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) proc = self.exit_stack.enter_context(servers.managed_portserver()) proc.unresponsive = True self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, 'python -m scripts.run_portserver ' '--portserver_unix_socket_address %s' % ( common.PORTSERVER_SOCKET_FILEPATH), ) self.assertEqual(proc.signals_received, [signal.SIGINT]) self.assertEqual(proc.terminate_count, 1) self.assertEqual(proc.kill_count, 1) def test_managed_webpack_compiler_in_watch_mode_when_build_succeeds(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'abc', b'Built at: 123', b'def'])) str_io = io.StringIO() self.exit_stack.enter_context(contextlib.redirect_stdout(str_io)) logs = self.exit_stack.enter_context(self.capture_logging()) proc = self.exit_stack.enter_context(servers.managed_webpack_compiler( watch_mode=True)) self.exit_stack.close() self.assert_proc_was_managed_as_expected(logs, proc.pid) self.assertEqual(len(popen_calls), 1) self.assertIn('--color', popen_calls[0].program_args) self.assertIn('--watch', popen_calls[0].program_args) self.assertIn('--progress', popen_calls[0].program_args) self.assert_matches_regexps(str_io.getvalue().strip().split('\n'), [ 'Starting new Webpack Compiler', 'abc', 'Built at: 123', 'def', 'Stopping Webpack Compiler', ]) def test_managed_webpack_compiler_in_watch_mode_raises_when_not_built(self): # NOTE: The 'Built at: ' message is never printed. self.exit_stack.enter_context(self.swap_popen(outputs=[b'abc', b'def'])) str_io = io.StringIO() self.exit_stack.enter_context(contextlib.redirect_stdout(str_io)) self.assertRaisesRegexp( IOError, 'First build never completed', lambda: self.exit_stack.enter_context( servers.managed_webpack_compiler(watch_mode=True))) self.assert_matches_regexps(str_io.getvalue().strip().split('\n'), [ 'Starting new Webpack Compiler', 'abc', 'def', 'Stopping Webpack Compiler', ]) def test_managed_webpack_compiler_uses_explicit_config_path(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( config_path='config.json')) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s --config config.json' % ( common.NODE_BIN_PATH, common.WEBPACK_BIN_PATH)) def test_managed_webpack_compiler_uses_prod_source_maps_config(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( use_prod_env=True, use_source_maps=True)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s --config %s' % ( common.NODE_BIN_PATH, common.WEBPACK_BIN_PATH, common.WEBPACK_PROD_SOURCE_MAPS_CONFIG)) def test_managed_webpack_compiler_uses_prod_config(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( use_prod_env=True, use_source_maps=False)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s --config %s' % ( common.NODE_BIN_PATH, common.WEBPACK_BIN_PATH, common.WEBPACK_PROD_CONFIG)) def test_managed_webpack_compiler_uses_dev_source_maps_config(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( use_prod_env=False, use_source_maps=True)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s --config %s' % ( common.NODE_BIN_PATH, common.WEBPACK_BIN_PATH, common.WEBPACK_DEV_SOURCE_MAPS_CONFIG)) def test_managed_webpack_compiler_uses_dev_config(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( use_prod_env=False, use_source_maps=False)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s --config %s' % ( common.NODE_BIN_PATH, common.WEBPACK_BIN_PATH, common.WEBPACK_DEV_CONFIG)) def test_managed_webpack_compiler_with_max_old_space_size(self): popen_calls = self.exit_stack.enter_context(self.swap_popen( outputs=[b'Built at: 123'])) self.exit_stack.enter_context(servers.managed_webpack_compiler( max_old_space_size=2056)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertIn('--max-old-space-size=2056', popen_calls[0].program_args) def test_managed_webdriver_with_explicit_chrome_version(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Linux')) self.exit_stack.enter_context(self.swap_with_checks( subprocess, 'check_call', lambda _: None, expected_args=[ ( [common.NODE_BIN_PATH, common.WEBDRIVER_MANAGER_BIN_PATH, 'update', '--versions.chrome', '123'], ), ])) self.exit_stack.enter_context(self.swap_with_checks( common, 'wait_for_port_to_be_in_use', lambda _: None, expected_args=[(4444,)])) self.exit_stack.enter_context( servers.managed_webdriver_server(chrome_version='123')) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s start --versions.chrome 123 --quiet --standalone' % ( common.NODE_BIN_PATH, common.WEBDRIVER_MANAGER_BIN_PATH)) def test_managed_webdriver_on_mac_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Darwin')) self.exit_stack.enter_context(self.swap_to_always_return( subprocess, 'check_call')) self.exit_stack.enter_context(self.swap_with_checks( subprocess, 'check_output', lambda _: b'4.5.6.78', expected_args=[ ( ['/Applications/Google Chrome.app/Contents/MacOS' '/Google Chrome', '--version'], ), ])) self.exit_stack.enter_context(self.swap_with_checks( utils, 'url_open', lambda _: mock.Mock(read=lambda: b'4.5.6'), expected_args=[ ( 'https://chromedriver.storage.googleapis.com' '/LATEST_RELEASE_4.5.6', ), ])) self.exit_stack.enter_context(self.swap_with_checks( common, 'wait_for_port_to_be_in_use', lambda _: None, expected_args=[(4444,)])) self.exit_stack.enter_context(servers.managed_webdriver_server()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s start --versions.chrome 4.5.6 --quiet --standalone' % ( common.NODE_BIN_PATH, common.WEBDRIVER_MANAGER_BIN_PATH)) def test_managed_webdriver_on_non_mac_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Linux')) self.exit_stack.enter_context(self.swap_to_always_return( subprocess, 'check_call')) self.exit_stack.enter_context(self.swap_with_checks( subprocess, 'check_output', lambda _: b'1.2.3.45', expected_args=[ (['google-chrome', '--version'],), ])) self.exit_stack.enter_context(self.swap_with_checks( utils, 'url_open', lambda _: mock.Mock(read=lambda: b'1.2.3'), expected_args=[ ( 'https://chromedriver.storage.googleapis.com' '/LATEST_RELEASE_1.2.3', ), ])) self.exit_stack.enter_context(self.swap_with_checks( common, 'wait_for_port_to_be_in_use', lambda _: None, expected_args=[(4444,)])) self.exit_stack.enter_context(servers.managed_webdriver_server()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s start --versions.chrome 1.2.3 --quiet --standalone' % ( common.NODE_BIN_PATH, common.WEBDRIVER_MANAGER_BIN_PATH)) def test_managed_webdriver_fails_to_get_chrome_version(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Linux')) self.exit_stack.enter_context(self.swap_to_always_raise( subprocess, 'check_output', error=OSError)) self.exit_stack.enter_context(self.swap_with_checks( common, 'wait_for_port_to_be_in_use', lambda _: None, called=False)) expected_regexp = 'Failed to execute "google-chrome --version" command' with self.assertRaisesRegexp(Exception, expected_regexp): self.exit_stack.enter_context(servers.managed_webdriver_server()) self.assertEqual(len(popen_calls), 0) def test_managed_webdriver_on_window_os(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(self.swap(common, 'OS_NAME', 'Windows')) self.exit_stack.enter_context(self.swap_to_always_return( subprocess, 'check_call')) self.exit_stack.enter_context(self.swap_to_always_return( subprocess, 'check_output', value=b'1.2.3.45')) self.exit_stack.enter_context(self.swap_to_always_return( utils, 'url_open', value=mock.Mock(read=lambda: b'1.2.3'))) self.exit_stack.enter_context(self.swap_to_always_return( common, 'is_x64_architecture', value=True)) self.exit_stack.enter_context(self.swap_with_checks( common, 'inplace_replace_file_context', lambda *_: contextlib.nullcontext(), expected_args=[ ( common.CHROME_PROVIDER_FILE_PATH, re.escape('this.osArch = os.arch();'), 'this.osArch = "x64";', ), ( common.GECKO_PROVIDER_FILE_PATH, re.escape('this.osArch = os.arch();'), 'this.osArch = "x64";', ), ])) self.exit_stack.enter_context(self.swap_with_checks( common, 'wait_for_port_to_be_in_use', lambda _: None, expected_args=[(4444,)])) self.exit_stack.enter_context(servers.managed_webdriver_server()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].program_args, '%s %s start --versions.chrome 1.2.3 --quiet --standalone' % ( common.NODE_BIN_PATH, common.WEBDRIVER_MANAGER_BIN_PATH)) def test_managed_protractor_with_invalid_sharding_instances(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) with self.assertRaisesRegexp(ValueError, 'should be larger than 0'): self.exit_stack.enter_context( servers.managed_protractor_server(sharding_instances=0)) with self.assertRaisesRegexp(ValueError, 'should be larger than 0'): self.exit_stack.enter_context( servers.managed_protractor_server(sharding_instances=-1)) self.exit_stack.close() self.assertEqual(len(popen_calls), 0) def test_managed_protractor(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(servers.managed_protractor_server()) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual(popen_calls[0].kwargs, {'shell': True}) program_args = popen_calls[0].program_args self.assertIn( '%s --unhandled-rejections=strict %s %s' % ( common.NODE_BIN_PATH, common.PROTRACTOR_BIN_PATH, common.PROTRACTOR_CONFIG_FILE_PATH), program_args) self.assertNotIn('--inspect-brk', program_args) self.assertIn('--params.devMode=True', program_args) self.assertIn('--suite full', program_args) def test_managed_protractor_with_explicit_args(self): popen_calls = self.exit_stack.enter_context(self.swap_popen()) self.exit_stack.enter_context(servers.managed_protractor_server( suite_name='abc', sharding_instances=3, debug_mode=True, dev_mode=False, stdout=subprocess.PIPE)) self.exit_stack.close() self.assertEqual(len(popen_calls), 1) self.assertEqual( popen_calls[0].kwargs, {'shell': True, 'stdout': subprocess.PIPE}) program_args = popen_calls[0].program_args # From debug_mode=True. self.assertIn('--inspect-brk', program_args) # From sharding_instances=3. self.assertIn('--capabilities.shardTestFiles=True', program_args) self.assertIn('--capabilities.maxInstances=3', program_args) # From dev_mode=True. self.assertIn('--params.devMode=False', program_args) # From suite='full'. self.assertIn('--suite abc', program_args)