core-python/tests/test_memory_components.py

"""Tests for top-level memory infrastructure helpers."""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from memory.long_term.db_store import DbStore  # noqa: E402
from memory.vector_store.vector_db_client import VectorDbClient  # noqa: E402


def test_vector_db_client_persists_and_searches(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.setattr(
        "memory.vector_store.embeddings.embed_text",
        lambda text: [1.0, 0.0] if "api" in text.lower() else [0.9, 0.1],
    )
    client = VectorDbClient(storage_path=str(tmp_path / "vectors.json"))
    client.add("memory", "Python API klients ar retry", {"source": "test"})

    matches = client.search("memory", "API klients", top_k=3)

    assert matches
    assert matches[0]["metadata"]["source"] == "test"
    assert "API" in matches[0]["text"]


@pytest.mark.asyncio
async def test_db_store_rejects_unknown_table_name() -> None:
    store = DbStore(dsn="postgresql://example.invalid/test")

    result = await store.query("users; DROP TABLE users;", limit=10)

    assert result == []