basicmachines-co-basic-memory/tests/api/v2/test_schema_router.py

"""Tests for v2 schema router endpoints.

Tests the integration layer where ORM entities are converted to NoteData
and passed through the schema engine (infer, validate, diff).

Note: EntityType uses BeforeValidator(to_snake_case) so "Person" becomes "person"
in the database. All query params must use the stored (snake_case) form.
"""

from textwrap import dedent

import pytest
from httpx import AsyncClient

from basic_memory.models import Project
from basic_memory.schemas.base import Entity as EntitySchema


# --- Helpers ---


async def create_person_entities(entity_service, search_service):
    """Create multiple person entities with observations and relations for testing.

    Returns the created entities after indexing them for search.
    """
    persons = [
        EntitySchema(
            title="Alice",
            directory="people",
            entity_type="person",
            content=dedent("""\
                ## Observations
                - [name] Alice Smith
                - [role] Engineer
                - [expertise] Python

                ## Relations
                - works_at [[Acme Corp]]
            """),
        ),
        EntitySchema(
            title="Bob",
            directory="people",
            entity_type="person",
            content=dedent("""\
                ## Observations
                - [name] Bob Jones
                - [role] Designer
                - [expertise] UI/UX

                ## Relations
                - works_at [[Acme Corp]]
            """),
        ),
        EntitySchema(
            title="Carol",
            directory="people",
            entity_type="person",
            content=dedent("""\
                ## Observations
                - [name] Carol Lee
                - [role] Manager

                ## Relations
                - works_at [[Globex]]
            """),
        ),
    ]

    created = []
    for schema in persons:
        entity, _ = await entity_service.create_or_update_entity(schema)
        await search_service.index_entity(entity)
        created.append(entity)

    return created


# --- Infer Endpoint Tests ---


@pytest.mark.asyncio
async def test_infer_schema(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
    entity_service,
    search_service,
):
    """Infer a schema from person notes with observations and relations."""
    await create_person_entities(entity_service, search_service)

    response = await client.post(
        f"{v2_project_url}/schema/infer",
        params={"entity_type": "person"},
    )

    assert response.status_code == 200
    data = response.json()

    assert data["entity_type"] == "person"
    assert data["notes_analyzed"] == 3
    assert isinstance(data["field_frequencies"], list)
    assert isinstance(data["suggested_schema"], dict)
    assert isinstance(data["suggested_required"], list)
    assert isinstance(data["suggested_optional"], list)

    # "name" and "role" appear in all 3 notes -> should be in frequencies
    freq_names = [f["name"] for f in data["field_frequencies"]]
    assert "name" in freq_names
    assert "role" in freq_names

    # "name" appears in 100% of notes -> should be suggested as required
    assert "name" in data["suggested_required"]
    assert "role" in data["suggested_required"]


@pytest.mark.asyncio
async def test_infer_schema_no_matching_notes(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
):
    """Infer returns empty result when no notes of the given type exist."""
    response = await client.post(
        f"{v2_project_url}/schema/infer",
        params={"entity_type": "nonexistent"},
    )

    assert response.status_code == 200
    data = response.json()
    assert data["notes_analyzed"] == 0
    assert data["field_frequencies"] == []
    assert data["suggested_schema"] == {}
    assert data["suggested_required"] == []
    assert data["suggested_optional"] == []


@pytest.mark.asyncio
async def test_infer_schema_includes_relations(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
    entity_service,
    search_service,
):
    """Infer detects relation types as fields alongside observations."""
    await create_person_entities(entity_service, search_service)

    response = await client.post(
        f"{v2_project_url}/schema/infer",
        params={"entity_type": "person"},
    )

    assert response.status_code == 200
    data = response.json()

    # "works_at" relation appears in all 3 notes -> should be in frequencies
    freq_names = [f["name"] for f in data["field_frequencies"]]
    assert "works_at" in freq_names

    # Check it's identified as a relation source
    works_at = next(f for f in data["field_frequencies"] if f["name"] == "works_at")
    assert works_at["source"] == "relation"
    assert works_at["count"] == 3


# --- Validate Endpoint Tests ---


