add utests

Author: anon

qubesadmin/tests/tools/qvm_shutdown.py

@@ -222,3 +222,185 @@ def test_015_wait_all_kill_timeout(self):
             qubesadmin.tools.qvm_shutdown.main(
                 ['--wait', '--all', '--timeout=1'], app=self.app)
         self.assertAllCalled()
+
+    def test_005_force(self):
+        '''test --force sends force flag to shutdown call'''
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00some-vm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Shutdown', 'force', None)] = b'0\x00'
+        qubesadmin.tools.qvm_shutdown.main(
+            ['--force', 'some-vm'], app=self.app)
+        self.assertAllCalled()
+
+    def test_006_dry_run(self):
+        '''test --dry-run skips shutdown calls'''
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00some-vm class=AppVM state=Running\n'
+        qubesadmin.tools.qvm_shutdown.main(
+            ['--dry-run', 'some-vm'], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(qubesadmin.tools.qvm_shutdown.have_events,
+        'Events not present')
+    def test_011_wait_retry(self):
+        '''test --wait retries VMs whose shutdown request failed'''
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            'qubesadmin.events.EventsDispatcher._get_events_reader',
+            mock_events)
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader([
+            # round 1: wait for some-vm
+            b'1\0\0connection-established\0\0',
+            b'1\0some-vm\0domain-shutdown\0\0',
+            # round 2: wait for other-vm
+            b'1\0\0connection-established\0\0',
+            b'1\0other-vm\0domain-shutdown\0\0',
+            ])
+
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00' \
+            b'some-vm class=AppVM state=Running\n' \
+            b'other-vm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Shutdown', None, None)] = \
+            b'0\x00'
+        # other-vm fails first attempt, succeeds on retry
+        self.app.expected_calls[
+            ('other-vm', 'admin.vm.Shutdown', None, None)] = [
+            b'2\x00QubesException\x00\x00Shutdown refused\x00',
+            b'0\x00',
+        ]
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.CurrentState', None, None)] = [
+            b'0\x00power_state=Running',
+            b'0\x00power_state=Halted',
+        ]
+        self.app.expected_calls[
+            ('other-vm', 'admin.vm.CurrentState', None, None)] = [
+            b'0\x00power_state=Running',
+            b'0\x00power_state=Halted',
+        ]
+        qubesadmin.tools.qvm_shutdown.main(
+            ['--wait', 'some-vm', 'other-vm'], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(qubesadmin.tools.qvm_shutdown.have_events,
+        'Events not present')
+    def test_013_wait_all_shutdown_fail(self):
+        '''test --wait exits with error when all shutdown requests fail'''
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00some-vm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Shutdown', None, None)] = \
+            b'2\x00QubesException\x00\x00Shutdown refused\x00'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.CurrentState', None, None)] = \
+            b'0\x00power_state=Running'
+        with self.assertRaises(SystemExit):
+            qubesadmin.tools.qvm_shutdown.main(
+                ['--wait', 'some-vm'], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(qubesadmin.tools.qvm_shutdown.have_events,
+        'Events not present')
+    def test_016_wait_kill_exception(self):
+        '''test --wait timeout where kill raises QubesException'''
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            'qubesadmin.events.EventsDispatcher._get_events_reader',
+            mock_events)
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader([
+            b'1\0\0connection-established\0\0',
+            ])
+
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00some-vm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Shutdown', None, None)] = \
+            b'0\x00'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Kill', None, None)] = \
+            b'2\x00QubesException\x00\x00Kill failed\x00'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.CurrentState', None, None)] = [
+            b'0\x00power_state=Running',
+            b'0\x00power_state=Running',
+        ]
+        with self.assertRaises(SystemExit):
+            qubesadmin.tools.qvm_shutdown.main(
+                ['--wait', '--timeout=1', 'some-vm'], app=self.app)
+        self.assertAllCalled()
+
+    @unittest.skipUnless(qubesadmin.tools.qvm_shutdown.have_events,
+        'Events not present')
+    def test_017_wait_dispvm_na(self):
+        '''test --wait treats DispVM with NA power state as shut down'''
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        mock_events = unittest.mock.AsyncMock()
+        patch = unittest.mock.patch(
+            'qubesadmin.events.EventsDispatcher._get_events_reader',
+            mock_events)
+        patch.start()
+        self.addCleanup(patch.stop)
+        mock_events.side_effect = qubesadmin.tests.tools.MockEventsReader([
+            b'1\0\0connection-established\0\0',
+            b'1\0disp123\0domain-shutdown\0\0',
+            ])
+
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00disp123 class=DispVM state=Running\n'
+        self.app.expected_calls[
+            ('disp123', 'admin.vm.Shutdown', None, None)] = \
+            b'0\x00'
+        self.app.expected_calls[
+            ('disp123', 'admin.vm.CurrentState', None, None)] = [
+            b'0\x00power_state=Running',
+            # failed_domains: first get_power_state() != 'Halted',
+            # then klass == 'DispVM' triggers second get_power_state()
+            b'0\x00power_state=NA',
+            b'0\x00power_state=NA',
+        ]
+        qubesadmin.tools.qvm_shutdown.main(
+            ['--wait', 'disp123'], app=self.app)
+        self.assertAllCalled()
+
+    def test_018_wait_polling_fallback(self):
+        '''test --wait uses polling when events are unavailable'''
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        self.app.expected_calls[
+            ('dom0', 'admin.vm.List', None, None)] = \
+            b'0\x00some-vm class=AppVM state=Running\n'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.Shutdown', None, None)] = \
+            b'0\x00'
+        self.app.expected_calls[
+            ('some-vm', 'admin.vm.CurrentState', None, None)] = [
+            b'0\x00power_state=Halted',
+            b'0\x00power_state=Halted',
+        ]
+        with unittest.mock.patch.object(
+                qubesadmin.tools.qvm_shutdown, 'have_events', False):
+            qubesadmin.tools.qvm_shutdown.main(
+                ['--wait', 'some-vm'], app=self.app)
+        self.assertAllCalled()