memory-gateway/tests/test_memory_gateway_skill.py

from __future__ import annotations

import importlib.util
import json
from pathlib import Path
from urllib.error import HTTPError

import pytest


SCRIPT_PATH = (
    Path(__file__).parents[1]
    / "skill"
    / "memory-gateway-agent"
    / "scripts"
    / "memory_gateway.py"
)


def load_cli():
    spec = importlib.util.spec_from_file_location("memory_gateway_skill_cli", SCRIPT_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class FakeResponse:
    def __init__(self, body: dict[str, object], status: int = 200) -> None:
        self.body = json.dumps(body).encode()
        self.status = status

    def __enter__(self):
        return self

    def __exit__(self, *args: object) -> None:
        return None

    def read(self) -> bytes:
        return self.body

    def close(self) -> None:
        return None


def test_settings_read_gateway_credentials_from_environment(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    cli = load_cli()
    monkeypatch.setenv("MEMORY_GATEWAY_BASE_URL", "http://gateway.test/")
    monkeypatch.setenv("MEMORY_GATEWAY_USER_ID", "u_agent")
    monkeypatch.setenv("MEMORY_GATEWAY_USER_KEY", "uk_secret")

    settings = cli.Settings.from_env()

    assert settings.base_url == "http://gateway.test"
    assert settings.user_id == "u_agent"
    assert settings.user_key == "uk_secret"


def test_json_request_sends_authenticated_payload(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    cli = load_cli()
    captured: dict[str, object] = {}

    def fake_urlopen(request, timeout):
        captured["url"] = request.full_url
        captured["method"] = request.method
        captured["body"] = json.loads(request.data)
        captured["content_type"] = request.headers["Content-type"]
        captured["timeout"] = timeout
        return FakeResponse({"results": []})

    monkeypatch.setattr(cli, "urlopen", fake_urlopen)
    client = cli.MemoryGatewayClient(
        "http://gateway.test",
        user_id="u_agent",
        user_key="uk_secret",
        timeout=9,
    )

    result = client.search("contract", scopes=["resources"], top_k=5)

    assert result == {"results": []}
    assert captured == {
        "url": "http://gateway.test/memories/search",
        "method": "POST",
        "body": {
            "user_id": "u_agent",
            "user_key": "uk_secret",
            "query": "contract",
            "scope": ["resources"],
            "top_k": 5,
            "app_id": "default",
            "project_id": "default",
        },
        "content_type": "application/json",
        "timeout": 9,
    }


def test_upload_builds_multipart_request_without_exposing_file_uri(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    cli = load_cli()
    upload = tmp_path / "note.txt"
    upload.write_text("remember this", encoding="utf-8")
    captured: dict[str, object] = {}

    def fake_urlopen(request, timeout):
        captured["url"] = request.full_url
        captured["method"] = request.method
        captured["body"] = request.data
        captured["content_type"] = request.headers["Content-type"]
        return FakeResponse(
            {
                "resource_id": "r_1",
                "uri": "resource://u_agent/r_1",
                "status": "extracted",
            }
        )

    monkeypatch.setattr(cli, "urlopen", fake_urlopen)
    client = cli.MemoryGatewayClient(
        "http://gateway.test",
        user_id="u_agent",
        user_key="uk_secret",
    )

    result = client.upload_resource(upload, title="Agent note")

    body = captured["body"]
    assert isinstance(body, bytes)
    assert captured["url"] == "http://gateway.test/resources"
    assert captured["method"] == "POST"
    assert str(captured["content_type"]).startswith("multipart/form-data; boundary=")
    assert b'name="user_id"' in body
    assert b"u_agent" in body
    assert b'name="file"; filename="note.txt"' in body
    assert b"remember this" in body
    assert b"file://" not in body
    assert result["uri"] == "resource://u_agent/r_1"


def test_http_error_raises_gateway_error_without_leaking_user_key(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    cli = load_cli()

    def fake_urlopen(request, timeout):
        raise HTTPError(
            request.full_url,
            401,
            "Unauthorized",
            hdrs=None,
            fp=FakeResponse({"detail": "invalid user credentials"}),
        )

    monkeypatch.setattr(cli, "urlopen", fake_urlopen)
    client = cli.MemoryGatewayClient(
        "http://gateway.test",
        user_id="u_agent",
        user_key="uk_super_secret",
    )

    with pytest.raises(cli.GatewayError) as exc_info:
        client.list_resources()

    message = str(exc_info.value)
    assert "401" in message
    assert "invalid user credentials" in message
    assert "uk_super_secret" not in message


def test_load_messages_accepts_large_inline_json() -> None:
    cli = load_cli()
    value = json.dumps(
        [
            {
                "sender_id": "u_agent",
                "role": "user",
                "timestamp": 1234567890123,
                "content": "x" * 5000,
            }
        ]
    )

    messages = cli._load_json_array(value)

    assert messages[0]["content"] == "x" * 5000


def test_search_requires_conversation_id_for_current_chat_scope() -> None:
    cli = load_cli()
    client = cli.MemoryGatewayClient(
        "http://gateway.test",
        user_id="u_agent",
        user_key="uk_secret",
    )

    with pytest.raises(cli.GatewayError, match="conversation_id"):
        client.search("what did we discuss", scopes=["current_chat"])