83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import threading
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from mas_cli.main import main
|
||
|
|
from mas_core.discovery import Response, discover, http_transport
|
||
|
|
from mas_core.statute import lint
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
STATIC = ROOT / "examples/statute/great-service.example"
|
||
|
|
STATUTE = STATIC / ".well-known/mas"
|
||
|
|
|
||
|
|
|
||
|
|
def run(capsys, *argv):
|
||
|
|
code = main(list(argv))
|
||
|
|
return code, capsys.readouterr().out
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_json(capsys):
|
||
|
|
code, out = run(capsys, "parse", "ticket!callback", "--json")
|
||
|
|
d = json.loads(out)
|
||
|
|
assert code == 0 and d["required_profile"] == "P2"
|
||
|
|
assert run(capsys, "parse", "ticket!callback", "--profile", "p1")[0] == 1
|
||
|
|
assert run(capsys, "parse", "ticket=PR42$1")[0] == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_explain_separates_layers_and_never_guesses(capsys, monkeypatch):
|
||
|
|
import mas_cli.main as cli
|
||
|
|
monkeypatch.setattr(cli, "discover", lambda host, **kw: discover(
|
||
|
|
host, transport=lambda u, t: Response(404, {}, b"")))
|
||
|
|
code, out = run(capsys, "explain", "bernd.worsch@example.org")
|
||
|
|
assert code == 0
|
||
|
|
for section in ("Syntax:", "Discovery:", "Semantics:", "Unknowns:"):
|
||
|
|
assert section in out
|
||
|
|
assert "no valid MAS statute found" in out and "NotFound" in out
|
||
|
|
assert "kind: opaque" in out and "person" not in out
|
||
|
|
|
||
|
|
|
||
|
|
def test_explain_with_statute_file(capsys):
|
||
|
|
code, out = run(capsys, "explain", "ticket=GS-41872+comments@great-service.example",
|
||
|
|
"--statute", str(STATUTE), "--json")
|
||
|
|
sem = json.loads(out)["semantics"]
|
||
|
|
assert sem["kind"] == "resource" and sem["resource_id"] == "GS-41872" and sem["facet"] == "comments"
|
||
|
|
|
||
|
|
|
||
|
|
def test_lint_exit_codes(capsys):
|
||
|
|
st = ROOT / "mas_conformance/statutes"
|
||
|
|
assert run(capsys, "lint-statute", str(STATUTE), "--host", "great-service.example")[0] == 0
|
||
|
|
code, out = run(capsys, "lint-statute", str(st / "semantic-p2-disabled-modes.json"))
|
||
|
|
assert code == 1 and "semantic error" in out
|
||
|
|
code, out = run(capsys, "lint-statute", str(st / "schema-version.json"))
|
||
|
|
assert code == 1 and "schema error" in out
|
||
|
|
|
||
|
|
|
||
|
|
def test_conformance_command(capsys):
|
||
|
|
code, out = run(capsys, "test")
|
||
|
|
assert code == 0 and "FAIL" not in out
|
||
|
|
|
||
|
|
|
||
|
|
def test_static_statute_lints_clean():
|
||
|
|
assert lint(json.loads(STATUTE.read_text()), "great-service.example") == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_static_statute_served_and_discovered():
|
||
|
|
spec = importlib.util.spec_from_file_location("serve", ROOT / "examples/statute/serve.py")
|
||
|
|
serve = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(serve)
|
||
|
|
# Serve a copy whose namespace is the local test host.
|
||
|
|
srv = serve.make_server(str(STATIC), port=0)
|
||
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||
|
|
try:
|
||
|
|
port = srv.server_address[1]
|
||
|
|
r = discover("127.0.0.1", scheme="http", port=port)
|
||
|
|
assert not r.ok and r.kind.value == "NamespaceMismatch" # served for great-service.example
|
||
|
|
r = discover("great-service.example", scheme="http", port=port,
|
||
|
|
transport=lambda url, t: http_transport(url.replace("great-service.example", "127.0.0.1"), t))
|
||
|
|
assert r.ok and r.statute.interaction_mode_enabled("callback")
|
||
|
|
finally:
|
||
|
|
srv.shutdown()
|