diff --git a/tests/test_full_stack_contract.py b/tests/test_full_stack_contract.py index 59cee34..7831816 100644 --- a/tests/test_full_stack_contract.py +++ b/tests/test_full_stack_contract.py @@ -1,7 +1,14 @@ +import os from pathlib import Path +import pytest + +from a2a_pack import LocalRunContext, PlatformUserAuth from a2a_pack.cli.local import load_local_project +import agent as launch_agent +from agent import FetchResult, LaunchCheckStudioV1 + ROOT = Path(__file__).resolve().parents[1] @@ -12,17 +19,128 @@ def test_full_stack_product_contract(): frontend = (ROOT / "frontend" / "src" / "App.jsx").read_text(encoding="utf-8") assert "frontend:" in manifest and "mount: /app" in manifest + assert "public: false" in manifest assert "resources:" in manifest and "databases:" in manifest assert "migrations:" in manifest and "db/migrations" in manifest assert "PlatformUserAuth" in source assert "AgentPlatformResources" in source and "AgentDatabase(" in source - assert "Skill runner" not in frontend, "replace the generic scaffold with the product workflow" + assert "LLMProvisioning" not in source + assert "ctx.llm" not in source + assert "A2A_LITELLM_KEY" not in source and "OPENAI_API_KEY" not in source + assert "Skill runner" not in frontend + assert "DATABASE_URL" not in frontend migrations = list((ROOT / "db" / "migrations").glob("*.sql")) assert migrations and all(path.read_text(encoding="utf-8").strip() for path in migrations) card = load_local_project(ROOT).agent_cls().card() + skill_names = {skill.name for skill in card.skills} + assert {"audit_url", "get_audit"}.issubset(skill_names) databases = card.runtime.platform_resources.databases assert databases, "live Agent Card must declare its managed database" + assert databases[0].scope == "user" assert databases[0].migrations is not None assert databases[0].migrations.path == "db/migrations" + + +def _ctx(): + return LocalRunContext( + auth=PlatformUserAuth(sub="test-user", user_id=42, email="test@example.com"), + task_id="test-run", + ) + + +@pytest.mark.asyncio +async def test_acceptance_success_reload_and_receipt(monkeypatch): + os.environ.pop("DATABASE_URL", None) + stored = {} + + async def fake_host_allowed(host, port): + return False + + async def fake_fetch(url, max_redirects, max_bytes): + return FetchResult( + final_url="https://example.com/", + status_code=200, + headers={ + "content-type": "text/html", + "strict-transport-security": "max-age=31536000", + "x-content-type-options": "nosniff", + "referrer-policy": "strict-origin-when-cross-origin", + }, + body=b""" + + Example Domain + + + + Example + """, + redirect_chain=[], + ) + + async def fake_robots(url): + return {"present": True, "status_code": 200, "checked_url": "https://example.com/robots.txt", "bytes_read": 20} + + async def fake_save(tenant, audit_id, payload, receipt): + stored[(tenant, audit_id)] = payload + assert receipt["receipt_id"] + assert receipt["kind"] == "launchcheck.audit.v1" + + async def fake_load(tenant, audit_id): + return stored.get((tenant, audit_id)) + + monkeypatch.setattr(launch_agent, "_host_is_disallowed", fake_host_allowed) + monkeypatch.setattr(launch_agent, "_fetch_bounded", fake_fetch) + monkeypatch.setattr(launch_agent, "_check_robots", fake_robots) + monkeypatch.setattr(launch_agent, "_save_audit_and_receipt", fake_save) + monkeypatch.setattr(launch_agent, "_load_audit", fake_load) + + app = LaunchCheckStudioV1() + result = await app.local_invoke("audit_url", auth=_ctx().auth, audit_id="studio-launch-v1", url="https://example.com") + assert result["ok"] is True + assert result["audit_id"] == "studio-launch-v1" + assert result["target_url"] == "https://example.com/" + assert result["report"]["reachable"] is True + assert result["receipt"]["receipt_id"].startswith("launchcheck:studio-launch-v1:") + + reopened = await app.local_invoke("get_audit", auth=_ctx().auth, audit_id="studio-launch-v1") + assert reopened["ok"] is True + assert reopened["target_url"] == "https://example.com/" + assert reopened["report"]["reachable"] is True + + +@pytest.mark.asyncio +async def test_acceptance_failure_rejects_loopback_without_internals(monkeypatch): + calls = {"fetch": 0} + + async def fake_fetch(*args, **kwargs): + calls["fetch"] += 1 + raise AssertionError("network must not be called for private targets") + + monkeypatch.setattr(launch_agent, "_fetch_bounded", fake_fetch) + app = LaunchCheckStudioV1() + result = await app.local_invoke("audit_url", auth=_ctx().auth, audit_id="studio-launch-invalid", url="http://127.0.0.1:80") + assert result["ok"] is False + assert result["code"] == "disallowed_private_target" + assert calls["fetch"] == 0 + assert "127.0.0.1" in result["target_url"] + assert "Traceback" not in str(result) + assert "DATABASE_URL" not in str(result) + + +@pytest.mark.asyncio +async def test_url_policy_rejects_dangerous_inputs(monkeypatch): + assert (await launch_agent._validate_public_url("file:///etc/passwd"))["code"] == "invalid_scheme" + assert (await launch_agent._validate_public_url("https://user:pass@example.com"))["code"] == "credentials_not_allowed" + assert (await launch_agent._validate_public_url("http://169.254.169.254/latest/meta-data"))["code"] == "disallowed_private_target" + + async def fake_getaddrinfo(host, port, type): + return [(None, None, None, None, ("10.0.0.5", port))] + + class FakeLoop: + def getaddrinfo(self, host, port, type): + return fake_getaddrinfo(host, port, type) + + monkeypatch.setattr(launch_agent.asyncio, "get_running_loop", lambda: FakeLoop()) + assert (await launch_agent._validate_public_url("https://rebinding.test"))["code"] == "disallowed_private_target"