174 lines
7.1 KiB
Python
174 lines
7.1 KiB
Python
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_public_addresses(host, port):
|
|
return ("93.184.216.34",)
|
|
|
|
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, "_resolve_public_addresses", fake_public_addresses)
|
|
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"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validated_dns_addresses_are_pinned_for_connection(monkeypatch):
|
|
calls = {"count": 0}
|
|
|
|
async def fake_getaddrinfo(host, port, type):
|
|
calls["count"] += 1
|
|
return [(None, None, None, None, ("93.184.216.34", 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())
|
|
validation = await launch_agent._validate_public_url("https://rebinding.test/path")
|
|
|
|
assert validation["ok"] is True
|
|
assert validation["addresses"] == ("93.184.216.34",)
|
|
resolver = launch_agent._PinnedResolver(
|
|
validation["hostname"], validation["addresses"]
|
|
)
|
|
resolved = await resolver.resolve(validation["hostname"], validation["port"])
|
|
assert [item["host"] for item in resolved] == ["93.184.216.34"]
|
|
assert calls["count"] == 1
|