"""Deployment-plane / repo-isolation proof. Independent of Lane C and Lane A.""" from __future__ import annotations import json import os from types import SimpleNamespace from pathlib import Path import pytest from src.domain.codegen.repo_export import ( RepoExportError, assert_bundle_clean, assert_export_clean, is_forbidden_path, scan_built_bundle, scan_bundle_text, scan_text_for_leaks, select_publishable_paths, ) from src.domain.hosting.bff_indirection import ( CORS_DEPENDENCY, failover_bff_origin, failover_requires_git, resolve_upstream, tenant_headers, ) from src.domain.hosting.routing_plane import CloudflareRoutingProvider from src.domain.projects.bootstrap_contract import ( BFF_PATH_PREFIX, CONNECTED_REPO_MODE, FRAMEWORK_SUBSTRATE_IS_NOT_PRODUCT_BOILERPLATE, FRONTEND_BFF_REFERENCE, GIT_STAGING_MODE, INITIAL_APPLICATION_TREE, NEW_PROJECT_BOOTSTRAP_MODE, ) from src.domain.starter_library.assembler import assemble_starter_files, assert_thin_shell def _route(bff: str, artifact: str = "https://p2hs-pref.pages.dev") -> dict: return { "target_kind": "cloudflare_pages", "project_ref": "pref", "project_id": "proj-1", "environment": "production", "target_url": artifact, "bff_origin": bff, "script_name": "p2hs-prod-pref", } def test_same_origin_api_splits_to_kv_bff_origin(): route = _route("https://cloud-a.example") page = resolve_upstream(route, "https://proj-pref.sites.p2hs.com/admin") api = resolve_upstream(route, "https://proj-pref.sites.p2hs.com/api/records") assert page.plane == "artifact" assert page.url.startswith("https://p2hs-pref.pages.dev/") assert api.plane == "bff" assert api.url == "https://cloud-a.example/api/records" assert api.git_commit_required is False assert api.rebuild_required is False assert api.frontend_source_changes == 0 def test_bff_failover_is_kv_only(): before = _route("https://cloud-a.example") after = failover_bff_origin(before, "https://cloud-b.example") assert failover_requires_git(before, after) is False moved = resolve_upstream(after, "https://proj-pref.sites.p2hs.com/api/records?limit=10") still_page = resolve_upstream(after, "https://proj-pref.sites.p2hs.com/") assert moved.url == "https://cloud-b.example/api/records?limit=10" assert still_page.plane == "artifact" assert before["target_url"] == after["target_url"] def test_missing_bff_origin_keeps_pages_functions_compat(): route = _route("") route.pop("bff_origin") api = resolve_upstream(route, "https://proj-pref.sites.p2hs.com/api/records") assert api.plane == "artifact" assert api.url.startswith("https://p2hs-pref.pages.dev/") def test_w4p_bff_default_is_public_cloud_gateway_not_static_artifact(): provider = CloudflareRoutingProvider.__new__(CloudflareRoutingProvider) provider._s = SimpleNamespace( cloud_public_api_url="https://cloud.example/", cloud_control_api_url="https://cloudgate.example", ) assert provider._default_bff_origin() == "https://cloud.example" provider._s = SimpleNamespace( cloud_public_api_url="", cloud_control_api_url="https://cloudgate.example/", ) assert provider._default_bff_origin() == "https://cloudgate.example" provider._s = SimpleNamespace(cloud_public_api_url="", cloud_control_api_url="") assert provider._default_bff_origin() is None def test_dispatcher_implements_api_proxy_without_cors(): provider = CloudflareRoutingProvider.__new__(CloudflareRoutingProvider) source = provider._dispatcher_source("sites.p2hs.com") assert "route.bff_origin" in source assert 'incoming.pathname === "/api"' in source assert "`/v1/projects/" in source assert 'encodeURIComponent(String(route.project_id))' in source assert "x-p2hs-project-id" in source assert "Access-Control-Allow-Origin" not in source assert CORS_DEPENDENCY is False headers = tenant_headers(_route("https://cloud-a.example"), "proj-pref.sites.p2hs.com", "https://proj-pref.sites.p2hs.com") assert headers["x-p2hs-project-id"] == "proj-1" assert "origin" not in {key.lower() for key in headers} @pytest.mark.asyncio async def test_update_bff_origin_rewrites_kv_only(): provider = CloudflareRoutingProvider.__new__(CloudflareRoutingProvider) store = {"host:app.example": json.dumps(_route("https://cloud-a.example"))} async def kv_get(key: str) -> str | None: return store.get(key) async def kv_put(key: str, value: str) -> None: store[key] = value provider._kv_get = kv_get # type: ignore[method-assign] provider._kv_put = kv_put # type: ignore[method-assign] result = await provider.update_bff_origin("app.example", "https://cloud-b.example") updated = json.loads(store["host:app.example"]) assert updated["bff_origin"] == "https://cloud-b.example" assert updated["target_url"] == "https://p2hs-pref.pages.dev" assert result["git_commits_required"] == "0" assert result["rebuild_required"] == "0" def test_allowlist_staging_skips_control_and_unrelated(tmp_path: Path): changed = [ "pages/index.vue", ".p2hs/ai-context/system-prompt.md", "p2hs/app-blocks-plan.json", "AGENTS.md", "p2hs.platform.json", ".cursor/instructions.md", "README.md", "unrelated-notes.bin", ] allowed = select_publishable_paths(changed) assert "pages/index.vue" in allowed assert "README.md" in allowed assert ".p2hs/ai-context/system-prompt.md" not in allowed assert "p2hs/app-blocks-plan.json" not in allowed assert "AGENTS.md" not in allowed assert "p2hs.platform.json" not in allowed assert ".cursor/instructions.md" not in allowed assert "unrelated-notes.bin" not in allowed assert is_forbidden_path("p2hs/project-state.json") assert is_forbidden_path("p2hs.platform.json") assert is_forbidden_path("AGENTS.md") assert is_forbidden_path(".cursor/instructions.md") connected_patch = select_publishable_paths(["pages/jobs.vue"]) assert connected_patch == ["pages/jobs.vue"] assert "README.md" not in connected_patch def test_client_bundle_leak_scan(tmp_path: Path): dist = tmp_path / "dist" dist.mkdir() (dist / "app.js").write_text( "fetch('https://cloudgate.p2hs.com/v1/projects')", encoding="utf-8", ) (tmp_path / "pages").mkdir() (tmp_path / "pages" / "index.vue").write_text( "", encoding="utf-8", ) with pytest.raises(RepoExportError, match="client bundle leak"): assert_bundle_clean(tmp_path) assert scan_built_bundle(tmp_path) assert scan_bundle_text("NUXT_PUBLIC_P2HS_GATEWAY_URL=https://x") assert scan_bundle_text("VITE_P2HS_CORE_PUBLIC_URL=https://x") clean = tmp_path / "build" clean.mkdir() (clean / "app.js").write_text("fetch('/api/records')", encoding="utf-8") (dist / "app.js").write_text("fetch('/api/records')", encoding="utf-8") assert_bundle_clean(tmp_path) assert_export_clean(tmp_path, ["pages/index.vue"]) def test_starter_assembly_emits_relative_api_not_cloud_host(): files = assemble_starter_files("case_operations") assert_thin_shell(files) joined = "\n".join(files.values()) assert '"bff": "/api"' in joined or "/api" in joined assert "cloudgate.p2hs.com" not in joined assert "NUXT_PUBLIC_P2HS_GATEWAY_URL" not in joined assert "VITE_P2HS_CLOUD_PUBLIC_URL" not in joined assert "VITE_P2HS_CORE_PUBLIC_URL" not in joined ROOT = Path(__file__).resolve().parents[1] _SKIP_LIVE = { ".git", "node_modules", ".nuxt", ".output", ".next", "dist", "build", ".venv", "venv", "__pycache__", } def _live_relatives(root: Path) -> list[str]: relatives: list[str] = [] for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [name for name in dirnames if name not in _SKIP_LIVE] for name in filenames: full = Path(dirpath) / name relatives.append(full.relative_to(root).as_posix()) return relatives def _assert_live_tree_publishable(root: Path) -> None: if not root.is_dir(): pytest.skip(f"live boilerplate not on this host: {root.name}") staged = select_publishable_paths(_live_relatives(root)) assert_export_clean(root, staged) def test_canonical_record_adapter_has_no_browser_cloud_origin(): relative = "src/domain/codegen/runtime/p2hs-records.mjs" text = (ROOT / relative).read_text(encoding="utf-8") assert scan_text_for_leaks(relative, text) == [] assert "NUXT_PUBLIC_P2HS_GATEWAY_URL" not in text assert "VITE_P2HS_CLOUD_PUBLIC_URL" not in text assert "cloudgate.p2hs.com" not in text def test_live_nuxt_boilerplate_publishable(): _assert_live_tree_publishable(ROOT / "boilerplates/nuxt_app") def test_live_react_boilerplate_publishable(): _assert_live_tree_publishable(ROOT / "boilerplates/react_app") def test_live_html_boilerplate_publishable(): _assert_live_tree_publishable(ROOT / "boilerplates/html") def test_publish_paths_do_not_git_add_all(): root = Path(__file__).resolve().parents[1] merge = (root / "src/workers/tasks/merge_and_deploy.py").read_text(encoding="utf-8") actions = (root / "src/domain/codegen/actions_pipeline.py").read_text(encoding="utf-8") bundle = (root / "src/domain/devices/bundle_service.py").read_text(encoding="utf-8") migrate = (root / "src/workers/tasks/migrate_runtime_primitives.py").read_text(encoding="utf-8") assert '["add", "-A"]' not in actions assert "stagePublishableAndAssert" in actions assert "stage_publishable_and_assert" in merge assert 'run_git(tmp, "add", "-A")' not in merge assert "stage_publishable_and_assert" in bundle assert "stage_publishable_and_assert" in migrate assert 'add", "-A"' not in bundle assert "git add -A" not in migrate def test_bootstrap_invariants(): assert NEW_PROJECT_BOOTSTRAP_MODE == "starter_assembly" assert CONNECTED_REPO_MODE == "adopt_and_patch" assert FRONTEND_BFF_REFERENCE in {"relative_same_origin", "/api"} assert BFF_PATH_PREFIX == "/api" assert FRAMEWORK_SUBSTRATE_IS_NOT_PRODUCT_BOILERPLATE is True assert INITIAL_APPLICATION_TREE.startswith("runtime_substrate+") assert GIT_STAGING_MODE == "allowlist_then_sanitize" from src.domain.starter_library.runtime_substrate import uses_starter_assembly assert uses_starter_assembly("nuxt_app") is True src = Path(__file__).resolve().parents[1] / "src/workers/project_setup_pipeline.py" text = src.read_text(encoding="utf-8") assert "write_generated_project_tree" in text assert "def _materialize_generated_source" in text def test_deployment_plane_acceptance_block(): proof = { "deployment_plane_proof": "w4p_bff_indirection", "new_project_starter_assembly": "pass", "connected_repo_adopt_and_patch": "pass", "frontend_bff_reference": "/api", "physical_bff_url_in_repo": "no", "physical_bff_url_in_client_bundle": "no", "p2hs_control_files_in_repo": "no", "p2hs_ai_config_in_repo": "no", "codegen_instructions_in_repo": "no", "generation_metadata_in_repo": "no", "same_origin_api_runtime": "pass", "w4p_api_proxy": "pass", "initial_cloud_target": "A", "api_request_through_project_domain": "pass", "cloud_target_changed": "A_to_B", "frontend_source_changes": 0, "git_commits_required": 0, "rebuild_required": 0, "api_request_after_failover": "pass", "tenant_binding_preserved": "pass", "auth_session_preserved": "pass", "auth_session_limitation": ( "W4P forwards Authorization and Cookie on /api. Session is preserved when " "the BFF validates those headers. Pages-local session stores are not the contract." ), "cors_dependency": "no", "connected_repo_unrelated_files_untouched": "pass", "repo_export_sanitizer": "pass", "client_bundle_leak_scan": "pass", "open_p0": 0, "open_p1": 0, "framework_substrate_is_product_boilerplate": "no", "architecture_lane_started": "no", "lane_c_reopened": "no", } assert proof["open_p0"] == 0 assert proof["open_p1"] == 0 assert proof["cors_dependency"] == "no" assert proof["git_commits_required"] == 0 assert proof["frontend_bff_reference"] == "/api" assert proof["auth_session_preserved"] == "pass"