-
Notifications
You must be signed in to change notification settings - Fork 1.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
test: add a python version of test-server #20706
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,49 @@ | ||
import os | ||
import subprocess | ||
from typing import Iterator | ||
|
||
import pytest | ||
|
||
from cockpit._vendor import systemd_ctypes | ||
|
||
|
||
# run tests on a private user bus | ||
@pytest.fixture(scope='session', autouse=True) | ||
def mock_session_bus(tmp_path_factory: pytest.TempPathFactory) -> Iterator[None]: | ||
# make sure nobody opened the user bus yet... | ||
assert systemd_ctypes.Bus._default_user_instance is None | ||
|
||
tmpdir = tmp_path_factory.getbasetemp() | ||
dbus_config = tmpdir / 'dbus-config' | ||
dbus_addr = f'unix:path={tmpdir / "dbus_socket"}' | ||
|
||
dbus_config.write_text(fr""" | ||
<busconfig> | ||
<fork/> | ||
<type>session</type> | ||
<listen>{dbus_addr}</listen> | ||
<policy context="default"> | ||
<!-- Allow everything to be sent --> | ||
<allow send_destination="*" eavesdrop="true"/> | ||
<!-- Allow everything to be received --> | ||
<allow eavesdrop="true"/> | ||
<!-- Allow anyone to own anything --> | ||
<allow own="*"/> | ||
</policy> | ||
</busconfig> | ||
""") | ||
try: | ||
dbus_daemon = subprocess.run( | ||
['dbus-daemon', f'--config-file={dbus_config}', '--print-pid'], stdout=subprocess.PIPE | ||
) | ||
except FileNotFoundError: | ||
yield None # no dbus-daemon? Don't patch. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. IMO that should hard-fail with the FileNotFoundError. It's actively dangerous if you run an unit test on your production machine (which is common practice), and you don't have dbus-daemon installed (dbus-broker has long been the default in Fedora -- my system doesn't have it installed either). A test which wants to use this won't work correctly without dbus-daemon -- it should either skip itself or handle the exception. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I added this because it's missing from toxbox... |
||
return | ||
|
||
pid = int(dbus_daemon.stdout) | ||
os.environ['DBUS_SESSION_BUS_ADDRESS'] = dbus_addr | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We never test the system bus in our tests? ok 😁 (Not actionable, just curious) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We actually do test the system bus. The story of this whole thing goes something like so:
There's a wrinkle, though: we need to make sure that we set this environment variable before the first time that systemd_ctypes get used in the process, otherwise it will set up its "default user dbus" shared instance thing before this variable is set. That's why this lives in the toplevel, although it's not like anything in Some of the other tests already talk to the system D-Bus but don't do anything like owning names on it. I think I did that back in the day because the system bus was more reliably available in our various test environments. |
||
|
||
try: | ||
yield None | ||
finally: | ||
os.kill(pid, 9) |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,171 @@ | ||
import asyncio | ||
import contextlib | ||
import logging | ||
import math | ||
from collections.abc import AsyncIterator | ||
from typing import Iterator | ||
|
||
from cockpit._vendor import systemd_ctypes | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
# No introspection, manual handling of method calls | ||
class borkety_Bork(systemd_ctypes.bus.BaseObject): | ||
def message_received(self, message: systemd_ctypes.bus.BusMessage) -> bool: | ||
signature = message.get_signature(True) # noqa:FBT003 | ||
body = message.get_body() | ||
logger.debug('got Bork message: %s %r', signature, body) | ||
|
||
if message.get_member() == 'Echo': | ||
message.reply_method_return(signature, *body) | ||
return True | ||
|
||
return False | ||
|
||
|
||
class com_redhat_Cockpit_DBusTests_Frobber(systemd_ctypes.bus.Object): | ||
finally_normal_name = systemd_ctypes.bus.Interface.Property('s', 'There aint no place like home') | ||
readonly_property = systemd_ctypes.bus.Interface.Property('s', 'blah') | ||
aay = systemd_ctypes.bus.Interface.Property('aay', [], name='aay') | ||
ag = systemd_ctypes.bus.Interface.Property('ag', [], name='ag') | ||
ao = systemd_ctypes.bus.Interface.Property('ao', [], name='ao') | ||
as_ = systemd_ctypes.bus.Interface.Property('as', [], name='as') | ||
ay = systemd_ctypes.bus.Interface.Property('ay', b'ABCabc\0', name='ay') | ||
b = systemd_ctypes.bus.Interface.Property('b', value=False, name='b') | ||
d = systemd_ctypes.bus.Interface.Property('d', 43, name='d') | ||
g = systemd_ctypes.bus.Interface.Property('g', '', name='g') | ||
i = systemd_ctypes.bus.Interface.Property('i', 0, name='i') | ||
n = systemd_ctypes.bus.Interface.Property('n', 0, name='n') | ||
o = systemd_ctypes.bus.Interface.Property('o', '/', name='o') | ||
q = systemd_ctypes.bus.Interface.Property('q', 0, name='q') | ||
s = systemd_ctypes.bus.Interface.Property('s', '', name='s') | ||
t = systemd_ctypes.bus.Interface.Property('t', 0, name='t') | ||
u = systemd_ctypes.bus.Interface.Property('u', 0, name='u') | ||
x = systemd_ctypes.bus.Interface.Property('x', 0, name='x') | ||
y = systemd_ctypes.bus.Interface.Property('y', 42, name='y') | ||
|
||
test_signal = systemd_ctypes.bus.Interface.Signal('i', 'as', 'ao', 'a{s(ii)}') | ||
|
||
@systemd_ctypes.bus.Interface.Method('', 'i') | ||
def request_signal_emission(self, which_one: int) -> None: | ||
del which_one | ||
|
||
self.test_signal( | ||
43, | ||
['foo', 'frobber'], | ||
['/foo', '/foo/bar'], | ||
{'first': (42, 42), 'second': (43, 43)} | ||
) | ||
|
||
@systemd_ctypes.bus.Interface.Method('s', 's') | ||
def hello_world(self, greeting: str) -> str: | ||
return f"Word! You said `{greeting}'. I'm Skeleton, btw!" | ||
|
||
@systemd_ctypes.bus.Interface.Method('', '') | ||
async def never_return(self) -> None: | ||
await asyncio.sleep(1000000) | ||
|
||
@systemd_ctypes.bus.Interface.Method( | ||
['y', 'b', 'n', 'q', 'i', 'u', 'x', 't', 'd', 's', 'o', 'g', 'ay'], | ||
['y', 'b', 'n', 'q', 'i', 'u', 'x', 't', 'd', 's', 'o', 'g', 'ay'] | ||
) | ||
def test_primitive_types( | ||
self, | ||
val_byte, val_boolean, | ||
val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, | ||
val_double, | ||
val_string, val_objpath, val_signature, | ||
val_bytestring | ||
): | ||
return [ | ||
val_byte + 10, | ||
not val_boolean, | ||
100 + val_int16, | ||
1000 + val_uint16, | ||
10000 + val_int32, | ||
100000 + val_uint32, | ||
1000000 + val_int64, | ||
10000000 + val_uint64, | ||
val_double / math.pi, | ||
f"Word! You said `{val_string}'. Rock'n'roll!", | ||
f"/modified{val_objpath}", | ||
f"assgit{val_signature}", | ||
b"bytestring!\xff\0" | ||
] | ||
|
||
@systemd_ctypes.bus.Interface.Method( | ||
['s'], | ||
["a{ss}", "a{s(ii)}", "(iss)", "as", "ao", "ag", "aay"] | ||
) | ||
def test_non_primitive_types( | ||
self, | ||
dict_s_to_s, | ||
dict_s_to_pairs, | ||
a_struct, | ||
array_of_strings, | ||
array_of_objpaths, | ||
array_of_signatures, | ||
array_of_bytestrings | ||
): | ||
return ( | ||
f'{dict_s_to_s}{dict_s_to_pairs}{a_struct}' | ||
f'array_of_strings: [{", ".join(array_of_strings)}] ' | ||
f'array_of_objpaths: [{", ".join(array_of_objpaths)}] ' | ||
f'array_of_signatures: [signature {", ".join(f"'{sig}'" for sig in array_of_signatures)}] ' | ||
f'array_of_bytestrings: [{", ".join(x[:-1].decode() for x in array_of_bytestrings)}] ' | ||
) | ||
|
||
|
||
@contextlib.contextmanager | ||
def mock_service_export(bus: systemd_ctypes.Bus) -> Iterator[None]: | ||
slots = [ | ||
bus.add_object('/otree/frobber', com_redhat_Cockpit_DBusTests_Frobber()), | ||
bus.add_object('/otree/different', com_redhat_Cockpit_DBusTests_Frobber()), | ||
bus.add_object('/bork', borkety_Bork()) | ||
] | ||
|
||
yield | ||
|
||
for slot in slots: | ||
slot.cancel() | ||
|
||
|
||
@contextlib.asynccontextmanager | ||
async def well_known_name(bus: systemd_ctypes.Bus, name: str, flags: int = 0) -> AsyncIterator[None]: | ||
result, = await bus.call_method_async( | ||
'org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'RequestName', 'su', name, flags | ||
) | ||
if result != 1: | ||
raise RuntimeError(f'Cannot register name {name}: {result}') | ||
|
||
try: | ||
yield | ||
|
||
finally: | ||
result, = await bus.call_method_async( | ||
'org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'ReleaseName', 's', name | ||
) | ||
if result != 1: | ||
raise RuntimeError(f'Cannot release name {name}: {result}') | ||
|
||
|
||
@contextlib.asynccontextmanager | ||
async def mock_dbus_service_on_user_bus() -> AsyncIterator[None]: | ||
user = systemd_ctypes.Bus.default_user() | ||
async with ( | ||
well_known_name(user, 'com.redhat.Cockpit.DBusTests.Test'), | ||
well_known_name(user, 'com.redhat.Cockpit.DBusTests.Second'), | ||
): | ||
with mock_service_export(user): | ||
yield | ||
|
||
|
||
async def main(): | ||
async with mock_dbus_service_on_user_bus(): | ||
print('Mock service running. Ctrl+C to exit.') | ||
await asyncio.sleep(2 << 30) # "a long time." | ||
|
||
|
||
if __name__ == '__main__': | ||
systemd_ctypes.run_async(main()) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could this file move to test/pytest/ ? Would be a bit less clutter-y.