import json import subprocess import sys from pathlib import Path import pytest from src.domain.codegen.backend_guard import validate_port_artifacts from src.domain.codegen.port_adapter import ( LOCK_PATH, catalog_adapter_files, install_catalog_adapters, port_install_status, ) from src.integrations.platform_ports import ( CATALOG, QUEUE_REQUEST_TYPES, allowed_request_types, api_surface, entitled_ports, ) def test_only_executable_jobs_can_be_authorized(): assert 'record.query' not in allowed_request_types({}) assert 'data.import' not in allowed_request_types({}) assert 'voice.call.handle' not in QUEUE_REQUEST_TYPES assert 'livechat.message' not in QUEUE_REQUEST_TYPES assert 'email.send' in allowed_request_types({}) assert allowed_request_types({'cloud_allowed_request_types': []}) == set() assert allowed_request_types({'cloud_allowed_request_types': ['invented']}) == set() assert allowed_request_types({'cloud_enabled': False}) == set() assert 'voice.call.handle' in allowed_request_types({'cloud_capabilities': {'voice.enabled': True}}) def test_catalog_3_drops_stub_trio_from_projection(): stub_ports = ( 'jobs.marketing_subscribe', 'jobs.crm_contact_upsert', 'jobs.custom_provider_request', ) port_ids = {p['port_id'] for p in CATALOG['ports']} # Stub trio dropped at catalog 3.1.0; version has since advanced (additive # changes only) but the trio must stay dropped. assert tuple(int(part) for part in CATALOG['version'].split('.')) >= (3, 1, 0) for port_id in stub_ports: assert port_id not in port_ids for request_type in ( "marketing.subscribe", "crm.contact.upsert", "custom.provider.request", "pdf.generate", ): assert request_type not in QUEUE_REQUEST_TYPES assert request_type not in allowed_request_types({}) @pytest.mark.parametrize('framework', ['html', 'react_app', 'nuxt_app']) def test_every_projected_port_is_installed_and_pinned(tmp_path, framework): ports = entitled_ports({'cloud_capabilities': {'voice.enabled': True}}, 'pro') context = {'ports': ports, 'project_id': 'project', 'workspace_id': 'workspace', 'client_id': 'client'} install_catalog_adapters(str(tmp_path), framework, context) lock = json.loads((tmp_path / LOCK_PATH).read_text()) for port in ports: assert port['port_id'] in lock['ports'] for kind in (['pages', 'nuxt'] if framework == 'nuxt_app' else ['pages']): assert (tmp_path / port['bff']['files'][kind]).is_file() validate_port_artifacts(str(tmp_path), lock) surface = api_surface(ports) assert sum(len(group) for group in surface.values()) == len(ports) for group in surface.values(): for operation in group.values(): assert operation['schema'], f"{operation['port_id']} is missing its contract schema" assert surface['auth']['password_reset_request']['path'] == '/api/auth/forgot-password' assert surface['jobs']['email_send']['path'] == '/api/jobs/email.send' queue = [p for p in CATALOG['ports'] if p['cloud']['transport'] == 'queue'] assert queue assert all(p['cloud']['path'] == '/v1/projects/:project_id/jobs' for p in queue) assert all(p['bff']['auth'] == 'staff' for p in queue) assert {p['port_id']: p['entitlement'] for p in CATALOG['ports'] if p['port_id'].startswith('records.')}['records.create'] == 'cloud' def test_cloud_contract_manifest_is_locally_verifiable(tmp_path): install_catalog_adapters(str(tmp_path), "nuxt_app") contract = json.loads((tmp_path / "p2hs/cloud-contract.json").read_text()) assert contract["schema_version"] == 1 assert contract["validation"]["network_required"] is False assert contract["ports"] assert all( port["cloud"]["transport"] == "runtime" or port["bff"]["method"] in {"GET", "HEAD", "DELETE"} or ( isinstance(port["schema"].get("request_body"), dict) and isinstance(port["schema"].get("response"), dict) ) for port in contract["ports"] ) result = subprocess.run( ["node", "p2hs/verify-cloud-contract.mjs"], cwd=tmp_path, capture_output=True, text=True, check=False, ) assert result.returncode == 0, result.stderr assert "P2HS_CLOUD_CONTRACT_OK" in result.stdout @pytest.mark.parametrize('shadow', [ 'server/api/auth/signin.ts', 'server/api/auth/signin/index.post.ts', 'server/api/auth/[name].post.ts', 'server/api/[...path].ts', 'functions/api/storage/[bucket]/files/index.js', 'functions/api/[[anything]].js', ]) def test_shadow_routes_fail_even_with_platform_marker(tmp_path, shadow): install_catalog_adapters(str(tmp_path), 'nuxt_app') lock = json.loads((tmp_path / LOCK_PATH).read_text()) target = tmp_path / shadow target.parent.mkdir(parents=True, exist_ok=True) target.write_text('// p2hs-records\nexport default () => ({ success: true });') with pytest.raises(ValueError, match='shadows'): validate_port_artifacts(str(tmp_path), lock) def test_modified_or_deleted_artifacts_and_lock_fail(tmp_path): files = catalog_adapter_files('nuxt_app') lock = json.loads(files[LOCK_PATH]) for rel in ['server/utils/p2hs-ports.mjs', 'server/api/auth/signin.post.ts', LOCK_PATH]: install_catalog_adapters(str(tmp_path), 'nuxt_app') (tmp_path / rel).write_text('{}') with pytest.raises(ValueError, match='modified|lock was changed'): validate_port_artifacts(str(tmp_path), lock) def test_revoked_entitlements_remove_installed_routes(tmp_path): install_catalog_adapters(str(tmp_path), 'nuxt_app', {'ports': entitled_ports(plan='pro')}) assert (tmp_path / 'server/api/ai/complete.post.ts').exists() install_catalog_adapters(str(tmp_path), 'nuxt_app', {'ports': entitled_ports({'cloud_enabled': False})}) assert not (tmp_path / 'server/api/ai/complete.post.ts').exists() assert not (tmp_path / 'server/api/auth/signin.post.ts').exists() validate_port_artifacts(str(tmp_path)) def test_port_install_status_detects_stale_checkout(tmp_path): empty = port_install_status(str(tmp_path), "nuxt_app") assert empty["stale"] is True assert empty["installed_version"] is None assert empty["catalog_version"] == CATALOG["version"] install_catalog_adapters(str(tmp_path), "nuxt_app") current = port_install_status(str(tmp_path), "nuxt_app") assert current["stale"] is False (tmp_path / "server/utils/p2hs-ports.mjs").write_text("// stale\n", encoding="utf-8") assert port_install_status(str(tmp_path), "nuxt_app")["stale"] is True def test_reproject_after_provider_edit_restores_platform_owned_routes(tmp_path): """A later codegen pass must repair routes an older app/provider overwrote.""" install_catalog_adapters(str(tmp_path), "nuxt_app") auth_route = tmp_path / "server/api/auth/signin.post.ts" users_route = tmp_path / "server/api/admin/users.ts" auth_route.write_text("export default () => ({ local: true })\n", encoding="utf-8") users_route.write_text("export default () => ({ local: true })\n", encoding="utf-8") install_catalog_adapters(str(tmp_path), "nuxt_app") assert "p2hs-ports" in auth_route.read_text(encoding="utf-8") users_source = users_route.read_text(encoding="utf-8") assert "p2hs-auth-users" in users_source assert 'getCookie(event, P2HS_ACCESS_COOKIE)' in users_source validate_port_artifacts(str(tmp_path)) def test_catalog_has_no_uninstalled_app_port(): for port in CATALOG['ports']: assert port['bff'] or port['audience'] == 'external_runtime' assert port['cloud']['transport'] in {'queue', 'http', 'runtime'} def test_core_snapshot_matches_cloud_when_both_checkouts_available(): cloud = Path(__file__).resolve().parents[2] / 'P2HSCloud/packages/shared/src/platform-port-catalog.json' if not cloud.exists(): pytest.skip('Cross-repository parity is checked when the Cloud checkout is available') core = Path(__file__).resolve().parents[1] / 'src/integrations/contracts/platform-port-catalog.json' assert core.read_bytes() == cloud.read_bytes()