coulomb/markitect-main
2
0
Fork 0
Code Issues 60 Pull Requests Actions 1 Projects 12 Releases 1 Wiki Activity

feat(prompts): implement Phase 5 - Dependency Tracking (FR-6)

Browse Source 
Add directed dependency graph with cycle detection, topological sort,
and query service for finding dependents/dependencies transitively.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-02-09 13:18:18 +01:00
parent c56c92c815
commit 9ce157400e
13 changed files with 3021 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
markitect/prompts/dependencies/__init__.py Normal file
View File

@@ -0,0 +1,40 @@
"""
Dependency tracking for prompt artifacts.
Implements FR-6: Dependency Tracking
- FR-6.1: Directed dependency edges between artifacts
- FR-6.2: Cross-space dependency graph
- FR-6.3: Circular dependency detection
"""
from markitect.prompts.dependencies.models import (
EdgeType,
DependencyEdge,
DependencyGraph,
CircularDependencyError,
)
from markitect.prompts.dependencies.repository import (
IDependencyRepository,
SQLiteDependencyRepository,
DependencyRepositoryError,
DuplicateDependencyError,
)
from markitect.prompts.dependencies.graph import GraphBuilder
from markitect.prompts.dependencies.queries import DependencyQueryService
__all__ = [
# Models
"EdgeType",
"DependencyEdge",
"DependencyGraph",
"CircularDependencyError",
# Repository
"IDependencyRepository",
"SQLiteDependencyRepository",
"DependencyRepositoryError",
"DuplicateDependencyError",
# Graph
"GraphBuilder",
# Queries
"DependencyQueryService",
]

145
markitect/prompts/dependencies/graph.py Normal file
View File

@@ -0,0 +1,145 @@
"""
Graph builder for bridging ephemeral manifest edges to persistent storage.
Bridges RunManifest ephemeral DependencyEdge instances (3-field) to
persistent DependencyEdge storage (6-field), and builds DependencyGraph
from repository data.
"""
from typing import List, Optional, Set
from markitect.prompts.dependencies.models import (
DependencyEdge,
DependencyGraph,
EdgeType,
)
from markitect.prompts.dependencies.repository import IDependencyRepository
from markitect.prompts.execution.manifest import (
DependencyEdge as ManifestDependencyEdge,
RunManifest,
)
class GraphBuilder:
"""
Bridges ephemeral manifest edges to persistent dependency storage.
Extracts dependency edges from RunManifest instances, persists them
via IDependencyRepository, and builds in-memory DependencyGraph
instances for analysis.
"""
def __init__(self, repository: IDependencyRepository):
"""
Initialize with dependency repository.
Args:
repository: Repository for persisting dependency edges
"""
self._repository = repository
def extract_edges(self, manifest: RunManifest) -> List[DependencyEdge]:
"""
Extract persistent DependencyEdge instances from a RunManifest.
Converts the ephemeral 3-field DependencyEdge from the manifest
into persistent 6-field DependencyEdge instances.
Args:
manifest: RunManifest containing ephemeral edges
Returns:
List of persistent DependencyEdge instances
"""
edges = []
for manifest_edge in manifest.dependency_edges:
edge_type = EdgeType(manifest_edge.edge_type)
edge = DependencyEdge.create(
source_artifact_id=manifest_edge.source_id,
target_artifact_id=manifest_edge.target_id,
run_id=manifest.run_id,
edge_type=edge_type,
)
edges.append(edge)
return edges
def persist_edges(self, manifest: RunManifest) -> List[DependencyEdge]:
"""
Extract edges from manifest and persist them to the repository.
Args:
manifest: RunManifest containing ephemeral edges
Returns:
List of persisted DependencyEdge instances
"""
edges = self.extract_edges(manifest)
persisted = []
for edge in edges:
created = self._repository.create(edge)
persisted.append(created)
return persisted
def build_graph(
self,
artifact_ids: Optional[Set[str]] = None,
) -> DependencyGraph:
"""
Build an in-memory DependencyGraph from repository data.
Args:
artifact_ids: Optional set of artifact IDs to scope the graph.
If None, builds graph from all edges.
Returns:
DependencyGraph instance
"""
graph = DependencyGraph()
if artifact_ids is not None:
# Build scoped subgraph
for artifact_id in artifact_ids:
edges = self._repository.get_by_source(artifact_id)
for edge in edges:
if edge.target_artifact_id in artifact_ids:
graph.add_edge(
edge.source_artifact_id,
edge.target_artifact_id,
edge.edge_type,
)
# Ensure node exists even if it has no edges
if artifact_id not in graph._forward:
graph._forward[artifact_id] = set()
if artifact_id not in graph._reverse:
graph._reverse[artifact_id] = set()
else:
# Build full graph
all_edges = self._repository.get_all()
for edge in all_edges:
graph.add_edge(
edge.source_artifact_id,
edge.target_artifact_id,
edge.edge_type,
)
return graph
def build_graph_for_run(self, run_id: str) -> DependencyGraph:
"""
Build a DependencyGraph from edges of a specific run.
Args:
run_id: Run identifier
Returns:
DependencyGraph for the run's edges
"""
graph = DependencyGraph()
edges = self._repository.get_by_run(run_id)
for edge in edges:
graph.add_edge(
edge.source_artifact_id,
edge.target_artifact_id,
edge.edge_type,
)
return graph

308
markitect/prompts/dependencies/models.py Normal file
View File

