| #!/usr/bin/env vpython3 |
| # Copyright 2024 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import contextlib |
| import datetime |
| import pathlib |
| import unittest |
| import os |
| import signal |
| import socket |
| import subprocess |
| import sys |
| import tempfile |
| import time |
| |
| import fast_local_dev_server as server |
| |
| sys.path.append(os.path.join(os.path.dirname(__file__), 'gyp')) |
| from util import server_utils |
| |
| |
| class RegexTest(unittest.TestCase): |
| |
| def testBuildIdRegex(self): |
| self.assertRegex(server.FIRST_LOG_LINE.format(build_id='abc', outdir='PWD'), |
| server.BUILD_ID_RE) |
| |
| |
| def sendMessage(message): |
| with contextlib.closing(socket.socket(socket.AF_UNIX)) as sock: |
| sock.settimeout(1) |
| sock.connect(server_utils.SOCKET_ADDRESS) |
| server_utils.SendMessage(sock, message) |
| |
| |
| def pollServer(): |
| try: |
| sendMessage({'message_type': server_utils.POLL_HEARTBEAT}) |
| return True |
| except ConnectionRefusedError: |
| return False |
| |
| |
| def shouldSkip(): |
| if os.environ.get('ALLOW_EXTERNAL_BUILD_SERVER', None): |
| return False |
| return pollServer() |
| |
| |
| def callServer(args, check=True): |
| return subprocess.run([str(server_utils.SERVER_SCRIPT.absolute())] + args, |
| cwd=pathlib.Path(__file__).parent, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| check=check, |
| text=True, |
| timeout=3) |
| |
| |
| @contextlib.contextmanager |
| def blockingFifo(fifo_path='/tmp/.fast_local_dev_server_test.fifo'): |
| fifo_path = pathlib.Path(fifo_path) |
| try: |
| if not fifo_path.exists(): |
| os.mkfifo(fifo_path) |
| yield fifo_path |
| finally: |
| # Write to the fifo nonblocking to unblock other end. |
| try: |
| pipe = os.open(fifo_path, os.O_WRONLY | os.O_NONBLOCK) |
| os.write(pipe, b'') |
| os.close(pipe) |
| except OSError: |
| # Can't open non-blocking an unconnected pipe for writing. |
| pass |
| fifo_path.unlink(missing_ok=True) |
| |
| |
| class ServerStartedTest(unittest.TestCase): |
| build_id_counter = 0 |
| task_name_counter = 0 |
| |
| def __init__(self, *args, **kwargs): |
| super().__init__(*args, **kwargs) |
| self._tty_path = None |
| self._build_id = None |
| |
| def setUp(self): |
| if shouldSkip(): |
| self.skipTest("Cannot run test when server already running.") |
| self._process = subprocess.Popen( |
| [server_utils.SERVER_SCRIPT.absolute(), '--exit-on-idle', '--quiet'], |
| start_new_session=True, |
| cwd=pathlib.Path(__file__).parent, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| text=True) |
| # pylint: disable=unused-variable |
| for attempt in range(5): |
| if pollServer(): |
| break |
| time.sleep(0.05) |
| |
| def tearDown(self): |
| self._process.terminate() |
| stdout, _ = self._process.communicate() |
| if stdout != '': |
| self.fail(f'build server should be silent but it output:\n{stdout}') |
| |
| @contextlib.contextmanager |
| def _register_build(self): |
| with tempfile.NamedTemporaryFile() as f: |
| build_id = f'BUILD_ID_{ServerStartedTest.build_id_counter}' |
| os.environ['AUTONINJA_BUILD_ID'] = build_id |
| os.environ['AUTONINJA_STDOUT_NAME'] = f.name |
| ServerStartedTest.build_id_counter += 1 |
| build_proc = subprocess.Popen( |
| [sys.executable, '-c', 'import time; time.sleep(100)']) |
| callServer( |
| ['--register-build', build_id, '--builder-pid', |
| str(build_proc.pid)]) |
| self._tty_path = f.name |
| self._build_id = build_id |
| try: |
| yield |
| finally: |
| self._tty_path = None |
| self._build_id = None |
| del os.environ['AUTONINJA_BUILD_ID'] |
| del os.environ['AUTONINJA_STDOUT_NAME'] |
| build_proc.kill() |
| build_proc.wait() |
| |
| def sendTask(self, cmd, stamp_path=None): |
| if stamp_path: |
| _stamp_file = pathlib.Path(stamp_path) |
| else: |
| _stamp_file = pathlib.Path('/tmp/.test.stamp') |
| _stamp_file.touch() |
| |
| name_prefix = f'{self._build_id}-{ServerStartedTest.task_name_counter}' |
| sendMessage({ |
| 'name': f'{name_prefix}: {" ".join(cmd)}', |
| 'message_type': server_utils.ADD_TASK, |
| 'cmd': cmd, |
| # So that logfiles do not clutter cwd. |
| 'cwd': '/tmp/', |
| 'build_id': self._build_id, |
| 'stamp_file': _stamp_file.name, |
| }) |
| ServerStartedTest.task_name_counter += 1 |
| |
| def getTtyContents(self): |
| return pathlib.Path(self._tty_path).read_text() |
| |
| def getBuildInfo(self): |
| build_info = server.query_build_info(self._build_id)['builds'][0] |
| pending_tasks = build_info['pending_tasks'] |
| completed_tasks = build_info['completed_tasks'] |
| return pending_tasks, completed_tasks |
| |
| def waitForTasksDone(self, timeout_seconds=3): |
| timeout_duration = datetime.timedelta(seconds=timeout_seconds) |
| start_time = datetime.datetime.now() |
| while True: |
| pending_tasks, completed_tasks = self.getBuildInfo() |
| |
| if completed_tasks > 0 and pending_tasks == 0: |
| return |
| |
| current_time = datetime.datetime.now() |
| duration = current_time - start_time |
| if duration > timeout_duration: |
| raise TimeoutError('Timed out waiting for pending tasks ' + |
| f'[{pending_tasks}/{pending_tasks+completed_tasks}]') |
| time.sleep(0.1) |
| |
| def testRunsQuietTask(self): |
| with self._register_build(): |
| self.sendTask(['true']) |
| self.waitForTasksDone() |
| self.assertEqual(self.getTtyContents(), '') |
| |
| def testRunsNoisyTask(self): |
| with self._register_build(): |
| self.sendTask(['echo', 'some_output']) |
| self.waitForTasksDone() |
| tty_contents = self.getTtyContents() |
| self.assertIn('some_output', tty_contents) |
| |
| def testStampFileDeletedOnFailedTask(self): |
| with self._register_build(): |
| stamp_file = pathlib.Path('/tmp/.failed_task.stamp') |
| self.sendTask(['echo', 'some_output'], stamp_path=stamp_file) |
| self.waitForTasksDone() |
| self.assertFalse(stamp_file.exists()) |
| |
| def testStampFileNotDeletedOnSuccess(self): |
| with self._register_build(): |
| stamp_file = pathlib.Path('/tmp/.successful_task.stamp') |
| self.sendTask(['true'], stamp_path=stamp_file) |
| self.waitForTasksDone() |
| self.assertTrue(stamp_file.exists()) |
| |
| def testWaitForBuildServerCall(self): |
| with self._register_build(): |
| callServer(['--wait-for-build', self._build_id]) |
| self.assertEqual(self.getTtyContents(), '') |
| |
| def testWaitForIdleServerCall(self): |
| with self._register_build(): |
| self.sendTask(['true']) |
| proc_result = callServer(['--wait-for-idle']) |
| self.assertIn('All', proc_result.stdout) |
| self.assertEqual('', self.getTtyContents()) |
| |
| def testCancelBuildServerCall(self): |
| with self._register_build(): |
| callServer(['--cancel-build', self._build_id]) |
| self.assertEqual(self.getTtyContents(), '') |
| |
| def testBuildStatusServerCall(self): |
| with self._register_build(): |
| proc_result = callServer(['--print-status', self._build_id]) |
| self.assertEqual(proc_result.stdout, '') |
| |
| proc_result = callServer(['--print-status-all']) |
| self.assertIn(self._build_id, proc_result.stdout) |
| |
| self.sendTask(['true']) |
| self.waitForTasksDone() |
| |
| proc_result = callServer(['--print-status', self._build_id]) |
| self.assertEqual('', proc_result.stdout) |
| |
| proc_result = callServer(['--print-status-all']) |
| self.assertIn('has 1 registered build', proc_result.stdout) |
| self.assertIn('1/1', proc_result.stdout) |
| |
| with blockingFifo() as fifo_path: |
| # cat gets stuck until we open the other end of the fifo. |
| self.sendTask(['cat', str(fifo_path)]) |
| proc_result = callServer(['--print-status', self._build_id]) |
| self.assertIn('is still 1 static analysis job', proc_result.stdout) |
| self.assertIn('--wait-for-idle', proc_result.stdout) |
| proc_result = callServer(['--print-status-all']) |
| self.assertIn('1/2', proc_result.stdout) |
| |
| self.waitForTasksDone() |
| callServer(['--cancel-build', self._build_id]) |
| self.waitForTasksDone() |
| proc_result = callServer(['--print-status', self._build_id]) |
| self.assertEqual('', proc_result.stdout) |
| |
| proc_result = callServer(['--print-status-all']) |
| self.assertIn('Main build completed', proc_result.stdout) |
| |
| def testServerCancelsRunningTasks(self): |
| output_stamp = pathlib.Path('/tmp/.deleteme.stamp') |
| with blockingFifo() as fifo_path: |
| self.assertFalse(output_stamp.exists()) |
| # dd blocks on fifo so task never finishes inside with block. |
| with self._register_build(): |
| self.sendTask(['dd', f'if={str(fifo_path)}', f'of={str(output_stamp)}']) |
| callServer(['--cancel-build', self._build_id]) |
| self.waitForTasksDone() |
| self.assertFalse(output_stamp.exists()) |
| |
| def testKeyboardInterrupt(self): |
| os.kill(self._process.pid, signal.SIGINT) |
| self._process.wait(timeout=1) |
| |
| |
| class ServerNotStartedTest(unittest.TestCase): |
| |
| def setUp(self): |
| if pollServer(): |
| self.skipTest("Cannot run test when server already running.") |
| |
| def testWaitForBuildServerCall(self): |
| proc_result = callServer(['--wait-for-build', 'invalid-build-id']) |
| self.assertIn('No server running', proc_result.stdout) |
| |
| def testBuildStatusServerCall(self): |
| proc_result = callServer(['--print-status-all']) |
| self.assertIn('No server running', proc_result.stdout) |
| |
| |
| if __name__ == '__main__': |
| # Suppress logging messages. |
| unittest.main(buffer=True) |
