diff --git a/pkg/base1/test-dbus-common.js b/pkg/base1/test-dbus-common.js
index cd795fb1dc6b..61038eb9ef67 100644
--- a/pkg/base1/test-dbus-common.js
+++ b/pkg/base1/test-dbus-common.js
@@ -232,7 +232,7 @@ export function common_dbus_tests(channel_options, bus_name) { // eslint-disable
assert.ok(false, "should not be reached");
} catch (ex) {
assert.equal(ex.name, "org.freedesktop.DBus.Error.UnknownMethod", "error name");
- assert.equal(ex.message, "Method UnimplementedMethod is not implemented on interface com.redhat.Cockpit.DBusTests.Frobber", "error message");
+ assert.equal(ex.message, "Unknown method UnimplementedMethod or interface com.redhat.Cockpit.DBusTests.Frobber.", "error message");
}
});
diff --git a/pkg/base1/test-http.js b/pkg/base1/test-http.js
index db508143560e..20fd4188f7fa 100644
--- a/pkg/base1/test-http.js
+++ b/pkg/base1/test-http.js
@@ -210,6 +210,10 @@ QUnit.test("headers", assert => {
.get("/mock/headers", null, { Header1: "booo", Header2: "yay value" })
.response((status, headers) => {
assert.equal(status, 201, "status code");
+
+ delete headers['Content-Type'];
+ delete headers.Date;
+ delete headers.Server;
assert.deepEqual(headers, {
Header1: "booo",
Header2: "yay value",
@@ -248,6 +252,10 @@ QUnit.test("connection headers", assert => {
.get("/mock/headers", null, { Header2: "yay value", Header0: "extra" })
.response((status, headers) => {
assert.equal(status, 201, "status code");
+
+ delete headers['Content-Type'];
+ delete headers.Date;
+ delete headers.Server;
assert.deepEqual(headers, {
Header0: "extra",
Header1: "booo",
diff --git a/test/common/tap-cdp b/test/common/tap-cdp
index 9efe34bfb682..dd0e512b07e6 100755
--- a/test/common/tap-cdp
+++ b/test/common/tap-cdp
@@ -18,9 +18,7 @@
#
import argparse
-import os
import re
-import subprocess
import sys
from cdp import CDP
@@ -28,31 +26,8 @@ from cdp import CDP
tap_line_re = re.compile(r'^(ok [0-9]+|not ok [0-9]+|bail out!|[0-9]+\.\.[0-9]+|# )', re.IGNORECASE)
parser = argparse.ArgumentParser(description="A CDP driver for QUnit which outputs TAP")
-parser.add_argument("server", help="path to the test-server and the test page to run", nargs=argparse.REMAINDER)
-
-# Strip prefix from url
-# We need this to compensate for automake test generation behavior:
-# The tests are called with the path (relative to the build directory) of the testfile,
-# but from the build directory. Some tests make assumptions regarding the structure of the
-# filename. In order to make sure that they receive the same name, regardless of actual
-# build directory location, we need to strip that prefix (path from build to source directory)
-# from the filename
-parser.add_argument("--strip", dest="strip", help="strip prefix from test file paths")
-
-opts = parser.parse_args()
-
-# argparse sometimes forgets to remove this on argparse.REMAINDER args
-if opts.server[0] == '--':
- opts.server = opts.server[1:]
-
-# The test file is the last argument, but 'server' might contain arbitrary
-# amount of options. We cannot express this with argparse, so take it apart
-# manually.
-opts.test = opts.server[-1]
-opts.server = opts.server[:-1]
-
-if opts.strip and opts.test.startswith(opts.strip):
- opts.test = opts.test[len(opts.strip):]
+parser.add_argument("url", help="url to the test to run")
+args = parser.parse_args()
cdp = CDP("C.utf8")
@@ -62,21 +37,7 @@ except SystemError:
print('1..0 # skip web browser not found')
sys.exit(0)
-# pass the address through a separate fd, so that we can see g_debug() messages (which go to stdout)
-(addr_r, addr_w) = os.pipe()
-env = os.environ.copy()
-env["TEST_SERVER_ADDRESS_FD"] = str(addr_w)
-
-server = subprocess.Popen(opts.server,
- stdin=subprocess.DEVNULL,
- pass_fds=(addr_w,),
- close_fds=True,
- env=env)
-os.close(addr_w)
-address = os.read(addr_r, 1000).decode()
-os.close(addr_r)
-
-cdp.invoke("Page.navigate", url=address + '/' + opts.test)
+cdp.invoke("Page.navigate", url=args.url)
success = True
ignore_resource_errors = False
@@ -109,9 +70,6 @@ for t, message in cdp.read_log():
else:
print(message, file=sys.stderr)
-
-server.terminate()
-server.wait()
cdp.kill()
if not success:
diff --git a/test/pytest/mockservice.py b/test/pytest/mockservice.py
new file mode 100644
index 000000000000..77b82c6135a3
--- /dev/null
+++ b/test/pytest/mockservice.py
@@ -0,0 +1,170 @@
+import asyncio
+import contextlib
+import logging
+import math
+from collections.abc import AsyncIterator
+
+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.asynccontextmanager
+async def mock_service_export(bus: systemd_ctypes.Bus) -> AsyncIterator[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_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'),
+ ):
+ async with mock_service_export(user):
+ yield
+
+
+async def main():
+ async with mock_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())
diff --git a/test/pytest/test_browser.py b/test/pytest/test_browser.py
index 38dc0e904d4c..2e133d16c7e2 100644
--- a/test/pytest/test_browser.py
+++ b/test/pytest/test_browser.py
@@ -1,11 +1,12 @@
+import asyncio
import glob
import os
-import subprocess
-import sys
from typing import Iterable
import pytest
+from .testserver import test_server
+
SRCDIR = os.path.realpath(f'{__file__}/../../..')
BUILDDIR = os.environ.get('abs_builddir', SRCDIR)
@@ -28,40 +29,31 @@ def glob_py310(fnmatch: str, *, root_dir: str, recursive: bool = False) -> Itera
yield result[prefixlen:]
+@pytest.mark.asyncio
@pytest.mark.parametrize('html', glob_py310('**/test-*.html', root_dir=f'{SRCDIR}/qunit', recursive=True))
-def test_browser(html):
- if not os.path.exists(f'{BUILDDIR}/test-server'):
- pytest.skip('no test-server')
+async def test_browser(html: str) -> None:
if html in SKIP:
pytest.skip()
elif html in XFAIL:
pytest.xfail()
- if 'COVERAGE_RCFILE' in os.environ:
- coverage = ['coverage', 'run', '--parallel-mode', '--module']
- else:
- coverage = []
-
- # Merge 2>&1 so that pytest displays an interleaved log
- subprocess.run(['test/common/tap-cdp', f'{BUILDDIR}/test-server',
- sys.executable, '-m', *coverage, 'cockpit.bridge', '--debug',
- f'./qunit/{html}'], check=True, stderr=subprocess.STDOUT)
+ async with test_server() as url:
+ tap_cdp = await asyncio.subprocess.create_subprocess_exec(
+ 'test/common/tap-cdp', f'{url}qunit/{html}',
+ stderr=asyncio.subprocess.STDOUT
+ )
+ result = await tap_cdp.wait()
+ assert result == 0
# run test-timeformat.ts in different time zones: west/UTC/east
+@pytest.mark.asyncio
@pytest.mark.parametrize('tz', ['America/Toronto', 'Europe/London', 'UTC', 'Europe/Berlin', 'Australia/Sydney'])
-def test_timeformat_timezones(tz):
- if not os.path.exists(f'{BUILDDIR}/test-server'):
- pytest.skip('no test-server')
+async def test_timeformat_timezones(tz: str, monkeypatch: pytest.MonkeyPatch) -> None:
# this doesn't get built in rpm/deb package build environments, similar to test_browser()
built_test = './qunit/base1/test-timeformat.html'
if not os.path.exists(built_test):
pytest.skip(f'{built_test} not found')
- env = os.environ.copy()
- env['TZ'] = tz
-
- # Merge 2>&1 so that pytest displays an interleaved log
- subprocess.run(['test/common/tap-cdp', f'{BUILDDIR}/test-server',
- sys.executable, '-m', 'cockpit.bridge', '--debug',
- built_test], check=True, stderr=subprocess.STDOUT, env=env)
+ monkeypatch.setenv('TZ', tz)
+ await test_browser('base1/test-timeformat.html')
diff --git a/test/pytest/testserver.py b/test/pytest/testserver.py
new file mode 100644
index 000000000000..3a70b8089b9a
--- /dev/null
+++ b/test/pytest/testserver.py
@@ -0,0 +1,335 @@
+import argparse
+import asyncio
+import contextlib
+import json
+import logging
+import os
+import socket
+from collections.abc import AsyncIterator, Awaitable, Callable
+from pathlib import Path
+from typing import Never
+
+import aiohttp
+from aiohttp import web
+
+from cockpit._vendor import systemd_ctypes
+from cockpit.bridge import Bridge
+from cockpit.channel import Channel
+from cockpit.jsonutil import JsonObject, JsonValue, create_object, get_str
+from cockpit.protocol import CockpitProblem
+
+from .mockservice import mock_service_on_user_bus
+
+
+class WebTransport(asyncio.Transport):
+ def __init__(self, factory: Callable[[], asyncio.Protocol], ws: web.WebSocketResponse):
+ self.write_queue = asyncio.Queue[str | bytes | None]()
+ self.protocol = factory()
+ self.protocol.connection_made(self)
+ self.ws = ws
+
+ def check_binary(self, data: bytes) -> bool:
+ # We need to find out if that should be binary or text
+ channel, _, _ = data.partition(b'\n')
+ if not channel:
+ return False # control messages are always text
+
+ assert isinstance(self.protocol, Bridge)
+ endpoint = self.protocol.open_channels[channel.decode()]
+ assert isinstance(endpoint, Channel)
+ return endpoint.is_binary
+
+ def write(self, data: bytes) -> None:
+ # We know that cockpit.protocol always writes complete frames
+ header, _, body = data.partition(b'\n')
+ assert int(header) == len(body)
+
+ if self.check_binary(body):
+ self.write_queue.put_nowait(body)
+ else:
+ self.write_queue.put_nowait(body.decode())
+
+ def data_received(self, data: bytes):
+ # cockpit.protocol expects a frame length header
+ header = f'{len(data)}\n'.encode()
+ self.protocol.data_received(header + data)
+
+ async def process_write_queue(self) -> None:
+ while True:
+ item = await self.write_queue.get()
+ if isinstance(item, str):
+ await self.ws.send_str(item)
+ elif isinstance(item, bytes):
+ await self.ws.send_bytes(item)
+ else:
+ break
+
+ async def communicate(self) -> None:
+ write_task = asyncio.create_task(self.process_write_queue())
+
+ try:
+ async for msg in self.ws:
+ if msg.type == aiohttp.WSMsgType.TEXT:
+ self.data_received(msg.data.encode())
+ elif msg.type == aiohttp.WSMsgType.BINARY:
+ self.data_received(msg.data)
+ else:
+ print('weird message!', msg)
+ finally:
+ self.write_queue.put_nowait(None)
+ await write_task
+
+ def close(self) -> None:
+ pass
+
+
+routes = web.RouteTableDef()
+
+
+@routes.get(r'/favicon.ico')
+async def favicon_ico(request: web.Request) -> web.FileResponse:
+ del request
+ return web.FileResponse('src/branding/default/favicon.ico')
+
+
+SPLIT_UTF8_FRAMES = [
+ b"initial",
+ # split an é in the middle
+ b"first half \xc3",
+ b"\xa9 second half",
+ b"final"
+]
+
+
+@routes.get(r'/mock/info')
+async def mock_info(request: web.Request) -> web.Response:
+ del request
+ return web.json_response({
+ 'pybridge': True,
+ 'skip_slow_tests': 'COCKPIT_SKIP_SLOW_TESTS' in os.environ
+ })
+
+
+@routes.get(r'/mock/stream')
+async def mock_stream(request: web.Request) -> web.StreamResponse:
+ response = web.StreamResponse()
+ await response.prepare(request)
+
+ for i in range(10):
+ await response.write(f'{i} '.encode())
+
+ return response
+
+
+@routes.get(r'/mock/split-utf8')
+async def mock_split_utf8(request: web.Request) -> web.StreamResponse:
+ response = web.StreamResponse()
+ await response.prepare(request)
+
+ for chunk in SPLIT_UTF8_FRAMES:
+ await response.write(chunk)
+
+ return response
+
+
+@routes.get(r'/mock/truncated-utf8')
+async def mock_truncated_utf8(request: web.Request) -> web.StreamResponse:
+ response = web.StreamResponse()
+ await response.prepare(request)
+
+ for chunk in SPLIT_UTF8_FRAMES[0:2]:
+ await response.write(chunk)
+
+ return response
+
+
+@routes.get(r'/mock/headers')
+async def mock_headers(request: web.Request) -> web.Response:
+ headers = {k: v for k, v in request.headers.items() if k.startswith('Header')}
+ headers['Header3'] = 'three'
+ headers['Header4'] = 'marmalade'
+
+ return web.Response(status=201, text='Yoo Hoo', headers=headers)
+
+
+@routes.get(r'/mock/host')
+async def mock_host(request: web.Request) -> web.Response:
+ return web.Response(status=201, text='Yoo Hoo', headers={'Host': request.headers['Host']})
+
+
+@routes.get(r'/mock/headonly')
+async def mock_headonly(request: web.Request) -> web.Response:
+ if request.method != 'HEAD':
+ return web.Response(status=400, reason="Only HEAD allowed on this path")
+
+ input_data = request.headers.get('InputData')
+ if not input_data:
+ return web.Response(status=400, reason="Requires InputData header")
+
+ return web.Response(status=200, text='OK', headers={'InputDataLength': str(len(input_data))})
+
+
+@routes.get(r'/mock/qs')
+async def mock_qs(request: web.Request) -> web.Response:
+ return web.Response(text=request.query_string.replace(' ', '+'))
+
+
+@routes.get(r'/cockpit/channel/{csrf_token}')
+async def cockpit_channel(request: web.Request) -> web.StreamResponse:
+ response = web.StreamResponse()
+ await response.prepare(request)
+ return response
+
+
+class CockpitWebSocket(web.WebSocketResponse):
+ def __init__(self):
+ super().__init__(protocols=['cockpit1'])
+
+ async def send_control(self, _msg: JsonObject | None = None, **kwargs: JsonValue) -> None:
+ await self.send_str('\n' + json.dumps(create_object(_msg, kwargs)))
+
+ async def protocol_error(self) -> Never:
+ if not self.closed:
+ await self.send_control(command='close', problem='protocol-error')
+ await self.close()
+ raise web.HTTPBadRequest
+
+ async def receive_control(self, expected: str | None = None) -> JsonObject:
+ try:
+ message = await self.receive_json()
+ except TypeError:
+ await self.protocol_error()
+
+ command = get_str(message, 'command', None)
+ if command is None or (expected is not None and command != expected):
+ await self.protocol_error()
+
+ return message
+
+
+@routes.get(r'/cockpit/socket')
+async def cockpit_socket(request: web.Request) -> web.WebSocketResponse:
+ ws = CockpitWebSocket()
+ await ws.prepare(request)
+
+ try:
+ await ws.send_control(
+ command='init', version=1, host='localhost',
+ channel_seed='test-server', csrf_token='hunter2',
+ capabilities=['multi', 'credentials', 'binary'],
+ system={'version': '0'}
+ )
+
+ init = await ws.receive_control('init')
+ if init != {'command': 'init', 'version': 1}:
+ await ws.protocol_error()
+
+ args = argparse.Namespace(privileged=False, beipack=False)
+ transport = WebTransport(lambda: Bridge(args), ws)
+ # TODO: explicit-superuser handling
+ transport.data_received(b'\n' + json.dumps({
+ "command": "init",
+ "version": 1,
+ "host": "localhost"
+ }).encode())
+
+ await transport.communicate()
+
+ except CockpitProblem:
+ await ws.protocol_error()
+
+ return ws
+
+
+@routes.get('/')
+async def index(request: web.Request) -> web.Response:
+ del request
+ cases = Path('qunit').rglob('test-*.html')
+
+ result = (
+ """
+
+
+ Test cases
+
+
+
+ """ + '\n'.join(
+ f'- {case}
' for case in cases
+ ) + """
+
+
+
+ """
+ )
+
+ return web.Response(text=result, content_type='text/html')
+
+
+@routes.get(r'/{name:(pkg|dist|qunit)/.+}')
+async def serve_file(request: web.Request) -> web.FileResponse:
+ path = Path('.') / request.match_info['name']
+ return web.FileResponse(path)
+
+
+COMMON_HEADERS = {
+ "Cross-Origin-Resource-Policy": "same-origin",
+ "Referrer-Policy": "no-referrer",
+ "X-Content-Type-Options": "nosniff",
+ "X-DNS-Prefetch-Control": "off",
+ "X-Frame-Options": "sameorigin",
+}
+
+
+@web.middleware
+async def cockpit_middleware(
+ request: web.Request, handler: Callable[[web.Request], Awaitable[web.StreamResponse]]
+) -> web.StreamResponse:
+ try:
+ response = await handler(request)
+ except web.HTTPException as ex:
+ response = web.Response(
+ status=ex.status, reason=ex.reason, text=f'{ex.reason}
', content_type='text/html'
+ )
+
+ response.headers.update(COMMON_HEADERS)
+ return response
+
+
+@contextlib.asynccontextmanager
+async def test_server(addr: str = '127.0.0.1', port: int = 0) -> AsyncIterator[str]:
+ async with mock_service_on_user_bus():
+ app = web.Application(middlewares=[cockpit_middleware])
+ app.add_routes(routes)
+
+ runner = web.AppRunner(app)
+ await runner.setup()
+
+ listener = socket.socket()
+ listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ listener.bind((addr, port))
+ listener.listen()
+ site = web.SockSite(runner, listener)
+ await site.start()
+
+ addr, port = listener.getsockname()
+ yield f'http://{addr}:{port}/'
+
+ await runner.cleanup()
+
+
+async def main() -> None:
+ parser = argparse.ArgumentParser(description="Serve a single git repository via HTTP")
+ parser.add_argument('--addr', '-a', default='127.0.0.1', help="Address to bind to")
+ parser.add_argument('--port', '-p', type=int, default=8080, help="Port number to bind to")
+ args = parser.parse_args()
+
+ logging.basicConfig(level=logging.DEBUG)
+
+ async with test_server(args.addr, args.port) as url:
+ print(f"\n {url}\n\nCtrl+C to exit.")
+ await asyncio.sleep(1000000)
+
+
+if __name__ == '__main__':
+ systemd_ctypes.run_async(main())