tests/test_api.py

"""
tests/test_api.py — C3.1

Integration tests for REST endpoints using the Flask test client.
"""
import sys
import os
import json
import datetime

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(scope="module")
def flask_app(tmp_path_factory):
    tmp_dir = tmp_path_factory.mktemp("api_test_db")
    db_path = str(tmp_dir / "test.db")
    os.environ.pop("AGENTCACHE_SECRET", None)
    os.environ.pop("AGENTMEMORY_SECRET", None)

    from db import StateKV
    original_init = StateKV.__init__

    def patched_init(self, db_path_arg=None, **kwargs):
        original_init(self, db_path=db_path, **kwargs)

    StateKV.__init__ = patched_init
    import app as app_module
    os.environ.pop("AGENTCACHE_SECRET", None)
    os.environ.pop("AGENTMEMORY_SECRET", None)
    flask_application = app_module.create_app()
    StateKV.__init__ = original_init
    flask_application.config["TESTING"] = True
    return flask_application


@pytest.fixture(scope="module")
def client(flask_app):
    return flask_app.test_client()


def _now():
    return datetime.datetime.utcnow().isoformat() + "Z"


def _post(client, url, payload):
    return client.post(url, data=json.dumps(payload), content_type="application/json")


# ---------------------------------------------------------------------------
# POST /agentcache/agent/observe
# ---------------------------------------------------------------------------

class TestAgentObserve:
    def test_valid_payload_returns_201(self, client):
        resp = _post(client, "/agentcache/agent/observe", {
            "folderPath": "/home/user/test-project",
            "agentId": "kiro",
            "text": "Implemented new authentication middleware",
            "timestamp": _now(),
        })
        assert resp.status_code == 201
        data = resp.get_json()
        assert "observationId" in data
        assert data["observationId"].startswith("fobs_")

    def test_missing_folder_path_returns_400(self, client):
        resp = _post(client, "/agentcache/agent/observe", {
            "agentId": "kiro",
            "text": "Some work",
            "timestamp": _now(),
        })
        assert resp.status_code == 400

    def test_missing_agent_id_returns_400(self, client):
        resp = _post(client, "/agentcache/agent/observe", {
            "folderPath": "/home/user/proj",
            "text": "Some work",
            "timestamp": _now(),
        })
        assert resp.status_code == 400

    def test_missing_text_returns_400(self, client):
        resp = _post(client, "/agentcache/agent/observe", {
            "folderPath": "/home/user/proj",
            "agentId": "kiro",
            "timestamp": _now(),
        })
        assert resp.status_code == 400


# ---------------------------------------------------------------------------
# POST /agentcache/search
# ---------------------------------------------------------------------------

class TestSearch:
    def test_search_with_query_returns_200(self, client):
        # Seed data first
        _post(client, "/agentcache/agent/observe", {
            "folderPath": "/home/user/search-proj",
            "agentId": "kiro",
            "text": "Refactored the authentication system",
            "timestamp": _now(),
        })
        resp = _post(client, "/agentcache/search", {"query": "authentication"})
        assert resp.status_code == 200
        data = resp.get_json()
        assert isinstance(data, list) or isinstance(data, dict)

    def test_search_missing_query_returns_400(self, client):
        resp = _post(client, "/agentcache/search", {})
        assert resp.status_code == 400

    def test_search_empty_query_returns_400(self, client):
        resp = _post(client, "/agentcache/search", {"query": "   "})
        assert resp.status_code == 400


# ---------------------------------------------------------------------------
# GET /agentcache/folders
# ---------------------------------------------------------------------------

class TestFolders:
    def test_get_folders_returns_200(self, client):
        # Ensure at least one folder exists from earlier tests
        _post(client, "/agentcache/agent/observe", {
            "folderPath": "/home/user/folders-check",
            "agentId": "kiro",
            "text": "Check folders endpoint",
            "timestamp": _now(),
        })
        resp = client.get("/agentcache/folders")
        assert resp.status_code == 200
        data = resp.get_json()
        assert "folders" in data
        assert isinstance(data["folders"], list)


# ---------------------------------------------------------------------------
# GET /agentcache/health
# ---------------------------------------------------------------------------

class TestHealth:
    def test_health_returns_200(self, client):
        resp = client.get("/agentcache/health")
        assert resp.status_code == 200
        data = resp.get_json()
        assert "folderCount" in data
        assert "observationCount" in data
        assert "memoryCount" in data

    def test_health_status_ok(self, client):
        resp = client.get("/agentcache/health")
        data = resp.get_json()
        assert data.get("status") in ("ok", "degraded")


# ---------------------------------------------------------------------------
# GET /agentcache/livez
# ---------------------------------------------------------------------------

class TestLivez:
    def test_livez_returns_200_no_auth(self, client):
        resp = client.get("/agentcache/livez")
        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "ok"

    def test_livez_open_with_secret_set(self, client):
        os.environ["AGENTCACHE_SECRET"] = "test-secret-123"
        try:
            resp = client.get("/agentcache/livez")
            assert resp.status_code == 200
        finally:
            del os.environ["AGENTCACHE_SECRET"]


# ---------------------------------------------------------------------------
# Authentication
# ---------------------------------------------------------------------------

class TestAuthentication:
    def test_protected_endpoint_returns_401_with_wrong_token(self, client):
        os.environ["AGENTCACHE_SECRET"] = "correct-secret"
        try:
            resp = client.get(
                "/agentcache/audit",
                headers={"Authorization": "Bearer wrong-token"},
            )
            assert resp.status_code == 401
        finally:
            del os.environ["AGENTCACHE_SECRET"]

    def test_protected_endpoint_passes_with_correct_token(self, client):
        secret = "my-test-secret-xyz"
        os.environ["AGENTCACHE_SECRET"] = secret
        try:
            resp = client.get(
                "/agentcache/audit",
                headers={"Authorization": f"Bearer {secret}"},
            )
            assert resp.status_code == 200
        finally:
            del os.environ["AGENTCACHE_SECRET"]

    def test_livez_always_open_regardless_of_secret(self, client):
        os.environ["AGENTCACHE_SECRET"] = "any-secret"
        try:
            resp = client.get("/agentcache/livez")
            assert resp.status_code == 200
        finally:
            del os.environ["AGENTCACHE_SECRET"]


# ---------------------------------------------------------------------------
# Additional endpoint smoke tests
# ---------------------------------------------------------------------------

class TestMemoriesEndpoint:
    def test_memories_list_returns_200(self, client):
        resp = client.get("/agentcache/memories")
        assert resp.status_code == 200
        data = resp.get_json()
        assert "memories" in data

    def test_remember_valid_payload_returns_201(self, client):
        resp = _post(client, "/agentcache/remember", {
            "content": "API test memory content",
            "type": "fact",
        })
        assert resp.status_code == 201

    def test_forget_nonexistent_id(self, client):
        resp = _post(client, "/agentcache/forget", {"memoryId": "mem_nonexistent"})
        assert resp.status_code == 200

    def test_graph_endpoint_returns_200(self, client):
        resp = client.get("/agentcache/graph")
        assert resp.status_code == 200
        data = resp.get_json()
        assert "nodes" in data
        assert "edges" in data

    def test_mcp_tools_list_returns_200(self, client):
        resp = client.get("/agentcache/mcp/tools")
        assert resp.status_code == 200
        data = resp.get_json()
        assert "tools" in data
        tool_names = {t["name"] for t in data["tools"]}
        assert "agent_observe" in tool_names
        assert "cache_recall" in tool_names