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]
def test_full_stack_product_contract():
manifest = (ROOT / "a2a.yaml").read_text(encoding="utf-8")
source = (ROOT / "agent.py").read_text(encoding="utf-8")
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 "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)
audit_schema = next(skill.input_schema for skill in card.skills if skill.name == "audit_url")
assert set(audit_schema["properties"]) == {"audit_id", "url"}
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"