feat(spaces): implement Phase 0-1 of Information Space Service

Phase 0 - Project Organization:
- Create docs/PROJECT_STRUCTURE.md documenting codebase layout
- Create markitect/core/ with parser, serializer, document_manager, workspace
- Create markitect/schema/ consolidating 6 schema_*.py modules
- Create markitect/storage/ with database module
- Maintain backward compatibility via re-exports from original locations
- Add docs/roadmap/information-space-service/ with README and WORKPLAN

Phase 1 - Foundation (Weeks 1-3):
- Week 1: Core domain models (InformationSpace, SpaceDocument, SpaceConfig,
  SpaceMetadata, SpaceVariable, TransclusionReference, SpaceStatus)
- Week 2: Repository layer with interfaces (ISpaceRepository,
  IDocumentAssociationRepository, IVariableRepository, IReferenceRepository)
  and SQLite implementations with foreign key cascade deletes
- Week 3: SpaceService orchestration layer with full CRUD, document,
  variable, and reference tracking operations

Test coverage: 124 tests (25 model + 63 repository + 36 integration)

Capabilities delivered:
- CAP-001: InformationSpace entity with lifecycle management
- CAP-002: SpaceRepository CRUD with SQLite backing
- CAP-003: Document-Space associations with path-based organization
- CAP-004: Space metadata and configuration schemas
- CAP-005: Database schema with migrations

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-02-08 02:02:46 +01:00
parent 6ebcc0f60e
commit 9b12875681
45 changed files with 9818 additions and 4300 deletions

View File

View File

