Skip to content

Commit

Permalink
q-dev: replace overriding with patching
Browse files Browse the repository at this point in the history
  • Loading branch information
@piotrbartman
piotrbartman committed Nov 15, 2024
1 parent 665e9cd commit 738aa3a
Showing 1 changed file with 54 additions and 45 deletions.
99 changes: 54 additions & 45 deletions qubesusbproxy/tests.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -722,12 +722,13 @@ def test_010_on_qdb_change_multiple_assignments_including_full(self):
qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"pid": "did"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"pid": "did"}
)

def test_011_on_qdb_change_multiple_assignments_port_vs_dev(self):
back, front = self.added_assign_setup()
Expand All @@ -752,12 +753,13 @@ def test_011_on_qdb_change_multiple_assignments_port_vs_dev(self):
qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"pid": "any"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"pid": "any"}
)

def test_012_on_qdb_change_multiple_assignments_dev(self):
back, front = self.added_assign_setup()
Expand Down Expand Up @@ -785,12 +787,13 @@ def test_012_on_qdb_change_multiple_assignments_dev(self):
qubesusbproxy.core3ext.USBDevice(Port(back, "1-2", "usb"))
)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"any": "did"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"any": "did"}
)

def test_013_on_qdb_change_two_fronts(self):
back, front = self.added_assign_setup()
Expand Down Expand Up @@ -826,13 +829,14 @@ def test_014_failed_confirmation(self, socket):
socket.return_value = "allow:nonsense"

loop = asyncio.get_event_loop()
self.ext.attach_and_notify = AsyncMock()
loop.run_until_complete(
qubes.ext.utils.resolve_conflicts_and_attach(
self.ext, {"1-1": {front: assmnt, back: assmnt}}
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(
qubes.ext.utils.resolve_conflicts_and_attach(
self.ext, {"1-1": {front: assmnt, back: assmnt}}
)
)
)
self.ext.attach_and_notify.assert_not_called()
attach_and_notify.assert_not_called()

# call_socket_service returns coroutine
@unittest.mock.patch(
Expand All @@ -850,13 +854,14 @@ def test_015_successful_confirmation(self, socket):
socket.return_value = "allow:front-vm"

loop = asyncio.get_event_loop()
self.ext.attach_and_notify = AsyncMock()
loop.run_until_complete(
qubes.ext.utils.resolve_conflicts_and_attach(
self.ext, {"1-1": {front: assmnt, back: assmnt}}
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(
qubes.ext.utils.resolve_conflicts_and_attach(
self.ext, {"1-1": {front: assmnt, back: assmnt}}
)
)
)
self.ext.attach_and_notify.assert_called_once_with(front, assmnt)
attach_and_notify.assert_called_once_with(front, assmnt)

def test_016_on_qdb_change_ask(self):
back, front = self.added_assign_setup()
Expand Down Expand Up @@ -902,12 +907,13 @@ def test_020_on_startup_multiple_assignments_including_full(self):
qubesusbproxy.core3ext.USBDevice(Port(back, "1-1", "usb"))
)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"pid": "did"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"pid": "did"}
)

def test_021_on_startup_multiple_assignments_port_vs_dev(self):
back, front = self.added_assign_setup()
Expand All @@ -933,11 +939,12 @@ def test_021_on_startup_multiple_assignments_port_vs_dev(self):
)

loop = asyncio.get_event_loop()
self.ext.attach_and_notify = AsyncMock()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"pid": "any"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"pid": "any"}
)

def test_022_on_startup_multiple_assignments_dev(self):
back, front = self.added_assign_setup()
Expand Down Expand Up @@ -965,12 +972,13 @@ def test_022_on_startup_multiple_assignments_dev(self):
qubesusbproxy.core3ext.USBDevice(Port(back, "1-2", "usb"))
)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
self.ext.attach_and_notify.call_args[0][1].options, {"any": "did"}
)
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.assertEqual(
attach_and_notify.call_args[0][1].options, {"any": "did"}
)

def test_023_on_startup_already_attached(self):
back, front = self.added_assign_setup(attachment="sys-usb")
Expand All @@ -983,10 +991,11 @@ def test_023_on_startup_already_attached(self):
front.devices["usb"]._assigned.append(assmnt)
back.devices["usb"]._exposed.append(exp_dev)

self.ext.attach_and_notify = AsyncMock()
loop = asyncio.get_event_loop()
loop.run_until_complete(self.ext.on_domain_start(front, None))
self.ext.attach_and_notify.assert_not_called()
with (mock.patch.object(self.ext, "attach_and_notify")
as attach_and_notify):
loop.run_until_complete(self.ext.on_domain_start(front, None))
attach_and_notify.assert_not_called()


def list_tests():
Expand Down

0 comments on commit 738aa3a

Please sign in to comment.