tests/test_entrypoint_policy_contracts.py

"""Contratti di integrazione per policy tool nei diversi entry point backend.

Gli endpoint sono invocati direttamente con finti loop/provider. Ogni test misura la
policy effettivamente passata al loop o dimostra che un contratto letterale termina
prima di provider, notifiche, planner e tool.
"""
from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import os
import time
import sys
import types
import unittest
from unittest.mock import patch

from fastapi import HTTPException

_BACKEND = os.path.join(os.path.dirname(__file__), "..")
if _BACKEND not in sys.path:
    sys.path.insert(0, _BACKEND)


NO_TOOL_PROMPT = "Spiega il concetto senza usare strumenti, tool, rete o file."
LITERAL_PROMPT = (
    "Rispondi ESCLUSIVAMENTE con TEST_E2E_OK. Non usare strumenti, tool, "
    "comandi shell, file, rete, servizi esterni, azioni o card."
)


class _Request:
    def __init__(self, headers: dict[str, str] | None = None) -> None:
        self.headers = headers or {}


class _WebhookRequest(_Request):
    def __init__(self, raw_body: bytes, headers: dict[str, str]) -> None:
        super().__init__(headers)
        self._raw_body = raw_body

    async def body(self) -> bytes:
        return self._raw_body


class _CallbackRequest(_Request):
    async def json(self) -> dict[str, object]:
        raise AssertionError("bad secret must not parse payload")


class _RecordingLoop:
    calls: list[dict] = []

    def __init__(self, **_kwargs: object) -> None:
        pass

    async def run(self, **kwargs: object) -> dict[str, object]:
        self.__class__.calls.append(dict(kwargs))
        return {"success": True, "output": "safe textual result", "steps": []}


class _FakeAIClient:
    pass


def _fake_loop_modules() -> dict[str, types.ModuleType]:
    agents = types.ModuleType("agents")
    agents.__path__ = []  # type: ignore[attr-defined]
    unified = types.ModuleType("agents.unified_loop")
    unified.UnifiedAgentLoop = _RecordingLoop  # type: ignore[attr-defined]
    models = types.ModuleType("models")
    models.__path__ = []  # type: ignore[attr-defined]
    ai_client = types.ModuleType("models.ai_client")
    ai_client.AIClient = _FakeAIClient  # type: ignore[attr-defined]
    return {
        "agents": agents,
        "agents.unified_loop": unified,
        "models": models,
        "models.ai_client": ai_client,
    }


async def _noop(*_args: object, **_kwargs: object) -> None:
    return None


def _signed_webhook_request(
    payload: dict[str, object],
    *,
    event_id: str = "evt-000000000001",
    timestamp: int | None = None,
    signature: str | None = None,
) -> _WebhookRequest:
    raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    ts = str(int(time.time()) if timestamp is None else timestamp)
    signing = b"v1." + ts.encode("ascii") + b"." + event_id.encode("ascii") + b"." + raw
    expected = hmac.new(b"h" * 32, signing, hashlib.sha256).hexdigest()
    return _WebhookRequest(raw, {
        "x-webhook-id": event_id,
        "x-webhook-timestamp": ts,
        "x-webhook-signature": signature or f"sha256={expected}",
    })