@@ -0,0 +1,570 @@
"""
Integration tests for SpaceService.
Tests the full workflow of space operations including:
- Space creation and lifecycle management
- Document operations within spaces
- Variable management
- Reference tracking for cache invalidation
"""
import pytest
import tempfile
import os
from markitect.spaces import (
SpaceService,
InformationSpace,
SpaceDocument,
SpaceConfig,
SpaceMetadata,
SpaceStatus,
SqliteSpaceRepository,
SqliteDocumentRepository,
SqliteVariableRepository,
SqliteReferenceRepository,
)
@pytest.fixture
def temp_db():
"""Create a temporary database file for testing."""
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
yield path
if os.path.exists(path):
os.unlink(path)
@pytest.fixture
def space_service(temp_db):
"""Create a fully wired SpaceService for testing."""
return SpaceService(
space_repo=SqliteSpaceRepository(temp_db),
document_repo=SqliteDocumentRepository(temp_db),
variable_repo=SqliteVariableRepository(temp_db),
reference_repo=SqliteReferenceRepository(temp_db),
)
class TestSpaceLifecycle:
"""Tests for space lifecycle operations."""
def test_create_and_retrieve_space(self, space_service):
"""Test creating and retrieving a space."""
space = space_service.create_space(
name="my-docs",
description="My documentation",
)
assert space.name == "my-docs"
assert space.description == "My documentation"
assert space.status == SpaceStatus.DRAFT
# Retrieve by ID
retrieved = space_service.get_space(space.id)
assert retrieved is not None
assert retrieved.name == "my-docs"
# Retrieve by name
by_name = space_service.get_space_by_name("my-docs")
assert by_name is not None
assert by_name.id == space.id
def test_create_space_with_config_and_metadata(self, space_service):
"""Test creating a space with custom config and metadata."""
config = SpaceConfig(
theme="dark",
history_enabled=True,
enable_caching=False,
)
metadata = SpaceMetadata(
tags=["api", "v1"],
author="tester",
custom={"version": "1.0"},
)
space = space_service.create_space(
name="configured-space",
config=config,
metadata=metadata,
)
assert space.config.theme == "dark"
assert space.config.history_enabled is True
assert space.metadata.tags == ["api", "v1"]
assert space.metadata.author == "tester"
def test_update_space(self, space_service):
"""Test updating a space."""
space = space_service.create_space(name="original")
updated = space_service.update_space(
space.id,
name="updated",
description="New description",
)
assert updated.name == "updated"
assert updated.description == "New description"
# Verify persisted
retrieved = space_service.get_space(space.id)
assert retrieved.name == "updated"
def test_space_lifecycle_transitions(self, space_service):
"""Test space status transitions."""
space = space_service.create_space(name="lifecycle-test")
assert space.status == SpaceStatus.DRAFT
# Activate
activated = space_service.activate_space(space.id)
assert activated.status == SpaceStatus.ACTIVE
# Archive
archived = space_service.archive_space(space.id)
assert archived.status == SpaceStatus.ARCHIVED
def test_delete_space(self, space_service):
"""Test deleting a space."""
space = space_service.create_space(name="to-delete")
result = space_service.delete_space(space.id)
assert result is True
# Verify deleted
retrieved = space_service.get_space(space.id)
assert retrieved is None
def test_list_spaces_excludes_archived(self, space_service):
"""Test that list_spaces excludes archived by default."""
space1 = space_service.create_space(name="active")
space2 = space_service.create_space(name="archived")
space_service.archive_space(space2.id)
spaces = space_service.list_spaces()
assert len(spaces) == 1
assert spaces[0].name == "active"
# Include archived
all_spaces = space_service.list_spaces(include_archived=True)
assert len(all_spaces) == 2
class TestSpaceHierarchy:
"""Tests for space hierarchy operations."""
def test_create_child_space(self, space_service):
"""Test creating a child space."""
parent = space_service.create_space(name="parent")
child = space_service.create_space(
name="child",
parent_space_id=parent.id,
)
assert child.parent_space_id == parent.id
children = space_service.get_child_spaces(parent.id)
assert len(children) == 1
assert children[0].id == child.id
def test_create_nested_hierarchy(self, space_service):
"""Test creating a nested space hierarchy."""
root = space_service.create_space(name="root")
level1 = space_service.create_space(name="level1", parent_space_id=root.id)
level2 = space_service.create_space(name="level2", parent_space_id=level1.id)
# Verify hierarchy
root_children = space_service.get_child_spaces(root.id)
assert len(root_children) == 1
assert root_children[0].id == level1.id
level1_children = space_service.get_child_spaces(level1.id)
assert len(level1_children) == 1
assert level1_children[0].id == level2.id
def test_delete_space_with_children_cascade(self, space_service):
"""Test deleting a space cascades to children."""
parent = space_service.create_space(name="parent")
child = space_service.create_space(name="child", parent_space_id=parent.id)
space_service.delete_space(parent.id, cascade=True)
assert space_service.get_space(parent.id) is None
assert space_service.get_space(child.id) is None
def test_delete_space_with_children_no_cascade_raises(self, space_service):
"""Test deleting a space with children raises if cascade=False."""
parent = space_service.create_space(name="parent")
space_service.create_space(name="child", parent_space_id=parent.id)
with pytest.raises(ValueError, match="has 1 child"):
space_service.delete_space(parent.id, cascade=False)
class TestDocumentOperations:
"""Tests for document operations within spaces."""
def test_add_and_list_documents(self, space_service):
"""Test adding and listing documents."""
space = space_service.create_space(name="doc-space")
doc1 = space_service.add_document(
space.id,
space_path="/intro.md",
document_id="doc-1",
)
doc2 = space_service.add_document(
space.id,
space_path="/api/endpoints.md",
document_id="doc-2",
)
docs = space_service.list_documents(space.id)
assert len(docs) == 2
def test_get_document_by_path(self, space_service):
"""Test getting a document by its path."""
space = space_service.create_space(name="doc-space")
space_service.add_document(space.id, "/intro.md", document_id="doc-1")
doc = space_service.get_document_by_path(space.id, "/intro.md")
assert doc is not None
assert doc.document_id == "doc-1"
# Also works without leading slash
doc2 = space_service.get_document_by_path(space.id, "intro.md")
assert doc2 is not None
def test_move_document(self, space_service):
"""Test moving a document to a new path."""
space = space_service.create_space(name="doc-space")
doc = space_service.add_document(space.id, "/old-path.md")
moved = space_service.move_document(doc.id, "/new-path.md")
assert moved.space_path == "/new-path.md"
# Old path should not exist
old_doc = space_service.get_document_by_path(space.id, "/old-path.md")
assert old_doc is None
# New path should work
new_doc = space_service.get_document_by_path(space.id, "/new-path.md")
assert new_doc is not None
def test_remove_document(self, space_service):
"""Test removing a document."""
space = space_service.create_space(name="doc-space")
doc = space_service.add_document(space.id, "/to-remove.md")
result = space_service.remove_document(doc.id)
assert result is True
# Verify removed
retrieved = space_service.get_document(doc.id)
assert retrieved is None
def test_reorder_documents(self, space_service):
"""Test reordering documents."""
space = space_service.create_space(name="doc-space")
doc1 = space_service.add_document(space.id, "/a.md", order_index=0)
doc2 = space_service.add_document(space.id, "/b.md", order_index=1)
doc3 = space_service.add_document(space.id, "/c.md", order_index=2)
# Reorder: c, a, b
space_service.reorder_documents(space.id, [doc3.id, doc1.id, doc2.id])
docs = space_service.list_documents(space.id)
assert docs[0].id == doc3.id
assert docs[1].id == doc1.id
assert docs[2].id == doc2.id
def test_document_with_metadata(self, space_service):
"""Test document with custom metadata."""
space = space_service.create_space(name="doc-space")
doc = space_service.add_document(
space.id,
"/api.md",
metadata={"title": "API Reference", "order": 5},
)
retrieved = space_service.get_document(doc.id)
assert retrieved.metadata["title"] == "API Reference"
assert retrieved.metadata["order"] == 5
def test_update_document_hash(self, space_service):
"""Test updating document content hash."""
space = space_service.create_space(name="doc-space")
doc = space_service.add_document(space.id, "/content.md")
space_service.update_document_hash(doc.id, "hash123abc")
retrieved = space_service.get_document(doc.id)
assert retrieved.content_hash == "hash123abc"
class TestVariableOperations:
"""Tests for variable operations within spaces."""
def test_set_and_get_variable(self, space_service):
"""Test setting and getting a variable."""
space = space_service.create_space(name="var-space")
var = space_service.set_variable(space.id, "version", "1.0.0")
assert var.value == "1.0.0"
retrieved = space_service.get_variable(space.id, "version")
assert retrieved is not None
assert retrieved.value == "1.0.0"
def test_list_variables(self, space_service):
"""Test listing variables."""
space = space_service.create_space(name="var-space")
space_service.set_variable(space.id, "var1", "value1")
space_service.set_variable(space.id, "var2", "value2")
variables = space_service.list_variables(space.id)
assert len(variables) == 2
def test_list_variables_by_scope(self, space_service):
"""Test listing variables filtered by scope."""
space = space_service.create_space(name="var-space")
space_service.set_variable(space.id, "global", "g", scope="space")
space_service.set_variable(space.id, "local", "l", scope="document")
space_vars = space_service.list_variables(space.id, scope="space")
assert len(space_vars) == 1
assert space_vars[0].name == "global"
def test_delete_variable(self, space_service):
"""Test deleting a variable."""
space = space_service.create_space(name="var-space")
space_service.set_variable(space.id, "temp", "value")
result = space_service.delete_variable(space.id, "temp")
assert result is True
retrieved = space_service.get_variable(space.id, "temp")
assert retrieved is None
def test_get_variables_dict(self, space_service):
"""Test getting variables as a dictionary."""
space = space_service.create_space(name="var-space")
space_service.set_variable(space.id, "api_url", "https://api.example.com")
space_service.set_variable(space.id, "version", "2.0")
variables_dict = space_service.get_variables_dict(space.id)
assert variables_dict == {
"api_url": "https://api.example.com",
"version": "2.0",
}
def test_variable_with_complex_value(self, space_service):
"""Test variable with complex JSON value."""
space = space_service.create_space(name="var-space")
complex_value = {
"endpoints": ["/api/v1", "/api/v2"],
"config": {"timeout": 30},
}
space_service.set_variable(space.id, "api_config", complex_value)
retrieved = space_service.get_variable(space.id, "api_config")
assert retrieved.value == complex_value
class TestReferenceTracking:
"""Tests for transclusion reference tracking."""
def test_add_and_get_references(self, space_service):
"""Test adding and getting references."""
space = space_service.create_space(name="ref-space")
space_service.add_reference("doc-1", "shared-component", space.id)
space_service.add_reference("doc-2", "shared-component", space.id)
refs = space_service.get_references_to("shared-component", space.id)
assert len(refs) == 2
def test_get_references_from(self, space_service):
"""Test getting references from a source document."""
space = space_service.create_space(name="ref-space")
space_service.add_reference("doc-1", "component-a", space.id)
space_service.add_reference("doc-1", "component-b", space.id)
refs = space_service.get_references_from("doc-1", space.id)
assert len(refs) == 2
targets = [r.target_doc_id for r in refs]
assert "component-a" in targets
assert "component-b" in targets
def test_get_dependents(self, space_service):
"""Test getting dependent documents."""
space = space_service.create_space(name="ref-space")
space_service.add_reference("doc-1", "shared", space.id)
space_service.add_reference("doc-2", "shared", space.id)
space_service.add_reference("doc-3", "shared", space.id)
dependents = space_service.get_dependents("shared", space.id)
assert len(dependents) == 3
assert set(dependents) == {"doc-1", "doc-2", "doc-3"}
def test_clear_references_from(self, space_service):
"""Test clearing references from a source document."""
space = space_service.create_space(name="ref-space")
space_service.add_reference("doc-1", "a", space.id)
space_service.add_reference("doc-1", "b", space.id)
space_service.add_reference("doc-2", "a", space.id)
count = space_service.clear_references_from("doc-1", space.id)
assert count == 2
# doc-1 refs should be gone
refs1 = space_service.get_references_from("doc-1", space.id)
assert len(refs1) == 0
# doc-2 refs should still exist
refs2 = space_service.get_references_from("doc-2", space.id)
assert len(refs2) == 1
def test_remove_document_clears_references(self, space_service):
"""Test that removing a document clears its references."""
space = space_service.create_space(name="ref-space")
doc = space_service.add_document(space.id, "/source.md")
# Add reference from this document
space_service.add_reference(doc.id, "target", space.id)
# Verify reference exists
refs = space_service.get_references_from(doc.id, space.id)
assert len(refs) == 1
# Remove document
space_service.remove_document(doc.id)
# References should be cleared
refs = space_service.get_references_from(doc.id, space.id)
assert len(refs) == 0
class TestFullWorkflow:
"""End-to-end workflow tests."""
def test_documentation_space_workflow(self, space_service):
"""Test a complete documentation space workflow."""
# Create a documentation space
space = space_service.create_space(
name="api-docs",
description="API Documentation",
config=SpaceConfig(theme="minimal"),
metadata=SpaceMetadata(tags=["api", "v2"]),
)
# Add documents
intro = space_service.add_document(
space.id,
"/intro.md",
order_index=0,
metadata={"title": "Introduction"},
)
endpoints = space_service.add_document(
space.id,
"/api/endpoints.md",
order_index=1,
metadata={"title": "API Endpoints"},
)
auth = space_service.add_document(
space.id,
"/api/auth.md",
order_index=2,
metadata={"title": "Authentication"},
)
# Add variables for transclusion
space_service.set_variable(space.id, "api_base_url", "https://api.example.com")
space_service.set_variable(space.id, "version", "2.0")
# Track references (e.g., endpoints includes auth)
space_service.add_reference(endpoints.id, auth.id, space.id)
# Activate the space
space_service.activate_space(space.id)
# Get stats
stats = space_service.get_space_stats(space.id)
assert stats["document_count"] == 3
assert stats["variable_count"] == 2
assert stats["status"] == "active"
# Verify the space
retrieved = space_service.get_space(space.id)
assert retrieved.status == SpaceStatus.ACTIVE
# List documents in order
docs = space_service.list_documents(space.id)
assert len(docs) == 3
assert docs[0].space_path == "/intro.md"
# Get transclusion context
context = space_service.get_variables_dict(space.id)
assert context["api_base_url"] == "https://api.example.com"
# Check dependencies for cache invalidation
dependents = space_service.get_dependents(auth.id, space.id)
assert endpoints.id in dependents
def test_space_stats(self, space_service):
"""Test getting space statistics."""
space = space_service.create_space(name="stats-test")
space_service.add_document(space.id, "/doc1.md")
space_service.add_document(space.id, "/doc2.md")
space_service.set_variable(space.id, "var1", "value1")
space_service.create_space(name="child", parent_space_id=space.id)
stats = space_service.get_space_stats(space.id)
assert stats["name"] == "stats-test"
assert stats["document_count"] == 2
assert stats["variable_count"] == 1
assert stats["child_space_count"] == 1
class TestErrorHandling:
"""Tests for error handling scenarios."""
def test_create_space_empty_name_raises(self, space_service):
"""Test that empty name raises ValueError."""
with pytest.raises(ValueError, match="cannot be empty"):
space_service.create_space(name="")
with pytest.raises(ValueError, match="cannot be empty"):
space_service.create_space(name=" ")
def test_create_space_duplicate_name_raises(self, space_service):
"""Test that duplicate name raises ValueError."""
space_service.create_space(name="taken")
with pytest.raises(ValueError, match="already exists"):
space_service.create_space(name="taken")
def test_update_nonexistent_space_raises(self, space_service):
"""Test that updating non-existent space raises ValueError."""
with pytest.raises(ValueError, match="not found"):
space_service.update_space("non-existent", name="new-name")
def test_add_document_to_nonexistent_space_raises(self, space_service):
"""Test that adding document to non-existent space raises."""
with pytest.raises(ValueError, match="not found"):
space_service.add_document("non-existent", "/doc.md")
def test_set_variable_in_nonexistent_space_raises(self, space_service):
"""Test that setting variable in non-existent space raises."""
with pytest.raises(ValueError, match="not found"):
space_service.set_variable("non-existent", "var", "value")
def test_create_child_with_nonexistent_parent_raises(self, space_service):
"""Test that creating child with non-existent parent raises."""
with pytest.raises(ValueError, match="Parent space.*not found"):
space_service.create_space(name="orphan", parent_space_id="non-existent")

