udo/uce
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: port network suite to uce cli runner

Browse Source
This commit is contained in:
udo
2026-06-13 22:58:35 +00:00
parent b1856c1725
commit 6bb4f7f0ad
14 changed files with 541 additions and 891 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/README.md
+20 -69
View File
@@ -1,81 +1,32 @@
# Network Tests
# UCE CLI Tests
This directory contains a dependency-less vanilla Python 3 network test runner and plugin-style test cases. These are AI-generated so they might be garbage.
The network/runtime smoke suite is implemented as a UCE unit and invoked through the runtime CLI socket. The old Python plugin runner was retired so the tests exercise the same runtime path as local operational CLI commands.
## Runner
Run the default local UCE battery:
## Run
```bash
./tests/run_network_tests.py
/bin/python3 tests/run_network_tests.py
scripts/run_cli_tests.sh
scripts/run_cli_tests.sh --include-wasm-kill
scripts/run_cli_tests.sh --list
```
The default battery is aimed at public or cross-host LAN checks and currently focuses on stable published doc endpoints, the published `site/tests` public coverage pages, plus a basic TCP reachability check.
Internal-only cases are tagged `internal` and are not part of the default smoke path unless you ask for them explicitly.
`--include-wasm-kill` adds the wasm trap/loop/recurse kill cases and should be used when `WASM_BACKEND_ENABLED=1`.
For the local runtime, the target is localhost rather than `http://k-uce`.
The script reads `CLI_SOCKET_PATH` from `/etc/uce/settings.cfg` unless `UCE_CLI_SOCKET` is set.
Run against another machine on the local network:
## Implementation
```bash
/bin/python3 tests/run_network_tests.py --base-url http://k-uce --host-header uce.openfu.com
```
- Runner unit: `site/tests/cli_runner.uce`
- Bash wrapper: `scripts/run_cli_tests.sh`
Explicit localhost example:
The runner covers the former Python plugin behaviors:
```bash
/bin/python3 tests/run_network_tests.py --base-url http://localhost --host-header uce.openfu.com
```
- demo page smoke checks
- docs and starter HTTP smoke checks
- published `site/tests` suite pages
- security/header/session hardening checks
- starter parity checks
- TCP listener checks
- optional wasm kill checks
List discovered cases without running them:
```bash
/bin/python3 tests/run_network_tests.py --list
```
Run only one plugin or a subset of tagged tests:
```bash
/bin/python3 tests/run_network_tests.py --plugin uce_http_smoke
/bin/python3 tests/run_network_tests.py --tag tcp
/bin/python3 tests/run_network_tests.py --match session
/bin/python3 tests/run_network_tests.py --tag internal --include-internal
```
Override the network target:
```bash
/bin/python3 tests/run_network_tests.py --host 10.8.0.12 --port 80 --host-header uce.openfu.com
/bin/python3 tests/run_network_tests.py --base-url http://10.8.0.12:8080
```
Write a JSON report:
```bash
/bin/python3 tests/run_network_tests.py --json-report tests/last-report.json
```
## Plugin Contract
Each plugin is a plain Python file under `tests/plugins/` that exports:
```python
def register(registry):
...
```
Use `registry.case(name, callback, tags=[...])` to register cases.
Each callback receives a `TestContext` and may use helpers such as:
- `context.request(...)`
- `context.expect_status(...)`
- `context.expect_body_contains(...)`
- `context.tcp_connect(...)`
## Included Plugins
- `uce_http_smoke.py`: HTTP 200/body-marker checks for stable published doc pages
- `uce_site_suite.py`: HTTP 200/body-marker checks for the published `site/tests` coverage pages; local-only pages such as filesystem, ZIP, services, and tasks are tagged `internal`
- `uce_tcp_smoke.py`: basic TCP listener checks, with public smoke checks on port 80 and an internal-only HTTP/WebSocket listener probe on port `8080` tagged `internal`
The CLI unit sets HTTP status `500 CLI Tests Failed` when any case fails, so the bash wrapper exits nonzero via `curl --fail-with-body`.
tests/plugins/uce_demo_smoke.py
-44
View File
@@ -1,44 +0,0 @@
from pathlib import Path
from run_network_tests import TestFailure
# The demo pages are not regression suites, but every one of them must at
# least compile and render: a preprocessor or runtime change that breaks a
# demo page (e.g. literal text that looks like an entry point) should fail CI.
ERROR_MARKERS = [
"uce compile error",
"fatal signal during request",
"uncaught exception during request",
]
def _repo_root():
return Path(__file__).resolve().parents[2]
def _make_demo_case(page_path):
def run(context):
# sharedunit.uce exercises compiler_load_shared_unit() and can cold-compile
# through the native fallback in a fresh worker process; keep the smoke gate
# strict on status/body while allowing that one cold path enough time.
response = context.request(page_path, timeout=15.0 if page_path.endswith("/sharedunit.uce") else None)
if response.status != 200:
raise TestFailure("expected HTTP 200 for %s, got %s %s" % (page_path, response.status, response.reason))
body_lower = response.text.lower()
for marker in ERROR_MARKERS:
if marker in body_lower:
raise Exception("response body contained error marker %r for %s" % (marker, page_path))
return "HTTP 200 without error markers for %s" % page_path
return run
def register(registry):
demo_dir = _repo_root() / "site" / "demo"
for path in sorted(demo_dir.glob("*.uce")):
registry.case(
"demo page " + path.name,
_make_demo_case("/demo/" + path.name),
tags=["http", "smoke", "uce", "demo", "public"],
)
tests/plugins/uce_http_smoke.py
-42
View File
@@ -1,42 +0,0 @@
def register(registry):
pages = [
("doc index", "/doc/index.uce", "<html>"),
("doc singlepage", "/doc/singlepage.uce", "<html>"),
("doc component page", "/doc/index.uce?p=component", "component()"),
("doc regex page", "/doc/index.uce?p=regex_search", "regex_search"),
("doc xml page", "/doc/index.uce?p=xml_encode", "xml_encode"),
("doc yaml page", "/doc/index.uce?p=yaml_encode", "yaml_encode"),
("doc relative time page", "/doc/index.uce?p=time_format_relative", "time_format_relative"),
]
starter_pages = [
("starter home", "/examples/uce-starter/", 200, "Stunning Apps"),
("starter dashboard", "/examples/uce-starter/?dashboard", 200, "Dashboard"),
("starter dashboard ONCE assets reach head", "/examples/uce-starter/?dashboard", 200, "views/dashboard.css"),
("starter workspace nested route", "/examples/uce-starter/?workspace/projects", 200, "Workspace"),
("starter workspace ONCE assets reach head", "/examples/uce-starter/?workspace/projects", 200, "css/workspace.css"),
("starter ajax section", "/examples/uce-starter/?page2-section1", 200, "UCE starter AJAX fragment response"),
("starter route traversal blocked", "/examples/uce-starter/?../../../demo/index", 404, "The requested page does not exist."),
]
for name, path, needle in pages:
def make_case(page_path=path, expected_text=needle):
def run(context):
response = context.expect_status(page_path, 200)
context.expect_body_contains(response, expected_text)
return "HTTP 200 with expected body marker for %s" % page_path
return run
registry.case(name, make_case(), tags=["http", "smoke", "uce", "public"])
for name, path, status, needle in starter_pages:
def make_starter_case(page_path=path, expected_status=status, expected_text=needle):
def run(context):
response = context.expect_status(page_path, expected_status)
context.expect_body_contains(response, expected_text)
return "starter route %s returned HTTP %s with expected marker" % (page_path, expected_status)
return run
registry.case(name, make_starter_case(), tags=["http", "smoke", "uce", "public", "starter"])
tests/plugins/uce_security_smoke.py
-80
View File
@@ -1,80 +0,0 @@
import http.client
from run_network_tests import TestFailure
def _direct_http_request(path, headers=None):
connection = http.client.HTTPConnection("127.0.0.1", 8080, timeout=5.0)
try:
connection.request("GET", path, headers=headers or {})
response = connection.getresponse()
body = response.read().decode("utf-8", errors="replace")
return response.status, dict(response.getheaders()), body
finally:
connection.close()
def _frontend_http_request(path, headers=None):
request_headers = {"Host": "uce.openfu.com"}
request_headers.update(headers or {})
connection = http.client.HTTPConnection("127.0.0.1", 80, timeout=5.0)
try:
connection.request("GET", path, headers=request_headers)
response = connection.getresponse()
body = response.read().decode("utf-8", errors="replace")
return response.status, response.getheaders(), body
finally:
connection.close()
def register(registry):
def direct_http_rejects_dotdot(context):
status, headers, body = _direct_http_request("/../site/demo/hello.uce")
if status == 200 or "hello world" in body:
raise TestFailure("direct HTTP accepted dot-dot script traversal")
return "direct HTTP dot-dot traversal rejected with HTTP %s" % status
def direct_http_ignores_script_filename_header(context):
status, headers, body = _direct_http_request(
"/no-such-script.uce",
headers={"Script-Filename": "/Code/uce.openfu.com/uce/site/demo/hello.uce"},
)
if status == 200 or "hello world" in body:
raise TestFailure("direct HTTP trusted client Script-Filename header")
return "direct HTTP Script-Filename override rejected with HTTP %s" % status
def response_headers_are_sanitized(context):
response = context.request("/tests/security_headers.uce")
injected_headers = [
"X-UCE-Injected",
"X-UCE-Injected-Name",
"X-UCE-Cookie-Injected",
"X-UCE-Redirect-Injected",
"X-UCE-Status-Injected",
]
for header in injected_headers:
if header in response.headers:
raise TestFailure("CRLF header injection produced extra response header %s" % header)
location = response.headers.get("Location", "")
if "\r" in location or "\n" in location:
raise TestFailure("Location header contains raw CR/LF")
status, direct_headers, direct_body = _direct_http_request("/site/tests/security_headers.uce")
if status >= 500 and "security header sanitizer test" not in direct_body:
raise TestFailure("direct HTTP sanitizer test did not render successfully; status=%s" % status)
return "CRLF response header injection was sanitized"
def unknown_session_id_is_not_adopted(context):
attacker_id = "a" * 64
status, headers, body = _frontend_http_request("/tests/http.uce", headers={"Cookie": "uce-site-tests=" + attacker_id})
set_cookies = [value for name, value in headers if name.lower() == "set-cookie"]
session_cookies = [value for value in set_cookies if value.startswith("uce-site-tests=")]
if any(("uce-site-tests=" + attacker_id) in value for value in session_cookies):
raise TestFailure("session_start adopted caller supplied unknown session id")
if not session_cookies or not any("HttpOnly" in value and "SameSite=Lax" in value for value in session_cookies):
raise TestFailure("session_start did not issue a hardened replacement session cookie")
return "unknown caller-supplied session id was replaced"
registry.case("direct HTTP rejects dot-dot script traversal", direct_http_rejects_dotdot, tags=["security", "http", "internal"])
registry.case("direct HTTP ignores Script-Filename header", direct_http_ignores_script_filename_header, tags=["security", "http", "internal"])
registry.case("response headers sanitize CRLF", response_headers_are_sanitized, tags=["security", "http", "internal"])
registry.case("unknown session ids are not adopted", unknown_session_id_is_not_adopted, tags=["security", "http", "internal"])
tests/plugins/uce_site_suite.py
-109
View File
@@ -1,109 +0,0 @@
from pathlib import Path
from run_network_tests import TestFailure
ERROR_MARKERS = [
"compile error",
"runtime error",
"timed out acquiring compile lock",
"near line",
]
def _repo_root():
return Path(__file__).resolve().parents[2]
def _parse_manifest(manifest_path):
manifest = {}
if not manifest_path.exists():
return manifest
for raw_line in manifest_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#"):
continue
parts = [part.strip() for part in line.split("|")]
if len(parts) < 7:
continue
file_name, title, description, tags, expected, suite, index = parts[:7]
manifest[file_name] = {
"file": file_name,
"title": title,
"description": description,
"tags": tags.split(),
"expected": expected,
"suite": suite == "1",
"index": index == "1",
}
return manifest
def _case_name(file_name):
if file_name == "index.uce":
return "site tests index"
if file_name == "io.uce":
return "site tests filesystem"
if file_name == "websockets.ws.uce":
return "site tests websockets page"
stem = file_name
if stem.endswith(".uce"):
stem = stem[:-4]
if stem.endswith(".ws"):
stem = stem[:-3]
return "site tests " + stem.replace("_", " ")
def _make_missing_metadata_case(file_name):
def run(context):
raise TestFailure("site/tests/%s is missing from site/tests/manifest.txt" % file_name)
return run
def _make_suite_case(page_path, expected_title):
def run(context):
# services.uce starts/stops a local HTTP listener and may cold-compile the
# handler in a fresh worker; keep content checks strict but avoid a false
# timeout on the operational smoke path.
response = context.request(page_path, timeout=15.0 if page_path.endswith("/services.uce") else None)
if response.status != 200:
raise TestFailure("expected HTTP 200 for %s, got %s %s" % (page_path, response.status, response.reason))
context.expect_body_contains(response, expected_title)
body_lower = response.text.lower()
for marker in ERROR_MARKERS:
if marker in body_lower:
raise Exception("response body contained error marker %r for %s" % (marker, page_path))
if '<div class="tests-summary">' in body_lower and ('>fail</span>' in body_lower or '>failed 0<' not in body_lower):
raise Exception("site suite reported failed cases for %s" % page_path)
return "HTTP 200 with suite page marker and no failed cases for %s" % page_path
return run
def register(registry):
repo = _repo_root()
tests_dir = repo / "site" / "tests"
manifest = _parse_manifest(tests_dir / "manifest.txt")
for path in sorted(tests_dir.glob("*.uce")):
file_name = path.name
meta = manifest.get(file_name)
if not meta:
registry.case(
_case_name(file_name),
_make_missing_metadata_case(file_name),
tags=["http", "suite", "uce", "metadata", "internal"],
)
continue
if not meta["suite"]:
continue
tags = list(meta["tags"])
for tag in ["http", "suite", "uce"]:
if tag not in tags:
tags.append(tag)
registry.case(
_case_name(file_name),
_make_suite_case("/tests/" + file_name, meta["expected"] or meta["title"]),
tags=tags,
)
tests/plugins/uce_starter_parity.py
-66
View File
@@ -1,66 +0,0 @@
# Parity harness for the uce-starter example app.
#
# WASM-PROPOSAL Phase 3/5 exit criterion: `run_network_tests.py --match starter`
# must pass against the wasm worker backend. These cases are written against
# the native backend first so the bar exists — and is green — before the wasm
# worker does. Keep assertions backend-agnostic (rendered content only), so
# the same cases gate both backends unchanged.
ERROR_MARKERS = [
"uce compile error",
"fatal signal during request",
"uncaught exception during request",
]
BASE = "/examples/uce-starter/index.uce"
# (case name suffix, query-string route, expected <title>)
VIEWS = [
("landing", "", "Home | UCE Starter"),
("dashboard", "?dashboard", "Dashboard | UCE Starter"),
("gauges", "?gauges", "Gauges | UCE Starter"),
("features", "?features", "Features | UCE Starter"),
("components", "?page1", "Components | UCE Starter"),
("workspace", "?workspace", "Workspace | UCE Starter"),
]
def _check_body(body, title):
body_lower = body.lower()
for marker in ERROR_MARKERS:
if marker in body_lower:
raise Exception("response body contained error marker %r" % marker)
needle = "<title>%s</title>" % title
if needle not in body:
raise Exception("expected %r in response body" % needle)
def _make_view_case(name, query, title):
def run(context):
response = context.expect_status(BASE + query, 200)
_check_body(response.text, title)
return "rendered %s with expected title and no error markers" % name
return run
def _unknown_route_case(context):
# the starter app must render its own 404 view through the app shell,
# not fall through to a bare server error
response = context.expect_status(BASE + "?does-not-exist", 404)
_check_body(response.text, "404 Not Found | UCE Starter")
return "starter app rendered its own 404 page"
def register(registry):
for name, query, title in VIEWS:
registry.case(
"starter view " + name,
_make_view_case(name, query, title),
tags=["http", "uce", "starter", "parity"],
)
registry.case(
"starter unknown route 404",
_unknown_route_case,
tags=["http", "uce", "starter", "parity"],
)
tests/plugins/uce_tcp_smoke.py
-12
View File
@@ -1,12 +0,0 @@
def register(registry):
def port_80(context):
context.tcp_connect(port=80)
return "TCP connect succeeded on port 80"
registry.case("frontend port 80", port_80, tags=["tcp", "smoke", "uce", "public"])
def port_8080(context):
context.tcp_connect(port=8080)
return "TCP connect succeeded on port 8080"
registry.case("http websocket port 8080", port_8080, tags=["tcp", "uce", "internal"])
tests/plugins/uce_wasm_kill.py
-30
View File
@@ -1,30 +0,0 @@
import os
def register(registry):
# W4/W5 kill pages are only meaningful with the wasm backend enabled. Native
# requests may terminate the worker by design, so keep them out of the normal
# native suite unless the W5 harness opts in explicitly.
if os.environ.get("UCE_INCLUDE_WASM_KILL") != "1":
return
pages = [
# oob is __builtin_trap() → wasm `unreachable`, a signal-delivering trap.
# It crashed the worker until signals_based_traps(false) (see make_engine
# in src/wasm/worker.cpp); keeping it in the gate guards that fix.
("wasm kill trap", "/tests/wasm-kill/oob.uce", "unreachable"),
("wasm kill loop", "/tests/wasm-kill/loop.uce", "interrupt"),
("wasm kill recurse", "/tests/wasm-kill/recurse.uce", "wasm_kill_recurse"),
]
for name, path, marker in pages:
def make_case(page_path=path, expected_marker=marker):
def run(context):
response = context.expect_status(page_path, 500)
context.expect_body_contains(response, "wasm runtime error during request")
context.expect_body_contains(response, expected_marker)
# The worker should remain healthy after the trap.
health = context.expect_status("/demo/hello.uce", 200)
context.expect_body_contains(health, "hello world")
return "clean wasm trap page and post-trap health check for %s" % page_path
return run
registry.case(name, make_case(), tags=["http", "uce", "wasm", "kill", "internal"])
tests/run_network_tests.py
-360
View File
@@ -1,360 +0,0 @@
#!/bin/python3
import argparse
import http.client
import importlib.util
import json
import re
import socket
import sys
import time
from urllib.parse import urlparse
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Callable, Dict, Iterable, List, Optional
# When executed as a script this module is "__main__"; register it under its
# real name too, so plugins doing `from run_network_tests import TestFailure`
# get this module instance (and the same TestFailure class) instead of a
# re-imported copy whose exceptions the runner's except clause cannot catch.
sys.modules.setdefault("run_network_tests", sys.modules[__name__])
@dataclass
class Target:
scheme: str = "http"
host: str = "localhost"
port: int = 80
host_header: Optional[str] = None
timeout: float = 5.0
@property
def label(self) -> str:
return "%s://%s:%s" % (self.scheme, self.host, self.port)
@dataclass
class HttpResponse:
status: int
reason: str
headers: Dict[str, str]
body: bytes
duration_ms: float
@property
def text(self) -> str:
return self.body.decode("utf-8", errors="replace")
@dataclass
class TestResult:
name: str
ok: bool
summary: str
plugin: str
duration_ms: float = 0.0
details: str = ""
tags: List[str] = field(default_factory=list)
class TestFailure(Exception):
pass
class TestContext:
def __init__(self, target: Target, args: argparse.Namespace):
self.target = target
self.args = args
def request(
self,
path: str,
method: str = "GET",
headers: Optional[Dict[str, str]] = None,
body: Optional[bytes] = None,
timeout: Optional[float] = None,
) -> HttpResponse:
request_headers = dict(headers or {})
if self.target.host_header and "Host" not in request_headers:
request_headers["Host"] = self.target.host_header
connection_class = http.client.HTTPConnection
if self.target.scheme == "https":
connection_class = http.client.HTTPSConnection
started_at = time.perf_counter()
connection = connection_class(self.target.host, self.target.port, timeout=timeout or self.target.timeout)
try:
connection.request(method, path, body=body, headers=request_headers)
response = connection.getresponse()
content = response.read()
return HttpResponse(
status=response.status,
reason=response.reason,
headers=dict(response.getheaders()),
body=content,
duration_ms=(time.perf_counter() - started_at) * 1000.0,
)
finally:
connection.close()
def expect_status(self, path: str, expected_status: int = 200, method: str = "GET") -> HttpResponse:
response = self.request(path, method=method)
if response.status != expected_status:
raise TestFailure(
"expected HTTP %s for %s, got %s %s" % (
expected_status,
path,
response.status,
response.reason,
)
)
return response
def expect_body_contains(self, response: HttpResponse, needle: str) -> None:
if needle not in response.text:
raise TestFailure("response body did not contain expected text: %r" % needle)
def tcp_connect(self, host: Optional[str] = None, port: Optional[int] = None, timeout: Optional[float] = None) -> None:
sock = socket.create_connection(
(host or self.target.host, port or self.target.port),
timeout or self.target.timeout,
)
sock.close()
@dataclass
class NetworkTestCase:
name: str
callback: Callable[[TestContext], Optional[str]]
plugin: str
tags: List[str] = field(default_factory=list)
def run(self, context: TestContext) -> TestResult:
started_at = time.perf_counter()
try:
summary = self.callback(context) or "ok"
return TestResult(
name=self.name,
ok=True,
summary=summary,
plugin=self.plugin,
duration_ms=(time.perf_counter() - started_at) * 1000.0,
tags=list(self.tags),
)
except TestFailure as exc:
return TestResult(
name=self.name,
ok=False,
summary=str(exc),
plugin=self.plugin,
duration_ms=(time.perf_counter() - started_at) * 1000.0,
tags=list(self.tags),
)
except Exception as exc:
return TestResult(
name=self.name,
ok=False,
summary="unexpected error: %s" % exc,
plugin=self.plugin,
duration_ms=(time.perf_counter() - started_at) * 1000.0,
details=repr(exc),
tags=list(self.tags),
)
class TestRegistry:
def __init__(self):
self._cases = []
self._active_plugin = ""
def set_plugin(self, plugin_name: str) -> None:
self._active_plugin = plugin_name
def case(self, name: str, callback: Callable[[TestContext], Optional[str]], tags: Optional[Iterable[str]] = None) -> None:
if not self._active_plugin:
raise RuntimeError("plugin name was not set before registering tests")
self._cases.append(
NetworkTestCase(
name=name,
callback=callback,
plugin=self._active_plugin,
tags=list(tags or []),
)
)
@property
def cases(self) -> List[NetworkTestCase]:
return list(self._cases)
def discover_plugin_files(plugin_dir: Path) -> List[Path]:
if not plugin_dir.exists():
return []
return sorted(
path for path in plugin_dir.iterdir()
if path.is_file() and path.suffix == ".py" and path.name != "__init__.py"
)
def load_plugin_module(module_path: Path):
module_name = "network_test_plugin_%s" % module_path.stem
spec = importlib.util.spec_from_file_location(module_name, str(module_path))
if spec is None or spec.loader is None:
raise RuntimeError("could not load plugin %s" % module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def register_plugin_cases(registry: TestRegistry, module_path: Path) -> None:
module = load_plugin_module(module_path)
if not hasattr(module, "register"):
raise RuntimeError("plugin %s does not define register(registry)" % module_path.name)
registry.set_plugin(module_path.stem)
module.register(registry)
def filter_cases(
cases: List[NetworkTestCase],
plugins: List[str],
name_pattern: Optional[str],
tags: List[str],
include_internal: bool,
excluded_names: List[str],
) -> List[NetworkTestCase]:
filtered = list(cases)
if plugins:
filtered = [case for case in filtered if case.plugin in plugins]
if not include_internal:
filtered = [case for case in filtered if "internal" not in case.tags]
if name_pattern:
matcher = re.compile(name_pattern)
filtered = [case for case in filtered if matcher.search(case.name)]
if tags:
filtered = [case for case in filtered if any(tag in case.tags for tag in tags)]
if excluded_names:
excluded = set(excluded_names)
filtered = [case for case in filtered if case.name not in excluded]
return filtered
def print_case_list(cases: List[NetworkTestCase]) -> None:
for case in cases:
tag_suffix = ""
if case.tags:
tag_suffix = " tags=%s" % ",".join(case.tags)
print("%s:%s%s" % (case.plugin, case.name, tag_suffix))
def print_result(result: TestResult) -> None:
status = "PASS" if result.ok else "FAIL"
print("[%s] %s:%s (%.1f ms) - %s" % (
status,
result.plugin,
result.name,
result.duration_ms,
result.summary,
))
if result.details:
print(" %s" % result.details)
def write_json_report(results: List[TestResult], report_path: Path) -> None:
report_path.write_text(json.dumps([asdict(result) for result in results], indent=2) + "\n", encoding="utf-8")
def build_parser() -> argparse.ArgumentParser:
base_dir = Path(__file__).resolve().parent
parser = argparse.ArgumentParser(description="Dependency-less plugin-style network test runner")
parser.add_argument("--base-url", help="full base URL, for example http://k-uce or http://127.0.0.1:8080")
parser.add_argument("--host", default="localhost", help="target host or IP")
parser.add_argument("--port", type=int, default=80, help="target port")
parser.add_argument("--scheme", default="http", choices=["http", "https"], help="target scheme")
parser.add_argument("--host-header", default="uce.openfu.com", help="optional Host header override")
parser.add_argument("--timeout", type=float, default=5.0, help="per-request timeout in seconds")
parser.add_argument("--plugin-dir", default=str(base_dir / "plugins"), help="directory containing plugin files")
parser.add_argument("--plugin", action="append", default=[], help="limit to one or more plugin basenames")
parser.add_argument("--match", help="regex filter for case names")
parser.add_argument("--exclude", action="append", default=[], help="exclude an exact test case name after other filters")
parser.add_argument("--tag", action="append", default=[], help="only run cases with at least one matching tag")
parser.add_argument("--include-internal", action="store_true", help="include tests tagged internal in the run")
parser.add_argument("--list", action="store_true", help="list discovered tests without running them")
parser.add_argument("--fail-fast", action="store_true", help="stop on first failing test")
parser.add_argument("--json-report", help="write full results to a JSON file")
return parser
def build_target(args: argparse.Namespace) -> Target:
if args.base_url:
parsed = urlparse(args.base_url)
if parsed.scheme not in ("http", "https") or not parsed.hostname:
raise SystemExit("--base-url must include an http or https scheme and hostname")
default_port = 443 if parsed.scheme == "https" else 80
return Target(
scheme=parsed.scheme,
host=parsed.hostname,
port=parsed.port or default_port,
host_header=args.host_header or parsed.hostname,
timeout=args.timeout,
)
return Target(
scheme=args.scheme,
host=args.host,
port=args.port,
host_header=args.host_header or None,
timeout=args.timeout,
)
def main(argv: Optional[List[str]] = None) -> int:
args = build_parser().parse_args(argv)
plugin_dir = Path(args.plugin_dir).resolve()
plugin_files = discover_plugin_files(plugin_dir)
if not plugin_files:
print("No plugin files found in %s" % plugin_dir, file=sys.stderr)
return 2
registry = TestRegistry()
for plugin_path in plugin_files:
register_plugin_cases(registry, plugin_path)
cases = filter_cases(registry.cases, args.plugin, args.match, args.tag, args.include_internal, args.exclude)
if not cases:
print("No tests matched the selected filters", file=sys.stderr)
return 2
if args.list:
print_case_list(cases)
return 0
context = TestContext(
target=build_target(args),
args=args,
)
print("Running %s test(s) against %s" % (len(cases), context.target.label))
if context.target.host_header:
print("Host header: %s" % context.target.host_header)
results = []
for case in cases:
result = case.run(context)
results.append(result)
print_result(result)
if args.fail_fast and not result.ok:
break
if args.json_report:
write_json_report(results, Path(args.json_report).resolve())
failures = [result for result in results if not result.ok]
print("")
print("Summary: %s passed, %s failed" % (len(results) - len(failures), len(failures)))
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main())