274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""Tests for Caddyfile generation — subdomain-only routing model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from castle_core.config import CastleConfig, GatewayConfig
|
|
from castle_core.generators.caddyfile import (
|
|
compute_routes,
|
|
generate_caddyfile_from_registry,
|
|
)
|
|
from castle_core.manifest import (
|
|
CaddyDeployment,
|
|
DeploymentSpec,
|
|
ExposeSpec,
|
|
HttpExposeSpec,
|
|
HttpInternal,
|
|
ProgramSpec,
|
|
Reach,
|
|
RunPython,
|
|
SystemdDeployment,
|
|
)
|
|
from castle_core.registry import Deployment, NodeConfig, NodeRegistry
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_config(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Isolate the generator from the real ~/.castle config so static-frontend
|
|
routes don't leak into these registry-focused tests."""
|
|
import castle_core.config as config_mod
|
|
|
|
def _no_config(*args: object, **kwargs: object) -> object:
|
|
raise FileNotFoundError("isolated in tests")
|
|
|
|
monkeypatch.setattr(config_mod, "load_config", _no_config)
|
|
|
|
|
|
def _make_registry(
|
|
deployed: dict[str, Deployment] | None = None,
|
|
gateway_port: int = 9000,
|
|
gateway_tls: str | None = None,
|
|
gateway_domain: str | None = None,
|
|
acme_email: str | None = None,
|
|
public_domain: str | None = None,
|
|
) -> NodeRegistry:
|
|
reg = NodeRegistry(
|
|
node=NodeConfig(
|
|
hostname="test",
|
|
gateway_port=gateway_port,
|
|
gateway_tls=gateway_tls,
|
|
gateway_domain=gateway_domain,
|
|
acme_email=acme_email,
|
|
public_domain=public_domain,
|
|
),
|
|
)
|
|
for name, d in (deployed or {}).items():
|
|
d.name = name
|
|
reg.put(d)
|
|
return reg
|
|
|
|
|
|
def _dep(port: int, *, expose: bool, name: str | None = None, launcher: str = "python") -> Deployment:
|
|
"""A deployed service; exposed at <name>.<domain> when expose=True."""
|
|
return Deployment(
|
|
manager="systemd",
|
|
launcher=launcher,
|
|
run_cmd=["x"],
|
|
port=port,
|
|
subdomain=(name if expose else None),
|
|
)
|
|
|
|
|
|
def _acme(
|
|
deployed: dict[str, Deployment],
|
|
domain: str | None = "example.com",
|
|
public_domain: str | None = None,
|
|
) -> NodeRegistry:
|
|
return _make_registry(
|
|
gateway_tls="acme", gateway_domain=domain, acme_email="p@e.com",
|
|
public_domain=public_domain, deployed=deployed,
|
|
)
|
|
|
|
|
|
class TestAcmeMode:
|
|
"""gateway.tls=acme → one *.domain wildcard site; every service is a subdomain."""
|
|
|
|
def test_global_acme_block(self) -> None:
|
|
cf = generate_caddyfile_from_registry(_acme({"api": _dep(9020, expose=True, name="api")}))
|
|
assert "email p@e.com" in cf
|
|
assert "acme_dns cloudflare {env.CLOUDFLARE_API_TOKEN}" in cf
|
|
|
|
def test_service_is_a_subdomain_matcher(self) -> None:
|
|
cf = generate_caddyfile_from_registry(
|
|
_acme({"openclaw": _dep(18789, expose=True, name="openclaw")})
|
|
)
|
|
assert "*.example.com {" in cf
|
|
# Subdomain is the service name.
|
|
assert "@host_openclaw host openclaw.example.com" in cf
|
|
assert "reverse_proxy localhost:18789" in cf
|
|
|
|
def test_port_9000_redirects_to_dashboard(self) -> None:
|
|
cf = generate_caddyfile_from_registry(_acme({"api": _dep(9020, expose=True, name="api")}))
|
|
assert ":9000 {" in cf
|
|
assert "redir https://castle.example.com{uri}" in cf
|
|
|
|
def test_no_path_routes(self) -> None:
|
|
cf = generate_caddyfile_from_registry(_acme({"api": _dep(9020, expose=True, name="api")}))
|
|
assert "handle_path" not in cf
|
|
assert "redir /" not in cf # no path-prefix trailing-slash redirects
|
|
assert "auto_https off" not in cf
|
|
|
|
def test_unexposed_service_not_routed(self) -> None:
|
|
cf = generate_caddyfile_from_registry(_acme({"pg": _dep(5432, expose=False, name="pg")}))
|
|
assert "pg.example.com" not in cf
|
|
|
|
def test_staging_toggle(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CASTLE_ACME_STAGING", "1")
|
|
cf = generate_caddyfile_from_registry(_acme({"api": _dep(9020, expose=True, name="api")}))
|
|
assert "acme_ca https://acme-staging-v02.api.letsencrypt.org/directory" in cf
|
|
|
|
def test_static_frontend_is_a_file_server_subdomain(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# A static frontend is a `runner: static` service serving <source>/<root>.
|
|
import castle_core.config as config_mod
|
|
|
|
cfg = _config(
|
|
deployments={
|
|
"castle": CaddyDeployment(
|
|
manager="caddy", program="castle", root="dist"
|
|
)
|
|
},
|
|
programs={"castle": ProgramSpec(source="/data/repos/castle/app")},
|
|
)
|
|
monkeypatch.setattr(config_mod, "load_config", lambda *a, **k: cfg)
|
|
cf = generate_caddyfile_from_registry(_acme({}))
|
|
assert "@host_castle host castle.example.com" in cf
|
|
assert "root * /data/repos/castle/app/dist" in cf
|
|
assert "try_files {path} /index.html" in cf
|
|
assert "file_server" in cf
|
|
|
|
|
|
class TestPublicExposure:
|
|
"""Public deployments are served under the public zone; a `public_host`
|
|
override (apex / other zone) gets its own standalone site with its own cert."""
|
|
|
|
def _static_pub(self, name: str, public_host: str | None = None) -> Deployment:
|
|
return Deployment(
|
|
manager="caddy", run_cmd=[], subdomain=name,
|
|
static_root=f"/data/repos/{name}/public", public=True, public_host=public_host,
|
|
)
|
|
|
|
def test_default_public_uses_wildcard_site(self) -> None:
|
|
reg = _acme({"blog": self._static_pub("blog")}, public_domain="pub.example.org")
|
|
cf = generate_caddyfile_from_registry(reg)
|
|
assert "*.pub.example.org {" in cf
|
|
assert "@host_blog_pub host blog.pub.example.org" in cf
|
|
|
|
def test_public_host_override_is_standalone_apex_site(self) -> None:
|
|
reg = _acme({"payne-io": self._static_pub("payne-io", "payne.io")},
|
|
public_domain="pub.example.org")
|
|
cf = generate_caddyfile_from_registry(reg)
|
|
# An explicit apex site (not under the *.pub wildcard) so Caddy issues its
|
|
# own cert via DNS-01; file_server serves the same local dir directly.
|
|
assert "payne.io {" in cf
|
|
assert "root * /data/repos/payne-io/public" in cf
|
|
# The override host must NOT appear as a *.pub.example.org subdomain.
|
|
assert "payne-io.pub.example.org" not in cf
|
|
|
|
def test_override_and_default_coexist(self) -> None:
|
|
reg = _acme(
|
|
{"blog": self._static_pub("blog"),
|
|
"payne-io": self._static_pub("payne-io", "payne.io")},
|
|
public_domain="pub.example.org",
|
|
)
|
|
cf = generate_caddyfile_from_registry(reg)
|
|
assert "*.pub.example.org {" in cf # default wildcard block still present
|
|
assert "@host_blog_pub host blog.pub.example.org" in cf
|
|
assert "payne.io {" in cf # standalone apex site
|
|
|
|
def test_public_host_without_default_domain(self) -> None:
|
|
# No node-wide public_domain: only the override host gets a site.
|
|
reg = _acme({"payne-io": self._static_pub("payne-io", "payne.io")})
|
|
cf = generate_caddyfile_from_registry(reg)
|
|
assert "payne.io {" in cf
|
|
assert "*.None" not in cf
|
|
|
|
|
|
class TestOffMode:
|
|
"""No domain → HTTP-only control plane on :<port>: dashboard at / + /api → castle-api.
|
|
Other services are port-only (not routed)."""
|
|
|
|
def test_control_plane(self) -> None:
|
|
cf = generate_caddyfile_from_registry(
|
|
_make_registry(deployed={"castle-api": _dep(9020, expose=True, name="castle-api")})
|
|
)
|
|
assert "auto_https off" in cf
|
|
assert "handle_path /api/* {" in cf
|
|
assert "reverse_proxy localhost:9020" in cf
|
|
assert "handle {" in cf # dashboard catch-all (SPA fallback)
|
|
|
|
def test_other_services_not_routed_in_off_mode(self) -> None:
|
|
cf = generate_caddyfile_from_registry(
|
|
_make_registry(deployed={"litellm": _dep(4000, expose=True, name="litellm")})
|
|
)
|
|
assert "litellm" not in cf # no subdomains without a domain
|
|
|
|
|
|
def _service(port: int, *, expose: bool) -> SystemdDeployment:
|
|
return SystemdDeployment(
|
|
manager="systemd",
|
|
run=RunPython(launcher="python", program="svc"),
|
|
expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=port))),
|
|
reach=Reach.INTERNAL if expose else Reach.OFF,
|
|
)
|
|
|
|
|
|
def _config(
|
|
deployments: dict[str, DeploymentSpec], programs: dict[str, ProgramSpec] | None = None
|
|
) -> CastleConfig:
|
|
return CastleConfig(
|
|
root=None, # type: ignore[arg-type]
|
|
gateway=GatewayConfig(port=9000),
|
|
repo=None,
|
|
programs=programs or {},
|
|
deployments=deployments,
|
|
)
|
|
|
|
|
|
class TestConfigSourceOfTruth:
|
|
"""compute_routes derives exposure/port from castle.yaml (the checkbox), so a
|
|
regenerated Caddyfile tracks the spec, not a stale registry."""
|
|
|
|
def test_exposed_service_becomes_a_route(self) -> None:
|
|
routes = compute_routes(_make_registry(), _config({"claw": _service(18789, expose=True)}))
|
|
r = next(r for r in routes if r.name == "claw")
|
|
assert r.kind == "proxy"
|
|
assert r.address == "claw" # subdomain label = name
|
|
assert r.target == "localhost:18789"
|
|
|
|
def test_unexposed_service_has_no_route(self) -> None:
|
|
routes = compute_routes(_make_registry(), _config({"pg": _service(5432, expose=False)}))
|
|
assert not [r for r in routes if r.name == "pg"]
|
|
|
|
def test_tcp_service_has_no_http_route(self) -> None:
|
|
"""A raw-TCP service (reach: internal + expose.tcp) is reachable by
|
|
name+port via DNS, but must NOT get an HTTP gateway route."""
|
|
tcp = SystemdDeployment.model_validate(
|
|
{
|
|
"manager": "systemd",
|
|
"run": RunPython(launcher="python", program="pg"),
|
|
"reach": "internal",
|
|
"expose": {"tcp": {"port": 5432}},
|
|
}
|
|
)
|
|
assert tcp.tcp_port == 5432 and tcp.http_exposed is False
|
|
routes = compute_routes(_make_registry(), _config({"pg": tcp}))
|
|
assert not [r for r in routes if r.name == "pg"]
|
|
|
|
def test_config_port_overrides_stale_registry(self) -> None:
|
|
registry = _make_registry(
|
|
deployed={"claw": _dep(8001, expose=True, name="claw")}
|
|
)
|
|
config = _config({"claw": _service(8002, expose=True)})
|
|
routes = compute_routes(registry, config)
|
|
assert {r.target for r in routes if r.name == "claw"} == {"localhost:8002"}
|
|
|
|
def test_fallback_to_registry_without_config(self) -> None:
|
|
# load_config is isolated (raises) → generate uses the registry snapshot.
|
|
cf = generate_caddyfile_from_registry(
|
|
_acme({"claw": _dep(8001, expose=True, name="claw")})
|
|
)
|
|
assert "@host_claw host claw.example.com" in cf
|