@pytest.mark.asyncio
async def test_validate_with_inline_schema(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
    entity_service,
    search_service,
):
    """Validate a person note against an inline schema in its entity_metadata."""
    # Create a person entity with an inline schema in entity_metadata.
    # The resolver will pick this up as resolution path 1 (inline schema).
    entity, _ = await entity_service.create_or_update_entity(
        EntitySchema(
            title="Dave",
            directory="people",
            entity_type="person",
            entity_metadata={
                "schema": {"name": "string", "role": "string"},
            },
            content=dedent("""\
                ## Observations
                - [name] Dave Wilson
                - [role] Architect
            """),
        )
    )
    await search_service.index_entity(entity)

    response = await client.post(
        f"{v2_project_url}/schema/validate",
        params={"entity_type": "person"},
    )

    assert response.status_code == 200
    data = response.json()

    # Entity was found and validated
    assert data["total_notes"] == 1
    assert isinstance(data["results"], list)
    assert len(data["results"]) == 1

    result = data["results"][0]
    assert result["passed"] is True
    assert result["schema_entity"] == "person"

    # Both "name" and "role" should be present
    field_statuses = {fr["field_name"]: fr["status"] for fr in result["field_results"]}
    assert field_statuses["name"] == "present"
    assert field_statuses["role"] == "present"


@pytest.mark.asyncio
async def test_validate_missing_required_field(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
    entity_service,
    search_service,
):
    """Validate detects a missing required field and produces a warning."""
    # Schema requires "name" and "role", but note only has "name"
    entity, _ = await entity_service.create_or_update_entity(
        EntitySchema(
            title="Eve",
            directory="people",
            entity_type="person",
            entity_metadata={
                "schema": {"name": "string", "role": "string"},
            },
            content=dedent("""\
                ## Observations
                - [name] Eve Martinez
            """),
        )
    )
    await search_service.index_entity(entity)

    response = await client.post(
        f"{v2_project_url}/schema/validate",
        params={"entity_type": "person"},
    )

    assert response.status_code == 200
    data = response.json()
    assert data["total_notes"] == 1

    result = data["results"][0]
    # Default validation mode is "warn" so missing required field produces warning, not error
    assert result["passed"] is True
    assert len(result["warnings"]) > 0
    assert any("role" in w for w in result["warnings"])


@pytest.mark.asyncio
async def test_validate_no_matching_notes(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
):
    """Validate returns empty result when no notes of the given type exist."""
    response = await client.post(
        f"{v2_project_url}/schema/validate",
        params={"entity_type": "nonexistent"},
    )

    assert response.status_code == 200
    data = response.json()
    assert data["total_notes"] == 0
    assert data["results"] == []


# --- Diff Endpoint Tests ---


@pytest.mark.asyncio
async def test_diff_no_schema_found(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
):
    """Diff returns empty drift report when no schema exists for the type."""
    response = await client.get(
        f"{v2_project_url}/schema/diff/nonexistent",
    )

    assert response.status_code == 200
    data = response.json()
    assert data["entity_type"] == "nonexistent"
    assert data["new_fields"] == []
    assert data["dropped_fields"] == []
    assert data["cardinality_changes"] == []


@pytest.mark.asyncio
async def test_diff_with_schema_note(
    client: AsyncClient,
    test_project: Project,
    v2_project_url: str,
    entity_service,
    search_service,
):
    """Diff detects new fields and drift when a schema note exists."""
    # Create a schema note (type="schema" with entity/schema metadata).
    # The search_fn in the router looks for schema notes by searching
    # with types=["schema"] and then returns row.metadata from the search index.
    schema_entity, _ = await entity_service.create_or_update_entity(
        EntitySchema(
            title="Person Schema",
            directory="schemas",
            entity_type="schema",
            entity_metadata={
                "entity": "person",
                "version": 1,
                "schema": {"name": "string", "role": "string"},
            },
            content=dedent("""\
                ## Observations
                - [note] Schema definition for person entities
            """),
        )
    )
    await search_service.index_entity(schema_entity)

    # Create person entities with extra fields not in schema
    await create_person_entities(entity_service, search_service)

    response = await client.get(
        f"{v2_project_url}/schema/diff/person",
    )

    assert response.status_code == 200
    data = response.json()
    assert data["entity_type"] == "person"
    assert isinstance(data["new_fields"], list)
    assert isinstance(data["dropped_fields"], list)
    assert isinstance(data["cardinality_changes"], list)