a2a-source-edit: write tests/test_full_stack_contract.py

2026-07-18 09:22:08 +00:00
parent f395243e21
commit 3b6f66c556
1 changed files with 119 additions and 1 deletions
--- 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"""
+              <html><head>
+                <title>Example Domain</title>
+                <meta name='description' content='Example launch page'>
+                <meta name='viewport' content='width=device-width, initial-scale=1'>
+                <link rel='canonical' href='https://example.com/'>
+              </head><body>Example</body></html>
+            """,
+            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"