View File

@@ -0,0 +1 @@
"""Unit tests for the spaces package."""

View File

@@ -0,0 +1,299 @@
"""
Unit tests for Information Space models.
Tests the core domain models: InformationSpace, SpaceDocument, SpaceConfig, SpaceMetadata.
"""
import pytest
from datetime import datetime
from markitect.spaces.models import (
InformationSpace,
SpaceDocument,
SpaceConfig,
SpaceMetadata,
SpaceVariable,
TransclusionReference,
SpaceStatus,
)
class TestSpaceMetadata:
"""Tests for SpaceMetadata dataclass."""
def test_default_metadata(self):
"""Test default metadata values."""
metadata = SpaceMetadata()
assert metadata.tags == []
assert metadata.author is None
assert metadata.custom == {}
def test_metadata_with_values(self):
"""Test metadata with custom values."""
metadata = SpaceMetadata(
tags=["api", "docs"],
author="test-user",
custom={"version": "1.0"}
)
assert metadata.tags == ["api", "docs"]
assert metadata.author == "test-user"
assert metadata.custom["version"] == "1.0"
def test_metadata_to_dict(self):
"""Test metadata serialization."""
metadata = SpaceMetadata(tags=["test"], author="user")
data = metadata.to_dict()
assert data["tags"] == ["test"]
assert data["author"] == "user"
def test_metadata_from_dict(self):
"""Test metadata deserialization."""
data = {"tags": ["api"], "author": "admin", "custom": {"key": "value"}}
metadata = SpaceMetadata.from_dict(data)
assert metadata.tags == ["api"]
assert metadata.author == "admin"
assert metadata.custom["key"] == "value"
class TestSpaceConfig:
"""Tests for SpaceConfig dataclass."""
def test_default_config(self):
"""Test default configuration values."""
config = SpaceConfig()
assert config.default_variant == "hierarchical"
assert config.enable_caching is True
assert config.theme is None
assert config.history_enabled is False
assert config.history_backend == "git"
def test_config_with_history_enabled(self):
"""Test config with git history enabled."""
config = SpaceConfig(history_enabled=True, history_backend="git")
assert config.history_enabled is True
assert config.history_backend == "git"
def test_config_to_dict(self):
"""Test config serialization."""
config = SpaceConfig(theme="dark", enable_caching=False)
data = config.to_dict()
assert data["theme"] == "dark"
assert data["enable_caching"] is False
def test_config_from_dict(self):
"""Test config deserialization."""
data = {"default_variant": "flat", "history_enabled": True}
config = SpaceConfig.from_dict(data)
assert config.default_variant == "flat"
assert config.history_enabled is True
class TestSpaceDocument:
"""Tests for SpaceDocument dataclass."""
def test_default_document(self):
"""Test default document values."""
doc = SpaceDocument()
assert doc.id is not None
assert doc.space_path == ""
assert doc.order_index == 0
assert doc.metadata == {}
def test_document_with_values(self):
"""Test document with custom values."""
doc = SpaceDocument(
space_id="space-1",
document_id="doc-1",
space_path="/intro.md",
order_index=1,
content_hash="abc123"
)
assert doc.space_id == "space-1"
assert doc.space_path == "/intro.md"
assert doc.content_hash == "abc123"
def test_document_to_dict(self):
"""Test document serialization."""
doc = SpaceDocument(space_path="/test.md")
data = doc.to_dict()
assert data["space_path"] == "/test.md"
assert "id" in data
assert "added_at" in data
def test_document_from_dict(self):
"""Test document deserialization."""
data = {
"id": "doc-123",
"space_id": "space-1",
"space_path": "/api.md",
"order_index": 5
}
doc = SpaceDocument.from_dict(data)
assert doc.id == "doc-123"
assert doc.space_path == "/api.md"
assert doc.order_index == 5
class TestInformationSpace:
"""Tests for InformationSpace dataclass."""
def test_space_requires_name(self):
"""Test that space name is required."""
with pytest.raises(ValueError, match="Space name is required"):
InformationSpace(name="")
def test_space_default_values(self):
"""Test default space values."""
space = InformationSpace(name="test-space")
assert space.name == "test-space"
assert space.id is not None
assert space.status == SpaceStatus.DRAFT
assert space.description is None
assert space.parent_space_id is None
def test_space_with_config(self):
"""Test space with custom config."""
config = SpaceConfig(theme="minimal", history_enabled=True)
space = InformationSpace(
name="docs",
description="Documentation space",
config=config
)
assert space.config.theme == "minimal"
assert space.config.history_enabled is True
def test_space_activation(self):
"""Test space lifecycle transitions."""
space = InformationSpace(name="test")
assert space.status == SpaceStatus.DRAFT
space.activate()
assert space.status == SpaceStatus.ACTIVE
space.archive()
assert space.status == SpaceStatus.ARCHIVED
def test_space_touch_updates_timestamp(self):
"""Test that touch() updates the timestamp."""
space = InformationSpace(name="test")
original_updated = space.updated_at
import time
time.sleep(0.01) # Small delay to ensure timestamp changes
space.touch()
assert space.updated_at >= original_updated
def test_space_to_dict(self):
"""Test space serialization."""
space = InformationSpace(
name="api-docs",
description="API Documentation"
)
data = space.to_dict()
assert data["name"] == "api-docs"
assert data["description"] == "API Documentation"
assert data["status"] == "draft"
assert "id" in data
assert "created_at" in data
def test_space_from_dict(self):
"""Test space deserialization."""
data = {
"id": "space-123",
"name": "my-space",
"description": "Test space",
"status": "active",
"config": {"history_enabled": True},
"metadata": {"tags": ["test"]}
}
space = InformationSpace.from_dict(data)
assert space.id == "space-123"
assert space.name == "my-space"
assert space.status == SpaceStatus.ACTIVE
assert space.config.history_enabled is True
assert space.metadata.tags == ["test"]
def test_space_roundtrip_serialization(self):
"""Test that to_dict and from_dict are inverse operations."""
original = InformationSpace(
name="roundtrip-test",
description="Testing serialization",
config=SpaceConfig(theme="dark", history_enabled=True),
metadata=SpaceMetadata(tags=["test", "roundtrip"])
)
original.activate()
data = original.to_dict()
restored = InformationSpace.from_dict(data)
assert restored.name == original.name
assert restored.description == original.description
assert restored.status == original.status
assert restored.config.theme == original.config.theme
assert restored.metadata.tags == original.metadata.tags
class TestSpaceVariable:
"""Tests for SpaceVariable dataclass."""
def test_variable_creation(self):
"""Test variable creation."""
var = SpaceVariable(
space_id="space-1",
name="version",
value="1.0.0"
)
assert var.name == "version"
assert var.value == "1.0.0"
assert var.scope == "space"
def test_variable_to_dict(self):
"""Test variable serialization."""
var = SpaceVariable(
space_id="space-1",
name="config",
value={"key": "value"},
scope="document"
)
data = var.to_dict()
assert data["name"] == "config"
assert data["scope"] == "document"
class TestTransclusionReference:
"""Tests for TransclusionReference dataclass."""
def test_reference_creation(self):
"""Test transclusion reference creation."""
ref = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id="space-1"
)
assert ref.source_doc_id == "doc-1"
assert ref.target_doc_id == "doc-2"
assert ref.created_at is not None
def test_reference_to_dict(self):
"""Test reference serialization."""
ref = TransclusionReference(
source_doc_id="a",
target_doc_id="b",
space_id="s"
)
data = ref.to_dict()
assert "created_at" in data
assert data["source_doc_id"] == "a"
class TestSpaceStatus:
"""Tests for SpaceStatus enum."""
def test_status_values(self):
"""Test status enum values."""
assert SpaceStatus.DRAFT.value == "draft"
assert SpaceStatus.ACTIVE.value == "active"
assert SpaceStatus.ARCHIVED.value == "archived"
assert SpaceStatus.DELETED.value == "deleted"

