test: add HTTP upload test for systemd-report

Add a fake HTTP server (fake-report-server.py) that accepts JSON POST
requests and validates the report structure, and test cases in
TEST-74-AUX-UTILS.report.sh that exercise plain HTTP upload of both
metrics and facts.

Co-developed-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Zbigniew Jędrzejewski-Szmek
2026-04-16 21:12:20 +02:00
parent ce400ef2d9
commit 0f3bd778e0
3 changed files with 71 additions and 0 deletions
+1
View File
@@ -45,6 +45,7 @@ wrap=(
/usr/lib/systemd/tests/testdata/TEST-74-AUX-UTILS.units/proxy-echo.py
/usr/libexec/polkit-1/polkitd
/usr/lib/systemd/tests/integration-tests/TEST-74-AUX-UTILS/TEST-74-AUX-UTILS.units/fake-imds.py
/usr/lib/systemd/tests/integration-tests/TEST-74-AUX-UTILS/TEST-74-AUX-UTILS.units/fake-report-server.py
agetty
btrfs
capsh
@@ -0,0 +1,56 @@
#!/usr/bin/python3
# SPDX-License-Identifier: LGPL-2.1-or-later
import json, os, socket
from http.server import BaseHTTPRequestHandler, HTTPServer
def sd_notify(state: str) -> bool:
notify_socket = os.environ.get("NOTIFY_SOCKET")
if not notify_socket:
return False
if notify_socket.startswith("@"):
notify_socket = "\0" + notify_socket[1:]
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
sock.sendto(state.encode(), notify_socket)
except OSError:
return False
return True
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length)
# Validate JSON structure
try:
data = json.loads(body)
except json.JSONDecodeError:
self.send_error(400, "Invalid JSON")
return
print(f"JSON: {s if len(s := str(data)) < 80 else s[:40] + '…' + s[-40:]}")
if "metrics" not in data and "facts" not in data:
self.send_error(400, "Missing 'metrics' or 'facts' field")
return
response = json.dumps({"status": "ok"}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", len(response))
self.end_headers()
self.wfile.write(response)
def log_message(self, fmt, *args):
print(f"{self.address_string()} - {fmt % args}")
PORT = 8089
server = HTTPServer(("", PORT), Handler)
print(f"Serving on http://localhost:{PORT}/")
try:
sd_notify("READY=1")
server.serve_forever()
except KeyboardInterrupt:
print("\nStopped.")
+14
View File
@@ -65,3 +65,17 @@ systemctl start systemd-report-basic.socket
# Test facts via direct Varlink call on existing socket
varlinkctl --more call /run/systemd/report/io.systemd.Basic io.systemd.Facts.List {}
varlinkctl --more call /run/systemd/report/io.systemd.Basic io.systemd.Facts.Describe {}
# Test HTTP upload (plain http)
at_exit() {
set +e
systemctl stop fake-report-server
}
trap at_exit EXIT
systemd-run -p Type=notify --unit=fake-report-server \
/usr/lib/systemd/tests/integration-tests/TEST-74-AUX-UTILS/TEST-74-AUX-UTILS.units/fake-report-server.py
systemctl status fake-report-server
"$REPORT" metrics --url=http://localhost:8089/
"$REPORT" facts --url=http://localhost:8089/