98 lines
4.5 KiB
Python
98 lines
4.5 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
|
|
from a2a_pack import LocalRunContext, PlatformUserAuth
|
|
from a2a_pack.cli.local import load_local_project
|
|
|
|
from agent import (
|
|
BuildDashboardRequest,
|
|
ProductionProofV116HighUtili72553,
|
|
_analyze_updates,
|
|
_default_updates,
|
|
)
|
|
|
|
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")
|
|
frontend_client = (ROOT / "frontend" / "src" / "a2a.js").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 "runtime:" in manifest and "max_runtime_seconds: 600" in manifest
|
|
assert "PlatformUserAuth" in source
|
|
assert "AccountAccess(required=True, platform_skill_calls=3, after_trial=\"byok\")" in source
|
|
assert "AgentPlatformResources" in source and "AgentDatabase(" in source
|
|
assert "llm_provisioning" not in source
|
|
assert "Skill runner" not in frontend, "replace the generic scaffold with the product workflow"
|
|
assert "data-testid=\"agent-app\"" in frontend
|
|
assert "data-testid=\"agent-submit\"" in frontend
|
|
assert "data-testid=\"agent-result\"" in frontend
|
|
assert "data-testid=\"agent-download\"" in frontend
|
|
assert "unwrapInvokeResponse" in frontend_client, "unwrap the /invoke result envelope before rendering"
|
|
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()
|
|
databases = card.runtime.platform_resources.databases
|
|
assert databases, "live Agent Card must declare its managed database"
|
|
assert databases[0].migrations is not None
|
|
assert databases[0].migrations.path == "db/migrations"
|
|
assert card.runtime.account_access.required is True
|
|
assert card.runtime.account_access.platform_skill_calls == 3
|
|
assert card.runtime.pricing.caller_pays_llm is False
|
|
|
|
|
|
def test_analysis_returns_dashboard_table_filters_trends_exceptions():
|
|
request = BuildDashboardRequest(updates=_default_updates())
|
|
analysis = _analyze_updates(request)
|
|
assert len(analysis["rows"]) >= 1
|
|
assert analysis["filters"]["owners"] == ["Mateo", "Nora", "Priya"]
|
|
assert any(item["metric"] == "pipeline_total" and item["value"] == 288000 for item in analysis["trends"])
|
|
assert [item["account"] for item in analysis["exceptions"]][:2] == ["Summit Retail", "Acme Manufacturing"]
|
|
|
|
|
|
def test_build_dashboard_without_database_returns_preview_artifact_and_setup_required(monkeypatch):
|
|
monkeypatch.delenv("DATABASE_URL", raising=False)
|
|
agent = ProductionProofV116HighUtili72553()
|
|
ctx = LocalRunContext(
|
|
auth=PlatformUserAuth(sub="user-123", user_id=123, email="dev@example.local"),
|
|
task_id="test-task",
|
|
)
|
|
|
|
result = asyncio.run(agent.invoke("build_dashboard", ctx, request=BuildDashboardRequest()))
|
|
|
|
assert result.status == "setup_required"
|
|
assert result.dashboard_data
|
|
assert result.artifact is not None
|
|
assert result.artifact.name == "sales-ops-dashboard.csv"
|
|
assert result.artifact.media_type == "text/csv"
|
|
assert "account,owner,stage" in result.artifact.content
|
|
assert "sales-ops-dashboard.csv" in ctx.artifacts
|
|
assert result.receipt_id.startswith("rcpt_")
|
|
|
|
|
|
def test_history_without_database_is_structured_setup_required(monkeypatch):
|
|
monkeypatch.delenv("DATABASE_URL", raising=False)
|
|
agent = ProductionProofV116HighUtili72553()
|
|
ctx = LocalRunContext(auth=PlatformUserAuth(sub="user-123", user_id=123))
|
|
result = asyncio.run(agent.invoke("list_dashboard_history", ctx, limit=1000))
|
|
assert result.status == "setup_required"
|
|
assert result.history == []
|
|
|
|
|
|
def test_public_tool_schemas_are_bounded_and_mcp_named():
|
|
card = ProductionProofV116HighUtili72553().card()
|
|
skill_names = {skill.name for skill in card.skills}
|
|
assert {"build_dashboard", "list_dashboard_history", "run_scheduled_snapshot"}.issubset(skill_names)
|
|
build_skill = next(skill for skill in card.skills if skill.name == "build_dashboard")
|
|
schema_text = str(build_skill.input_schema)
|
|
assert "maxItems" in schema_text or "max_length" in schema_text
|
|
assert "request" in build_skill.input_schema.get("properties", {})
|