| 1 | from pathlib import Path |
| 2 | import signal |
| 3 | import unittest |
| 4 | from unittest.mock import Mock, patch |
| 5 | |
| 6 | import crash |
| 7 | |
| 8 | |
| 9 | class AbruptStopTest(unittest.TestCase): |
| 10 | def backend(self): |
| 11 | backend = Mock() |
| 12 | backend.runtime.return_value = (Path('/lab/owned'), Path('/lab/owned/qmp.sock'), Mock(), None) |
| 13 | backend.runtime.return_value[2].read_text.return_value = '1234' |
| 14 | backend.qmp.return_value = {'name': 'OneNote Windows 7 owned'} |
| 15 | backend.running.side_effect = [True, False, False] |
| 16 | backend.instance_path.return_value.exists.return_value = True |
| 17 | return backend |
| 18 | |
| 19 | def test_stops_identified_clone_and_preserves_machine(self): |
| 20 | backend = self.backend() |
| 21 | with patch.object(crash, 'vm', backend), patch.object(crash.subprocess, 'check_output', return_value='qemu-system-x86_64 -qmp unix:/lab/owned/qmp.sock'), patch.object(crash.os, 'kill') as stop: |
| 22 | result = crash.stop('windows', 'owned') |
| 23 | stop.assert_called_once_with(1234, signal.SIGKILL) |
| 24 | backend.load_instance.assert_called_once_with('owned') |
| 25 | backend.delete_instance.assert_not_called() |
| 26 | backend.shutdown.assert_not_called() |
| 27 | self.assertTrue(result['machine_preserved']) |
| 28 | |
| 29 | def test_refuses_mismatched_vm_identity(self): |
| 30 | backend = self.backend() |
| 31 | backend.qmp.return_value = {'name': 'OneNote Windows 7 another'} |
| 32 | with patch.object(crash, 'vm', backend), patch.object(crash.os, 'kill') as stop: |
| 33 | with self.assertRaisesRegex(ValueError, 'identity'): |
| 34 | crash.stop('windows', 'owned') |
| 35 | stop.assert_not_called() |
| 36 | |
| 37 | def test_refuses_pid_belonging_to_another_process(self): |
| 38 | backend = self.backend() |
| 39 | with patch.object(crash, 'vm', backend), patch.object(crash.subprocess, 'check_output', return_value='qemu-system-x86_64 -qmp unix:/lab/other/qmp.sock'), patch.object(crash.os, 'kill') as stop: |
| 40 | with self.assertRaisesRegex(ValueError, 'registered process'): |
| 41 | crash.stop('windows', 'owned') |
| 42 | stop.assert_not_called() |
| 43 | |
| 44 | |
| 45 | if __name__ == '__main__': |
| 46 | unittest.main() |