class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        from api.webhook_security import WebhookDeliveryStore
        _RecordingLoop.calls.clear()
        await WebhookDeliveryStore.reset_memory_for_test()

    async def test_telegram_callback_fails_closed_before_payload_or_outbound_work(self) -> None:
        from api import webhook

        with patch.dict(os.environ, {"TELEGRAM_WEBHOOK_SECRET": ""}, clear=False):
            result = await webhook.telegram_callback(_CallbackRequest())

        self.assertEqual(result, {"ok": True, "ignored": "bad_secret"})

    async def test_webhook_literal_contract_returns_before_loop_or_notification(self) -> None:
        from api import webhook
        import api.state as state

        request = _signed_webhook_request({"goal": LITERAL_PROMPT})
        env = {
            "WEBHOOK_TOKEN": "test-webhook-token",
            "WEBHOOK_HMAC_SECRET": "h" * 32,
            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
        }
        with patch.dict(os.environ, env, clear=False), \
             patch.object(state, "_sb", None), \
             patch.object(webhook, "_tg_start", side_effect=AssertionError("notification forbidden")), \
             patch.object(webhook, "_get_mem_manager", side_effect=AssertionError("provider forbidden")):
            result = await webhook.inbound_webhook("test-webhook-token", request)

        self.assertEqual(result["output"], "TEST_E2E_OK")
        self.assertEqual(result["steps"], 0)
        self.assertEqual(_RecordingLoop.calls, [])

    async def test_public_chat_literal_contract_returns_before_loop_or_notification(self) -> None:
        from api import webhook

        payload = webhook.PublicChatPayload(message=LITERAL_PROMPT, conversation_id="conv-safe")
        with patch.dict(os.environ, {"PUBLIC_API_TOKEN": "public-test-token"}, clear=False), \
             patch.object(webhook, "_tg_start", side_effect=AssertionError("notification forbidden")), \
             patch.object(webhook, "_get_mem_manager", side_effect=AssertionError("provider forbidden")):
            result = await webhook.public_chat(
                payload,
                _Request({"authorization": "Bearer public-test-token"}),
            )

        self.assertEqual(result["response"], "TEST_E2E_OK")
        self.assertEqual(result["conversation_id"], "conv-safe")
        self.assertEqual(_RecordingLoop.calls, [])

    async def test_webhook_and_public_chat_propagate_no_tool_policy_to_loop(self) -> None:
        from api import webhook
        import api.state as state

        fake_modules = _fake_loop_modules()
        env = {
            "WEBHOOK_TOKEN": "test-webhook-token",
            "PUBLIC_API_TOKEN": "public-test-token",
            "WEBHOOK_HMAC_SECRET": "h" * 32,
            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
        }
        with patch.dict(sys.modules, fake_modules), \
             patch.dict(os.environ, env, clear=False), \
             patch.object(state, "_sb", None), \
             patch.object(webhook, "_get_mem_manager", return_value=object()), \
             patch.object(webhook, "_get_executor", return_value=object()), \
             patch.object(webhook, "_get_planner", return_value=object()), \
             patch.object(webhook, "_tg_start", _noop), \
             patch.object(webhook, "_tg_done", _noop):
            webhook_result = await webhook.inbound_webhook(
                "test-webhook-token", _signed_webhook_request({"goal": NO_TOOL_PROMPT})
            )
            public_result = await webhook.public_chat(
                webhook.PublicChatPayload(message=NO_TOOL_PROMPT),
                _Request({"authorization": "Bearer public-test-token"}),
            )

        self.assertTrue(webhook_result["ok"])
        self.assertTrue(public_result["ok"])
        self.assertEqual(len(_RecordingLoop.calls), 2)
        self.assertTrue(all(call["allow_tools"] is False for call in _RecordingLoop.calls))

    async def test_webhook_replay_conflict_and_expired_timestamp_are_rejected_before_dispatch(self) -> None:
        from api import webhook
        import api.state as state

        env = {
            "WEBHOOK_TOKEN": "test-webhook-token",
            "WEBHOOK_HMAC_SECRET": "h" * 32,
            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
            "WEBHOOK_MAX_AGE_SECONDS": "300",
        }
        first = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001")
        duplicate = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001")
        conflict = _signed_webhook_request({"goal": LITERAL_PROMPT, "context": [{"content": "different"}]}, event_id="evt-replay-000001")
        expired = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-expired-0001", timestamp=int(time.time()) - 301)
        with patch.dict(os.environ, env, clear=False), patch.object(state, "_sb", None):
            first_response = await webhook.inbound_webhook("test-webhook-token", first)
            duplicate_response = await webhook.inbound_webhook("test-webhook-token", duplicate)
            with self.assertRaises(HTTPException) as conflict_error:
                await webhook.inbound_webhook("test-webhook-token", conflict)
            with self.assertRaises(HTTPException) as expired_error:
                await webhook.inbound_webhook("test-webhook-token", expired)

        self.assertEqual(first_response, duplicate_response)
        self.assertEqual(conflict_error.exception.status_code, 409)
        self.assertEqual(expired_error.exception.status_code, 401)

    async def test_webhook_bad_signature_is_rejected_before_pydantic_parse(self) -> None:
        from api import webhook

        request = _signed_webhook_request({"goal": LITERAL_PROMPT}, signature="sha256=" + "0" * 64)
        env = {"WEBHOOK_TOKEN": "test-webhook-token", "WEBHOOK_HMAC_SECRET": "h" * 32}
        with patch.dict(os.environ, env, clear=False), \
             patch.object(webhook.WebhookPayload, "model_validate_json", side_effect=AssertionError("must not parse")):
            with self.assertRaises(HTTPException) as signature_error:
                await webhook.inbound_webhook("test-webhook-token", request)

        self.assertEqual(signature_error.exception.status_code, 401)

    async def test_scheduler_literal_contract_skips_provider_initialization(self) -> None:
        from api import scheduler

        with patch.object(scheduler, "_get_ai_client", side_effect=AssertionError("provider forbidden"), create=True):
            result = await scheduler._run_goal(LITERAL_PROMPT)

        self.assertEqual(result, "TEST_E2E_OK")
        self.assertEqual(_RecordingLoop.calls, [])

    async def test_scheduler_propagates_no_tool_policy_to_loop(self) -> None:
        from api import scheduler
        import api.state as state

        fake_modules = _fake_loop_modules()
        async def fake_memory() -> object:
            return object()

        with patch.dict(sys.modules, fake_modules), \
             patch.object(state, "_get_ai_client", return_value=object()), \
             patch.object(state, "_get_mem_manager_async", fake_memory), \
             patch.object(state, "_get_executor", return_value=object()), \
             patch.object(state, "_get_planner", return_value=object()):
            result = await scheduler._run_goal(NO_TOOL_PROMPT, risk="safe")

        self.assertEqual(result, "safe textual result")
        self.assertEqual(len(_RecordingLoop.calls), 1)
        self.assertIs(_RecordingLoop.calls[0]["allow_tools"], False)

    async def test_restored_sse_task_rebuilds_literal_policy_before_task_start(self) -> None:
        from api import agent

        task_id = "resume-literal-policy"
        original = dict(agent._agent_tasks)
        agent._agent_tasks.clear()
        agent._agent_tasks[task_id] = {"id": task_id, "goal": LITERAL_PROMPT, "status": "QUEUED"}
        try:
            with patch.object(agent, "sb_update_status", _noop):
                response = await agent.stream_agent_task(task_id, _Request())
                payload = b"".join([
                    chunk.encode() if isinstance(chunk, str) else chunk
                    async for chunk in response.body_iterator
                ]).decode()
        finally:
            agent._agent_tasks.clear()
            agent._agent_tasks.update(original)

        self.assertIn('"event": "task_done"', payload)
        self.assertIn("TEST_E2E_OK", payload)
        self.assertNotIn("task_start", payload)
        self.assertTrue(agent._agent_tasks == original or task_id not in agent._agent_tasks)


if __name__ == "__main__":
    unittest.main()