basicmachines-co-basic-memory/test-int/test_schema/test_validator_integration.py

"""Integration tests for schema validation using real fixture files."""

import pytest

from basic_memory.schema.parser import parse_schema_note, parse_picoschema, SchemaDefinition
from basic_memory.schema.validator import validate_note
from basic_memory.schema.resolver import resolve_schema

from test_schema.helpers import (
    parse_frontmatter,
    parse_observations,
    parse_relations,
    VALID_DIR,
    WARNINGS_DIR,
    EDGE_CASES_DIR,
    SCHEMAS_DIR,
)


class TestValidNotesPassValidation:
    """Validate notes that should pass against the Person schema."""

    def test_paul_graham_passes(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = VALID_DIR / "paul-graham.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("paul-graham", schema, obs, rels)
        assert result.passed is True
        assert len(result.errors) == 0

    def test_paul_graham_name_present(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = VALID_DIR / "paul-graham.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("paul-graham", schema, obs, rels)
        name_result = next(r for r in result.field_results if r.field.name == "name")
        assert name_result.status == "present"
        assert "Paul Graham" in name_result.values

    def test_paul_graham_unmatched_observations(self, person_schema_frontmatter):
        """Extra observations should appear as unmatched, not warnings."""
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = VALID_DIR / "paul-graham.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("paul-graham", schema, obs, rels)
        assert "fact" in result.unmatched_observations

    def test_rich_hickey_passes(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = VALID_DIR / "rich-hickey.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("rich-hickey", schema, obs, rels)
        assert result.passed is True


class TestWarningNotes:
    """Notes that should produce warnings but still pass."""

    def test_missing_required_name_warns(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = WARNINGS_DIR / "missing-required.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("missing-required", schema, obs, rels)
        assert result.passed is True
        name_result = next(r for r in result.field_results if r.field.name == "name")
        assert name_result.status == "missing"
        assert any("name" in w for w in result.warnings)

    def test_wrong_enum_value_warns(self, book_schema_frontmatter):
        schema = parse_schema_note(book_schema_frontmatter)
        filepath = WARNINGS_DIR / "wrong-enum-value.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("wrong-enum", schema, obs, rels)
        genre_result = next(r for r in result.field_results if r.field.name == "genre")
        assert genre_result.status == "enum_mismatch"


class TestInlineSchemaValidation:
    """Validate notes with inline schemas."""

    def test_standup_with_inline_schema(self):
        filepath = VALID_DIR / "standup-2024-01-15.md"
        frontmatter = parse_frontmatter(filepath)
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        assert isinstance(frontmatter.get("schema"), dict)
        fields = parse_picoschema(frontmatter["schema"])
        schema = SchemaDefinition(
            entity=frontmatter.get("type", "unknown"),
            version=1,
            fields=fields,
            validation_mode="warn",
        )
        result = validate_note("standup", schema, obs, rels)
        assert result.passed is True


class TestEdgeCaseValidation:
    """Validation edge cases."""

    def test_empty_note_missing_all_fields(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = EDGE_CASES_DIR / "empty-note.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("empty-note", schema, obs, rels)
        assert all(r.status == "missing" for r in result.field_results)

    def test_unicode_fields_validate(self, person_schema_frontmatter):
        schema = parse_schema_note(person_schema_frontmatter)
        filepath = EDGE_CASES_DIR / "unicode-fields.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("unicode", schema, obs, rels)
        name_result = next(r for r in result.field_results if r.field.name == "name")
        assert name_result.status == "present"

    def test_array_single_value_passes(self, meeting_schema_frontmatter):
        schema = parse_schema_note(meeting_schema_frontmatter)
        filepath = EDGE_CASES_DIR / "array-single.md"
        obs = parse_observations(filepath)
        rels = parse_relations(filepath)
        result = validate_note("array-single", schema, obs, rels)
        attendees = next(r for r in result.field_results if r.field.name == "attendees")
        assert attendees.status == "present"
        assert len(attendees.values) == 1


class TestResolverIntegration:
    """Test schema resolution priority with fixture files."""

    @pytest.mark.asyncio
    async def test_inline_schema_overrides_type(self):
        filepath = EDGE_CASES_DIR / "inline-and-type.md"
        frontmatter = parse_frontmatter(filepath)

        async def mock_search(query: str) -> list[dict]:
            if query == "Person":
                return [parse_frontmatter(SCHEMAS_DIR / "Person.md")]
            return []

        schema = await resolve_schema(frontmatter, mock_search)
        assert schema is not None
        field_names = {f.name for f in schema.fields}
        assert "topic" in field_names
        assert "name" not in field_names

    @pytest.mark.asyncio
    async def test_explicit_ref_overrides_type(self):
        filepath = EDGE_CASES_DIR / "explicit-overrides-type.md"
        frontmatter = parse_frontmatter(filepath)

        async def mock_search(query: str) -> list[dict]:
            if query == "Meeting":
                return [parse_frontmatter(SCHEMAS_DIR / "Meeting.md")]
            if query == "Person":
                return [parse_frontmatter(SCHEMAS_DIR / "Person.md")]
            return []

        schema = await resolve_schema(frontmatter, mock_search)
        assert schema is not None
        assert schema.entity == "Meeting"

    @pytest.mark.asyncio
    async def test_no_schema_returns_none(self):
        filepath = VALID_DIR / "no-schema-note.md"
        frontmatter = parse_frontmatter(filepath)

        async def mock_search(query: str) -> list[dict]:
            return []

        schema = await resolve_schema(frontmatter, mock_search)
        assert schema is None