Bill Ballou
d6a79bf24a
Implement persistent SQLite database feature that allows scripts to query schedule data directly via SQL after loading XER files through MCP. Key changes: - Extend load_xer with db_path parameter for persistent database - Add get_database_info tool to retrieve database connection details - Add schema introspection with tables, columns, primary/foreign keys - Support WAL mode for concurrent read access - Use atomic write pattern to prevent corruption New features: - db_path=None: in-memory database (default, backward compatible) - db_path="": auto-generate path from XER filename (.sqlite extension) - db_path="/path/to/db": explicit persistent database path Response includes complete DatabaseInfo: - db_path: absolute path (or :memory:) - is_persistent: boolean - source_file: loaded XER path - loaded_at: ISO timestamp - schema: tables with columns, primary keys, foreign keys, row counts Closes: User Story 1, 2, 3 from 002-direct-db-access spec
144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
"""Contract tests for get_database_info MCP tool."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from xer_mcp.db import db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_db():
|
|
"""Reset database state for each test."""
|
|
if db.is_initialized:
|
|
db.close()
|
|
yield
|
|
if db.is_initialized:
|
|
db.close()
|
|
|
|
|
|
class TestGetDatabaseInfoContract:
|
|
"""Contract tests verifying get_database_info tool interface."""
|
|
|
|
async def test_get_database_info_returns_current_database(
|
|
self, tmp_path: Path, sample_xer_single_project: Path
|
|
) -> None:
|
|
"""get_database_info returns info about currently loaded database."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
from xer_mcp.tools.load_xer import load_xer
|
|
|
|
db_file = tmp_path / "schedule.db"
|
|
await load_xer(
|
|
file_path=str(sample_xer_single_project),
|
|
db_path=str(db_file),
|
|
)
|
|
|
|
result = await get_database_info()
|
|
|
|
assert "database" in result
|
|
assert result["database"]["db_path"] == str(db_file)
|
|
assert result["database"]["is_persistent"] is True
|
|
|
|
async def test_get_database_info_error_when_no_database(self) -> None:
|
|
"""get_database_info returns error when no database loaded."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
|
|
# Ensure database is not initialized
|
|
if db.is_initialized:
|
|
db.close()
|
|
|
|
result = await get_database_info()
|
|
|
|
assert "error" in result
|
|
assert result["error"]["code"] == "NO_FILE_LOADED"
|
|
|
|
async def test_get_database_info_includes_schema(
|
|
self, tmp_path: Path, sample_xer_single_project: Path
|
|
) -> None:
|
|
"""get_database_info includes schema information."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
from xer_mcp.tools.load_xer import load_xer
|
|
|
|
db_file = tmp_path / "schedule.db"
|
|
await load_xer(
|
|
file_path=str(sample_xer_single_project),
|
|
db_path=str(db_file),
|
|
)
|
|
|
|
result = await get_database_info()
|
|
|
|
assert "schema" in result["database"]
|
|
assert "tables" in result["database"]["schema"]
|
|
|
|
async def test_get_database_info_includes_loaded_at(
|
|
self, tmp_path: Path, sample_xer_single_project: Path
|
|
) -> None:
|
|
"""get_database_info includes loaded_at timestamp."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
from xer_mcp.tools.load_xer import load_xer
|
|
|
|
db_file = tmp_path / "schedule.db"
|
|
await load_xer(
|
|
file_path=str(sample_xer_single_project),
|
|
db_path=str(db_file),
|
|
)
|
|
|
|
result = await get_database_info()
|
|
|
|
assert "loaded_at" in result["database"]
|
|
# Should be ISO format timestamp
|
|
assert "T" in result["database"]["loaded_at"]
|
|
|
|
async def test_get_database_info_includes_source_file(
|
|
self, tmp_path: Path, sample_xer_single_project: Path
|
|
) -> None:
|
|
"""get_database_info includes source XER file path."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
from xer_mcp.tools.load_xer import load_xer
|
|
|
|
db_file = tmp_path / "schedule.db"
|
|
await load_xer(
|
|
file_path=str(sample_xer_single_project),
|
|
db_path=str(db_file),
|
|
)
|
|
|
|
result = await get_database_info()
|
|
|
|
assert result["database"]["source_file"] == str(sample_xer_single_project)
|
|
|
|
async def test_get_database_info_for_memory_database(
|
|
self, sample_xer_single_project: Path
|
|
) -> None:
|
|
"""get_database_info works for in-memory database."""
|
|
from xer_mcp.tools.get_database_info import get_database_info
|
|
from xer_mcp.tools.load_xer import load_xer
|
|
|
|
await load_xer(file_path=str(sample_xer_single_project))
|
|
|
|
result = await get_database_info()
|
|
|
|
assert "database" in result
|
|
assert result["database"]["db_path"] == ":memory:"
|
|
assert result["database"]["is_persistent"] is False
|
|
|
|
|
|
class TestGetDatabaseInfoToolSchema:
|
|
"""Tests for MCP tool schema."""
|
|
|
|
async def test_get_database_info_tool_registered(self) -> None:
|
|
"""get_database_info tool is registered with MCP server."""
|
|
from xer_mcp.server import list_tools
|
|
|
|
tools = await list_tools()
|
|
tool_names = [t.name for t in tools]
|
|
assert "get_database_info" in tool_names
|
|
|
|
async def test_get_database_info_tool_has_empty_input_schema(self) -> None:
|
|
"""get_database_info tool has no required inputs."""
|
|
from xer_mcp.server import list_tools
|
|
|
|
tools = await list_tools()
|
|
tool = next(t for t in tools if t.name == "get_database_info")
|
|
# Should have empty or no required properties
|
|
assert "required" not in tool.inputSchema or len(tool.inputSchema.get("required", [])) == 0
|