View File

@@ -0,0 +1,901 @@
"""
Unit tests for space repositories.
Tests the SQLite implementations of:
- ISpaceRepository (SqliteSpaceRepository)
- IDocumentAssociationRepository (SqliteDocumentRepository)
- IVariableRepository (SqliteVariableRepository)
- IReferenceRepository (SqliteReferenceRepository)
"""
import pytest
import tempfile
import os
from datetime import datetime
from markitect.spaces.models import (
InformationSpace,
SpaceDocument,
SpaceVariable,
TransclusionReference,
SpaceStatus,
SpaceConfig,
SpaceMetadata,
)
from markitect.spaces.repositories.sqlite import (
SqliteSpaceRepository,
SqliteDocumentRepository,
SqliteVariableRepository,
SqliteReferenceRepository,
initialize_space_tables,
)
@pytest.fixture
def temp_db():
"""Create a temporary database file for testing."""
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
yield path
if os.path.exists(path):
os.unlink(path)
@pytest.fixture
def space_repo(temp_db):
"""Create a SqliteSpaceRepository for testing."""
return SqliteSpaceRepository(temp_db)
@pytest.fixture
def doc_repo(temp_db):
"""Create a SqliteDocumentRepository for testing."""
return SqliteDocumentRepository(temp_db)
@pytest.fixture
def var_repo(temp_db):
"""Create a SqliteVariableRepository for testing."""
return SqliteVariableRepository(temp_db)
@pytest.fixture
def ref_repo(temp_db):
"""Create a SqliteReferenceRepository for testing."""
return SqliteReferenceRepository(temp_db)
class TestInitializeSpaceTables:
"""Tests for initialize_space_tables function."""
def test_creates_tables(self, temp_db):
"""Test that initialize_space_tables creates all required tables."""
import sqlite3
initialize_space_tables(temp_db)
conn = sqlite3.connect(temp_db)
cursor = conn.cursor()
# Check that all tables exist
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = {row[0] for row in cursor.fetchall()}
assert "spaces" in tables
assert "space_documents" in tables
assert "space_variables" in tables
assert "transclusion_references" in tables
conn.close()
def test_idempotent(self, temp_db):
"""Test that initialize_space_tables can be called multiple times."""
initialize_space_tables(temp_db)
initialize_space_tables(temp_db) # Should not raise
def test_creates_parent_directory(self):
"""Test that initialize_space_tables creates parent directories."""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = os.path.join(tmpdir, "subdir", "nested", "test.db")
initialize_space_tables(db_path)
assert os.path.exists(db_path)
class TestSqliteSpaceRepository:
"""Tests for SqliteSpaceRepository."""
def test_create_space(self, space_repo):
"""Test creating a new space."""
space = InformationSpace(name="test-space", description="A test space")
created = space_repo.create(space)
assert created.id == space.id
assert created.name == "test-space"
assert created.description == "A test space"
def test_create_space_duplicate_name_raises(self, space_repo):
"""Test that creating a space with duplicate name raises ValueError."""
space1 = InformationSpace(name="duplicate")
space_repo.create(space1)
space2 = InformationSpace(name="duplicate")
with pytest.raises(ValueError, match="already exists"):
space_repo.create(space2)
def test_get_by_id(self, space_repo):
"""Test retrieving a space by ID."""
space = InformationSpace(name="test-space")
space_repo.create(space)
retrieved = space_repo.get_by_id(space.id)
assert retrieved is not None
assert retrieved.id == space.id
assert retrieved.name == "test-space"
def test_get_by_id_not_found(self, space_repo):
"""Test that get_by_id returns None for non-existent space."""
result = space_repo.get_by_id("non-existent-id")
assert result is None
def test_get_by_name(self, space_repo):
"""Test retrieving a space by name."""
space = InformationSpace(name="named-space")
space_repo.create(space)
retrieved = space_repo.get_by_name("named-space")
assert retrieved is not None
assert retrieved.name == "named-space"
def test_get_by_name_not_found(self, space_repo):
"""Test that get_by_name returns None for non-existent space."""
result = space_repo.get_by_name("non-existent")
assert result is None
def test_list_all_empty(self, space_repo):
"""Test listing spaces when none exist."""
spaces = space_repo.list_all()
assert spaces == []
def test_list_all(self, space_repo):
"""Test listing all spaces."""
space1 = InformationSpace(name="alpha")
space2 = InformationSpace(name="beta")
space_repo.create(space1)
space_repo.create(space2)
spaces = space_repo.list_all()
assert len(spaces) == 2
names = [s.name for s in spaces]
assert "alpha" in names
assert "beta" in names
def test_list_all_excludes_archived_by_default(self, space_repo):
"""Test that list_all excludes archived spaces by default."""
space1 = InformationSpace(name="active-space")
space2 = InformationSpace(name="archived-space")
space2.archive()
space_repo.create(space1)
space_repo.create(space2)
spaces = space_repo.list_all()
assert len(spaces) == 1
assert spaces[0].name == "active-space"
def test_list_all_includes_archived_when_requested(self, space_repo):
"""Test that list_all includes archived spaces when requested."""
space1 = InformationSpace(name="active-space")
space2 = InformationSpace(name="archived-space")
space2.archive()
space_repo.create(space1)
space_repo.create(space2)
spaces = space_repo.list_all(include_archived=True)
assert len(spaces) == 2
def test_update_space(self, space_repo):
"""Test updating a space."""
space = InformationSpace(name="original")
space_repo.create(space)
space.description = "Updated description"
updated = space_repo.update(space)
assert updated.description == "Updated description"
# Verify persisted
retrieved = space_repo.get_by_id(space.id)
assert retrieved.description == "Updated description"
def test_update_nonexistent_raises(self, space_repo):
"""Test that updating a non-existent space raises ValueError."""
space = InformationSpace(name="non-existent")
with pytest.raises(ValueError, match="does not exist"):
space_repo.update(space)
def test_delete_space(self, space_repo):
"""Test deleting a space."""
space = InformationSpace(name="to-delete")
space_repo.create(space)
result = space_repo.delete(space.id)
assert result is True
# Verify deleted
retrieved = space_repo.get_by_id(space.id)
assert retrieved is None
def test_delete_nonexistent(self, space_repo):
"""Test that deleting a non-existent space returns False."""
result = space_repo.delete("non-existent-id")
assert result is False
def test_exists(self, space_repo):
"""Test checking if a space exists."""
space = InformationSpace(name="existing")
space_repo.create(space)
assert space_repo.exists(space.id) is True
assert space_repo.exists("non-existent") is False
def test_get_children(self, space_repo):
"""Test getting child spaces."""
parent = InformationSpace(name="parent")
space_repo.create(parent)
child1 = InformationSpace(name="child1", parent_space_id=parent.id)
child2 = InformationSpace(name="child2", parent_space_id=parent.id)
space_repo.create(child1)
space_repo.create(child2)
children = space_repo.get_children(parent.id)
assert len(children) == 2
names = [c.name for c in children]
assert "child1" in names
assert "child2" in names
def test_get_children_empty(self, space_repo):
"""Test getting children when none exist."""
parent = InformationSpace(name="lonely-parent")
space_repo.create(parent)
children = space_repo.get_children(parent.id)
assert children == []
def test_space_with_config_and_metadata(self, space_repo):
"""Test creating and retrieving a space with config and metadata."""
config = SpaceConfig(theme="dark", history_enabled=True)
metadata = SpaceMetadata(tags=["api", "docs"], author="tester")
space = InformationSpace(
name="configured-space",
config=config,
metadata=metadata,
)
space_repo.create(space)
retrieved = space_repo.get_by_id(space.id)
assert retrieved.config.theme == "dark"
assert retrieved.config.history_enabled is True
assert retrieved.metadata.tags == ["api", "docs"]
assert retrieved.metadata.author == "tester"
def test_space_status_persistence(self, space_repo):
"""Test that space status is persisted correctly."""
space = InformationSpace(name="lifecycle-test")
space.activate()
space_repo.create(space)
retrieved = space_repo.get_by_id(space.id)
assert retrieved.status == SpaceStatus.ACTIVE
class TestSqliteDocumentRepository:
"""Tests for SqliteDocumentRepository."""
def test_add_document(self, doc_repo, space_repo):
"""Test adding a document to a space."""
# First create a space
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(
space_id=space.id,
document_id="doc-123",
space_path="/intro.md",
)
added = doc_repo.add_document(doc)
assert added.id == doc.id
assert added.space_path == "/intro.md"
def test_add_document_duplicate_path_raises(self, doc_repo, space_repo):
"""Test that adding a document with duplicate path raises ValueError."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc1 = SpaceDocument(space_id=space.id, space_path="/same.md")
doc_repo.add_document(doc1)
doc2 = SpaceDocument(space_id=space.id, space_path="/same.md")
with pytest.raises(ValueError, match="already exists"):
doc_repo.add_document(doc2)
def test_get_document(self, doc_repo, space_repo):
"""Test getting a document by ID."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/test.md")
doc_repo.add_document(doc)
retrieved = doc_repo.get_document(doc.id)
assert retrieved is not None
assert retrieved.space_path == "/test.md"
def test_get_document_not_found(self, doc_repo):
"""Test that get_document returns None for non-existent document."""
result = doc_repo.get_document("non-existent")
assert result is None
def test_get_by_space_path(self, doc_repo, space_repo):
"""Test getting a document by space path."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/api/docs.md")
doc_repo.add_document(doc)
retrieved = doc_repo.get_by_space_path(space.id, "/api/docs.md")
assert retrieved is not None
assert retrieved.id == doc.id
def test_get_by_space_path_not_found(self, doc_repo, space_repo):
"""Test that get_by_space_path returns None for non-existent path."""
space = InformationSpace(name="test-space")
space_repo.create(space)
result = doc_repo.get_by_space_path(space.id, "/non-existent.md")
assert result is None
def test_list_by_space(self, doc_repo, space_repo):
"""Test listing documents in a space."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc1 = SpaceDocument(space_id=space.id, space_path="/first.md", order_index=0)
doc2 = SpaceDocument(space_id=space.id, space_path="/second.md", order_index=1)
doc_repo.add_document(doc1)
doc_repo.add_document(doc2)
docs = doc_repo.list_by_space(space.id)
assert len(docs) == 2
assert docs[0].space_path == "/first.md"
assert docs[1].space_path == "/second.md"
def test_list_by_space_empty(self, doc_repo, space_repo):
"""Test listing documents when none exist."""
space = InformationSpace(name="empty-space")
space_repo.create(space)
docs = doc_repo.list_by_space(space.id)
assert docs == []
def test_update_document(self, doc_repo, space_repo):
"""Test updating a document."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/old.md")
doc_repo.add_document(doc)
doc.content_hash = "newhash123"
updated = doc_repo.update_document(doc)
assert updated.content_hash == "newhash123"
# Verify persisted
retrieved = doc_repo.get_document(doc.id)
assert retrieved.content_hash == "newhash123"
def test_update_nonexistent_raises(self, doc_repo):
"""Test that updating a non-existent document raises ValueError."""
doc = SpaceDocument(space_path="/non-existent.md")
with pytest.raises(ValueError, match="does not exist"):
doc_repo.update_document(doc)
def test_remove_document(self, doc_repo, space_repo):
"""Test removing a document."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/to-remove.md")
doc_repo.add_document(doc)
result = doc_repo.remove_document(doc.id)
assert result is True
# Verify removed
retrieved = doc_repo.get_document(doc.id)
assert retrieved is None
def test_remove_nonexistent(self, doc_repo):
"""Test that removing a non-existent document returns False."""
result = doc_repo.remove_document("non-existent")
assert result is False
def test_move_document(self, doc_repo, space_repo):
"""Test moving a document to a new path."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/old-path.md")
doc_repo.add_document(doc)
moved = doc_repo.move_document(doc.id, "/new-path.md")
assert moved.space_path == "/new-path.md"
# Verify old path no longer works
old_result = doc_repo.get_by_space_path(space.id, "/old-path.md")
assert old_result is None
# Verify new path works
new_result = doc_repo.get_by_space_path(space.id, "/new-path.md")
assert new_result is not None
def test_move_document_to_existing_path_raises(self, doc_repo, space_repo):
"""Test that moving to an existing path raises ValueError."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc1 = SpaceDocument(space_id=space.id, space_path="/first.md")
doc2 = SpaceDocument(space_id=space.id, space_path="/second.md")
doc_repo.add_document(doc1)
doc_repo.add_document(doc2)
with pytest.raises(ValueError, match="already exists"):
doc_repo.move_document(doc1.id, "/second.md")
def test_move_nonexistent_raises(self, doc_repo):
"""Test that moving a non-existent document raises ValueError."""
with pytest.raises(ValueError, match="does not exist"):
doc_repo.move_document("non-existent", "/new-path.md")
def test_reorder_documents(self, doc_repo, space_repo):
"""Test reordering documents within a space."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc1 = SpaceDocument(space_id=space.id, space_path="/a.md", order_index=0)
doc2 = SpaceDocument(space_id=space.id, space_path="/b.md", order_index=1)
doc3 = SpaceDocument(space_id=space.id, space_path="/c.md", order_index=2)
doc_repo.add_document(doc1)
doc_repo.add_document(doc2)
doc_repo.add_document(doc3)
# Reorder: c, a, b
doc_repo.reorder_documents(space.id, [doc3.id, doc1.id, doc2.id])
docs = doc_repo.list_by_space(space.id)
assert docs[0].id == doc3.id
assert docs[1].id == doc1.id
assert docs[2].id == doc2.id
def test_update_content_hash(self, doc_repo, space_repo):
"""Test updating content hash."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/test.md")
doc_repo.add_document(doc)
doc_repo.update_content_hash(doc.id, "newhash456")
retrieved = doc_repo.get_document(doc.id)
assert retrieved.content_hash == "newhash456"
def test_document_with_metadata(self, doc_repo, space_repo):
"""Test document with custom metadata."""
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(
space_id=space.id,
space_path="/with-meta.md",
metadata={"title": "Test Document", "version": "1.0"},
)
doc_repo.add_document(doc)
retrieved = doc_repo.get_document(doc.id)
assert retrieved.metadata["title"] == "Test Document"
assert retrieved.metadata["version"] == "1.0"
class TestSqliteVariableRepository:
"""Tests for SqliteVariableRepository."""
def test_set_variable(self, var_repo, space_repo):
"""Test setting a variable."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var = SpaceVariable(
space_id=space.id,
name="version",
value="1.0.0",
)
result = var_repo.set_variable(var)
assert result.name == "version"
assert result.value == "1.0.0"
def test_set_variable_overwrites(self, var_repo, space_repo):
"""Test that setting a variable with same name overwrites."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var1 = SpaceVariable(space_id=space.id, name="config", value="old")
var_repo.set_variable(var1)
var2 = SpaceVariable(space_id=space.id, name="config", value="new")
var_repo.set_variable(var2)
retrieved = var_repo.get_variable(space.id, "config")
assert retrieved.value == "new"
def test_get_variable(self, var_repo, space_repo):
"""Test getting a variable."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var = SpaceVariable(space_id=space.id, name="api_key", value="secret123")
var_repo.set_variable(var)
retrieved = var_repo.get_variable(space.id, "api_key")
assert retrieved is not None
assert retrieved.value == "secret123"
def test_get_variable_not_found(self, var_repo, space_repo):
"""Test that get_variable returns None for non-existent variable."""
space = InformationSpace(name="test-space")
space_repo.create(space)
result = var_repo.get_variable(space.id, "non-existent")
assert result is None
def test_list_variables(self, var_repo, space_repo):
"""Test listing variables in a space."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var1 = SpaceVariable(space_id=space.id, name="var1", value="a")
var2 = SpaceVariable(space_id=space.id, name="var2", value="b")
var_repo.set_variable(var1)
var_repo.set_variable(var2)
variables = var_repo.list_variables(space.id)
assert len(variables) == 2
names = [v.name for v in variables]
assert "var1" in names
assert "var2" in names
def test_list_variables_empty(self, var_repo, space_repo):
"""Test listing variables when none exist."""
space = InformationSpace(name="empty-space")
space_repo.create(space)
variables = var_repo.list_variables(space.id)
assert variables == []
def test_list_variables_with_scope_filter(self, var_repo, space_repo):
"""Test listing variables filtered by scope."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var1 = SpaceVariable(space_id=space.id, name="global", value="x", scope="space")
var2 = SpaceVariable(space_id=space.id, name="local", value="y", scope="document")
var_repo.set_variable(var1)
var_repo.set_variable(var2)
space_vars = var_repo.list_variables(space.id, scope="space")
assert len(space_vars) == 1
assert space_vars[0].name == "global"
doc_vars = var_repo.list_variables(space.id, scope="document")
assert len(doc_vars) == 1
assert doc_vars[0].name == "local"
def test_delete_variable(self, var_repo, space_repo):
"""Test deleting a variable."""
space = InformationSpace(name="test-space")
space_repo.create(space)
var = SpaceVariable(space_id=space.id, name="to-delete", value="bye")
var_repo.set_variable(var)
result = var_repo.delete_variable(space.id, "to-delete")
assert result is True
# Verify deleted
retrieved = var_repo.get_variable(space.id, "to-delete")
assert retrieved is None
def test_delete_nonexistent(self, var_repo, space_repo):
"""Test that deleting a non-existent variable returns False."""
space = InformationSpace(name="test-space")
space_repo.create(space)
result = var_repo.delete_variable(space.id, "non-existent")
assert result is False
def test_variable_with_complex_value(self, var_repo, space_repo):
"""Test variable with complex JSON value."""
space = InformationSpace(name="test-space")
space_repo.create(space)
complex_value = {
"endpoints": [
{"url": "/api/v1", "methods": ["GET", "POST"]},
{"url": "/api/v2", "methods": ["GET"]},
],
"config": {"timeout": 30, "retries": 3},
}
var = SpaceVariable(space_id=space.id, name="api_config", value=complex_value)
var_repo.set_variable(var)
retrieved = var_repo.get_variable(space.id, "api_config")
assert retrieved.value == complex_value
assert retrieved.value["endpoints"][0]["url"] == "/api/v1"
class TestSqliteReferenceRepository:
"""Tests for SqliteReferenceRepository."""
def test_add_reference(self, ref_repo, space_repo):
"""Test adding a transclusion reference."""
space = InformationSpace(name="test-space")
space_repo.create(space)
ref = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
result = ref_repo.add_reference(ref)
assert result.source_doc_id == "doc-1"
assert result.target_doc_id == "doc-2"
def test_add_reference_overwrites(self, ref_repo, space_repo):
"""Test that adding same reference overwrites (no duplicates)."""
space = InformationSpace(name="test-space")
space_repo.create(space)
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
ref_repo.add_reference(ref1)
# Add same reference again (should not raise)
ref2 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
ref_repo.add_reference(ref2)
# Should still only have one reference
refs = ref_repo.get_references_from("doc-1", space.id)
assert len(refs) == 1
def test_get_references_from(self, ref_repo, space_repo):
"""Test getting references from a source document."""
space = InformationSpace(name="test-space")
space_repo.create(space)
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
ref2 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-3",
space_id=space.id,
)
ref_repo.add_reference(ref1)
ref_repo.add_reference(ref2)
refs = ref_repo.get_references_from("doc-1", space.id)
assert len(refs) == 2
targets = [r.target_doc_id for r in refs]
assert "doc-2" in targets
assert "doc-3" in targets
def test_get_references_from_empty(self, ref_repo, space_repo):
"""Test getting references when none exist."""
space = InformationSpace(name="test-space")
space_repo.create(space)
refs = ref_repo.get_references_from("non-existent", space.id)
assert refs == []
def test_get_references_to(self, ref_repo, space_repo):
"""Test getting references to a target document."""
space = InformationSpace(name="test-space")
space_repo.create(space)
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="shared-doc",
space_id=space.id,
)
ref2 = TransclusionReference(
source_doc_id="doc-2",
target_doc_id="shared-doc",
space_id=space.id,
)
ref_repo.add_reference(ref1)
ref_repo.add_reference(ref2)
refs = ref_repo.get_references_to("shared-doc", space.id)
assert len(refs) == 2
sources = [r.source_doc_id for r in refs]
assert "doc-1" in sources
assert "doc-2" in sources
def test_clear_references_from(self, ref_repo, space_repo):
"""Test clearing references from a source document."""
space = InformationSpace(name="test-space")
space_repo.create(space)
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
ref2 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-3",
space_id=space.id,
)
ref_repo.add_reference(ref1)
ref_repo.add_reference(ref2)
count = ref_repo.clear_references_from("doc-1", space.id)
assert count == 2
# Verify cleared
refs = ref_repo.get_references_from("doc-1", space.id)
assert refs == []
def test_clear_references_from_empty(self, ref_repo, space_repo):
"""Test clearing references when none exist."""
space = InformationSpace(name="test-space")
space_repo.create(space)
count = ref_repo.clear_references_from("non-existent", space.id)
assert count == 0
def test_get_dependents(self, ref_repo, space_repo):
"""Test getting dependent documents."""
space = InformationSpace(name="test-space")
space_repo.create(space)
# doc-1 and doc-2 both reference shared-component
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="shared-component",
space_id=space.id,
)
ref2 = TransclusionReference(
source_doc_id="doc-2",
target_doc_id="shared-component",
space_id=space.id,
)
ref_repo.add_reference(ref1)
ref_repo.add_reference(ref2)
dependents = ref_repo.get_dependents("shared-component", space.id)
assert len(dependents) == 2
assert "doc-1" in dependents
assert "doc-2" in dependents
def test_get_dependents_empty(self, ref_repo, space_repo):
"""Test getting dependents when none exist."""
space = InformationSpace(name="test-space")
space_repo.create(space)
dependents = ref_repo.get_dependents("orphan-doc", space.id)
assert dependents == []
def test_references_isolated_by_space(self, ref_repo, space_repo):
"""Test that references are isolated by space."""
space1 = InformationSpace(name="space-1")
space2 = InformationSpace(name="space-2")
space_repo.create(space1)
space_repo.create(space2)
# Same source/target IDs in different spaces
ref1 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space1.id,
)
ref2 = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space2.id,
)
ref_repo.add_reference(ref1)
ref_repo.add_reference(ref2)
# Each space should have its own reference
refs1 = ref_repo.get_references_from("doc-1", space1.id)
refs2 = ref_repo.get_references_from("doc-1", space2.id)
assert len(refs1) == 1
assert len(refs2) == 1
assert refs1[0].space_id == space1.id
assert refs2[0].space_id == space2.id
class TestCascadeDelete:
"""Test cascade delete behavior."""
def test_deleting_space_cascades_to_documents(self, temp_db):
"""Test that deleting a space also deletes its documents."""
space_repo = SqliteSpaceRepository(temp_db)
doc_repo = SqliteDocumentRepository(temp_db)
space = InformationSpace(name="test-space")
space_repo.create(space)
doc = SpaceDocument(space_id=space.id, space_path="/test.md")
doc_repo.add_document(doc)
# Delete space
space_repo.delete(space.id)
# Document should also be gone
retrieved = doc_repo.get_document(doc.id)
assert retrieved is None
def test_deleting_space_cascades_to_variables(self, temp_db):
"""Test that deleting a space also deletes its variables."""
space_repo = SqliteSpaceRepository(temp_db)
var_repo = SqliteVariableRepository(temp_db)
space = InformationSpace(name="test-space")
space_repo.create(space)
var = SpaceVariable(space_id=space.id, name="var", value="val")
var_repo.set_variable(var)
# Delete space
space_repo.delete(space.id)
# Variable should also be gone
retrieved = var_repo.get_variable(space.id, "var")
assert retrieved is None
def test_deleting_space_cascades_to_references(self, temp_db):
"""Test that deleting a space also deletes its references."""
space_repo = SqliteSpaceRepository(temp_db)
ref_repo = SqliteReferenceRepository(temp_db)
space = InformationSpace(name="test-space")
space_repo.create(space)
ref = TransclusionReference(
source_doc_id="doc-1",
target_doc_id="doc-2",
space_id=space.id,
)
ref_repo.add_reference(ref)
# Delete space
space_repo.delete(space.id)
# Reference should also be gone
refs = ref_repo.get_references_from("doc-1", space.id)
assert refs == []