simple-agent/tests/test_a2a_http_integration.py

from __future__ import annotations

from dataclasses import replace
from types import SimpleNamespace

from fastapi import FastAPI
from fastapi import HTTPException
from fastapi.testclient import TestClient

import app.a2a.router as a2a_module
from app.config import get_settings
from app.security.auth import AuthBackend


class _DummyCoreService:
    def check_availability(
        self,
        start: str,
        end: str,
        calendar_ids: list[str] | None,
    ) -> SimpleNamespace:
        checked = calendar_ids or ["primary"]
        return SimpleNamespace(
            start=start,
            end=end,
            available=True,
            busy_slots=[],
            checked_calendars=checked,
        )

    def available_meeting_intervals(
        self,
        start: str,
        end: str,
        calendar_ids: list[str] | None,
    ) -> SimpleNamespace:
        checked = calendar_ids or ["primary"]
        return SimpleNamespace(
            start=start,
            end=end,
            timezone="Europe/Paris",
            meeting_intervals=[
                SimpleNamespace(
                    start="2026-03-10T08:30:00+01:00",
                    end="2026-03-10T09:30:00+01:00",
                )
            ],
            checked_calendars=checked,
        )


class _ScopeAwareAuthBackend:
    def authenticate(
        self,
        *,
        x_api_key: str | None,
        authorization: str | None,
        required_scopes: set[str],
    ) -> None:
        if required_scopes == {"available_meeting_intervals:read"}:
            return None
        raise HTTPException(status_code=403, detail="insufficient_scope")


def _build_test_app() -> FastAPI:
    app = FastAPI()
    app.include_router(a2a_module.router)
    return app


def test_a2a_agent_card_endpoint(monkeypatch) -> None:
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.get("/.well-known/agent-card.json")

    assert response.status_code == 200
    assert response.headers["A2A-Version"] == "1.0"
    payload = response.json()
    assert payload["protocolVersion"] == "1.0"
    assert payload["url"].endswith("/a2a/rpc")


def test_a2a_send_message_requires_auth(monkeypatch) -> None:
    auth_settings = replace(
        get_settings(),
        auth_mode="api_key",
        agent_api_key="integration-key",
        auth_jwt_secret="",
    )
    monkeypatch.setattr(a2a_module, "auth_backend", AuthBackend(auth_settings))
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.post(
            "/a2a/rpc",
            json={
                "jsonrpc": "2.0",
                "id": "r1",
                "method": "SendMessage",
                "params": {
                    "start": "2026-03-10T09:00:00+01:00",
                    "end": "2026-03-10T10:00:00+01:00",
                },
            },
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["error"]["code"] == -32001


def test_a2a_send_message_with_api_key(monkeypatch) -> None:
    auth_settings = replace(
        get_settings(),
        auth_mode="api_key",
        agent_api_key="integration-key",
        auth_jwt_secret="",
    )
    monkeypatch.setattr(a2a_module, "auth_backend", AuthBackend(auth_settings))
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.post(
            "/a2a/rpc",
            headers={"X-API-Key": "integration-key"},
            json={
                "jsonrpc": "2.0",
                "id": "r2",
                "method": "SendMessage",
                "params": {
                    "start": "2026-03-10T09:00:00+01:00",
                    "end": "2026-03-10T10:00:00+01:00",
                    "calendar_ids": ["primary"],
                },
            },
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["error"] is None
    assert payload["result"]["availability"]["available"] is True
    assert payload["result"]["availability"]["checked_calendars"] == ["primary"]


def test_a2a_send_message_available_meeting_intervals_with_api_key(monkeypatch) -> None:
    auth_settings = replace(
        get_settings(),
        auth_mode="api_key",
        agent_api_key="integration-key",
        auth_jwt_secret="",
    )
    monkeypatch.setattr(a2a_module, "auth_backend", AuthBackend(auth_settings))
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.post(
            "/a2a/rpc",
            headers={"X-API-Key": "integration-key"},
            json={
                "jsonrpc": "2.0",
                "id": "r3",
                "method": "SendMessage",
                "params": {
                    "action": "available_meeting_intervals",
                    "start": "2026-03-10T08:00:00+01:00",
                    "end": "2026-03-10T10:00:00+01:00",
                    "calendar_ids": ["primary"],
                },
            },
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["error"] is None
    assert payload["result"]["type"] == "available_meeting_intervals.result"
    assert payload["result"]["meeting_intervals"]["timezone"] == "Europe/Paris"
    assert payload["result"]["meeting_intervals"]["meeting_intervals"] == [
        {
            "start": "2026-03-10T08:30:00+01:00",
            "end": "2026-03-10T09:30:00+01:00",
        }
    ]


def test_a2a_send_message_meeting_intervals_scope_enforced(monkeypatch) -> None:
    monkeypatch.setattr(a2a_module, "auth_backend", _ScopeAwareAuthBackend())
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.post(
            "/a2a/rpc",
            json={
                "jsonrpc": "2.0",
                "id": "r4",
                "method": "SendMessage",
                "params": {
                    "action": "available_meeting_intervals",
                    "start": "2026-03-10T08:00:00+01:00",
                    "end": "2026-03-10T10:00:00+01:00",
                },
            },
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["error"] is None
    assert payload["result"]["type"] == "available_meeting_intervals.result"


def test_a2a_send_message_defaults_to_availability_scope(monkeypatch) -> None:
    monkeypatch.setattr(a2a_module, "auth_backend", _ScopeAwareAuthBackend())
    monkeypatch.setattr(a2a_module, "core_service", _DummyCoreService())
    app = _build_test_app()

    with TestClient(app) as client:
        response = client.post(
            "/a2a/rpc",
            json={
                "jsonrpc": "2.0",
                "id": "r5",
                "method": "SendMessage",
                "params": {
                    "start": "2026-03-10T08:00:00+01:00",
                    "end": "2026-03-10T10:00:00+01:00",
                },
            },
        )

    assert response.status_code == 200
    payload = response.json()
    assert payload["error"]["code"] == -32001