@@ -0,0 +1,308 @@
"""
Dependency tracking models for prompt artifact graphs.
Implements FR-6: Dependency Tracking
- FR-6.1: Directed dependency edges between artifacts
- FR-6.2: Cross-space dependency graph
- FR-6.3: Circular dependency detection
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, Any, List, Optional, Set, Tuple
import uuid
class EdgeType(Enum):
"""Type of dependency relationship between artifacts."""
REQUIRES = "requires"
GENERATES = "generates"
INCLUDES = "includes"
class CircularDependencyError(Exception):
"""
Raised when a circular dependency is detected in the graph.
Attributes:
cycle: List of node IDs forming the cycle
"""
def __init__(self, message: str, cycle: Optional[List[str]] = None):
super().__init__(message)
self.cycle = cycle or []
@dataclass
class DependencyEdge:
"""
Persistent dependency edge between artifacts.
Extends the ephemeral DependencyEdge from execution/manifest.py
with persistence fields (id, run_id, created_at).
Attributes:
id: Unique edge identifier
source_artifact_id: Source artifact ID
target_artifact_id: Target artifact ID
run_id: ID of the run that created this edge
edge_type: Type of dependency relationship
created_at: When this edge was created
"""
id: str
source_artifact_id: str
target_artifact_id: str
run_id: str
edge_type: EdgeType
created_at: datetime = field(default_factory=datetime.utcnow)
@classmethod
def create(
cls,
source_artifact_id: str,
target_artifact_id: str,
run_id: str,
edge_type: EdgeType,
) -> "DependencyEdge":
"""
Create a new dependency edge.
Args:
source_artifact_id: Source artifact ID
target_artifact_id: Target artifact ID
run_id: Run that established this dependency
edge_type: Type of dependency
Returns:
New DependencyEdge instance
"""
return cls(
id=str(uuid.uuid4()),
source_artifact_id=source_artifact_id,
target_artifact_id=target_artifact_id,
run_id=run_id,
edge_type=edge_type,
)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"id": self.id,
"source_artifact_id": self.source_artifact_id,
"target_artifact_id": self.target_artifact_id,
"run_id": self.run_id,
"edge_type": self.edge_type.value,
"created_at": self.created_at.isoformat(),
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "DependencyEdge":
"""Create from dictionary."""
return cls(
id=data["id"],
source_artifact_id=data["source_artifact_id"],
target_artifact_id=data["target_artifact_id"],
run_id=data["run_id"],
edge_type=EdgeType(data["edge_type"]),
created_at=datetime.fromisoformat(data["created_at"]),
)
# DFS node coloring for cycle detection
_WHITE = 0 # Unvisited
_GRAY = 1 # In current DFS path
_BLACK = 2 # Fully processed
class DependencyGraph:
"""
In-memory directed graph for dependency analysis.
Maintains forward and reverse adjacency lists for efficient
traversal in both directions. Supports cycle detection via
3-color DFS and topological sorting.
Implements FR-6.2: Cross-space dependency graph
Implements FR-6.3: Circular dependency detection
"""
def __init__(self) -> None:
self._forward: Dict[str, Set[str]] = {}
self._reverse: Dict[str, Set[str]] = {}
self._edge_types: Dict[Tuple[str, str], EdgeType] = {}
@property
def nodes(self) -> Set[str]:
"""All nodes in the graph."""
return set(self._forward.keys()) | set(self._reverse.keys())
@property
def edge_count(self) -> int:
"""Total number of edges."""
return sum(len(targets) for targets in self._forward.values())
def add_edge(
self,
source: str,
target: str,
edge_type: EdgeType = EdgeType.REQUIRES,
) -> None:
"""
Add a directed edge from source to target.
Args:
source: Source node ID
target: Target node ID
edge_type: Type of dependency
"""
if source not in self._forward:
self._forward[source] = set()
if target not in self._forward:
self._forward[target] = set()
if source not in self._reverse:
self._reverse[source] = set()
if target not in self._reverse:
self._reverse[target] = set()
self._forward[source].add(target)
self._reverse[target].add(source)
self._edge_types[(source, target)] = edge_type
def get_successors(self, node: str) -> Set[str]:
"""Get direct successors (dependencies) of a node."""
return set(self._forward.get(node, set()))
def get_predecessors(self, node: str) -> Set[str]:
"""Get direct predecessors (dependents) of a node."""
return set(self._reverse.get(node, set()))
def get_edge_type(self, source: str, target: str) -> Optional[EdgeType]:
"""Get the edge type between two nodes."""
return self._edge_types.get((source, target))
def has_edge(self, source: str, target: str) -> bool:
"""Check if an edge exists between source and target."""
return target in self._forward.get(source, set())
def detect_cycles(self) -> List[List[str]]:
"""
Detect all cycles in the graph using 3-color DFS.
Returns:
List of cycles, where each cycle is a list of node IDs
"""
color: Dict[str, int] = {node: _WHITE for node in self.nodes}
parent: Dict[str, Optional[str]] = {node: None for node in self.nodes}
cycles: List[List[str]] = []
def _dfs(node: str) -> None:
color[node] = _GRAY
for neighbor in self._forward.get(node, set()):
if color.get(neighbor, _WHITE) == _GRAY:
# Back edge found - extract cycle
cycle = [neighbor]
current = node
while current != neighbor:
cycle.append(current)
current = parent[current]
cycle.append(neighbor)
cycle.reverse()
cycles.append(cycle)
elif color.get(neighbor, _WHITE) == _WHITE:
parent[neighbor] = node
_dfs(neighbor)
color[node] = _BLACK
for node in self.nodes:
if color.get(node, _WHITE) == _WHITE:
_dfs(node)
return cycles
def has_cycle(self) -> bool:
"""Check if the graph contains any cycles."""
color: Dict[str, int] = {node: _WHITE for node in self.nodes}
def _dfs(node: str) -> bool:
color[node] = _GRAY
for neighbor in self._forward.get(node, set()):
if color.get(neighbor, _WHITE) == _GRAY:
return True
if color.get(neighbor, _WHITE) == _WHITE and _dfs(neighbor):
return True
color[node] = _BLACK
return False
for node in self.nodes:
if color.get(node, _WHITE) == _WHITE:
if _dfs(node):
return True
return False
def topological_sort(self) -> List[str]:
"""
Topologically sort the graph nodes.
Returns:
Nodes in topological order (dependencies before dependents)
Raises:
CircularDependencyError: If graph contains cycles
"""
color: Dict[str, int] = {node: _WHITE for node in self.nodes}
result: List[str] = []
cycle_path: List[str] = []
def _dfs(node: str) -> bool:
color[node] = _GRAY
cycle_path.append(node)
for neighbor in sorted(self._forward.get(node, set())):
if color.get(neighbor, _WHITE) == _GRAY:
# Extract cycle for error reporting
idx = cycle_path.index(neighbor)
cycle = cycle_path[idx:] + [neighbor]
raise CircularDependencyError(
f"Circular dependency detected: {' -> '.join(cycle)}",
cycle=cycle,
)
if color.get(neighbor, _WHITE) == _WHITE:
_dfs(neighbor)
cycle_path.pop()
color[node] = _BLACK
result.append(node)
for node in sorted(self.nodes):
if color.get(node, _WHITE) == _WHITE:
_dfs(node)
result.reverse()
return result
def get_subgraph(self, node_ids: Set[str]) -> "DependencyGraph":
"""
Extract a subgraph containing only the specified nodes.
Args:
node_ids: Set of node IDs to include
Returns:
New DependencyGraph with only the specified nodes and their edges
"""
subgraph = DependencyGraph()
for source in node_ids:
for target in self._forward.get(source, set()):
if target in node_ids:
edge_type = self._edge_types.get(
(source, target), EdgeType.REQUIRES
)
subgraph.add_edge(source, target, edge_type)
# Ensure isolated nodes are present
for node_id in node_ids:
if node_id not in subgraph._forward:
subgraph._forward[node_id] = set()
if node_id not in subgraph._reverse:
subgraph._reverse[node_id] = set()
return subgraph

221
markitect/prompts/dependencies/queries.py Normal file
View File

@@ -0,0 +1,221 @@
"""
Dependency query service for analyzing artifact dependency graphs.
Provides operations for finding dependents, dependencies, transitive
closures, dependency chains, cycle detection, and build ordering.
"""
from collections import deque
from typing import List, Optional, Set
from markitect.prompts.dependencies.models import (
CircularDependencyError,
DependencyGraph,
EdgeType,
)
from markitect.prompts.dependencies.repository import IDependencyRepository
from markitect.prompts.dependencies.graph import GraphBuilder
class DependencyQueryService:
"""
Service for querying dependency relationships between artifacts.
Provides direct and transitive dependency lookups, dependency chain
computation via BFS, cycle detection, pre-validation, and build
order (topological sort).
"""
def __init__(self, repository: IDependencyRepository):
"""
Initialize with dependency repository.
Args:
repository: Repository for reading dependency edges
"""
self._repository = repository
self._graph_builder = GraphBuilder(repository)
def find_dependents(self, artifact_id: str) -> Set[str]:
"""
Find direct dependents of an artifact (who depends on it).
Args:
artifact_id: Artifact to find dependents of
Returns:
Set of artifact IDs that directly depend on this artifact
"""
edges = self._repository.get_by_target(artifact_id)
return {edge.source_artifact_id for edge in edges}
def find_dependencies(self, artifact_id: str) -> Set[str]:
"""
Find direct dependencies of an artifact (what it depends on).
Args:
artifact_id: Artifact to find dependencies of
Returns:
Set of artifact IDs that this artifact directly depends on
"""
edges = self._repository.get_by_source(artifact_id)
return {edge.target_artifact_id for edge in edges}
def find_transitive_dependents(self, artifact_id: str) -> Set[str]:
"""
Find all transitive dependents of an artifact (full upstream impact).
Uses BFS to traverse the reverse dependency graph.
Args:
artifact_id: Artifact to find transitive dependents of
Returns:
Set of all artifact IDs that transitively depend on this artifact
"""
graph = self._graph_builder.build_graph()
visited: Set[str] = set()
queue = deque([artifact_id])
while queue:
current = queue.popleft()
for predecessor in graph.get_predecessors(current):
if predecessor not in visited:
visited.add(predecessor)
queue.append(predecessor)
return visited
def find_transitive_dependencies(self, artifact_id: str) -> Set[str]:
"""
Find all transitive dependencies of an artifact (full dependency tree).
Uses BFS to traverse the forward dependency graph.
Args:
artifact_id: Artifact to find transitive dependencies of
Returns:
Set of all artifact IDs this artifact transitively depends on
"""
graph = self._graph_builder.build_graph()
visited: Set[str] = set()
queue = deque([artifact_id])
while queue:
current = queue.popleft()
for successor in graph.get_successors(current):
if successor not in visited:
visited.add(successor)
queue.append(successor)
return visited
def get_dependency_chain(
self,
source_id: str,
target_id: str,
) -> Optional[List[str]]:
"""
Find a dependency chain between two artifacts using BFS.
Args:
source_id: Starting artifact ID
target_id: Target artifact ID
Returns:
List of artifact IDs forming the chain from source to target,
or None if no path exists
"""
graph = self._graph_builder.build_graph()
if source_id == target_id:
return [source_id]
visited: Set[str] = {source_id}
queue: deque[List[str]] = deque([[source_id]])
while queue:
path = queue.popleft()
current = path[-1]
for successor in graph.get_successors(current):
if successor == target_id:
return path + [successor]
if successor not in visited:
visited.add(successor)
queue.append(path + [successor])
return None
def detect_circular_dependencies(self) -> List[List[str]]:
"""
Detect all circular dependencies in the graph.
Returns:
List of cycles, where each cycle is a list of artifact IDs
"""
graph = self._graph_builder.build_graph()
return graph.detect_cycles()
def would_create_cycle(
self,
source_id: str,
target_id: str,
) -> bool:
"""
Check if adding an edge would create a cycle.
Pre-validation check before persisting a new dependency edge.
Args:
source_id: Proposed source artifact ID
target_id: Proposed target artifact ID
Returns:
True if adding this edge would create a cycle
"""
graph = self._graph_builder.build_graph()
# Adding source -> target creates a cycle if target can reach source
# (i.e., there's already a path from target to source)
if source_id == target_id:
return True
visited: Set[str] = set()
queue = deque([target_id])
while queue:
current = queue.popleft()
if current == source_id:
return True
for successor in graph.get_successors(current):
if successor not in visited:
visited.add(successor)
queue.append(successor)
return False
def get_build_order(
self,
artifact_ids: Optional[Set[str]] = None,
) -> List[str]:
"""
Get artifacts in build order (topological sort).
Dependencies appear before the artifacts that depend on them.
Args:
artifact_ids: Optional set of artifact IDs to include.
If None, returns build order for all artifacts.
Returns:
List of artifact IDs in build order
Raises:
CircularDependencyError: If graph contains cycles
"""
graph = self._graph_builder.build_graph(artifact_ids)
order = graph.topological_sort()
order.reverse() # Dependencies before dependents
return order

429
markitect/prompts/dependencies/repository.py Normal file
View File

@@ -0,0 +1,429 @@
"""
Repository interfaces and SQLite implementation for dependency persistence.
Implements FR-6.1: Directed dependency edges between artifacts.
Mirrors the SQLiteArtifactRepository pattern for consistency.
"""
import sqlite3
import json
from abc import ABC, abstractmethod
from pathlib import Path
from typing import List, Optional
from datetime import datetime
from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
class DependencyRepositoryError(Exception):
"""Base exception for dependency repository errors."""
pass
class DuplicateDependencyError(DependencyRepositoryError):
"""Raised when attempting to create a duplicate dependency edge."""
pass
# SQL Schema for dependency tables
DEPENDENCY_TABLES_SQL = """
CREATE TABLE IF NOT EXISTS prompt_dependencies (
id TEXT PRIMARY KEY,
source_artifact_id TEXT NOT NULL,
target_artifact_id TEXT NOT NULL,
run_id TEXT NOT NULL,
edge_type TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(source_artifact_id, target_artifact_id, run_id)
);
CREATE INDEX IF NOT EXISTS idx_deps_source ON prompt_dependencies(source_artifact_id);
CREATE INDEX IF NOT EXISTS idx_deps_target ON prompt_dependencies(target_artifact_id);
CREATE INDEX IF NOT EXISTS idx_deps_run ON prompt_dependencies(run_id);
CREATE INDEX IF NOT EXISTS idx_deps_type ON prompt_dependencies(edge_type);
CREATE INDEX IF NOT EXISTS idx_deps_source_target ON prompt_dependencies(source_artifact_id, target_artifact_id);
"""
def initialize_dependency_tables(db_path: str) -> None:
"""
Initialize the dependency-related database tables.
Args:
db_path: Path to the SQLite database file
"""
db_dir = Path(db_path).parent
if db_dir and not db_dir.exists():
db_dir.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(db_path)
try:
conn.executescript(DEPENDENCY_TABLES_SQL)
conn.commit()
finally:
conn.close()
class IDependencyRepository(ABC):
"""
Abstract interface for dependency edge persistence.
Implements FR-6.1: Directed dependency edges between artifacts.
"""
@abstractmethod
def create(self, edge: DependencyEdge) -> DependencyEdge:
"""
Persist a new dependency edge.
Args:
edge: Dependency edge to create
Returns:
Created edge
Raises:
DuplicateDependencyError: If edge with same source+target+run exists
DependencyRepositoryError: On other persistence errors
"""
pass
@abstractmethod
def get_by_id(self, edge_id: str) -> Optional[DependencyEdge]:
"""
Retrieve edge by ID.
Args:
edge_id: Edge identifier
Returns:
Edge if found, None otherwise
"""
pass
@abstractmethod
def get_by_source(self, source_artifact_id: str) -> List[DependencyEdge]:
"""
Get all edges originating from a source artifact.
Args:
source_artifact_id: Source artifact ID
Returns:
List of edges from this source
"""
pass
@abstractmethod
def get_by_target(self, target_artifact_id: str) -> List[DependencyEdge]:
"""
Get all edges pointing to a target artifact.
Args:
target_artifact_id: Target artifact ID
Returns:
List of edges to this target
"""
pass
@abstractmethod
def get_by_run(self, run_id: str) -> List[DependencyEdge]:
"""
Get all edges created by a specific run.
Args:
run_id: Run identifier
Returns:
List of edges from this run
"""
pass
@abstractmethod
def get_by_type(self, edge_type: EdgeType) -> List[DependencyEdge]:
"""
Get all edges of a specific type.
Args:
edge_type: Edge type to filter by
Returns:
List of edges of this type
"""
pass
@abstractmethod
def get_all(self) -> List[DependencyEdge]:
"""
Get all dependency edges.
Returns:
List of all edges
"""
pass
@abstractmethod
def delete(self, edge_id: str) -> bool:
"""
Delete a dependency edge.
Args:
edge_id: Edge identifier
Returns:
True if deleted, False if not found
"""
pass
@abstractmethod
def delete_by_run(self, run_id: str) -> int:
"""
Delete all edges created by a specific run.
Args:
run_id: Run identifier
Returns:
Number of edges deleted
"""
pass
class SQLiteDependencyRepository(IDependencyRepository):
"""
SQLite implementation of dependency repository.
Provides persistent storage for dependency edges with query support
by source, target, run, and edge type.
"""
def __init__(self, db_path: str):
"""
Initialize repository with database path.
Args:
db_path: Path to SQLite database file
"""
self.db_path = db_path
initialize_dependency_tables(db_path)
def _get_connection(self) -> sqlite3.Connection:
"""Get a database connection."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _row_to_edge(self, row: sqlite3.Row) -> DependencyEdge:
"""Convert database row to DependencyEdge instance."""
return DependencyEdge(
id=row["id"],
source_artifact_id=row["source_artifact_id"],
target_artifact_id=row["target_artifact_id"],
run_id=row["run_id"],
edge_type=EdgeType(row["edge_type"]),
created_at=datetime.fromisoformat(row["created_at"]),
)
def create(self, edge: DependencyEdge) -> DependencyEdge:
"""
Persist a new dependency edge.
Args:
edge: Dependency edge to create
Returns:
Created edge
Raises:
DuplicateDependencyError: If edge with same source+target+run exists
DependencyRepositoryError: On other persistence errors
"""
conn = self._get_connection()
try:
conn.execute(
"""
INSERT INTO prompt_dependencies (
id, source_artifact_id, target_artifact_id,
run_id, edge_type, created_at
) VALUES (?, ?, ?, ?, ?, ?)
""",
(
edge.id,
edge.source_artifact_id,
edge.target_artifact_id,
edge.run_id,
edge.edge_type.value,
edge.created_at.isoformat(),
),
)
conn.commit()
return edge
except sqlite3.IntegrityError as e:
if "UNIQUE constraint" in str(e):
raise DuplicateDependencyError(
f"Dependency edge from '{edge.source_artifact_id}' to "
f"'{edge.target_artifact_id}' already exists for run '{edge.run_id}'"
)
raise DependencyRepositoryError(f"Database integrity error: {e}")
except Exception as e:
raise DependencyRepositoryError(f"Failed to create dependency edge: {e}")
finally:
conn.close()
def get_by_id(self, edge_id: str) -> Optional[DependencyEdge]:
"""
Retrieve edge by ID.
Args:
edge_id: Edge identifier
Returns:
Edge if found, None otherwise
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_dependencies WHERE id = ?",
(edge_id,),
)
row = cursor.fetchone()
return self._row_to_edge(row) if row else None
finally:
conn.close()
def get_by_source(self, source_artifact_id: str) -> List[DependencyEdge]:
"""
Get all edges originating from a source artifact.
Args:
source_artifact_id: Source artifact ID
Returns:
List of edges from this source
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_dependencies WHERE source_artifact_id = ?",
(source_artifact_id,),
)
return [self._row_to_edge(row) for row in cursor.fetchall()]
finally:
conn.close()
def get_by_target(self, target_artifact_id: str) -> List[DependencyEdge]:
"""
Get all edges pointing to a target artifact.
Args:
target_artifact_id: Target artifact ID
Returns:
List of edges to this target
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_dependencies WHERE target_artifact_id = ?",
(target_artifact_id,),
)
return [self._row_to_edge(row) for row in cursor.fetchall()]
finally:
conn.close()
def get_by_run(self, run_id: str) -> List[DependencyEdge]:
"""
Get all edges created by a specific run.
Args:
run_id: Run identifier
Returns:
List of edges from this run
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_dependencies WHERE run_id = ?",
(run_id,),
)
return [self._row_to_edge(row) for row in cursor.fetchall()]
finally:
conn.close()
def get_by_type(self, edge_type: EdgeType) -> List[DependencyEdge]:
"""
Get all edges of a specific type.
Args:
edge_type: Edge type to filter by
Returns:
List of edges of this type
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_dependencies WHERE edge_type = ?",
(edge_type.value,),
)
return [self._row_to_edge(row) for row in cursor.fetchall()]
finally:
conn.close()
def get_all(self) -> List[DependencyEdge]:
"""
Get all dependency edges.
Returns:
List of all edges
"""
conn = self._get_connection()
try:
cursor = conn.execute("SELECT * FROM prompt_dependencies")
return [self._row_to_edge(row) for row in cursor.fetchall()]
finally:
conn.close()
def delete(self, edge_id: str) -> bool:
"""
Delete a dependency edge.
Args:
edge_id: Edge identifier
Returns:
True if deleted, False if not found
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"DELETE FROM prompt_dependencies WHERE id = ?",
(edge_id,),
)
conn.commit()
return cursor.rowcount > 0
finally:
conn.close()
def delete_by_run(self, run_id: str) -> int:
"""
Delete all edges created by a specific run.
Args:
run_id: Run identifier
Returns:
Number of edges deleted
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"DELETE FROM prompt_dependencies WHERE run_id = ?",
(run_id,),
)
conn.commit()
return cursor.rowcount
finally:
conn.close()

36
migrations/prompts/004_create_dependencies.sql Normal file
View File

@@ -0,0 +1,36 @@
-- Migration 004: Create prompt dependencies table
-- Implements FR-6: Dependency Tracking
-- FR-6.1: Directed dependency edges between artifacts
-- FR-6.2: Cross-space dependency graph
-- Date: 2026-02-08
-- Prompt dependency edges table
CREATE TABLE IF NOT EXISTS prompt_dependencies (
id TEXT PRIMARY KEY,
source_artifact_id TEXT NOT NULL,
target_artifact_id TEXT NOT NULL,
run_id TEXT NOT NULL,
edge_type TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(source_artifact_id, target_artifact_id, run_id)
);
-- Note: Foreign keys documented but not enforced (matches existing pattern)
-- source_artifact_id REFERENCES prompt_artifacts(id)
-- target_artifact_id REFERENCES prompt_artifacts(id)
-- run_id REFERENCES prompt_runs(id)
-- Indexes for performance
CREATE INDEX IF NOT EXISTS idx_deps_source ON prompt_dependencies(source_artifact_id);
CREATE INDEX IF NOT EXISTS idx_deps_target ON prompt_dependencies(target_artifact_id);
CREATE INDEX IF NOT EXISTS idx_deps_run ON prompt_dependencies(run_id);
CREATE INDEX IF NOT EXISTS idx_deps_type ON prompt_dependencies(edge_type);
CREATE INDEX IF NOT EXISTS idx_deps_source_target ON prompt_dependencies(source_artifact_id, target_artifact_id);
-- Column documentation:
-- id: Unique UUID for this dependency edge
-- source_artifact_id: Artifact that depends on the target
-- target_artifact_id: Artifact that is depended upon
-- run_id: The execution run that established this dependency
-- edge_type: requires, generates, includes
-- created_at: When this dependency was recorded

1
tests/integration/prompts/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Integration tests for prompt dependency tracking."""

