| 1 | import json |
| 2 | import subprocess |
| 3 | from pathlib import Path |
| 4 | import tempfile |
| 5 | import unittest |
| 6 | from unittest.mock import patch |
| 7 | |
| 8 | import native_runner as runner |
| 9 | |
| 10 | |
| 11 | class NativeRunnerTest(unittest.TestCase): |
| 12 | def test_failed_client_build_cannot_start_a_lab_with_stale_binaries(self): |
| 13 | import native_collaboration as collaboration |
| 14 | with tempfile.TemporaryDirectory() as temporary: |
| 15 | root = Path(temporary) |
| 16 | for profile, failed in (('debug', 'debug'), ('release', 'debug'), ('release', 'release')): |
| 17 | output = root / (profile + '-' + failed) |
| 18 | results = [subprocess.CompletedProcess(['cargo'], 1)] |
| 19 | if failed == 'release': results.insert(0, subprocess.CompletedProcess(['cargo'], 0)) |
| 20 | with patch.object(collaboration, 'linux_vm') as linux, \ |
| 21 | patch.object(collaboration.subprocess, 'run', side_effect=results) as build: |
| 22 | linux.instance_path.return_value = root / 'absent' |
| 23 | with self.assertRaises(subprocess.CalledProcessError): |
| 24 | collaboration.replay(output, 'fixture', client_profile=profile) |
| 25 | self.assertEqual(build.call_count, len(results)) |
| 26 | linux.create_instance.assert_not_called() |
| 27 | linux.launch.assert_not_called() |
| 28 | receipt = json.loads((output / f'build-{failed}.json').read_text()) |
| 29 | self.assertEqual(receipt['exit'], 1) |
| 30 | self.assertEqual('--release' in receipt['command'], failed == 'release') |
| 31 | |
| 32 | def test_shutdown_timeout_still_removes_the_owned_clone(self): |
| 33 | with tempfile.TemporaryDirectory() as temporary: |
| 34 | output = Path(temporary) |
| 35 | manifest = output / 'instance.json' |
| 36 | with patch.object(runner, 'vm') as vm, patch.object(runner, 'install_agent', return_value={}): |
| 37 | vm.instance_path.return_value = manifest |
| 38 | vm.create_instance.side_effect = lambda _: manifest.touch() |
| 39 | vm.delete_instance.side_effect = lambda _: manifest.unlink() |
| 40 | vm.running.side_effect = [True, False] |
| 41 | vm.shutdown.side_effect = SystemExit('Windows did not stop') |
| 42 | with runner.clone(output) as name: |
| 43 | vm.create_instance.assert_called_once_with(name) |
| 44 | vm.qmp.assert_called_once_with(name, 'quit') |
| 45 | vm.delete_instance.assert_called_once_with(name) |
| 46 | self.assertEqual(json.loads((output / 'teardown.json').read_text()), {'absent': True}) |
| 47 | |
| 48 | def test_command_timeout_is_a_failure_even_with_stdout(self): |
| 49 | with tempfile.TemporaryDirectory() as temporary: |
| 50 | output = Path(temporary) |
| 51 | with patch.object(runner.windows, 'do_cmd', return_value={ |
| 52 | 'exit': None, 'stdout': 'Read 1 pages.', 'error': 'timed out', |
| 53 | }): |
| 54 | with self.assertRaisesRegex(RuntimeError, 'timed out'): |
| 55 | runner.command('fixture', 'read', output) |
| 56 | result = json.loads((output / 'commands.jsonl').read_text()) |
| 57 | self.assertIsNone(result['exit']) |
| 58 | |
| 59 | def test_failure_still_tears_down_the_owned_clone(self): |
| 60 | with tempfile.TemporaryDirectory() as temporary: |
| 61 | output = Path(temporary) |
| 62 | manifest = output / 'instance.json' |
| 63 | with patch.object(runner.vm, 'instance_path', return_value=manifest), \ |
| 64 | patch.object(runner.vm, 'create_instance', side_effect=lambda _: manifest.touch()), \ |
| 65 | patch.object(runner.vm, 'start_instance'), \ |
| 66 | patch.object(runner.vm, 'wait_instance'), \ |
| 67 | patch.object(runner.vm, 'running', return_value=True), \ |
| 68 | patch.object(runner.vm, 'shutdown') as shutdown, \ |
| 69 | patch.object(runner.vm, 'delete_instance', side_effect=lambda _: manifest.unlink()), \ |
| 70 | patch.object(runner, 'install_agent', return_value={}), \ |
| 71 | patch.object(runner.windows, 'do_health', return_value={}), \ |
| 72 | patch.object(runner.windows, 'do_shot', side_effect=RuntimeError('unreachable')): |
| 73 | with self.assertRaisesRegex(RuntimeError, 'capture failed'): |
| 74 | with runner.clone(output): |
| 75 | raise RuntimeError('capture failed') |
| 76 | shutdown.assert_called_once() |
| 77 | self.assertEqual(json.loads((output / 'teardown.json').read_text()), {'absent': True}) |
| 78 | |
| 79 | def test_output_inside_notebook_does_not_modify_source(self): |
| 80 | with tempfile.TemporaryDirectory() as temporary: |
| 81 | notebook = Path(temporary) |
| 82 | output = notebook / 'new-export' |
| 83 | with self.assertRaisesRegex(ValueError, 'outside'): |
| 84 | runner.capture(notebook, output) |
| 85 | self.assertFalse(output.exists()) |
| 86 | |
| 87 | |
| 88 | if __name__ == '__main__': |
| 89 | unittest.main() |