Source code for qubesadmin.tests.events

# -*- encoding: utf8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
#                               <[email protected]>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.



import socket
import subprocess
import qubesadmin.tests
import unittest
try:
    # qubesadmin.events require python3, so this tests can also use python3 features
    import asyncio
    import unittest.mock
    import qubesadmin.events
except ImportError:
    # don't run any tests on python2
    def load_tests(loader, tests, pattern):
        return unittest.TestSuite()
    # don't fail on coroutine decorator
    class asyncio(object):
        @staticmethod
        def coroutine(f):
            return f


[docs]class TC_00_Events(qubesadmin.tests.QubesTestCase):
[docs] def setUp(self): super().setUp() self.app = unittest.mock.MagicMock() self.dispatcher = qubesadmin.events.EventsDispatcher(self.app)
[docs] def test_000_handler_specific(self): handler = unittest.mock.Mock() self.dispatcher.add_handler('some-event', handler) self.dispatcher.handle('', 'some-event', arg1='value1') handler.assert_called_once_with(None, 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('test-vm', 'some-event', arg1='value1') handler.assert_called_once_with( self.app.domains.get_blind('test-vm'), 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('', 'other-event', arg1='value1') self.assertFalse(handler.called) self.dispatcher.remove_handler('some-event', handler) self.dispatcher.handle('', 'some-event', arg1='value1') self.assertFalse(handler.called)
[docs] def test_001_handler_glob(self): handler = unittest.mock.Mock() self.dispatcher.add_handler('*', handler) self.dispatcher.handle('', 'some-event', arg1='value1') handler.assert_called_once_with(None, 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('test-vm', 'some-event', arg1='value1') handler.assert_called_once_with( self.app.domains.get_blind('test-vm'), 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('', 'other-event', arg1='value1') handler.assert_called_once_with(None, 'other-event', arg1='value1') handler.reset_mock() self.dispatcher.remove_handler('*', handler) self.dispatcher.handle('', 'some-event', arg1='value1') self.assertFalse(handler.called)
[docs] def test_002_handler_glob_partial(self): handler = unittest.mock.Mock() self.dispatcher.add_handler('some-*', handler) self.dispatcher.handle('', 'some-event', arg1='value1') handler.assert_called_once_with(None, 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('test-vm', 'some-event', arg1='value1') handler.assert_called_once_with( self.app.domains.get_blind('test-vm'), 'some-event', arg1='value1') handler.reset_mock() self.dispatcher.handle('', 'other-event', arg1='value1') self.assertFalse(handler.called) handler.reset_mock() self.dispatcher.remove_handler('some-*', handler) self.dispatcher.handle('', 'some-event', arg1='value1') self.assertFalse(handler.called)
[docs] @asyncio.coroutine def mock_get_events_reader(self, stream, cleanup_func, expected_vm, vm=None): self.assertEqual(expected_vm, vm) return stream, cleanup_func
[docs] @asyncio.coroutine def send_events(self, stream, events): for event in events: stream.feed_data(event) # don't use yield from... sleep = asyncio.sleep(0.01) for x in iter(lambda: sleep.send(None), None): yield x stream.feed_eof()
[docs] def test_010_listen_for_events(self): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) stream = asyncio.StreamReader() cleanup_func = unittest.mock.Mock() self.dispatcher._get_events_reader = \ lambda vm: self.mock_get_events_reader(stream, cleanup_func, None, vm) handler = unittest.mock.Mock() self.dispatcher.add_handler('some-event', handler) events = [ b'1\0\0some-event\0arg1\0value1\0\0', b'1\0some-vm\0some-event\0arg1\0value1\0\0', b'1\0some-vm\0some-event\0arg_without_value\0\0arg2\0value\0\0', b'1\0some-vm\0other-event\0\0', ] asyncio.ensure_future(self.send_events(stream, events)) loop.run_until_complete(self.dispatcher.listen_for_events( reconnect=False)) self.assertEqual(handler.mock_calls, [ unittest.mock.call(None, 'some-event', arg1='value1'), unittest.mock.call( self.app.domains.get_blind('some-vm'), 'some-event', arg1='value1'), unittest.mock.call( self.app.domains.get_blind('some-vm'), 'some-event', arg_without_value='', arg2='value'), ]) cleanup_func.assert_called_once_with() loop.close()
[docs] def mock_open_unix_connection(self, expected_path, sock, path): self.assertEqual(expected_path, path) return asyncio.open_connection(sock=sock)
[docs] def read_all(self, sock): buf = b'' for data in iter(lambda: sock.recv(4096), b''): buf += data return buf
[docs] def test_020_get_events_reader_local(self): self.app.qubesd_connection_type = 'socket' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) sock1, sock2 = socket.socketpair() with unittest.mock.patch('asyncio.open_unix_connection', lambda path: self.mock_open_unix_connection( qubesadmin.config.QUBESD_SOCKET, sock1, path)): task = asyncio.ensure_future(self.dispatcher._get_events_reader()) reader = asyncio.ensure_future(loop.run_in_executor(None, self.read_all, sock2)) loop.run_until_complete(asyncio.wait([task, reader])) self.assertEqual(reader.result(), b'dom0\0admin.Events\0dom0\0\0') self.assertIsInstance(task.result()[0], asyncio.StreamReader) cleanup_func = task.result()[1] cleanup_func() sock2.close() # run socket cleanup functions loop.stop() loop.run_forever() loop.close()
[docs] def test_021_get_events_reader_local_vm(self): self.app.qubesd_connection_type = 'socket' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) sock1, sock2 = socket.socketpair() vm = unittest.mock.Mock() vm.name = 'test-vm' with unittest.mock.patch('asyncio.open_unix_connection', lambda path: self.mock_open_unix_connection( qubesadmin.config.QUBESD_SOCKET, sock1, path)): task = asyncio.ensure_future(self.dispatcher._get_events_reader(vm)) reader = asyncio.ensure_future(loop.run_in_executor(None, self.read_all, sock2)) loop.run_until_complete(asyncio.wait([task, reader])) self.assertEqual(reader.result(), b'dom0\0admin.Events\0test-vm\0\0') self.assertIsInstance(task.result()[0], asyncio.StreamReader) cleanup_func = task.result()[1] cleanup_func() sock2.close() # run socket cleanup functions loop.stop() loop.run_forever() loop.close()
[docs] @asyncio.coroutine def mock_coroutine(self, mock, *args, **kwargs): return mock(*args, **kwargs)
[docs] def test_022_get_events_reader_remote(self): self.app.qubesd_connection_type = 'qrexec' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) mock_proc = unittest.mock.Mock() with unittest.mock.patch('asyncio.create_subprocess_exec', lambda *args, **kwargs: self.mock_coroutine(mock_proc, *args, **kwargs)): task = asyncio.ensure_future(self.dispatcher._get_events_reader()) loop.run_until_complete(task) self.assertEqual(mock_proc.mock_calls, [ unittest.mock.call('qrexec-client-vm', 'dom0', 'admin.Events', stdin=subprocess.PIPE, stdout=subprocess.PIPE), unittest.mock.call().stdin.write_eof() ]) self.assertEqual(task.result()[0], mock_proc().stdout) cleanup_func = task.result()[1] cleanup_func() unittest.mock.call().kill.assert_called_once_with() loop.close()
[docs] def test_023_get_events_reader_remote_vm(self): self.app.qubesd_connection_type = 'qrexec' loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) mock_proc = unittest.mock.Mock() vm = unittest.mock.Mock() vm.name = 'test-vm' with unittest.mock.patch('asyncio.create_subprocess_exec', lambda *args, **kwargs: self.mock_coroutine(mock_proc, *args, **kwargs)): task = asyncio.ensure_future(self.dispatcher._get_events_reader(vm)) loop.run_until_complete(task) self.assertEqual(mock_proc.mock_calls, [ unittest.mock.call('qrexec-client-vm', 'test-vm', 'admin.Events', stdin=subprocess.PIPE, stdout=subprocess.PIPE), unittest.mock.call().stdin.write_eof() ]) self.assertEqual(task.result()[0], mock_proc().stdout) cleanup_func = task.result()[1] cleanup_func() unittest.mock.call().kill.assert_called_once_with() loop.close()
[docs] def test_030_events_device(self): handler = unittest.mock.Mock() self.dispatcher.add_handler('device-attach:test', handler) self.dispatcher.handle('test-vm', 'device-attach:test', device='test-vm2:dev', options='{}') vm = self.app.domains.get_blind('test-vm') dev = self.app.domains.get_blind('test-vm2').devices['test']['dev'] handler.assert_called_once_with(vm, 'device-attach:test', device=dev, options='{}')