274
tests/integration/prompts/test_circular_detection.py Normal file
View File

@@ -0,0 +1,274 @@
"""
Integration tests for circular dependency detection.
Tests cycle detection with real DB, pre-validation before
persisting edges, and error detail reporting.
"""
import pytest
import tempfile
from pathlib import Path
from markitect.prompts.models import Artifact, ArtifactType
from markitect.prompts.repositories.sqlite import SQLiteArtifactRepository
from markitect.prompts.dependencies.models import (
DependencyEdge,
EdgeType,
CircularDependencyError,
)
from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
from markitect.prompts.dependencies.graph import GraphBuilder
from markitect.prompts.dependencies.queries import DependencyQueryService
@pytest.fixture
def temp_db():
"""Create temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
yield db_path
Path(db_path).unlink(missing_ok=True)
@pytest.fixture
def artifact_repo(temp_db):
"""Create artifact repository."""
return SQLiteArtifactRepository(temp_db)
@pytest.fixture
def dep_repo(temp_db):
"""Create dependency repository."""
return SQLiteDependencyRepository(temp_db)
@pytest.fixture
def builder(dep_repo):
"""Create GraphBuilder."""
return GraphBuilder(dep_repo)
@pytest.fixture
def query_service(dep_repo):
"""Create DependencyQueryService."""
return DependencyQueryService(dep_repo)
def _create_artifact(artifact_repo, space_id, name):
"""Helper to create and persist an artifact."""
artifact = Artifact.create(
space_id=space_id,
name=name,
content=f"content for {name}",
artifact_type=ArtifactType.CONTENT,
)
return artifact_repo.create(artifact)
def _persist_edge(dep_repo, src_id, tgt_id, run_id="run-1"):
"""Helper to persist a dependency edge."""
edge = DependencyEdge.create(
source_artifact_id=src_id,
target_artifact_id=tgt_id,
run_id=run_id,
edge_type=EdgeType.REQUIRES,
)
return dep_repo.create(edge)
class TestCycleDetectionWithDB:
"""Tests for cycle detection using real database storage."""
def test_detect_simple_cycle(self, artifact_repo, dep_repo, query_service):
"""Test detecting a simple 2-node cycle in persisted graph."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_a.id, run_id="run-2")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
def test_detect_three_node_cycle(self, artifact_repo, dep_repo, query_service):
"""Test detecting a 3-node cycle A -> B -> C -> A."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_c.id)
_persist_edge(dep_repo, art_c.id, art_a.id, run_id="run-2")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
# All three artifacts should be in the cycle
cycle_nodes = set(cycles[0][:-1])
assert art_a.id in cycle_nodes
assert art_b.id in cycle_nodes
assert art_c.id in cycle_nodes
def test_no_false_positives(self, artifact_repo, dep_repo, query_service):
"""Test no false positive cycle detection in DAG."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
art_d = _create_artifact(artifact_repo, "space-1", "d")
# Diamond: A -> B, A -> C, B -> D, C -> D
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_a.id, art_c.id, run_id="run-2")
_persist_edge(dep_repo, art_b.id, art_d.id, run_id="run-3")
_persist_edge(dep_repo, art_c.id, art_d.id, run_id="run-4")
cycles = query_service.detect_circular_dependencies()
assert cycles == []
class TestPreValidation:
"""Tests for pre-validation (would_create_cycle) before persisting edges."""
def test_validate_safe_edge(self, artifact_repo, dep_repo, query_service):
"""Test pre-validation accepts safe edge."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
_persist_edge(dep_repo, art_a.id, art_b.id)
# B -> C would not create a cycle
assert query_service.would_create_cycle(art_b.id, art_c.id) is False
def test_validate_cycle_creating_edge(
self, artifact_repo, dep_repo, query_service
):
"""Test pre-validation rejects edge that would create a cycle."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_c.id, run_id="run-2")
# C -> A would create a cycle
assert query_service.would_create_cycle(art_c.id, art_a.id) is True
def test_validate_before_persist_workflow(
self, artifact_repo, dep_repo, query_service
):
"""Test the full validate-then-persist workflow."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
# Persist A -> B
_persist_edge(dep_repo, art_a.id, art_b.id)
# Validate and persist B -> C (safe)
assert query_service.would_create_cycle(art_b.id, art_c.id) is False
_persist_edge(dep_repo, art_b.id, art_c.id, run_id="run-2")
# Validate C -> A (would create cycle, don't persist)
assert query_service.would_create_cycle(art_c.id, art_a.id) is True
# Don't persist - graph remains acyclic
# Verify graph is still cycle-free
assert query_service.detect_circular_dependencies() == []
class TestCycleErrorDetails:
"""Tests for cycle error detail reporting."""
def test_topological_sort_error_contains_cycle(
self, artifact_repo, dep_repo, builder
):
"""Test that topological sort error includes cycle details."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_c.id, run_id="run-2")
_persist_edge(dep_repo, art_c.id, art_a.id, run_id="run-3")
graph = builder.build_graph()
with pytest.raises(CircularDependencyError) as exc_info:
graph.topological_sort()
error = exc_info.value
assert error.cycle is not None
assert len(error.cycle) >= 3
# The cycle should contain our artifact IDs
cycle_set = set(error.cycle)
assert art_a.id in cycle_set
assert art_b.id in cycle_set
assert art_c.id in cycle_set
def test_error_message_readable(self, artifact_repo, dep_repo, builder):
"""Test that error message is human-readable."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_a.id, run_id="run-2")
graph = builder.build_graph()
with pytest.raises(CircularDependencyError) as exc_info:
graph.topological_sort()
message = str(exc_info.value)
assert "Circular dependency detected" in message
assert "->" in message
class TestCrossSpaceCycles:
"""Tests for cycle detection across information spaces."""
def test_cross_space_cycle(self, artifact_repo, dep_repo, query_service):
"""Test detecting cycles that span multiple spaces."""
# space-a/X -> space-b/Y -> space-a/X (cycle across spaces)
art_x = _create_artifact(artifact_repo, "space-a", "x")
art_y = _create_artifact(artifact_repo, "space-b", "y")
_persist_edge(dep_repo, art_x.id, art_y.id)
_persist_edge(dep_repo, art_y.id, art_x.id, run_id="run-2")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
def test_cross_space_transitive_cycle(
self, artifact_repo, dep_repo, query_service
):
"""Test detecting transitive cycles across multiple spaces."""
art_a = _create_artifact(artifact_repo, "space-a", "a")
art_b = _create_artifact(artifact_repo, "space-b", "b")
art_c = _create_artifact(artifact_repo, "space-c", "c")
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_c.id, run_id="run-2")
_persist_edge(dep_repo, art_c.id, art_a.id, run_id="run-3")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
# Pre-validation should also catch it
# If we had A -> B -> C already, adding C -> A would be caught
assert query_service.would_create_cycle(art_c.id, art_a.id) is True
def test_cross_space_no_false_positive(
self, artifact_repo, dep_repo, query_service
):
"""Test no false positives across spaces."""
art_a = _create_artifact(artifact_repo, "space-a", "a")
art_b = _create_artifact(artifact_repo, "space-b", "b")
art_c = _create_artifact(artifact_repo, "space-c", "c")
# Linear: A -> B -> C (no cycle)
_persist_edge(dep_repo, art_a.id, art_b.id)
_persist_edge(dep_repo, art_b.id, art_c.id, run_id="run-2")
cycles = query_service.detect_circular_dependencies()
assert cycles == []

321
tests/integration/prompts/test_dependency_graph.py Normal file
View File

@@ -0,0 +1,321 @@
"""
Integration tests for dependency graph full workflow.
Tests the complete workflow: artifacts + edges + graph + queries
across multiple information spaces.
"""
import pytest
import tempfile
from pathlib import Path
from markitect.prompts.models import Artifact, ArtifactType
from markitect.prompts.repositories.sqlite import SQLiteArtifactRepository
from markitect.prompts.dependencies.models import (
DependencyEdge,
DependencyGraph,
EdgeType,
)
from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
from markitect.prompts.dependencies.graph import GraphBuilder
from markitect.prompts.dependencies.queries import DependencyQueryService
from markitect.prompts.execution.manifest import RunManifest
@pytest.fixture
def temp_db():
"""Create temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
yield db_path
Path(db_path).unlink(missing_ok=True)
@pytest.fixture
def artifact_repo(temp_db):
"""Create artifact repository."""
return SQLiteArtifactRepository(temp_db)
@pytest.fixture
def dep_repo(temp_db):
"""Create dependency repository (shares same DB)."""
return SQLiteDependencyRepository(temp_db)
@pytest.fixture
def builder(dep_repo):
"""Create GraphBuilder."""
return GraphBuilder(dep_repo)
@pytest.fixture
def query_service(dep_repo):
"""Create DependencyQueryService."""
return DependencyQueryService(dep_repo)
def _create_artifact(artifact_repo, space_id, name, content="test content"):
"""Helper to create and persist an artifact."""
artifact = Artifact.create(
space_id=space_id,
name=name,
content=content,
artifact_type=ArtifactType.CONTENT,
)
return artifact_repo.create(artifact)
class TestFullWorkflow:
"""Tests for the complete dependency tracking workflow."""
def test_manifest_to_graph_workflow(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test full workflow: create artifacts, manifest, persist edges, query."""
# Step 1: Create artifacts in a space
art_a = _create_artifact(artifact_repo, "space-1", "artifact-a", "content-a")
art_b = _create_artifact(artifact_repo, "space-1", "artifact-b", "content-b")
art_c = _create_artifact(artifact_repo, "space-1", "artifact-c", "content-c")
# Step 2: Create a manifest with dependency edges
manifest = RunManifest.create(
run_id="run-1",
template_id=art_a.id,
template_name="artifact-a",
template_digest=art_a.content_digest,
)
manifest.add_dependency_edge(art_a.id, art_b.id, "requires")
manifest.add_dependency_edge(art_b.id, art_c.id, "requires")
# Step 3: Persist edges from manifest
persisted = builder.persist_edges(manifest)
assert len(persisted) == 2
# Step 4: Build graph
graph = builder.build_graph()
assert graph.has_edge(art_a.id, art_b.id)
assert graph.has_edge(art_b.id, art_c.id)
# Step 5: Query dependencies
deps = query_service.find_dependencies(art_a.id)
assert art_b.id in deps
transitive_deps = query_service.find_transitive_dependencies(art_a.id)
assert art_b.id in transitive_deps
assert art_c.id in transitive_deps
# Step 6: Query dependents
dependents = query_service.find_dependents(art_c.id)
assert art_b.id in dependents
transitive_dependents = query_service.find_transitive_dependents(art_c.id)
assert art_a.id in transitive_dependents
assert art_b.id in transitive_dependents
def test_build_order_workflow(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test build order across a dependency graph."""
art_a = _create_artifact(artifact_repo, "space-1", "lib-core")
art_b = _create_artifact(artifact_repo, "space-1", "lib-utils")
art_c = _create_artifact(artifact_repo, "space-1", "app")
# app depends on lib-utils, lib-utils depends on lib-core
manifest = RunManifest.create(
run_id="run-build",
template_id=art_c.id,
template_name="app",
template_digest=art_c.content_digest,
)
manifest.add_dependency_edge(art_c.id, art_b.id, "requires")
manifest.add_dependency_edge(art_b.id, art_a.id, "requires")
builder.persist_edges(manifest)
# Build order should have core first, then utils, then app
order = query_service.get_build_order()
assert order.index(art_a.id) < order.index(art_b.id)
assert order.index(art_b.id) < order.index(art_c.id)
def test_dependency_chain_workflow(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test finding dependency chains between artifacts."""
art_a = _create_artifact(artifact_repo, "space-1", "source")
art_b = _create_artifact(artifact_repo, "space-1", "intermediate")
art_c = _create_artifact(artifact_repo, "space-1", "target")
manifest = RunManifest.create(
run_id="run-chain",
template_id=art_a.id,
template_name="source",
template_digest=art_a.content_digest,
)
manifest.add_dependency_edge(art_a.id, art_b.id, "requires")
manifest.add_dependency_edge(art_b.id, art_c.id, "requires")
builder.persist_edges(manifest)
chain = query_service.get_dependency_chain(art_a.id, art_c.id)
assert chain is not None
assert chain[0] == art_a.id
assert chain[-1] == art_c.id
assert len(chain) == 3
class TestCrossSpaceGraph:
"""Tests for dependency tracking across information spaces (FR-6.2)."""
def test_cross_space_dependencies(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test dependencies between artifacts in different spaces."""
# Create artifacts in different spaces
shared = _create_artifact(artifact_repo, "shared", "common-lib")
team_a = _create_artifact(artifact_repo, "team-a", "service-a")
team_b = _create_artifact(artifact_repo, "team-b", "service-b")
# Both services depend on shared lib
manifest = RunManifest.create(
run_id="run-cross-space",
template_id=team_a.id,
template_name="service-a",
template_digest=team_a.content_digest,
)
manifest.add_dependency_edge(team_a.id, shared.id, "requires")
manifest.add_dependency_edge(team_b.id, shared.id, "requires")
builder.persist_edges(manifest)
# Shared lib should have both services as dependents
dependents = query_service.find_dependents(shared.id)
assert team_a.id in dependents
assert team_b.id in dependents
# Each service should have shared as a dependency
assert shared.id in query_service.find_dependencies(team_a.id)
assert shared.id in query_service.find_dependencies(team_b.id)
def test_cross_space_transitive(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test transitive dependencies across spaces."""
# shared/foundation -> shared/utils -> team-a/app
foundation = _create_artifact(artifact_repo, "shared", "foundation")
utils = _create_artifact(artifact_repo, "shared", "utils")
app = _create_artifact(artifact_repo, "team-a", "app")
manifest = RunManifest.create(
run_id="run-trans",
template_id=app.id,
template_name="app",
template_digest=app.content_digest,
)
manifest.add_dependency_edge(app.id, utils.id, "requires")
manifest.add_dependency_edge(utils.id, foundation.id, "requires")
builder.persist_edges(manifest)
# App should transitively depend on foundation
transitive = query_service.find_transitive_dependencies(app.id)
assert utils.id in transitive
assert foundation.id in transitive
def test_cross_space_build_order(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test build order across spaces."""
core = _create_artifact(artifact_repo, "shared", "core")
api = _create_artifact(artifact_repo, "team-a", "api")
web = _create_artifact(artifact_repo, "team-b", "web")
manifest = RunManifest.create(
run_id="run-order",
template_id=web.id,
template_name="web",
template_digest=web.content_digest,
)
manifest.add_dependency_edge(api.id, core.id, "requires")
manifest.add_dependency_edge(web.id, api.id, "requires")
builder.persist_edges(manifest)
order = query_service.get_build_order()
assert order.index(core.id) < order.index(api.id)
assert order.index(api.id) < order.index(web.id)
class TestMultipleRuns:
"""Tests for dependency tracking across multiple runs."""
def test_edges_from_multiple_runs(
self, artifact_repo, dep_repo, builder, query_service
):
"""Test edges from multiple runs are combined in graph."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
# Run 1: A -> B
manifest1 = RunManifest.create(
run_id="run-1",
template_id=art_a.id,
template_name="a",
template_digest=art_a.content_digest,
)
manifest1.add_dependency_edge(art_a.id, art_b.id, "requires")
builder.persist_edges(manifest1)
# Run 2: B -> C
manifest2 = RunManifest.create(
run_id="run-2",
template_id=art_b.id,
template_name="b",
template_digest=art_b.content_digest,
)
manifest2.add_dependency_edge(art_b.id, art_c.id, "requires")
builder.persist_edges(manifest2)
# Full graph should have A -> B -> C
graph = builder.build_graph()
assert graph.has_edge(art_a.id, art_b.id)
assert graph.has_edge(art_b.id, art_c.id)
# Transitive query should work across runs
transitive = query_service.find_transitive_dependencies(art_a.id)
assert art_b.id in transitive
assert art_c.id in transitive
def test_run_scoped_graph(self, artifact_repo, dep_repo, builder):
"""Test building graph scoped to a specific run."""
art_a = _create_artifact(artifact_repo, "space-1", "a")
art_b = _create_artifact(artifact_repo, "space-1", "b")
art_c = _create_artifact(artifact_repo, "space-1", "c")
# Persist edges for different runs
edge1 = DependencyEdge.create(
source_artifact_id=art_a.id,
target_artifact_id=art_b.id,
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
dep_repo.create(edge1)
edge2 = DependencyEdge.create(
source_artifact_id=art_b.id,
target_artifact_id=art_c.id,
run_id="run-2",
edge_type=EdgeType.GENERATES,
)
dep_repo.create(edge2)
# Graph for run-1 only
graph1 = builder.build_graph_for_run("run-1")
assert graph1.has_edge(art_a.id, art_b.id)
assert not graph1.has_edge(art_b.id, art_c.id)
# Graph for run-2 only
graph2 = builder.build_graph_for_run("run-2")
assert not graph2.has_edge(art_a.id, art_b.id)
assert graph2.has_edge(art_b.id, art_c.id)

410
tests/unit/prompts/test_dependency_models.py Normal file
View File

@@ -0,0 +1,410 @@
"""
Unit tests for dependency tracking models.
Tests EdgeType, DependencyEdge, DependencyGraph operations,
cycle detection, and topological sort.
"""
import pytest
from datetime import datetime
from markitect.prompts.dependencies.models import (
EdgeType,
DependencyEdge,
DependencyGraph,
CircularDependencyError,
)
class TestEdgeType:
"""Tests for EdgeType enum."""
def test_edge_type_values(self):
"""Test edge type enum values."""
assert EdgeType.REQUIRES.value == "requires"
assert EdgeType.GENERATES.value == "generates"
assert EdgeType.INCLUDES.value == "includes"
def test_edge_type_from_value(self):
"""Test creating edge type from string value."""
assert EdgeType("requires") == EdgeType.REQUIRES
assert EdgeType("generates") == EdgeType.GENERATES
assert EdgeType("includes") == EdgeType.INCLUDES
def test_edge_type_invalid_value(self):
"""Test invalid edge type raises ValueError."""
with pytest.raises(ValueError):
EdgeType("invalid")
class TestDependencyEdge:
"""Tests for persistent DependencyEdge."""
def test_create_edge(self):
"""Test creating a dependency edge via factory method."""
edge = DependencyEdge.create(
source_artifact_id="artifact-1",
target_artifact_id="artifact-2",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
assert edge.id # UUID generated
assert edge.source_artifact_id == "artifact-1"
assert edge.target_artifact_id == "artifact-2"
assert edge.run_id == "run-1"
assert edge.edge_type == EdgeType.REQUIRES
assert isinstance(edge.created_at, datetime)
def test_create_edge_unique_ids(self):
"""Test each created edge gets a unique ID."""
edge1 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="r1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="r1",
edge_type=EdgeType.REQUIRES,
)
assert edge1.id != edge2.id
def test_edge_to_dict(self):
"""Test edge serialization to dict."""
edge = DependencyEdge.create(
source_artifact_id="artifact-1",
target_artifact_id="artifact-2",
run_id="run-1",
edge_type=EdgeType.GENERATES,
)
data = edge.to_dict()
assert data["id"] == edge.id
assert data["source_artifact_id"] == "artifact-1"
assert data["target_artifact_id"] == "artifact-2"
assert data["run_id"] == "run-1"
assert data["edge_type"] == "generates"
assert "created_at" in data
def test_edge_from_dict(self):
"""Test edge deserialization from dict."""
data = {
"id": "edge-123",
"source_artifact_id": "artifact-1",
"target_artifact_id": "artifact-2",
"run_id": "run-1",
"edge_type": "includes",
"created_at": "2026-01-01T00:00:00",
}
edge = DependencyEdge.from_dict(data)
assert edge.id == "edge-123"
assert edge.source_artifact_id == "artifact-1"
assert edge.target_artifact_id == "artifact-2"
assert edge.run_id == "run-1"
assert edge.edge_type == EdgeType.INCLUDES
assert edge.created_at == datetime(2026, 1, 1)
def test_edge_roundtrip(self):
"""Test serialization roundtrip preserves data."""
original = DependencyEdge.create(
source_artifact_id="src",
target_artifact_id="tgt",
run_id="run",
edge_type=EdgeType.REQUIRES,
)
restored = DependencyEdge.from_dict(original.to_dict())
assert restored.id == original.id
assert restored.source_artifact_id == original.source_artifact_id
assert restored.target_artifact_id == original.target_artifact_id
assert restored.run_id == original.run_id
assert restored.edge_type == original.edge_type
class TestDependencyGraph:
"""Tests for DependencyGraph."""
def test_empty_graph(self):
"""Test empty graph properties."""
graph = DependencyGraph()
assert len(graph.nodes) == 0
assert graph.edge_count == 0
assert not graph.has_cycle()
def test_add_edge(self):
"""Test adding an edge to the graph."""
graph = DependencyGraph()
graph.add_edge("A", "B", EdgeType.REQUIRES)
assert "A" in graph.nodes
assert "B" in graph.nodes
assert graph.edge_count == 1
assert graph.has_edge("A", "B")
assert not graph.has_edge("B", "A")
def test_add_multiple_edges(self):
"""Test adding multiple edges."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "C")
graph.add_edge("A", "C")
assert len(graph.nodes) == 3
assert graph.edge_count == 3
def test_get_successors(self):
"""Test getting direct successors."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "C")
assert graph.get_successors("A") == {"B", "C"}
assert graph.get_successors("B") == {"C"}
assert graph.get_successors("C") == set()
def test_get_predecessors(self):
"""Test getting direct predecessors."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "C")
assert graph.get_predecessors("A") == set()
assert graph.get_predecessors("B") == {"A"}
assert graph.get_predecessors("C") == {"A", "B"}
def test_get_successors_nonexistent_node(self):
"""Test getting successors of non-existent node."""
graph = DependencyGraph()
assert graph.get_successors("X") == set()
def test_get_predecessors_nonexistent_node(self):
"""Test getting predecessors of non-existent node."""
graph = DependencyGraph()
assert graph.get_predecessors("X") == set()
def test_get_edge_type(self):
"""Test getting edge type between nodes."""
graph = DependencyGraph()
graph.add_edge("A", "B", EdgeType.REQUIRES)
graph.add_edge("B", "C", EdgeType.GENERATES)
assert graph.get_edge_type("A", "B") == EdgeType.REQUIRES
assert graph.get_edge_type("B", "C") == EdgeType.GENERATES
assert graph.get_edge_type("A", "C") is None
def test_has_edge(self):
"""Test edge existence check."""
graph = DependencyGraph()
graph.add_edge("A", "B")
assert graph.has_edge("A", "B")
assert not graph.has_edge("B", "A")
assert not graph.has_edge("A", "C")
class TestDependencyGraphCycleDetection:
"""Tests for DependencyGraph cycle detection."""
def test_no_cycle_linear(self):
"""Test no cycle in linear graph A -> B -> C."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "C")
assert not graph.has_cycle()
assert graph.detect_cycles() == []
def test_no_cycle_diamond(self):
"""Test no cycle in diamond graph A -> B,C -> D."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "D")
graph.add_edge("C", "D")
assert not graph.has_cycle()
def test_simple_cycle(self):
"""Test detecting simple cycle A -> B -> A."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "A")
assert graph.has_cycle()
cycles = graph.detect_cycles()
assert len(cycles) > 0
def test_three_node_cycle(self):
"""Test detecting 3-node cycle A -> B -> C -> A."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "C")
graph.add_edge("C", "A")
assert graph.has_cycle()
cycles = graph.detect_cycles()
assert len(cycles) > 0
# The cycle should contain A, B, C
cycle_nodes = set(cycles[0][:-1]) # Last element repeats the start
assert cycle_nodes == {"A", "B", "C"}
def test_self_loop(self):
"""Test detecting self-loop A -> A."""
graph = DependencyGraph()
graph.add_edge("A", "A")
assert graph.has_cycle()
def test_cycle_with_acyclic_parts(self):
"""Test cycle detection in graph with both cyclic and acyclic parts."""
graph = DependencyGraph()
# Acyclic part
graph.add_edge("X", "Y")
# Cyclic part
graph.add_edge("A", "B")
graph.add_edge("B", "C")
graph.add_edge("C", "A")
assert graph.has_cycle()
class TestDependencyGraphTopologicalSort:
"""Tests for DependencyGraph topological sort."""
def test_topological_sort_linear(self):
"""Test topological sort of linear graph."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "C")
order = graph.topological_sort()
assert order.index("A") < order.index("B")
assert order.index("B") < order.index("C")
def test_topological_sort_diamond(self):
"""Test topological sort of diamond graph."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("A", "C")
graph.add_edge("B", "D")
graph.add_edge("C", "D")
order = graph.topological_sort()
assert order.index("A") < order.index("B")
assert order.index("A") < order.index("C")
assert order.index("B") < order.index("D")
assert order.index("C") < order.index("D")
def test_topological_sort_single_node(self):
"""Test topological sort with single node."""
graph = DependencyGraph()
graph.add_edge("A", "B")
order = graph.topological_sort()
assert len(order) == 2
assert order[0] == "A"
assert order[1] == "B"
def test_topological_sort_disconnected(self):
"""Test topological sort with disconnected components."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("C", "D")
order = graph.topological_sort()
assert len(order) == 4
assert order.index("A") < order.index("B")
assert order.index("C") < order.index("D")
def test_topological_sort_with_cycle_raises_error(self):
"""Test topological sort raises error on cycle."""
graph = DependencyGraph()
graph.add_edge("A", "B")
graph.add_edge("B", "C")
graph.add_edge("C", "A")
with pytest.raises(CircularDependencyError) as exc_info:
graph.topological_sort()
assert exc_info.value.cycle is not None
assert len(exc_info.value.cycle) > 0
def test_topological_sort_empty_graph(self):
"""Test topological sort of empty graph."""
graph = DependencyGraph()
assert graph.topological_sort() == []
class TestDependencyGraphSubgraph:
"""Tests for DependencyGraph subgraph extraction."""
def test_subgraph_extraction(self):
"""Test extracting a subgraph."""
graph = DependencyGraph()
graph.add_edge("A", "B", EdgeType.REQUIRES)
graph.add_edge("B", "C", EdgeType.GENERATES)
graph.add_edge("C", "D", EdgeType.INCLUDES)
subgraph = graph.get_subgraph({"A", "B", "C"})
assert subgraph.has_edge("A", "B")
assert subgraph.has_edge("B", "C")
assert not subgraph.has_edge("C", "D")
assert "D" not in subgraph.nodes
def test_subgraph_preserves_edge_types(self):
"""Test subgraph preserves edge types."""
graph = DependencyGraph()
graph.add_edge("A", "B", EdgeType.REQUIRES)
graph.add_edge("B", "C", EdgeType.GENERATES)
subgraph = graph.get_subgraph({"A", "B", "C"})
assert subgraph.get_edge_type("A", "B") == EdgeType.REQUIRES
assert subgraph.get_edge_type("B", "C") == EdgeType.GENERATES
def test_subgraph_isolated_nodes(self):
"""Test subgraph includes isolated nodes."""
graph = DependencyGraph()
graph.add_edge("A", "B")
subgraph = graph.get_subgraph({"A", "B", "C"})
assert "C" in subgraph.nodes
assert subgraph.get_successors("C") == set()
def test_subgraph_empty(self):
"""Test extracting empty subgraph."""
graph = DependencyGraph()
graph.add_edge("A", "B")
subgraph = graph.get_subgraph(set())
assert len(subgraph.nodes) == 0
class TestCircularDependencyError:
"""Tests for CircularDependencyError."""
def test_error_with_cycle(self):
"""Test error with cycle information."""
error = CircularDependencyError(
"Cycle detected",
cycle=["A", "B", "C", "A"],
)
assert str(error) == "Cycle detected"
assert error.cycle == ["A", "B", "C", "A"]
def test_error_without_cycle(self):
"""Test error without cycle information."""
error = CircularDependencyError("Cycle detected")
assert error.cycle == []
def test_error_is_exception(self):
"""Test that CircularDependencyError is an Exception."""
with pytest.raises(CircularDependencyError):
raise CircularDependencyError("test")

324
tests/unit/prompts/test_dependency_queries.py Normal file
View File

@@ -0,0 +1,324 @@
"""
Unit tests for DependencyQueryService.
Tests direct/transitive dependents and dependencies, dependency chains,
cycle validation, would_create_cycle, and build order.
"""
import pytest
import tempfile
from pathlib import Path
from markitect.prompts.dependencies.models import (
DependencyEdge,
DependencyGraph,
EdgeType,
CircularDependencyError,
)
from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
from markitect.prompts.dependencies.queries import DependencyQueryService
@pytest.fixture
def temp_db():
"""Create temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
yield db_path
Path(db_path).unlink(missing_ok=True)
@pytest.fixture
def repository(temp_db):
"""Create repository instance with temp database."""
return SQLiteDependencyRepository(temp_db)
@pytest.fixture
def query_service(repository):
"""Create DependencyQueryService instance."""
return DependencyQueryService(repository)
def _create_edge(repository, src, tgt, run_id="run-1", edge_type=EdgeType.REQUIRES):
"""Helper to create and persist a dependency edge."""
edge = DependencyEdge.create(
source_artifact_id=src,
target_artifact_id=tgt,
run_id=run_id,
edge_type=edge_type,
)
return repository.create(edge)
class TestFindDependents:
"""Tests for find_dependents (who depends on X)."""
def test_find_direct_dependents(self, repository, query_service):
"""Test finding direct dependents of an artifact."""
_create_edge(repository, "A", "B")
_create_edge(repository, "C", "B")
dependents = query_service.find_dependents("B")
assert dependents == {"A", "C"}
def test_find_dependents_none(self, repository, query_service):
"""Test finding dependents when none exist."""
_create_edge(repository, "A", "B")
dependents = query_service.find_dependents("A")
assert dependents == set()
def test_find_dependents_nonexistent(self, query_service):
"""Test finding dependents of non-existent artifact."""
dependents = query_service.find_dependents("nonexistent")
assert dependents == set()
class TestFindDependencies:
"""Tests for find_dependencies (what X depends on)."""
def test_find_direct_dependencies(self, repository, query_service):
"""Test finding direct dependencies of an artifact."""
_create_edge(repository, "A", "B")
_create_edge(repository, "A", "C")
dependencies = query_service.find_dependencies("A")
assert dependencies == {"B", "C"}
def test_find_dependencies_none(self, repository, query_service):
"""Test finding dependencies when none exist."""
_create_edge(repository, "A", "B")
dependencies = query_service.find_dependencies("B")
assert dependencies == set()
class TestTransitiveDependents:
"""Tests for find_transitive_dependents."""
def test_transitive_dependents(self, repository, query_service):
"""Test finding transitive dependents (upstream impact)."""
# A -> B -> C
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
# Who is transitively affected by C?
transitive = query_service.find_transitive_dependents("C")
assert transitive == {"A", "B"}
def test_transitive_dependents_diamond(self, repository, query_service):
"""Test transitive dependents in diamond graph."""
# A -> B -> D, A -> C -> D
_create_edge(repository, "A", "B")
_create_edge(repository, "A", "C")
_create_edge(repository, "B", "D")
_create_edge(repository, "C", "D")
transitive = query_service.find_transitive_dependents("D")
assert transitive == {"A", "B", "C"}
def test_transitive_dependents_none(self, repository, query_service):
"""Test transitive dependents when none exist."""
_create_edge(repository, "A", "B")
transitive = query_service.find_transitive_dependents("A")
assert transitive == set()
class TestTransitiveDependencies:
"""Tests for find_transitive_dependencies."""
def test_transitive_dependencies(self, repository, query_service):
"""Test finding transitive dependencies (full dep tree)."""
# A -> B -> C
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
# What does A transitively depend on?
transitive = query_service.find_transitive_dependencies("A")
assert transitive == {"B", "C"}
def test_transitive_dependencies_complex(self, repository, query_service):
"""Test transitive dependencies in complex graph."""
# A -> B -> D, A -> C -> D, D -> E
_create_edge(repository, "A", "B")
_create_edge(repository, "A", "C")
_create_edge(repository, "B", "D")
_create_edge(repository, "C", "D")
_create_edge(repository, "D", "E")
transitive = query_service.find_transitive_dependencies("A")
assert transitive == {"B", "C", "D", "E"}
def test_transitive_dependencies_leaf(self, repository, query_service):
"""Test transitive dependencies of a leaf node."""
_create_edge(repository, "A", "B")
transitive = query_service.find_transitive_dependencies("B")
assert transitive == set()
class TestDependencyChain:
"""Tests for get_dependency_chain (BFS path finding)."""
def test_direct_chain(self, repository, query_service):
"""Test finding a direct dependency chain."""
_create_edge(repository, "A", "B")
chain = query_service.get_dependency_chain("A", "B")
assert chain == ["A", "B"]
def test_multi_hop_chain(self, repository, query_service):
"""Test finding a multi-hop dependency chain."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
_create_edge(repository, "C", "D")
chain = query_service.get_dependency_chain("A", "D")
assert chain is not None
assert chain[0] == "A"
assert chain[-1] == "D"
assert len(chain) == 4
def test_no_chain_exists(self, repository, query_service):
"""Test when no chain exists between artifacts."""
_create_edge(repository, "A", "B")
_create_edge(repository, "C", "D")
chain = query_service.get_dependency_chain("A", "D")
assert chain is None
def test_same_source_target(self, query_service):
"""Test chain from node to itself."""
chain = query_service.get_dependency_chain("A", "A")
assert chain == ["A"]
def test_shortest_chain(self, repository, query_service):
"""Test BFS finds shortest chain."""
# A -> B -> C (length 3)
# A -> C (length 2, shorter)
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
_create_edge(repository, "A", "C")
chain = query_service.get_dependency_chain("A", "C")
assert chain == ["A", "C"]
class TestCircularDependencyDetection:
"""Tests for detect_circular_dependencies."""
def test_no_cycles(self, repository, query_service):
"""Test detection when no cycles exist."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
cycles = query_service.detect_circular_dependencies()
assert cycles == []
def test_detect_simple_cycle(self, repository, query_service):
"""Test detecting a simple cycle."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "A")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
def test_detect_three_node_cycle(self, repository, query_service):
"""Test detecting a 3-node cycle."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
_create_edge(repository, "C", "A")
cycles = query_service.detect_circular_dependencies()
assert len(cycles) > 0
class TestWouldCreateCycle:
"""Tests for would_create_cycle pre-validation."""
def test_would_not_create_cycle(self, repository, query_service):
"""Test edge that would not create a cycle."""
_create_edge(repository, "A", "B")
assert query_service.would_create_cycle("B", "C") is False
def test_would_create_cycle_direct(self, repository, query_service):
"""Test edge that would create a direct cycle."""
_create_edge(repository, "A", "B")
assert query_service.would_create_cycle("B", "A") is True
def test_would_create_cycle_transitive(self, repository, query_service):
"""Test edge that would create a transitive cycle."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
assert query_service.would_create_cycle("C", "A") is True
def test_self_reference_creates_cycle(self, query_service):
"""Test self-reference always creates a cycle."""
assert query_service.would_create_cycle("A", "A") is True
def test_safe_parallel_edge(self, repository, query_service):
"""Test parallel forward edge is safe."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
# Adding A -> C doesn't create a cycle
assert query_service.would_create_cycle("A", "C") is False
class TestBuildOrder:
"""Tests for get_build_order (topological sort)."""
def test_simple_build_order(self, repository, query_service):
"""Test simple build order (dependencies before dependents)."""
# A requires B, B requires C → build C first, then B, then A
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
order = query_service.get_build_order()
assert order.index("C") < order.index("B")
assert order.index("B") < order.index("A")
def test_build_order_diamond(self, repository, query_service):
"""Test build order with diamond dependencies."""
# A requires B and C, B and C require D → build D first, then B/C, then A
_create_edge(repository, "A", "B")
_create_edge(repository, "A", "C")
_create_edge(repository, "B", "D")
_create_edge(repository, "C", "D")
order = query_service.get_build_order()
assert order.index("D") < order.index("B")
assert order.index("D") < order.index("C")
assert order.index("B") < order.index("A")
assert order.index("C") < order.index("A")
def test_build_order_scoped(self, repository, query_service):
"""Test build order with scoped artifact set."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
_create_edge(repository, "C", "D")
order = query_service.get_build_order(artifact_ids={"A", "B", "C"})
assert len(order) == 3
assert "D" not in order
assert order.index("C") < order.index("B")
assert order.index("B") < order.index("A")
def test_build_order_with_cycle_raises_error(self, repository, query_service):
"""Test build order raises error on cycle."""
_create_edge(repository, "A", "B")
_create_edge(repository, "B", "C")
_create_edge(repository, "C", "A")
with pytest.raises(CircularDependencyError):
query_service.get_build_order()
def test_build_order_empty(self, query_service):
"""Test build order with no edges."""
order = query_service.get_build_order()
assert order == []

281
tests/unit/prompts/test_dependency_repository.py Normal file
View File

@@ -0,0 +1,281 @@
"""
Unit tests for dependency repository.
Tests CRUD operations, duplicate handling, and query by
source, target, run, and type.
"""
import pytest
import tempfile
from pathlib import Path
from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
from markitect.prompts.dependencies.repository import (
SQLiteDependencyRepository,
DuplicateDependencyError,
)
@pytest.fixture
def temp_db():
"""Create temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
yield db_path
Path(db_path).unlink(missing_ok=True)
@pytest.fixture
def repository(temp_db):
"""Create repository instance with temp database."""
return SQLiteDependencyRepository(temp_db)
@pytest.fixture
def sample_edge():
"""Create sample dependency edge for testing."""
return DependencyEdge.create(
source_artifact_id="artifact-1",
target_artifact_id="artifact-2",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
class TestSQLiteDependencyRepository:
"""Tests for SQLiteDependencyRepository."""
def test_create_edge(self, repository, sample_edge):
"""Test creating a dependency edge."""
created = repository.create(sample_edge)
assert created.id == sample_edge.id
assert created.source_artifact_id == "artifact-1"
assert created.target_artifact_id == "artifact-2"
def test_create_duplicate_edge_raises_error(self, repository):
"""Test creating duplicate edge raises error."""
edge1 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="r1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
edge2 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="r1",
edge_type=EdgeType.REQUIRES,
)
with pytest.raises(DuplicateDependencyError, match="already exists"):
repository.create(edge2)
def test_same_edge_different_runs_allowed(self, repository):
"""Test same source/target with different runs is allowed."""
edge1 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="run-2",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
repository.create(edge2) # Should not raise
def test_get_by_id(self, repository, sample_edge):
"""Test retrieving edge by ID."""
repository.create(sample_edge)
retrieved = repository.get_by_id(sample_edge.id)
assert retrieved is not None
assert retrieved.id == sample_edge.id
assert retrieved.source_artifact_id == sample_edge.source_artifact_id
assert retrieved.edge_type == sample_edge.edge_type
def test_get_by_id_not_found(self, repository):
"""Test retrieving non-existent edge returns None."""
assert repository.get_by_id("nonexistent") is None
def test_get_by_source(self, repository):
"""Test querying edges by source artifact."""
edge1 = DependencyEdge.create(
source_artifact_id="src-1",
target_artifact_id="tgt-1",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="src-1",
target_artifact_id="tgt-2",
run_id="run-1",
edge_type=EdgeType.GENERATES,
)
edge3 = DependencyEdge.create(
source_artifact_id="src-2",
target_artifact_id="tgt-3",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
repository.create(edge2)
repository.create(edge3)
edges = repository.get_by_source("src-1")
assert len(edges) == 2
assert all(e.source_artifact_id == "src-1" for e in edges)
def test_get_by_target(self, repository):
"""Test querying edges by target artifact."""
edge1 = DependencyEdge.create(
source_artifact_id="src-1",
target_artifact_id="tgt-1",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="src-2",
target_artifact_id="tgt-1",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
repository.create(edge2)
edges = repository.get_by_target("tgt-1")
assert len(edges) == 2
assert all(e.target_artifact_id == "tgt-1" for e in edges)
def test_get_by_run(self, repository):
"""Test querying edges by run ID."""
edge1 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="c",
target_artifact_id="d",
run_id="run-1",
edge_type=EdgeType.GENERATES,
)
edge3 = DependencyEdge.create(
source_artifact_id="e",
target_artifact_id="f",
run_id="run-2",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
repository.create(edge2)
repository.create(edge3)
edges = repository.get_by_run("run-1")
assert len(edges) == 2
assert all(e.run_id == "run-1" for e in edges)
def test_get_by_type(self, repository):
"""Test querying edges by type."""
edge1 = DependencyEdge.create(
source_artifact_id="a",
target_artifact_id="b",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
edge2 = DependencyEdge.create(
source_artifact_id="c",
target_artifact_id="d",
run_id="run-1",
edge_type=EdgeType.GENERATES,
)
edge3 = DependencyEdge.create(
source_artifact_id="e",
target_artifact_id="f",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge1)
repository.create(edge2)
repository.create(edge3)
requires_edges = repository.get_by_type(EdgeType.REQUIRES)
assert len(requires_edges) == 2
assert all(e.edge_type == EdgeType.REQUIRES for e in requires_edges)
generates_edges = repository.get_by_type(EdgeType.GENERATES)
assert len(generates_edges) == 1
def test_get_all(self, repository):
"""Test getting all edges."""
for i in range(3):
edge = DependencyEdge.create(
source_artifact_id=f"src-{i}",
target_artifact_id=f"tgt-{i}",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
all_edges = repository.get_all()
assert len(all_edges) == 3
def test_get_all_empty(self, repository):
"""Test getting all edges from empty repository."""
assert repository.get_all() == []
def test_delete_edge(self, repository, sample_edge):
"""Test deleting an edge."""
repository.create(sample_edge)
assert repository.delete(sample_edge.id) is True
assert repository.get_by_id(sample_edge.id) is None
def test_delete_nonexistent_edge(self, repository):
"""Test deleting non-existent edge returns False."""
assert repository.delete("nonexistent") is False
def test_delete_by_run(self, repository):
"""Test deleting all edges for a run."""
for i in range(3):
edge = DependencyEdge.create(
source_artifact_id=f"src-{i}",
target_artifact_id=f"tgt-{i}",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
other_edge = DependencyEdge.create(
source_artifact_id="other-src",
target_artifact_id="other-tgt",
run_id="run-2",
edge_type=EdgeType.REQUIRES,
)
repository.create(other_edge)
deleted = repository.delete_by_run("run-1")
assert deleted == 3
assert len(repository.get_by_run("run-1")) == 0
assert len(repository.get_by_run("run-2")) == 1
def test_delete_by_run_no_matches(self, repository):
"""Test deleting by run with no matches returns 0."""
assert repository.delete_by_run("nonexistent") == 0
def test_edge_type_preserved(self, repository):
"""Test that edge type is correctly stored and retrieved."""
for edge_type in EdgeType:
edge = DependencyEdge.create(
source_artifact_id=f"src-{edge_type.value}",
target_artifact_id=f"tgt-{edge_type.value}",
run_id="run-1",
edge_type=edge_type,
)
repository.create(edge)
retrieved = repository.get_by_id(edge.id)
assert retrieved.edge_type == edge_type

231
tests/unit/prompts/test_graph_builder.py Normal file
View File

@@ -0,0 +1,231 @@
"""
Unit tests for GraphBuilder.
Tests extracting edges from manifest, persisting them,
building graphs, and artifact-scoped subgraphs.
"""
import pytest
import tempfile
from pathlib import Path
from markitect.prompts.dependencies.models import (
DependencyEdge,
DependencyGraph,
EdgeType,
)
from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
from markitect.prompts.dependencies.graph import GraphBuilder
from markitect.prompts.execution.manifest import (
DependencyEdge as ManifestDependencyEdge,
RunManifest,
)
@pytest.fixture
def temp_db():
"""Create temporary database for testing."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = f.name
yield db_path
Path(db_path).unlink(missing_ok=True)
@pytest.fixture
def repository(temp_db):
"""Create repository instance with temp database."""
return SQLiteDependencyRepository(temp_db)
@pytest.fixture
def builder(repository):
"""Create GraphBuilder instance."""
return GraphBuilder(repository)
@pytest.fixture
def sample_manifest():
"""Create a sample RunManifest with dependency edges."""
manifest = RunManifest.create(
run_id="run-1",
template_id="template-1",
template_name="test-template",
template_digest="abc123",
)
manifest.add_dependency_edge("artifact-1", "artifact-2", "requires")
manifest.add_dependency_edge("artifact-2", "artifact-3", "generates")
manifest.add_dependency_edge("artifact-1", "artifact-3", "includes")
return manifest
class TestGraphBuilderExtract:
"""Tests for extracting edges from manifests."""
def test_extract_edges_from_manifest(self, builder, sample_manifest):
"""Test extracting persistent edges from a manifest."""
edges = builder.extract_edges(sample_manifest)
assert len(edges) == 3
assert all(isinstance(e, DependencyEdge) for e in edges)
def test_extracted_edge_fields(self, builder, sample_manifest):
"""Test extracted edges have correct fields."""
edges = builder.extract_edges(sample_manifest)
# First edge: artifact-1 -> artifact-2, requires
edge = edges[0]
assert edge.source_artifact_id == "artifact-1"
assert edge.target_artifact_id == "artifact-2"
assert edge.run_id == "run-1"
assert edge.edge_type == EdgeType.REQUIRES
assert edge.id # UUID assigned
def test_extracted_edges_have_unique_ids(self, builder, sample_manifest):
"""Test each extracted edge gets a unique ID."""
edges = builder.extract_edges(sample_manifest)
ids = [e.id for e in edges]
assert len(set(ids)) == len(ids)
def test_extract_edges_empty_manifest(self, builder):
"""Test extracting from manifest with no edges."""
manifest = RunManifest.create(
run_id="run-empty",
template_id="template-1",
template_name="empty",
template_digest="xyz",
)
edges = builder.extract_edges(manifest)
assert edges == []
def test_extract_edges_preserves_types(self, builder, sample_manifest):
"""Test all edge types are correctly mapped."""
edges = builder.extract_edges(sample_manifest)
types = [e.edge_type for e in edges]
assert EdgeType.REQUIRES in types
assert EdgeType.GENERATES in types
assert EdgeType.INCLUDES in types
class TestGraphBuilderPersist:
"""Tests for persisting edges from manifests."""
def test_persist_edges(self, builder, repository, sample_manifest):
"""Test persisting edges to repository."""
persisted = builder.persist_edges(sample_manifest)
assert len(persisted) == 3
# Verify in repository
all_edges = repository.get_all()
assert len(all_edges) == 3
def test_persisted_edges_retrievable(self, builder, repository, sample_manifest):
"""Test persisted edges can be retrieved from repository."""
builder.persist_edges(sample_manifest)
edges_from_source = repository.get_by_source("artifact-1")
assert len(edges_from_source) == 2 # artifact-1 -> artifact-2 and artifact-3
edges_from_run = repository.get_by_run("run-1")
assert len(edges_from_run) == 3
class TestGraphBuilderBuildGraph:
"""Tests for building dependency graphs."""
def test_build_graph_from_all_edges(self, builder, repository):
"""Test building graph from all stored edges."""
# Manually create some edges
for src, tgt in [("A", "B"), ("B", "C"), ("A", "C")]:
edge = DependencyEdge.create(
source_artifact_id=src,
target_artifact_id=tgt,
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
graph = builder.build_graph()
assert len(graph.nodes) == 3
assert graph.has_edge("A", "B")
assert graph.has_edge("B", "C")
assert graph.has_edge("A", "C")
def test_build_graph_scoped(self, builder, repository):
"""Test building scoped graph with subset of artifacts."""
for src, tgt in [("A", "B"), ("B", "C"), ("C", "D")]:
edge = DependencyEdge.create(
source_artifact_id=src,
target_artifact_id=tgt,
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
graph = builder.build_graph(artifact_ids={"A", "B", "C"})
assert graph.has_edge("A", "B")
assert graph.has_edge("B", "C")
assert not graph.has_edge("C", "D")
assert "D" not in graph.nodes
def test_build_graph_scoped_includes_isolated_nodes(self, builder, repository):
"""Test scoped graph includes nodes with no edges in scope."""
edge = DependencyEdge.create(
source_artifact_id="A",
target_artifact_id="B",
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
graph = builder.build_graph(artifact_ids={"A", "B", "C"})
assert "C" in graph.nodes
def test_build_graph_empty(self, builder):
"""Test building graph from empty repository."""
graph = builder.build_graph()
assert len(graph.nodes) == 0
assert graph.edge_count == 0
def test_build_graph_for_run(self, builder, repository):
"""Test building graph from a specific run's edges."""
# Run 1 edges
for src, tgt in [("A", "B"), ("B", "C")]:
edge = DependencyEdge.create(
source_artifact_id=src,
target_artifact_id=tgt,
run_id="run-1",
edge_type=EdgeType.REQUIRES,
)
repository.create(edge)
# Run 2 edges
edge = DependencyEdge.create(
source_artifact_id="X",
target_artifact_id="Y",
run_id="run-2",
edge_type=EdgeType.GENERATES,
)
repository.create(edge)
graph = builder.build_graph_for_run("run-1")
assert graph.has_edge("A", "B")
assert graph.has_edge("B", "C")
assert not graph.has_edge("X", "Y")
def test_build_graph_preserves_edge_types(self, builder, repository):
"""Test built graph preserves edge types."""
edge = DependencyEdge.create(
source_artifact_id="A",
target_artifact_id="B",
run_id="run-1",
edge_type=EdgeType.GENERATES,
)
repository.create(edge)
graph = builder.build_graph()
assert graph.get_edge_type("A", "B") == EdgeType.GENERATES