csv-answers-studio-v1/tests/test_full_stack_contract.py

import asyncio
import base64
from pathlib import Path

from a2a_pack import LocalRunContext, PlatformUserAuth
from a2a_pack.cli.local import load_local_project
from a2a_pack.mcp.server import MCPServer

import agent as agent_module
from agent import CsvAnswersStudioV1, _LOCAL_RECEIPTS, _reset_local_state_for_tests

ROOT = Path(__file__).resolve().parents[1]
AUTH = PlatformUserAuth(sub="test-user", user_id=42, email="tester@example.com", scopes=["agent:invoke"])
SUCCESS_CSV = "month,revenue\nJanuary,100\nFebruary,120\nMarch,150\n"
BAD_CSV = "month,revenue\nJanuary,\"100\n"


def run(coro):
    return asyncio.run(coro)


def make_ctx(task_id="test-task"):
    return LocalRunContext(auth=AUTH, task_id=task_id, caller="pytest")


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")
    browser_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 "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 and "ctx.llm" not in source
    assert "Skill runner" not in frontend
    assert "CSVAnswers Studio" in frontend
    assert "data_base64" in browser_client
    forbidden = ["A2A_LITELLM_KEY", "OPENAI_API_KEY", "DATABASE_URL", ".svc.cluster.local"]
    assert not any(secret in frontend + browser_client for secret in forbidden)

    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 {"analyze_csv", "get_analysis", "analyze_csv_upload_base64", "analyze_csv_file", "list_analyses"} <= skill_names
    analyze_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_csv")
    assert set(analyze_schema["required"]) == {"analysis_id", "csv_text", "question"}
    assert analyze_schema["properties"]["analysis_id"]["maxLength"] == 128
    assert analyze_schema["properties"]["csv_text"]["maxLength"] == agent_module.MAX_CSV_BYTES
    assert analyze_schema["properties"]["question"]["maxLength"] == 500
    upload_schema = next(skill.input_schema for skill in card.skills if skill.name == "analyze_csv_file")
    assert upload_schema["properties"]["file"]["x-a2a-file-upload"]["max_bytes"] == agent_module.MAX_CSV_BYTES
    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 test_success_reload_and_receipt_persist_locally(monkeypatch):
    monkeypatch.setattr(agent_module, "_connect", lambda: None)
    _reset_local_state_for_tests()
    app = CsvAnswersStudioV1()

    result = run(app.invoke("analyze_csv", make_ctx(), analysis_id="studio-csv-v1", csv_text=SUCCESS_CSV, question="Which month has the highest revenue?"))
    assert result.ok is True
    assert result.analysis_id == "studio-csv-v1"
    assert result.answer == "March has the highest revenue at 150."
    assert result.row_count == 3
    assert result.saved is True
    assert result.receipt_id
    assert _LOCAL_RECEIPTS and _LOCAL_RECEIPTS[-1]["analysis_id"] == "studio-csv-v1"

    reopened = run(app.invoke("get_analysis", make_ctx("reload"), analysis_id="studio-csv-v1"))
    assert reopened.ok is True
    assert reopened.analysis_id == "studio-csv-v1"
    assert reopened.answer == "March has the highest revenue at 150."
    assert reopened.row_count == 3


def test_malformed_csv_rejected_deterministically(monkeypatch):
    monkeypatch.delenv("DATABASE_URL", raising=False)
    _reset_local_state_for_tests()
    app = CsvAnswersStudioV1()

    result = run(app.invoke("analyze_csv", make_ctx(), analysis_id="studio-csv-invalid", csv_text=BAD_CSV, question="Which month has the highest revenue?"))
    assert result.ok is False
    assert result.code == "malformed_csv"
    assert not _LOCAL_RECEIPTS


def test_browser_base64_bridge(monkeypatch):
    monkeypatch.setattr(agent_module, "_connect", lambda: None)
    _reset_local_state_for_tests()
    app = CsvAnswersStudioV1()
    upload = {
        "filename": "revenue.csv",
        "media_type": "text/csv",
        "data_base64": base64.b64encode(SUCCESS_CSV.encode()).decode(),
    }

    result = run(app.invoke("analyze_csv_upload_base64", make_ctx(), analysis_id="studio-csv-v1", upload=upload, question="Which month has the highest revenue?"))
    assert result.ok is True
    assert result.answer == "March has the highest revenue at 150."


def test_mcp_tools_list_and_call(monkeypatch):
    monkeypatch.setattr(agent_module, "_connect", lambda: None)
    _reset_local_state_for_tests()
    app = CsvAnswersStudioV1()
    server = MCPServer(app, context_builder=lambda skill_name: make_ctx(f"mcp-{skill_name}"))

    tools = run(server.handle({"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}))
    names = {tool["name"] for tool in tools["result"]["tools"]}
    assert "analyze_csv" in names and "get_analysis" in names

    called = run(server.handle({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": {
            "name": "analyze_csv",
            "arguments": {
                "analysis_id": "studio-csv-v1",
                "csv_text": SUCCESS_CSV,
                "question": "Which month has the highest revenue?",
            },
        },
    }))
    structured = called["result"]["structuredContent"]
    assert structured["ok"] is True
    assert structured["answer"] == "March has the highest revenue at 150."

    reopened = run(server.handle({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tools/call",
        "params": {"name": "get_analysis", "arguments": {"analysis_id": "studio-csv-v1"}},
    }))
    assert reopened["result"]["structuredContent"]["ok"] is True


def test_database_unavailable_fails_closed(monkeypatch):
    def unavailable():
        raise RuntimeError("database unavailable")

    monkeypatch.setattr(agent_module, "_connect", unavailable)
    app = CsvAnswersStudioV1()

    result = run(
        app.invoke(
            "analyze_csv",
            make_ctx(),
            analysis_id="studio-csv-db-failure",
            csv_text=SUCCESS_CSV,
            question="Which month has the highest revenue?",
        )
    )

    assert result.ok is False
    assert result.code == "database_error"
    assert result.saved is False
    assert "RuntimeError" not in (result.message or "")