mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
380 lines
11 KiB
Python
380 lines
11 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from ampg.config import load_config
|
|
from ampg.manifest import fixture_manifest, write_fixture_manifests
|
|
|
|
|
|
class FixtureManifestTest(unittest.TestCase):
|
|
def test_manifest_contains_route_checks_for_enabled_protocols(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>Hi</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "example"
|
|
domain = "example.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
canonical_url = "https://example.test"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.protocols.clearnet]
|
|
enabled = true
|
|
renderer = "clearnet"
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
onion_location = "auto"
|
|
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
|
|
[site.protocols.gemini]
|
|
enabled = true
|
|
renderer = "gemtext"
|
|
|
|
[site.protocols.ipfs]
|
|
enabled = true
|
|
renderer = "clearnet"
|
|
cid = "bafyexample"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_config(config_path)
|
|
manifest = fixture_manifest(config, config.sites[0])
|
|
written = write_fixture_manifests(config)
|
|
|
|
self.assertEqual("ampg.fixture-manifest.v1", manifest["schema"])
|
|
self.assertEqual(["example"], [result.site_id for result in written])
|
|
self.assertEqual(5, written[0].fixture_count)
|
|
|
|
path = root / "out/ampg-fixture-manifest.json"
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
by_protocol = {fixture["protocol"]: fixture for fixture in data["fixtures"]}
|
|
|
|
self.assertEqual("https://example.test/", by_protocol["clearnet"]["url"])
|
|
self.assertEqual("http://example.onion/", by_protocol["tor"]["url"])
|
|
self.assertEqual("placeholder", by_protocol["tor"]["address_status"])
|
|
self.assertEqual("gemini://example.test/", by_protocol["gemini"]["url"])
|
|
self.assertEqual("ipfs://bafyexample", by_protocol["ipfs"]["url"])
|
|
self.assertEqual(
|
|
{
|
|
"transport": "ipfs",
|
|
"profile": "ipfs",
|
|
"isolation": "transport-profile",
|
|
},
|
|
by_protocol["ipfs"]["checks"],
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
"identity": "none",
|
|
"payments": "none",
|
|
"public_allowed": True,
|
|
"realtime": False,
|
|
"tier": "static",
|
|
},
|
|
by_protocol["ipfs"]["interaction"],
|
|
)
|
|
|
|
def test_manifest_includes_configured_interaction_policy(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>Store</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "store"
|
|
domain = "store.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
tier = "transactional"
|
|
identity = "http-session"
|
|
payments = "server-invoice"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_config(config_path)
|
|
manifest = fixture_manifest(config, config.sites[0])
|
|
|
|
self.assertEqual(
|
|
{
|
|
"identity": "http-session",
|
|
"payments": "server-invoice",
|
|
"public_allowed": True,
|
|
"realtime": False,
|
|
"tier": "transactional",
|
|
},
|
|
manifest["fixtures"][0]["interaction"],
|
|
)
|
|
|
|
def test_manifest_expands_public_route_policies(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>Game</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "game"
|
|
domain = "game.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
canonical_url = "https://game.test"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.interactions]
|
|
default_tier = "static"
|
|
deny_routes = ["/admin/*"]
|
|
|
|
[[site.interactions.route]]
|
|
match = "/play/*"
|
|
tier = "interactive-lite"
|
|
|
|
[[site.interactions.route]]
|
|
match = "/checkout/*"
|
|
tier = "transactional"
|
|
identity = "http-session"
|
|
payments = "server-invoice"
|
|
|
|
[[site.interactions.route]]
|
|
match = "/admin/*"
|
|
tier = "internal"
|
|
public_allowed = false
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_config(config_path)
|
|
manifest = fixture_manifest(config, config.sites[0])
|
|
|
|
self.assertEqual(3, len(config.sites[0].interactions.routes))
|
|
self.assertEqual(3, len(manifest["fixtures"]))
|
|
by_path = {fixture.get("route", {}).get("fixture_path", "/"): fixture for fixture in manifest["fixtures"]}
|
|
self.assertEqual("http://game.onion/play/", by_path["/play/"]["url"])
|
|
self.assertEqual("interactive-lite", by_path["/play/"]["interaction"]["tier"])
|
|
self.assertEqual("http://game.onion/checkout/", by_path["/checkout/"]["url"])
|
|
self.assertEqual("transactional", by_path["/checkout/"]["interaction"]["tier"])
|
|
self.assertEqual("server-invoice", by_path["/checkout/"]["interaction"]["payments"])
|
|
|
|
def test_manifest_imports_app_route_manifest(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>App</h1>", encoding="utf-8")
|
|
(root / "routes.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema": "ampg.route-manifest.v1",
|
|
"default_tier": "interactive-lite",
|
|
"deny_routes": ["/admin/*"],
|
|
"routes": [
|
|
{"match": "/play/*"},
|
|
{
|
|
"match": "/checkout/*",
|
|
"tier": "transactional",
|
|
"identity": "http-session",
|
|
"payments": "server-invoice",
|
|
},
|
|
{
|
|
"match": "/webhooks/*",
|
|
"tier": "internal",
|
|
"public_allowed": False,
|
|
},
|
|
],
|
|
},
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "app"
|
|
domain = "app.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.interactions]
|
|
route_manifest = "./routes.json"
|
|
deny_routes = ["/api/internal/*"]
|
|
|
|
[[site.interactions.route]]
|
|
match = "/account/*"
|
|
tier = "identity"
|
|
identity = "http-session"
|
|
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_config(config_path)
|
|
manifest = fixture_manifest(config, config.sites[0])
|
|
|
|
site = config.sites[0]
|
|
self.assertEqual("interactive-lite", site.interactions.default_tier)
|
|
self.assertEqual(("/admin/*", "/api/internal/*"), site.interactions.deny_routes)
|
|
self.assertEqual(4, len(site.interactions.routes))
|
|
self.assertEqual(4, len(manifest["fixtures"]))
|
|
|
|
by_path = {fixture.get("route", {}).get("fixture_path", "/"): fixture for fixture in manifest["fixtures"]}
|
|
self.assertEqual("interactive-lite", by_path["/play/"]["interaction"]["tier"])
|
|
self.assertEqual("http://app.i2p/play/", by_path["/play/"]["url"])
|
|
self.assertEqual("transactional", by_path["/checkout/"]["interaction"]["tier"])
|
|
self.assertEqual("server-invoice", by_path["/checkout/"]["interaction"]["payments"])
|
|
self.assertEqual("identity", by_path["/account/"]["interaction"]["tier"])
|
|
self.assertNotIn("/webhooks/", by_path)
|
|
|
|
def test_route_manifest_schema_is_required(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>App</h1>", encoding="utf-8")
|
|
(root / "routes.json").write_text('{"routes": []}', encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "app"
|
|
domain = "app.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[site.interactions]
|
|
route_manifest = "./routes.json"
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "ampg.route-manifest.v1"):
|
|
load_config(config_path)
|
|
|
|
def test_manifest_omits_routes_above_protocol_max_tier(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>App</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
"""
|
|
[[site]]
|
|
id = "app"
|
|
domain = "app.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
[[site.interactions.route]]
|
|
match = "/play/*"
|
|
tier = "interactive-lite"
|
|
|
|
[[site.interactions.route]]
|
|
match = "/checkout/*"
|
|
tier = "transactional"
|
|
|
|
[[site.interactions.route]]
|
|
match = "/live/*"
|
|
tier = "realtime"
|
|
realtime = true
|
|
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
max_tier = "interactive-lite"
|
|
|
|
[site.protocols.gemini]
|
|
enabled = true
|
|
renderer = "gemtext"
|
|
|
|
[site.protocols.ipfs]
|
|
enabled = true
|
|
renderer = "clearnet"
|
|
cid = "bafyexample"
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_config(config_path)
|
|
manifest = fixture_manifest(config, config.sites[0])
|
|
|
|
by_protocol = {}
|
|
for fixture in manifest["fixtures"]:
|
|
by_protocol.setdefault(fixture["protocol"], []).append(
|
|
fixture.get("route", {}).get("fixture_path", "/")
|
|
)
|
|
|
|
self.assertEqual(["/", "/play/"], by_protocol["tor"])
|
|
self.assertEqual(["/", "/play/"], by_protocol["gemini"])
|
|
self.assertEqual(["/"], by_protocol["ipfs"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|