mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Drew Cain
3415fd1014
Signed-off-by: Drew Cain <groksrc@gmail.com> Signed-off-by: phernandez <paul@basicmachines.co> Co-authored-by: phernandez <paul@basicmachines.co>
627 lines
17 KiB
Python
627 lines
17 KiB
Python
"""Tests for schema MCP tools (validate, infer, diff).
|
|
|
|
Covers the tool function logic including success paths and error/exception paths.
|
|
The success-path tests use the full ASGI stack via the app fixture.
|
|
Error-path tests monkeypatch SchemaClient methods to trigger the except branch.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from basic_memory.mcp.tools.schema import schema_validate, schema_infer, schema_diff
|
|
from basic_memory.mcp.tools.write_note import write_note
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
|
|
def _write_schema_file(project_path: Path, filename: str, content: str):
|
|
"""Write a markdown file directly to disk (bypasses write_note frontmatter generation)."""
|
|
path = project_path / filename
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
|
|
PERSON_SCHEMA = """\
|
|
---
|
|
title: Person
|
|
type: schema
|
|
entity: person
|
|
version: 1
|
|
schema:
|
|
name: string, full name
|
|
role?: string, job title
|
|
settings:
|
|
validation: warn
|
|
---
|
|
|
|
# Person
|
|
|
|
Schema for person entities.
|
|
"""
|
|
|
|
|
|
PERSON_NOTE = """\
|
|
---
|
|
title: {name}
|
|
type: person
|
|
permalink: people/{permalink}
|
|
---
|
|
|
|
# {name}
|
|
|
|
## Observations
|
|
- [name] {name}
|
|
- [role] Engineer
|
|
"""
|
|
|
|
|
|
# --- Success-path tests (full ASGI stack) ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_by_type(app, test_project, sync_service):
|
|
"""Validate all notes of a given entity type."""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA)
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
# Sync so the database picks up the files
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Validation: person" in result
|
|
assert "Notes: 1" in result
|
|
assert "**Alice**" in result
|
|
assert "valid" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_json_output(app, test_project, sync_service):
|
|
"""JSON output returns a dict with full structured data."""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA)
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
output_format="json",
|
|
)
|
|
|
|
assert isinstance(result, dict)
|
|
assert result["total_notes"] == 1
|
|
assert result["valid_count"] == 1
|
|
assert len(result["results"]) == 1
|
|
assert result["results"][0]["note_identifier"] == "Alice"
|
|
assert result["results"][0]["passed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_picoschema_modifier_descriptions(app, test_project, sync_service):
|
|
"""Modifier descriptions should not become literal field names."""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(
|
|
project_path,
|
|
"schemas/PicoTest.md",
|
|
"""\
|
|
---
|
|
title: PicoTest
|
|
type: schema
|
|
entity: pico_test
|
|
schema:
|
|
name: string
|
|
status(enum, current state): [active, inactive]
|
|
tags(array, list of tags): string
|
|
settings:
|
|
validation: warn
|
|
---
|
|
|
|
# PicoTest
|
|
""",
|
|
)
|
|
_write_schema_file(
|
|
project_path,
|
|
"pico/PicoTest1.md",
|
|
"""\
|
|
---
|
|
title: PicoTest1
|
|
type: pico_test
|
|
permalink: pico/pico-test-1
|
|
---
|
|
|
|
# PicoTest1
|
|
|
|
## Observations
|
|
- [name] PicoTest1
|
|
- [status] active
|
|
- [tags] foo
|
|
- [tags] bar
|
|
""",
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
note_type="pico_test",
|
|
project=test_project.name,
|
|
output_format="json",
|
|
)
|
|
|
|
assert isinstance(result, dict)
|
|
note_result = result["results"][0]
|
|
field_statuses = {fr["field_name"]: fr["status"] for fr in note_result["field_results"]}
|
|
assert result["valid_count"] == 1
|
|
assert note_result["warnings"] == []
|
|
assert note_result["unmatched_observations"] == {}
|
|
assert field_statuses == {
|
|
"name": "present",
|
|
"status": "present",
|
|
"tags": "present",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_by_identifier(app, test_project, sync_service):
|
|
"""Validate a specific note by identifier."""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA)
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
identifier="people/alice",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "**Alice**" in result
|
|
assert "valid" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_by_title(app, test_project, sync_service):
|
|
"""Validate a specific note by title (not permalink).
|
|
|
|
Regression test for issue #33: schema_validate(identifier="Note Title")
|
|
returned 0 notes because the router only searched by permalink.
|
|
"""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA)
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
# Use the title "Alice" instead of the permalink "people/alice"
|
|
result = await schema_validate(
|
|
identifier="Alice",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "**Alice**" in result
|
|
assert "Notes: 1" in result
|
|
assert "valid" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_identifier_no_schema_returns_guidance(
|
|
app, test_project, sync_service
|
|
):
|
|
"""When a note exists but no schema is defined, return guidance.
|
|
|
|
Regression test for issue #33: when validating a single note by identifier
|
|
and no schema exists, the tool should return guidance instead of an empty report.
|
|
"""
|
|
project_path = Path(test_project.path)
|
|
|
|
# Create a person note but no schema note
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
identifier="Alice",
|
|
project=test_project.name,
|
|
)
|
|
|
|
# Should return guidance string about missing schema
|
|
assert isinstance(result, str)
|
|
assert "No Schema Found" in result
|
|
assert "person" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_infer(app, test_project, sync_service):
|
|
"""Infer a schema from existing notes."""
|
|
project_path = Path(test_project.path)
|
|
|
|
for name in ["Alice", "Bob", "Charlie"]:
|
|
_write_schema_file(
|
|
project_path,
|
|
f"people/{name}.md",
|
|
PERSON_NOTE.format(name=name, permalink=name.lower()),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_infer(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Inference: person" in result
|
|
assert "Notes analyzed: 3" in result
|
|
assert "Field Frequencies" in result
|
|
assert "**name**" in result
|
|
assert "**role**" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_diff(app, test_project, sync_service):
|
|
"""Detect drift between schema and actual usage."""
|
|
project_path = Path(test_project.path)
|
|
|
|
_write_schema_file(project_path, "schemas/Person.md", PERSON_SCHEMA)
|
|
|
|
# Create a person with an extra "hobby" field not in the schema
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Dave.md",
|
|
"""\
|
|
---
|
|
title: Dave
|
|
type: person
|
|
permalink: people/dave
|
|
---
|
|
|
|
# Dave
|
|
|
|
## Observations
|
|
- [name] Dave
|
|
- [role] Manager
|
|
- [hobby] Chess
|
|
""",
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_diff(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Drift: person" in result
|
|
# Dave has a "hobby" field not in the schema, so drift should be detected
|
|
assert "**hobby**" in result
|
|
|
|
|
|
# --- write_note metadata → schema workflow ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_note_metadata_creates_schema_note(app, test_project, sync_service):
|
|
"""Create a schema note via write_note(metadata=...), then validate against it.
|
|
|
|
Proves the end-to-end workflow: write_note → sync → schema_validate.
|
|
"""
|
|
project_path = Path(test_project.path)
|
|
|
|
# 1. Create person notes via direct file write (content under test is the schema)
|
|
for name in ["Alice", "Bob"]:
|
|
_write_schema_file(
|
|
project_path,
|
|
f"people/{name}.md",
|
|
PERSON_NOTE.format(name=name, permalink=name.lower()),
|
|
)
|
|
|
|
# 2. Create the schema note via write_note with metadata
|
|
await write_note(
|
|
title="Person",
|
|
directory="schemas",
|
|
note_type="schema",
|
|
content="# Person\n\nSchema for person entities.",
|
|
metadata={
|
|
"entity": "person",
|
|
"version": 1,
|
|
"schema": {"name": "string", "role?": "string"},
|
|
"settings": {"validation": "warn"},
|
|
},
|
|
project=test_project.name,
|
|
)
|
|
|
|
# 3. Sync picks up person notes written directly to disk
|
|
await sync_service.sync(project_path)
|
|
|
|
# 4. Validate — schema_validate should find the schema and validate person notes
|
|
result = await schema_validate(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Validation: person" in result
|
|
assert "valid" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_title_mismatch_finds_by_metadata(app, test_project, sync_service):
|
|
"""Schema lookup works even when the schema title doesn't match the entity type.
|
|
|
|
Regression test: the old text-search approach failed when the schema note's title
|
|
(e.g. "Employee Schema") didn't textually match the entity type ("employee").
|
|
The metadata-based lookup matches on entity_metadata['entity'] instead.
|
|
"""
|
|
project_path = Path(test_project.path)
|
|
|
|
# Schema title "Employee Schema" != entity type "employee"
|
|
_write_schema_file(
|
|
project_path,
|
|
"schemas/EmployeeSchema.md",
|
|
"""\
|
|
---
|
|
title: Employee Schema
|
|
type: schema
|
|
entity: employee
|
|
version: 1
|
|
schema:
|
|
name: string, full name
|
|
department?: string, department name
|
|
settings:
|
|
validation: warn
|
|
---
|
|
|
|
# Employee Schema
|
|
|
|
Schema for employee entities.
|
|
""",
|
|
)
|
|
|
|
# Create employee notes
|
|
for name, dept in [("Alice", "Engineering"), ("Bob", "Marketing")]:
|
|
_write_schema_file(
|
|
project_path,
|
|
f"employees/{name}.md",
|
|
f"""\
|
|
---
|
|
title: {name}
|
|
type: employee
|
|
permalink: employees/{name.lower()}
|
|
---
|
|
|
|
# {name}
|
|
|
|
## Observations
|
|
- [name] {name}
|
|
- [department] {dept}
|
|
""",
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
# Validate — must find "Employee Schema" via entity_metadata['entity'] == "employee"
|
|
result = await schema_validate(
|
|
note_type="employee",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Validation: employee" in result
|
|
assert "Notes: 2" in result
|
|
assert "Valid: 2" in result
|
|
# Both notes have name + department, schema requires name and optionally department
|
|
assert "**Alice**" in result
|
|
assert "**Bob**" in result
|
|
|
|
|
|
# --- Empty schema guard ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_infer_empty_schema_returns_guidance(app, test_project, sync_service):
|
|
"""When notes exist but no fields meet the threshold, return guidance instead of data."""
|
|
project_path = Path(test_project.path)
|
|
|
|
# Create notes with completely different observation categories so no field
|
|
# reaches the 25% threshold across all notes
|
|
for i, (name, category) in enumerate(
|
|
[
|
|
("alpha", "color"),
|
|
("bravo", "shape"),
|
|
("charlie", "size"),
|
|
("delta", "weight"),
|
|
("echo", "temp"),
|
|
]
|
|
):
|
|
_write_schema_file(
|
|
project_path,
|
|
f"things/{name}.md",
|
|
f"""\
|
|
---
|
|
title: {name}
|
|
type: widget
|
|
permalink: things/{name}
|
|
---
|
|
|
|
# {name}
|
|
|
|
## Observations
|
|
- [{category}] some value
|
|
""",
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_infer(
|
|
note_type="widget",
|
|
project=test_project.name,
|
|
)
|
|
|
|
# Should return guidance string, not an InferenceReport
|
|
assert isinstance(result, str)
|
|
assert "No Schema Pattern Found" in result
|
|
assert "widget" in result
|
|
assert "Suggestions" in result
|
|
|
|
|
|
# --- No schema found guards ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_no_notes_returns_guidance(app, test_project, sync_service):
|
|
"""When no notes of the requested type exist, return guidance on creating notes."""
|
|
result = await schema_validate(
|
|
note_type="employee",
|
|
project=test_project.name,
|
|
)
|
|
|
|
# Should return guidance about creating notes, not about missing schema
|
|
assert isinstance(result, str)
|
|
assert "No Notes Found" in result
|
|
assert "employee" in result
|
|
assert "write_note" in result
|
|
assert "search_notes" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_no_schema_returns_guidance(app, test_project, sync_service):
|
|
"""When notes exist but no schema is defined, return guidance on creating one."""
|
|
project_path = Path(test_project.path)
|
|
|
|
# Create person notes but no schema note
|
|
for name in ["Alice", "Bob"]:
|
|
_write_schema_file(
|
|
project_path,
|
|
f"people/{name}.md",
|
|
PERSON_NOTE.format(name=name, permalink=name.lower()),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_validate(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
# Should return guidance string, not a ValidationReport
|
|
assert isinstance(result, str)
|
|
assert "No Schema Found" in result
|
|
assert "person" in result
|
|
assert "schema_infer" in result
|
|
assert "How to Create a Schema" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_diff_no_schema_returns_guidance(app, test_project, sync_service):
|
|
"""When no schema exists for the type, return guidance on creating one."""
|
|
project_path = Path(test_project.path)
|
|
|
|
# Create person notes but no schema note
|
|
_write_schema_file(
|
|
project_path,
|
|
"people/Alice.md",
|
|
PERSON_NOTE.format(name="Alice", permalink="alice"),
|
|
)
|
|
|
|
await sync_service.sync(project_path)
|
|
|
|
result = await schema_diff(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
# Should return guidance string, not a DriftReport
|
|
assert isinstance(result, str)
|
|
assert "No Schema Found" in result
|
|
assert "person" in result
|
|
assert "schema_infer" in result
|
|
assert "How to Create a Schema" in result
|
|
|
|
|
|
# --- Error-path tests (monkeypatched SchemaClient) ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_validate_error_returns_guidance(app, test_project):
|
|
"""When SchemaClient.validate raises, the tool returns a troubleshooting string."""
|
|
mock_validate = AsyncMock(side_effect=RuntimeError("connection lost"))
|
|
|
|
with patch("basic_memory.mcp.clients.schema.SchemaClient.validate", mock_validate):
|
|
result = await schema_validate(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Validation Failed" in result
|
|
assert "Troubleshooting" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_infer_error_returns_guidance(app, test_project):
|
|
"""When SchemaClient.infer raises, the tool returns a troubleshooting string."""
|
|
mock_infer = AsyncMock(side_effect=RuntimeError("db unavailable"))
|
|
|
|
with patch("basic_memory.mcp.clients.schema.SchemaClient.infer", mock_infer):
|
|
result = await schema_infer(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Inference Failed" in result
|
|
assert "Troubleshooting" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_diff_error_returns_guidance(app, test_project):
|
|
"""When SchemaClient.diff raises, the tool returns a troubleshooting string."""
|
|
mock_diff = AsyncMock(side_effect=RuntimeError("network error"))
|
|
|
|
with patch("basic_memory.mcp.clients.schema.SchemaClient.diff", mock_diff):
|
|
result = await schema_diff(
|
|
note_type="person",
|
|
project=test_project.name,
|
|
)
|
|
|
|
assert isinstance(result, str)
|
|
assert "Schema Diff Failed" in result
|
|
assert "Troubleshooting" in result
|