""" v1.5.0 Feature B — Site Setup Wizard model + helper unit tests. Coverage matrix (Section 6.4 of the v1.5.0 plan): - SiteCreate field validation * domain regex normalisation (M10) * acme mode REQUIRES apply_immediately=true (M22) * acme mode REQUIRES frontend.mode='http' AND bind_port=80 (HTTP-01 challenge; Round 10 micro-finding) - BackendStep / FrontendStep * '_' system-prefix rejection * https_redirect & redirect_rules mutual-exclusion (M19) - _strip_pem_from_payload (draft persistence; M14/M9) These tests stay 100% in the Pydantic model surface — no DB, no FastAPI TestClient — so they run hermetically alongside the existing pure-unit tests in this directory. """ import pytest from pydantic import ValidationError from models.site_wizard import ( BackendStep, FrontendStep, SiteCreate, ServerStep, SSLChoice, _strip_pem_from_payload, ) # ---------------------------------------------------------------------------- # Helper builders # ---------------------------------------------------------------------------- def _backend(name="be_test", **overrides): return {"name": name, **overrides} def _server(server_name="srv1", **overrides): return { "server_name": server_name, "server_address": "10.0.0.1", "server_port": 8080, **overrides, } def _frontend(name="fe_http", mode="http", bind_port=80, **overrides): return {"name": name, "mode": mode, "bind_port": bind_port, **overrides} _DUMMY_CERT_PEM = "-----BEGIN CERTIFICATE-----\nMIIBdummy\n-----END CERTIFICATE-----" _DUMMY_KEY_PEM = "-----BEGIN PRIVATE KEY-----\nMIIBdummy\n-----END PRIVATE KEY-----" def _ssl_for_mode(ssl_mode): """Bulgu #31: 'upload' / 'existing' modes need shape-valid payloads.""" if ssl_mode == "upload": return { "mode": "upload", "name": "cert-test", "certificate_content": _DUMMY_CERT_PEM, "private_key_content": _DUMMY_KEY_PEM, } if ssl_mode == "existing": return {"mode": "existing", "ssl_certificate_id": 1} return {"mode": ssl_mode} def _payload(ssl_mode="none", apply_immediately=False, **overrides): base = { "cluster_id": 1, "domains": ["www.example.com"], "backend": _backend(), "servers": [_server()], "frontend": _frontend(), "ssl": _ssl_for_mode(ssl_mode), "apply_immediately": apply_immediately, } base.update(overrides) return base # ---------------------------------------------------------------------------- # Domain validation # ---------------------------------------------------------------------------- def test_payload_normalises_domain_to_lowercase(): p = SiteCreate(**_payload(domains=["WWW.Example.COM"]) | {"domains": ["WWW.Example.COM"]}) assert p.domains == ["www.example.com"] def test_payload_rejects_garbage_domain(): with pytest.raises(ValidationError): SiteCreate(**(_payload() | {"domains": ["not a domain"]})) def test_payload_accepts_wildcard_subdomain(): p = SiteCreate(**(_payload() | {"domains": ["*.example.com"]})) assert p.domains == ["*.example.com"] def test_payload_rejects_empty_domains_list(): with pytest.raises(ValidationError): SiteCreate(**(_payload() | {"domains": []})) # ---------------------------------------------------------------------------- # Backend / frontend system-prefix rejection # ---------------------------------------------------------------------------- def test_backend_rejects_underscore_prefix(): with pytest.raises(ValidationError): BackendStep(**_backend(name="_acme_challenge_backend")) def test_frontend_rejects_underscore_prefix(): with pytest.raises(ValidationError): FrontendStep(**_frontend(name="_internal_fe")) def test_backend_accepts_normal_name(): bs = BackendStep(**_backend(name="be_legitimate")) assert bs.name == "be_legitimate" def test_backend_rejects_invalid_chars(): with pytest.raises(ValidationError): BackendStep(**_backend(name="be with spaces")) # ---------------------------------------------------------------------------- # https_redirect / redirect_rules mutual exclusion (M19) # ---------------------------------------------------------------------------- def test_frontend_rejects_https_redirect_with_redirect_rules(): with pytest.raises(ValidationError, match="mutually exclusive"): FrontendStep(**_frontend( https_redirect=True, redirect_rules=[{"from": "/", "to": "https://x"}], )) def test_frontend_accepts_https_redirect_alone(): fe = FrontendStep(**_frontend(https_redirect=True)) assert fe.https_redirect is True assert fe.redirect_rules == [] def test_frontend_accepts_redirect_rules_alone(): fe = FrontendStep(**_frontend( https_redirect=False, redirect_rules=[{"from": "/", "to": "https://x"}], )) assert fe.https_redirect is False assert len(fe.redirect_rules) == 1 # ---------------------------------------------------------------------------- # ACME mode enforcement (M22 + Round 10 micro-finding) # ---------------------------------------------------------------------------- def test_acme_mode_requires_apply_immediately(): with pytest.raises(ValidationError, match="apply_immediately=true"): SiteCreate(**_payload(ssl_mode="acme", apply_immediately=False)) def test_acme_mode_requires_frontend_mode_http(): with pytest.raises(ValidationError, match="frontend.mode='http'"): SiteCreate(**(_payload(ssl_mode="acme", apply_immediately=True) | {"frontend": _frontend(mode="tcp")})) def test_acme_mode_allows_non_80_bind_port(): """Bulgu #34 (round-15 audit) — pre-fix the model validator hard- rejected any ACME payload with `frontend.bind_port != 80`. That blanket rule didn't fit the canonical enterprise pattern (one shared port-80 frontend host-routing many sites), so operators on multi-tenant clusters were stuck — port 80 collided with the existing shared frontend, and any other port was rejected here. The cluster-aware reachability check has moved to the route handler (`_validate_acme_port80_reachable`) where DB lookups are available; the model layer no longer blocks based on bind_port. """ p = SiteCreate(**(_payload(ssl_mode="acme", apply_immediately=True) | {"frontend": _frontend(bind_port=8080)})) assert p.ssl.mode == "acme" assert p.frontend.bind_port == 8080 def test_acme_mode_happy_path(): p = SiteCreate(**_payload(ssl_mode="acme", apply_immediately=True)) assert p.ssl.mode == "acme" assert p.apply_immediately is True def test_non_acme_mode_does_not_force_apply_immediately(): p = SiteCreate(**_payload(ssl_mode="none", apply_immediately=False)) assert p.apply_immediately is False def test_upload_mode_does_not_force_http_frontend(): """Upload mode should accept any frontend port — HTTPS bind is the cert's job. R14 #v2: must use distinct ports for HTTP vs HTTPS otherwise the bind_port==https_bind_port collision validator fires (correctly — HAProxy can't bind the same address:port twice). Pick 8080 / 8443 so we exercise BOTH the 'non-80 HTTP allowed' path AND the new distinct-port invariant simultaneously. """ p_kwargs = _payload(ssl_mode="upload", apply_immediately=False) | { "frontend": _frontend(bind_port=8080), "ssl": { "mode": "upload", "name": "x", "certificate_content": "-----BEGIN CERTIFICATE-----\nx\n-----END CERTIFICATE-----", "private_key_content": "-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----", "https_bind_port": 8443, }, } p = SiteCreate(**p_kwargs) assert p.frontend.bind_port == 8080 assert p.ssl.https_bind_port == 8443 # ---------------------------------------------------------------------------- # Bulgu #31: upload mode rejects empty/missing PEMs and missing cert name # ---------------------------------------------------------------------------- def test_upload_mode_rejects_empty_certificate_content(): payload = _payload(ssl_mode="upload") payload["ssl"] = {**payload["ssl"], "certificate_content": ""} with pytest.raises(ValidationError) as ei: SiteCreate(**payload) assert "certificate_content" in str(ei.value) def test_upload_mode_rejects_empty_private_key_content(): payload = _payload(ssl_mode="upload") payload["ssl"] = {**payload["ssl"], "private_key_content": ""} with pytest.raises(ValidationError) as ei: SiteCreate(**payload) assert "private_key_content" in str(ei.value) def test_upload_mode_rejects_non_pem_text(): payload = _payload(ssl_mode="upload") payload["ssl"] = {**payload["ssl"], "certificate_content": "not a pem"} with pytest.raises(ValidationError) as ei: SiteCreate(**payload) assert "PEM" in str(ei.value) or "BEGIN" in str(ei.value) def test_upload_mode_rejects_missing_cert_name(): payload = _payload(ssl_mode="upload") payload["ssl"] = {**payload["ssl"], "name": ""} with pytest.raises(ValidationError) as ei: SiteCreate(**payload) assert "ssl.name" in str(ei.value) or "name" in str(ei.value) def test_existing_mode_requires_ssl_certificate_id(): payload = _payload(ssl_mode="existing") payload["ssl"] = {"mode": "existing"} # missing ssl_certificate_id with pytest.raises(ValidationError) as ei: SiteCreate(**payload) assert "ssl_certificate_id" in str(ei.value) # ---------------------------------------------------------------------------- # SSLChoice basic shape # ---------------------------------------------------------------------------- def test_ssl_choice_default_https_bind_port_443(): # SSLChoice itself does NOT enforce PEM content (the model_validator on # SiteCreate does); SSLChoice only enforces mode literal. s = SSLChoice(mode="upload", name="x", certificate_content="X", private_key_content="Y") assert s.https_bind_port == 443 def test_ssl_choice_acme_default_auto_renew_true(): s = SSLChoice(mode="acme") assert s.auto_renew is True def test_ssl_choice_invalid_mode_rejected(): with pytest.raises(ValidationError): SSLChoice(mode="self-signed") # ---------------------------------------------------------------------------- # ServerStep # ---------------------------------------------------------------------------- def test_server_step_port_range_validated(): with pytest.raises(ValidationError): ServerStep(server_name="s", server_address="10.0.0.1", server_port=70000) def test_server_step_weight_default_100(): s = ServerStep(server_name="s", server_address="10.0.0.1", server_port=80) assert s.weight == 100 # ---------------------------------------------------------------------------- # _strip_pem_from_payload (M14/M9) # ---------------------------------------------------------------------------- def test_strip_pem_removes_private_key_content(): payload = { "ssl": { "mode": "upload", "name": "cert", "certificate_content": "-----BEGIN CERT-----", "private_key_content": "-----BEGIN PRIVATE KEY-----", "chain_content": "-----BEGIN CERT-----", }, } out = _strip_pem_from_payload(payload) assert out["ssl"]["private_key_content"] == "" assert out["ssl"]["certificate_content"] == "" assert out["ssl"]["chain_content"] == "" # Non-sensitive fields preserved. assert out["ssl"]["mode"] == "upload" assert out["ssl"]["name"] == "cert" def test_strip_pem_recurses_into_lists(): payload = { "servers": [ {"server_name": "s1", "private_key": "K1", "ssl_enabled": True}, {"server_name": "s2", "private_key": "K2", "ssl_enabled": False}, ], } out = _strip_pem_from_payload(payload) for s in out["servers"]: assert s["private_key"] == "" def test_strip_pem_preserves_unrelated_fields(): payload = { "domains": ["www.example.com"], "cluster_id": 1, "ssl": {"mode": "none"}, } out = _strip_pem_from_payload(payload) assert out == payload # No PEM fields present → identical def test_strip_pem_handles_primitive_values(): """Top-level primitives should pass through unchanged.""" assert _strip_pem_from_payload(123) == 123 assert _strip_pem_from_payload("hello") == "hello" assert _strip_pem_from_payload(None) is None assert _strip_pem_from_payload([1, 2, 3]) == [1, 2, 3]