"""Tests for Agent Manifest A–C: schema status, provision, record adapter.""" from __future__ import annotations import json import os import subprocess import tempfile from pathlib import Path from datetime import datetime, timezone from uuid import uuid4 import pytest from src.domain.codegen.entity_catalog import ( colliding_existing_key, resolve_database_view, ) from src.domain.codegen.platform_backend import build_platform_backend_capabilities from src.domain.codegen.prompt_wrapper import build_system_prompt from src.domain.codegen.provision_manifest import ( assert_compatible_entity, provision_entities, validate_manifest, ) from src.domain.codegen.record_adapter import install_record_adapter, record_adapter_files from src.domain.manifest.schemas import ( ManifestAISection, ManifestAuthSection, ManifestBffConventions, ManifestCredentialsSection, ManifestDataImportSection, ManifestDatabaseSection, ManifestEntitlementsSection, ManifestEnvVarsSection, ManifestFeaturePacksSection, ManifestFunctionsSection, ManifestKnowledgeBaseSection, ManifestPlatformSDK, ManifestProjectSection, ManifestQueryDslSection, ManifestRealtimeSection, ManifestResponse, ManifestStorageSection, ) from src.integrations.cloud_schema import normalize_cloud_entity RUNTIME = Path(__file__).resolve().parents[1] / "src" / "domain" / "codegen" / "runtime" / "p2hs-records.mjs" def _entity(**overrides): base = { "key": "universities", "label": "Universities", "rls_enabled": False, "fields": [ {"key": "name", "label": "Name", "type": "text", "required": True, "unique": False}, ], } base.update(overrides) return base def test_validate_manifest_accepts_entities_and_buckets(): data = validate_manifest({ "version": 1, "entities": [_entity()], "storage_buckets": [{"name": "uploads", "access_policy": "private", "rls_enabled": True}], }) assert data["entities"][0]["key"] == "universities" assert data["storage_buckets"][0]["name"] == "uploads" def test_validate_manifest_ignores_auth_block(): data = validate_manifest({ "version": 1, "storage_buckets": [{"name": "student-documents", "access_policy": "private", "rls_enabled": True}], "auth": { "enabled": True, "custom_roles": ["admin", "counselor"], "seed_users": [{"email": "owner@example.com", "role": "super_admin"}], }, }) assert data["storage_buckets"][0]["name"] == "student-documents" assert "auth" not in data def test_schema_status_defaults_and_compact_omitted(): unavailable = ManifestDatabaseSection() assert unavailable.schema_status == "unavailable" assert unavailable.schema_source == "none" assert unavailable.entities is None omitted = ManifestDatabaseSection(schema_status="omitted") assert omitted.schema_status == "omitted" empty = ManifestDatabaseSection(schema_status="empty", entities=[]) assert empty.schema_status == "empty" assert empty.persistence == "cloud_record_engine" def test_auth_user_directory_is_platform_only(): assert ManifestAuthSection(enabled=False).user_directory == "platform_only" assert ManifestAuthSection(enabled=False).admin_panel_path == "/admin" assert ManifestAuthSection(enabled=False).customer_login_path == "/login" assert ManifestAuthSection(enabled=False).admin_role == "admin" def test_empty_live_schema_does_not_wipe_cached_catalog(): cached = [_entity()] view = resolve_database_view(live_status="ok", live_entities=[], cached_entities=cached) assert view["schema_status"] == "ready" assert view["schema_source"] == "cached" assert view["entities"][0]["key"] == "universities" assert view["persist"] is None failed = resolve_database_view(live_status="error", live_entities=None, cached_entities=cached) assert failed["schema_status"] == "unavailable" assert failed["schema_source"] == "cached" assert failed["entities"][0]["key"] == "universities" def test_university_collides_with_universities(): assert colliding_existing_key("university", ["universities"]) == "universities" assert colliding_existing_key("universities", ["universities"]) is None assert colliding_existing_key("leads", ["universities"]) is None def test_compatible_existing_fields_are_noop_and_new_fields_are_additive(): live = _entity(status="active", fields=[ {"key": "name", "label": "Name", "type": "text", "required": True, "unique": False}, {"key": "legacy", "label": "Legacy", "type": "text", "required": False, "unique": False}, ]) declared = _entity(fields=[ {"key": "name", "label": "Name", "type": "text", "required": True, "unique": False}, {"key": "country", "label": "Country", "type": "text", "required": False, "unique": False}, ]) additive = assert_compatible_entity(declared, live) assert [f["key"] for f in additive] == ["country"] def test_incompatible_live_field_is_rejected(): live = _entity() declared = _entity(fields=[ {"key": "name", "label": "Name", "type": "number", "required": True, "unique": False}, ]) with pytest.raises(ValueError, match="Incompatible field"): assert_compatible_entity(declared, live) class _FakeCloud: def __init__(self, entities): self.entities = list(entities) self.created = [] self.patched = [] async def get_project_schema(self, project_id): return {"entities": self.entities} async def create_project_entity(self, project_id, definition): self.created.append(definition) self.entities.append({ "key": definition["entity_key"], "label": definition["name"], "fields": definition["schema"]["fields"], "rls_enabled": definition["settings"]["rls_enabled"], "status": "active", }) return {} async def patch_project_entity(self, project_id, entity_key, payload): self.patched.append((entity_key, payload)) for entity in self.entities: if entity["key"] == entity_key: entity["fields"] = payload["schema"]["fields"] return {} @pytest.mark.asyncio async def test_provision_entities_creates_then_adds_fields_idempotently(): cloud = _FakeCloud([]) created = await provision_entities(cloud, "prj", [_entity()]) assert created == [{"entity": "universities", "ok": True, "action": "created"}] assert cloud.created again = await provision_entities(cloud, "prj", [_entity()]) assert again == [{"entity": "universities", "ok": True, "action": "existing"}] assert not cloud.patched with_country = _entity(fields=[ {"key": "name", "label": "Name", "type": "text", "required": True, "unique": False}, {"key": "country", "label": "Country", "type": "text", "required": False, "unique": False}, ]) updated = await provision_entities(cloud, "prj", [with_country]) assert updated == [{"entity": "universities", "ok": True, "action": "updated"}] assert cloud.patched keys = [f["key"] for f in cloud.entities[0]["fields"]] assert keys == ["name", "country"] @pytest.mark.asyncio async def test_provision_entities_rejects_synonym_of_live_or_known_key(): cloud = _FakeCloud([_entity(status="active")]) with pytest.raises(ValueError, match="collides with existing 'universities'"): await provision_entities(cloud, "prj", [_entity(key="university", label="University")]) assert not cloud.created empty_cloud = _FakeCloud([]) with pytest.raises(ValueError, match="collides with existing 'universities'"): await provision_entities( empty_cloud, "prj", [_entity(key="university", label="University")], known_entities=[_entity()], ) assert not empty_cloud.created @pytest.mark.asyncio async def test_provision_entities_refuses_pack_owned_keys_before_create_or_patch(): cloud = _FakeCloud([_entity(status="active")]) with pytest.raises(ValueError, match="owned by an installed Feature Pack"): await provision_entities( cloud, "prj", [_entity()], protected_entity_keys={"universities"}, ) assert not cloud.created assert not cloud.patched empty = _FakeCloud([]) with pytest.raises(ValueError, match="owned by an installed Feature Pack"): await provision_entities( empty, "prj", [_entity()], protected_entity_keys={"universities"}, ) assert not empty.created @pytest.mark.asyncio async def test_provision_entities_refuses_when_cloud_schema_unavailable(): class _Down(_FakeCloud): async def get_project_schema(self, project_id): raise RuntimeError("Cloud schema is unavailable") cloud = _Down([]) with pytest.raises(ValueError, match="Cloud schema is unavailable"): await provision_entities(cloud, "prj", [_entity()], known_entities=[_entity()]) assert not cloud.created def test_normalize_cloud_entity_keeps_key_when_fields_are_incomplete(): normalized = normalize_cloud_entity({ "entity_key": "leads", "name": "Leads", "schema_json": json.dumps({"fields": [{"key": "email", "type": "text"}]}), "settings_json": json.dumps({"rls_enabled": True}), }) assert normalized["key"] == "leads" assert normalized["rls_enabled"] is True with pytest.raises(ValueError, match="fields are unavailable or malformed"): normalize_cloud_entity({"entity_key": "bad", "schema_json": "{}"}) with pytest.raises(ValueError): normalize_cloud_entity({"schema_json": "{}"}) @pytest.mark.parametrize("project_type", ["html", "react_app", "nuxt_app"]) def test_record_adapter_installs_pages_function_for_every_starter(project_type): files = record_adapter_files(project_type) assert "functions/_lib/p2hs-records.mjs" in files assert "functions/api/records/[[path]].js" in files assert "readCloudConfig" in files["functions/_lib/p2hs-records.mjs"] assert 'staffRoles = new Set(["admin", "super_admin"])' in files["functions/_lib/p2hs-records.mjs"] assert "readCloudConfig" in files["functions/api/records/[[path]].js"] if project_type == "nuxt_app": assert "server/api/records/[...path].ts" in files assert "readCloudConfig" in files["server/api/records/[...path].ts"] else: assert "server/api/records/[...path].ts" not in files with tempfile.TemporaryDirectory() as tmp: install_record_adapter(tmp, project_type) assert (Path(tmp) / "functions/api/records/[[path]].js").is_file() marker = Path(tmp) / "functions/_lib/p2hs-records.mjs" marker.write_text("owned-by-app", encoding="utf-8") install_record_adapter(tmp, project_type) assert marker.read_text(encoding="utf-8") == "owned-by-app" install_record_adapter(tmp, project_type, replace_local=True) assert "p2hs-records" in marker.read_text(encoding="utf-8") assert (Path(tmp) / ".p2hs/local-bak/functions/_lib/p2hs-records.mjs").is_file() def test_record_adapter_skips_non_pages_types(): assert record_adapter_files("shopify_app") == {} def test_injected_contract_and_playbook_describe_schema_status(): prompt = build_system_prompt("nuxt_app", "") assert "database.schema_status" in prompt assert "Do not invent a local database" in prompt assert "`entities`" in prompt assert "Reuse those exact entity keys" in prompt assert "never create a parallel table" in prompt manifest = ManifestResponse( generated_at=datetime.now(timezone.utc), project=ManifestProjectSection( id=str(uuid4()), name="Demo", workspace_id=str(uuid4()), plan="pro", cloud_gateway_url="https://cloud.example", ), credentials=ManifestCredentialsSection(), env_vars=ManifestEnvVarsSection(), database=ManifestDatabaseSection(schema_status="empty", entities=[]), auth=ManifestAuthSection(enabled=True), storage=ManifestStorageSection(), functions=ManifestFunctionsSection(), feature_packs=ManifestFeaturePacksSection(), connectors={}, integrations={}, connections={}, realtime=ManifestRealtimeSection(), ai=ManifestAISection(enabled=False), schedules=[], knowledge_base=ManifestKnowledgeBaseSection(), api_surface={}, bff_conventions=ManifestBffConventions(), platform_sdk=ManifestPlatformSDK(), entitlements=ManifestEntitlementsSection(plan="pro"), query_dsl=ManifestQueryDslSection(), data_import=ManifestDataImportSection(), ) contract = build_platform_backend_capabilities(manifest) assert contract["contract_version"] == "3" assert contract["backend"]["provider"] == "p2hs_cloud" assert contract["backend"]["customer_backend"] is None assert contract["backend"]["uses"] == "p2hs_cloud" assert contract["database"]["schema_status"] == "empty" assert contract["auth"]["user_directory"] == "platform_only" # Empty conventions do not advertise uninstalled ports. assert contract["bff_conventions"]["ports"] == [] assert contract["bff_conventions"]["records"] == {} def test_read_cloud_config_and_record_envelopes_via_node(): if not RUNTIME.is_file(): pytest.skip("adapter source missing") adapter = RUNTIME.as_posix().replace("\\", "/") script = f""" import {{ readCloudConfig, recordsRequest }} from 'file:///{adapter}'; const cfg = readCloudConfig({{ NUXT_PUBLIC_P2HS_CLOUD_PUBLIC_URL: "https://gw.example", VITE_P2HS_PROJECT_ID: "prj_1", CLOUD_SERVICE_TOKEN: "tok_secret", VITE_CLOUD_SERVICE_TOKEN: "leaked", }}); if (cfg.cloudGatewayUrl !== "https://gw.example") throw new Error("gateway"); if (cfg.projectId !== "prj_1") throw new Error("project"); if (cfg.cloudServiceToken !== "tok_secret") throw new Error("token must ignore public vite keys"); const fetchImpl = async (url, opts) => {{ if (url.endsWith("/auth/me")) return {{ ok: true, json: async () => ({{ user: {{ id: "u1", role: "admin" }} }}) }}; if (url.includes("/records/query")) return {{ ok: true, json: async () => ({{ data: {{ records: [{{ id: "r1" }}], next_cursor: null }} }}) }}; if (url.includes("/records/aggregate")) return {{ ok: true, json: async () => ({{ rows: [{{ total: 1 }}] }}) }}; if (opts.method === "DELETE") return {{ ok: true, json: async () => ({{ ok: true }}) }}; return {{ ok: true, json: async () => ({{ record: {{ id: "r1", data: {{ name: "A" }} }} }}) }}; }}; const query = await recordsRequest({{ config: cfg, authorization: "Bearer user", method: "POST", segments: ["universities", "query"], body: {{ limit: 10 }}, fetchImpl, }}); if (!Array.isArray(query.records) || query.total !== 1) throw new Error("query envelope"); const created = await recordsRequest({{ config: cfg, authorization: "Bearer user", method: "POST", segments: ["universities"], body: {{ data: {{ name: "A" }} }}, fetchImpl, }}); if (!created.record?.id) throw new Error("create envelope"); const deleted = await recordsRequest({{ config: cfg, authorization: "Bearer user", method: "DELETE", segments: ["universities", "r1"], fetchImpl, }}); if (deleted.success !== true) throw new Error("delete envelope"); console.log("ok"); """ result = subprocess.run( ["node", "--input-type=module", "-e", script], capture_output=True, text=True, env=os.environ.copy(), ) if result.returncode == 127 or "not recognized" in (result.stderr or "").lower(): pytest.skip("node is not available") assert result.returncode == 0, result.stderr assert "ok" in result.stdout