fastapi-gsap/tests/test_functions.py

"""Tests for governed serverless function module."""
import pytest


@pytest.mark.asyncio
async def test_function_catalogue(client):
    resp = await client.get("/functions/")
    assert resp.status_code == 200
    body = resp.json()
    ids = [f["function_id"] for f in body["functions"]]
    assert "billing-processor" in ids
    assert "echo-function" in ids
    assert "trigger_index" in body


@pytest.mark.asyncio
async def test_function_descriptor(client):
    resp = await client.get("/functions/billing-processor/")
    assert resp.status_code == 200
    desc = resp.json()
    assert desc["function_id"] == "billing-processor"
    assert "trigger_events" in desc
    assert "GSAP_CR_RECEIVED" in desc["trigger_events"]
    assert desc["corpus_entry_cid"].startswith("sha256:")


@pytest.mark.asyncio
async def test_knative_manifest(client):
    resp = await client.get("/functions/billing-processor/manifest/?image=gcr.io/test/billing:v1")
    assert resp.status_code == 200
    manifest = resp.json()
    assert manifest["kind"] == "Service"
    assert manifest["apiVersion"] == "serving.knative.dev/v1"
    assert manifest["metadata"]["name"] == "billing-processor"
    containers = manifest["spec"]["template"]["spec"]["containers"]
    assert containers[0]["image"] == "gcr.io/test/billing:v1"


@pytest.mark.asyncio
async def test_invoke_echo(client):
    resp = await client.post(
        "/functions/echo-function/invoke/",
        json={"event": {"msg": "hello", "value": 42}},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["success"] is True
    assert body["data"]["msg"] == "hello"
    assert body["data"]["value"] == 42
    assert body["function_id"] == "echo-function"
    assert body["duration_ms"] >= 0


@pytest.mark.asyncio
async def test_billing_completed(client):
    resp = await client.post(
        "/functions/billing-processor/invoke/",
        json={
            "event": {
                "outcome": "completed",
                "event_type": "gsap_cr_completed",
                "quantity": 2,
                "chronicle_cid": "sha256:abc123",
            },
        },
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["success"] is True
    assert body["data"]["billed"] is True
    assert body["data"]["amount"] == 0.01  # 0.005 * 2
    assert body["data"]["chronicle_cid"] == "sha256:abc123"


@pytest.mark.asyncio
async def test_billing_failed_not_billed(client):
    resp = await client.post(
        "/functions/billing-processor/invoke/",
        json={
            "event": {
                "outcome": "failed",
                "event_type": "gsap_cr_completed",
            },
        },
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["success"] is True
    assert body["data"]["billed"] is False


@pytest.mark.asyncio
async def test_dispatch_routes_to_billing(client):
    resp = await client.post(
        "/functions/dispatch/",
        json={
            "event_kind": "GSAP_CR_RECEIVED",
            "event_data": {
                "outcome": "completed",
                "event_type": "gsap_cr_completed",
                "quantity": 1,
            },
            "trigger_cid": "sha256:trigger001",
        },
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["event_kind"] == "GSAP_CR_RECEIVED"
    assert body["dispatched_count"] >= 1
    # billing-processor should be in the results
    billing_results = [r for r in body["results"] if r["function_id"] == "billing-processor"]
    assert len(billing_results) == 1
    assert billing_results[0]["success"] is True


@pytest.mark.asyncio
async def test_gsap_required_without_ac(client):
    """A gsap_required function invoked without gsap_context_id returns error."""
    # We need a function with gsap_required=True. Register one via the runtime.
    # Use the billing-processor with gsap_required toggled, or invoke directly.
    # Since billing-processor has gsap_required=False, we test via the runtime directly.
    from gsap_broker.functions.base import FunctionContext, FunctionPlugin, FunctionResult
    from gsap_broker.functions.registry import FunctionRegistry
    from gsap_broker.functions.runtime import FunctionRuntime
    from typing import Any

    class StrictFunction(FunctionPlugin):
        function_id = "strict-func"
        corpus_entry_cid = "sha256:" + "s" * 64
        capability_mask = 1
        trigger_events = []
        gsap_required = True
        chronicle_enabled = False
        max_duration_seconds = 5
        display_name = "Strict"
        description = "Requires GSAP"
        version = "0.1.0"

        async def handle(self, event: dict[str, Any], context: FunctionContext) -> FunctionResult:
            return FunctionResult(success=True, data=event)

        def health_check(self) -> bool:
            return True

    reg = FunctionRegistry()
    reg.register(StrictFunction())
    rt = FunctionRuntime(registry=reg)

    ctx = FunctionContext()  # No gsap_context_id
    result = await rt.invoke("strict-func", {"test": True}, ctx)
    assert result.success is False
    assert "GSAP context required" in (result.error or "")