ATOCore/tests/test_memory.py

"""Tests for Memory Core."""

import os
import tempfile

import pytest

import atocore.config as _config
from atocore.models.database import init_db


@pytest.fixture(autouse=True)
def isolated_db():
    """Give each test a completely isolated database."""
    tmpdir = tempfile.mkdtemp()
    os.environ["ATOCORE_DATA_DIR"] = tmpdir

    # Replace the global settings so all modules see the new data_dir
    _config.settings = _config.Settings()

    # Also reset any module-level references to the old settings
    import atocore.models.database
    # database.py now uses _config.settings dynamically, so no patch needed

    init_db()
    yield tmpdir


def test_create_memory(isolated_db):
    from atocore.memory.service import create_memory
    mem = create_memory("identity", "User is a mechanical engineer specializing in optics")
    assert mem.memory_type == "identity"
    assert mem.status == "active"
    assert mem.confidence == 1.0


def test_create_memory_invalid_type(isolated_db):
    from atocore.memory.service import create_memory
    with pytest.raises(ValueError, match="Invalid memory type"):
        create_memory("invalid_type", "some content")


def test_create_memory_dedup(isolated_db):
    from atocore.memory.service import create_memory
    m1 = create_memory("identity", "User is an engineer")
    m2 = create_memory("identity", "User is an engineer")
    assert m1.id == m2.id


def test_create_memory_dedup_is_project_scoped(isolated_db):
    from atocore.memory.service import create_memory
    m1 = create_memory("project", "Uses SQLite for local state", project="atocore")
    m2 = create_memory("project", "Uses SQLite for local state", project="openclaw")
    assert m1.id != m2.id


def test_project_is_persisted_and_filterable(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("project", "Uses SQLite for local state", project="atocore")
    create_memory("project", "Uses Postgres in production", project="openclaw")

    atocore_memories = get_memories(memory_type="project", project="atocore")
    assert len(atocore_memories) == 1
    assert atocore_memories[0].project == "atocore"


def test_get_memories_all(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("identity", "User is an engineer")
    create_memory("preference", "Prefers Python with type hints")
    create_memory("knowledge", "Zerodur has near-zero thermal expansion")

    mems = get_memories()
    assert len(mems) == 3


def test_get_memories_by_type(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("identity", "User is an engineer")
    create_memory("preference", "Prefers concise code")
    create_memory("preference", "Uses FastAPI for APIs")

    mems = get_memories(memory_type="preference")
    assert len(mems) == 2


def test_get_memories_active_only(isolated_db):
    from atocore.memory.service import create_memory, get_memories, invalidate_memory
    m = create_memory("knowledge", "Fact about optics")
    invalidate_memory(m.id)

    assert len(get_memories(active_only=True)) == 0
    assert len(get_memories(active_only=False)) == 1


def test_get_memories_min_confidence(isolated_db):
    from atocore.memory.service import create_memory, get_memories
    create_memory("knowledge", "High confidence fact", confidence=0.9)
    create_memory("knowledge", "Low confidence fact", confidence=0.3)

    high = get_memories(min_confidence=0.5)
    assert len(high) == 1
    assert high[0].confidence == 0.9


def test_update_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, update_memory
    mem = create_memory("knowledge", "Initial fact")
    update_memory(mem.id, content="Updated fact", confidence=0.8)

    mems = get_memories()
    assert len(mems) == 1
    assert mems[0].content == "Updated fact"
    assert mems[0].confidence == 0.8


def test_update_memory_rejects_duplicate_active_memory(isolated_db):
    from atocore.memory.service import create_memory, update_memory
    import pytest

    first = create_memory("knowledge", "Canonical fact", project="atocore")
    second = create_memory("knowledge", "Different fact", project="atocore")

    with pytest.raises(ValueError, match="duplicate active memory"):
        update_memory(second.id, content="Canonical fact")


def test_create_memory_validates_confidence(isolated_db):
    from atocore.memory.service import create_memory
    import pytest

    with pytest.raises(ValueError, match="Confidence must be between 0.0 and 1.0"):
        create_memory("knowledge", "Out of range", confidence=1.5)


def test_invalidate_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, invalidate_memory
    mem = create_memory("knowledge", "Wrong fact")
    invalidate_memory(mem.id)
    assert len(get_memories(active_only=True)) == 0


def test_supersede_memory(isolated_db):
    from atocore.memory.service import create_memory, get_memories, supersede_memory
    mem = create_memory("knowledge", "Old fact")
    supersede_memory(mem.id)

    mems = get_memories(active_only=False)
    assert len(mems) == 1
    assert mems[0].status == "superseded"


def test_memories_for_context(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "User is a senior mechanical engineer")
    create_memory("preference", "Prefers Python with type hints")

    text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=500)
    assert "--- AtoCore Memory ---" in text
    assert "[identity]" in text
    assert "[preference]" in text
    assert chars > 0


def test_memories_for_context_reserves_room_for_each_type(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "Identity entry that is intentionally long so it could consume the whole budget on its own")
    create_memory("preference", "Preference entry that should still appear")

    text, _ = get_memories_for_context(memory_types=["identity", "preference"], budget=120)
    assert "[preference]" in text


def test_memories_for_context_respects_actual_serialized_budget(isolated_db):
    from atocore.memory.service import create_memory, get_memories_for_context
    create_memory("identity", "Identity text that should fit the wrapper-aware memory budget calculation")
    create_memory("preference", "Preference text that should also fit")

    text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=140)
    assert chars == len(text)
    assert chars <= 140


def test_memories_for_context_empty(isolated_db):
    from atocore.memory.service import get_memories_for_context
    text, chars = get_memories_for_context()
    assert text == ""
    assert chars == 0