Compare commits
1 Commits
step35/150
...
step35/99-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3e6882b3ac |
95
ARCHITECTURE.md
Normal file
95
ARCHITECTURE.md
Normal file
@@ -0,0 +1,95 @@
|
||||
# Architecture: STEP35-compounding-intelligence-99
|
||||
|
||||
**Generated by:** `scripts/architecture_doc_generator.py`
|
||||
|
||||
## Entry Points
|
||||
- `scripts/architecture_doc_generator.py`
|
||||
- `scripts/refactoring_opportunity_finder.py`
|
||||
- `scripts/automation_opportunity_finder.py`
|
||||
- `scripts/bootstrapper.py`
|
||||
- `scripts/dead_code_detector.py`
|
||||
- `scripts/dedup.py`
|
||||
- `scripts/dependency_graph.py`
|
||||
- `scripts/freshness.py`
|
||||
- `scripts/gitea_issue_parser.py`
|
||||
- `scripts/harvester.py`
|
||||
- `scripts/improvement_proposals.py`
|
||||
- `scripts/knowledge_staleness_check.py`
|
||||
- `scripts/perf_bottleneck_finder.py`
|
||||
- `scripts/pr_complexity_scorer.py`
|
||||
- `scripts/priority_rebalancer.py`
|
||||
- `quality_gate.py`
|
||||
- `scripts/sampler.py`
|
||||
- `scripts/session_metadata.py`
|
||||
- `scripts/session_pair_harvester.py`
|
||||
- `scripts/session_reader.py`
|
||||
- `scripts/test_automation_opportunity_finder.py`
|
||||
- `scripts/test_bootstrapper.py`
|
||||
- `scripts/test_diff_analyzer.py`
|
||||
- `tests/test_freshness.py`
|
||||
- `scripts/test_gitea_issue_parser.py`
|
||||
- `scripts/test_harvest_prompt.py`
|
||||
- `scripts/test_harvest_prompt_comprehensive.py`
|
||||
- `scripts/test_harvester_pipeline.py`
|
||||
- `scripts/test_improvement_proposals.py`
|
||||
- `tests/test_knowledge_gap_identifier.py`
|
||||
- `scripts/test_knowledge_staleness.py`
|
||||
- `tests/test_quality_gate.py`
|
||||
- `scripts/test_refactoring_opportunity_finder.py`
|
||||
- `scripts/test_session_pair_harvester.py`
|
||||
- `scripts/validate_knowledge.py`
|
||||
|
||||
## Module Dependencies
|
||||
| Module | Imports |
|
||||
|--------|---------|
|
||||
| `quality_gate` | `quality_gate` |
|
||||
| `scripts.harvester` | `scripts.session_reader` |
|
||||
| `scripts.session_metadata` | `scripts.session_reader` |
|
||||
| `scripts.test_bootstrapper` | `scripts.bootstrapper` |
|
||||
| `scripts.test_harvester_pipeline` | `scripts.harvester, scripts.session_reader` |
|
||||
| `scripts.test_pr_complexity_scorer` | `scripts.pr_complexity_scorer` |
|
||||
| `scripts.test_priority_rebalancer` | `scripts.priority_rebalancer` |
|
||||
| `scripts.test_session_pair_harvester` | `scripts.session_pair_harvester` |
|
||||
| `tests.test_dedup` | `scripts.dedup` |
|
||||
| `tests.test_knowledge_gap_identifier` | `scripts.knowledge_gap_identifier` |
|
||||
| `tests.test_perf_bottleneck_finder` | `scripts.perf_bottleneck_finder` |
|
||||
| `tests.test_quality_gate` | `quality_gate` |
|
||||
|
||||
## ASCII Diagram
|
||||
```
|
||||
*quality_gate*
|
||||
└─> quality_gate
|
||||
*scripts.bootstrapper*
|
||||
*scripts.dedup*
|
||||
*scripts.harvester*
|
||||
└─> scripts.session_reader
|
||||
[scripts.knowledge_gap_identifier]
|
||||
*scripts.perf_bottleneck_finder*
|
||||
*scripts.pr_complexity_scorer*
|
||||
*scripts.priority_rebalancer*
|
||||
*scripts.session_metadata*
|
||||
└─> scripts.session_reader
|
||||
*scripts.session_pair_harvester*
|
||||
*scripts.session_reader*
|
||||
*scripts.test_bootstrapper*
|
||||
└─> scripts.bootstrapper
|
||||
*scripts.test_harvester_pipeline*
|
||||
└─> scripts.harvester
|
||||
└─> scripts.session_reader
|
||||
[scripts.test_pr_complexity_scorer]
|
||||
└─> scripts.pr_complexity_scorer
|
||||
[scripts.test_priority_rebalancer]
|
||||
└─> scripts.priority_rebalancer
|
||||
*scripts.test_session_pair_harvester*
|
||||
└─> scripts.session_pair_harvester
|
||||
[tests.test_dedup]
|
||||
└─> scripts.dedup
|
||||
*tests.test_knowledge_gap_identifier*
|
||||
└─> scripts.knowledge_gap_identifier
|
||||
[tests.test_perf_bottleneck_finder]
|
||||
└─> scripts.perf_bottleneck_finder
|
||||
*tests.test_quality_gate*
|
||||
└─> quality_gate
|
||||
```
|
||||
|
||||
_Generated automatically. Keep this file in sync with code changes by re-running the generator._
|
||||
179
scripts/architecture_doc_generator.py
Executable file
179
scripts/architecture_doc_generator.py
Executable file
@@ -0,0 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Architecture Doc Generator — 4.4
|
||||
|
||||
Analyzes codebase structure and generates an architecture overview:
|
||||
- Maps module dependencies (Python imports within the repo)
|
||||
- Identifies entry points (main guards, CLI scripts)
|
||||
- Generates ASCII diagram of module relationships
|
||||
- Produces one ARCHITECTURE.md per repo
|
||||
|
||||
Usage:
|
||||
python3 scripts/architecture_doc_generator.py [repo_root]
|
||||
|
||||
If no repo_root given, uses current directory.
|
||||
Outputs ARCHITECTURE.md to the repo root.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def scan_python_files(root: Path):
|
||||
"""Find all .py files under root, excluding tests/ and .git/."""
|
||||
py_files = []
|
||||
for path in root.rglob("*.py"):
|
||||
parts = path.parts
|
||||
if any(p.startswith('.') for p in parts if p != '.'):
|
||||
continue
|
||||
if 'test' in parts:
|
||||
continue
|
||||
if any(x in parts for x in ('venv', 'node_modules', '__pycache__', 'dist', 'build')):
|
||||
continue
|
||||
py_files.append(path)
|
||||
return sorted(py_files)
|
||||
|
||||
|
||||
def module_id(path: Path, root: Path) -> str:
|
||||
"""Return a readable module identifier."""
|
||||
rel = path.relative_to(root)
|
||||
if rel.parent == Path('.'):
|
||||
return path.stem
|
||||
return str(rel.with_suffix('')).replace('/', '.')
|
||||
|
||||
|
||||
def extract_imports(path: Path) -> list[str]:
|
||||
"""Extract top-level import names from a Python file."""
|
||||
try:
|
||||
text = path.read_text(errors='ignore')
|
||||
except Exception:
|
||||
return []
|
||||
imports = set()
|
||||
# import X or import X.Y.Z
|
||||
for m in re.finditer(r'^\s*import\s+([a-zA-Z0-9_.]+)', text, re.MULTILINE):
|
||||
imports.add(m.group(1).split('.')[0])
|
||||
# from X import Y (handles absolute and relative: from .X import Y)
|
||||
for m in re.finditer(r'^\s*from\s+(\.+)?([a-zA-Z0-9_.]+)\s+import', text, re.MULTILINE):
|
||||
imports.add(m.group(2).split('.')[0])
|
||||
return sorted(imports)
|
||||
|
||||
|
||||
def build_dependency_graph(py_files: list[Path], root: Path) -> dict[str, set[str]]:
|
||||
"""Build adjacency: local_module -> set(local_modules it imports)."""
|
||||
graph = defaultdict(set)
|
||||
# Collect all local module identifiers
|
||||
local_ids = set()
|
||||
for p in py_files:
|
||||
local_ids.add(module_id(p, root))
|
||||
|
||||
for path in py_files:
|
||||
src_mod = module_id(path, root)
|
||||
for imp in extract_imports(path):
|
||||
# Match import to a local module by stem or by full dotted prefix
|
||||
target = None
|
||||
# Exact match
|
||||
if imp in local_ids:
|
||||
target = imp
|
||||
else:
|
||||
# Find module whose stem equals imp, or whose dotted name ends with .imp
|
||||
for mid in local_ids:
|
||||
if mid.split('.')[-1] == imp or mid == imp:
|
||||
target = mid
|
||||
break
|
||||
if target:
|
||||
graph[src_mod].add(target)
|
||||
|
||||
return {k: sorted(v) for k, v in graph.items()}
|
||||
|
||||
|
||||
def find_entry_points(py_files: list[Path]) -> list[Path]:
|
||||
"""Files with if __name__ == '__main__' guard or executable scripts."""
|
||||
entries = []
|
||||
for path in py_files:
|
||||
try:
|
||||
text = path.read_text(errors='ignore')
|
||||
except Exception:
|
||||
continue
|
||||
if 'if __name__' in text and '__main__' in text:
|
||||
entries.append(path)
|
||||
return sorted(entries, key=lambda p: (not (p.stat().st_mode & 0o111), p.name))
|
||||
|
||||
|
||||
def ascii_diagram(graph: dict[str, list[str]], entries: list[Path], root: Path) -> str:
|
||||
"""Generate a simple ASCII box-and-arrow diagram."""
|
||||
lines = []
|
||||
entry_names = {module_id(p, root) for p in entries}
|
||||
# All nodes
|
||||
nodes = sorted(set(graph.keys()) | set().union(*graph.values()))
|
||||
for node in nodes:
|
||||
is_entry = node in entry_names
|
||||
label = f"*{node}*" if is_entry else f"[{node}]"
|
||||
lines.append(label)
|
||||
for dep in graph.get(node, []):
|
||||
lines.append(f" └─> {dep}")
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def generate_markdown(root: Path, graph: dict, entries: list[Path], diagram: str) -> str:
|
||||
root_name = root.name
|
||||
md = []
|
||||
md.append(f"# Architecture: {root_name}")
|
||||
md.append("")
|
||||
md.append("**Generated by:** `scripts/architecture_doc_generator.py`")
|
||||
md.append("")
|
||||
md.append("## Entry Points")
|
||||
if entries:
|
||||
for p in entries:
|
||||
rel = p.relative_to(root)
|
||||
md.append(f"- `{rel}`")
|
||||
else:
|
||||
md.append("_No entry points detected._")
|
||||
md.append("")
|
||||
md.append("## Module Dependencies")
|
||||
if graph:
|
||||
md.append("| Module | Imports |")
|
||||
md.append("|--------|---------|")
|
||||
for mod in sorted(graph.keys()):
|
||||
deps = ', '.join(sorted(graph[mod])) if graph[mod] else '_none_'
|
||||
md.append(f"| `{mod}` | `{deps}` |")
|
||||
else:
|
||||
md.append("_No dependencies detected._")
|
||||
md.append("")
|
||||
md.append("## ASCII Diagram")
|
||||
md.append("```")
|
||||
md.append(diagram)
|
||||
md.append("```")
|
||||
md.append("")
|
||||
md.append("_Generated automatically. Keep this file in sync with code changes by re-running the generator._")
|
||||
return '\n'.join(md)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Generate architecture documentation")
|
||||
parser.add_argument("repo_root", nargs="?", default=".", help="Repository root (default: current directory)")
|
||||
args = parser.parse_args()
|
||||
|
||||
root = Path(args.repo_root).resolve()
|
||||
py_files = scan_python_files(root)
|
||||
if not py_files:
|
||||
print("No Python files found — nothing to do.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
graph = build_dependency_graph(py_files, root)
|
||||
entries = find_entry_points(py_files)
|
||||
diagram = ascii_diagram(graph, entries, root)
|
||||
markdown = generate_markdown(root, graph, entries, diagram)
|
||||
|
||||
out_path = root / "ARCHITECTURE.md"
|
||||
out_path.write_text(markdown, encoding='utf-8')
|
||||
print(f"Written: {out_path}")
|
||||
print(f" Modules scanned: {len(py_files)}")
|
||||
print(f" Entry points: {len(entries)}")
|
||||
print(f" Dependency edges: {sum(len(v) for v in graph.values())}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,170 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Graph Query Engine — traverse the knowledge graph.
|
||||
|
||||
Usage:
|
||||
python3 scripts/graph_query.py neighbors <fact_id> [--knowledge-dir knowledge/]
|
||||
python3 scripts/graph_query.py path <from_id> <to_id> [--max-hops 10]
|
||||
python3 scripts/graph_query.py subgraph <fact_id> [--depth 2]
|
||||
python3 scripts/graph_query.py stats # Graph statistics
|
||||
|
||||
Outputs JSON to stdout.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from collections import defaultdict, deque
|
||||
from typing import Optional
|
||||
|
||||
# --- Graph building ---
|
||||
|
||||
def load_index(knowledge_dir: Path) -> dict:
|
||||
index_path = knowledge_dir / "index.json"
|
||||
if not index_path.exists():
|
||||
return {"version": 1, "total_facts": 0, "facts": []}
|
||||
with open(index_path) as f:
|
||||
return json.load(f)
|
||||
|
||||
def build_adjacency(facts: list[dict]) -> dict:
|
||||
"""Build undirected adjacency list from fact 'related' fields."""
|
||||
adj = defaultdict(set)
|
||||
id_to_fact = {}
|
||||
for fact in facts:
|
||||
fid = fact.get("id")
|
||||
if not fid:
|
||||
continue
|
||||
id_to_fact[fid] = fact
|
||||
for related_id in fact.get("related", []):
|
||||
adj[fid].add(related_id)
|
||||
adj[related_id].add(fid) # undirected
|
||||
return dict(adj), id_to_fact
|
||||
|
||||
# --- Queries ---
|
||||
|
||||
def query_neighbors(fact_id: str, adj: dict, id_to_fact: dict) -> dict:
|
||||
"""Return directly connected facts."""
|
||||
neighbors = list(adj.get(fact_id, set()))
|
||||
return {
|
||||
"query": "neighbors",
|
||||
"fact_id": fact_id,
|
||||
"neighbors": [
|
||||
{"id": nid, "fact": id_to_fact.get(nid, {}).get("fact", ""), "category": id_to_fact.get(nid, {}).get("category", "")}
|
||||
for nid in neighbors if nid in id_to_fact
|
||||
],
|
||||
"count": len(neighbors),
|
||||
}
|
||||
|
||||
def query_path(from_id: str, to_id: str, adj: dict, max_hops: int = 10) -> dict:
|
||||
"""Find shortest path between two facts using BFS."""
|
||||
if from_id not in adj or to_id not in adj:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": None, "error": "Fact not found in graph"}
|
||||
|
||||
if from_id == to_id:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": [from_id], "length": 0}
|
||||
|
||||
queue = deque([(from_id, [from_id])])
|
||||
visited = {from_id}
|
||||
|
||||
while queue:
|
||||
current, path = queue.popleft()
|
||||
if len(path) > max_hops:
|
||||
continue
|
||||
for neighbor in adj.get(current, []):
|
||||
if neighbor == to_id:
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": path + [to_id], "length": len(path)}
|
||||
if neighbor not in visited:
|
||||
visited.add(neighbor)
|
||||
queue.append((neighbor, path + [neighbor]))
|
||||
|
||||
return {"query": "path", "from": from_id, "to": to_id, "path": None, "error": f"No path found within {max_hops} hops"}
|
||||
|
||||
def query_subgraph(fact_id: str, adj: dict, id_to_fact: dict, depth: int = 2) -> dict:
|
||||
"""Extract connected subgraph within N hops."""
|
||||
if fact_id not in adj:
|
||||
return {"query": "subgraph", "fact_id": fact_id, "nodes": [], "edges": [], "error": "Fact not found"}
|
||||
|
||||
visited = set()
|
||||
queue = deque([(fact_id, 0)])
|
||||
subgraph_nodes = set()
|
||||
subgraph_edges = []
|
||||
|
||||
while queue:
|
||||
node, d = queue.popleft()
|
||||
if node in visited or d > depth:
|
||||
continue
|
||||
visited.add(node)
|
||||
subgraph_nodes.add(node)
|
||||
for neighbor in adj.get(node, []):
|
||||
subgraph_edges.append({"source": node, "target": neighbor})
|
||||
if neighbor not in visited:
|
||||
queue.append((neighbor, d + 1))
|
||||
|
||||
return {
|
||||
"query": "subgraph",
|
||||
"fact_id": fact_id,
|
||||
"depth": depth,
|
||||
"nodes": [
|
||||
{"id": nid, "fact": id_to_fact.get(nid, {}).get("fact", ""), "category": id_to_fact.get(nid, {}).get("category", "")}
|
||||
for nid in sorted(subgraph_nodes)
|
||||
],
|
||||
"edges": [{"source": e["source"], "target": e["target"]} for e in subgraph_edges],
|
||||
"node_count": len(subgraph_nodes),
|
||||
"edge_count": len(subgraph_edges),
|
||||
}
|
||||
|
||||
def query_stats(adj: dict, id_to_fact: dict) -> dict:
|
||||
"""Graph statistics."""
|
||||
return {
|
||||
"statistics": {
|
||||
"total_facts": len(id_to_fact),
|
||||
"total_edges": sum(len(neighbors) for neighbors in adj.values()) // 2,
|
||||
"connected_components": 0, # TODO: compute if needed
|
||||
"average_degree": sum(len(neighbors) for neighbors in adj.values()) / len(adj) if adj else 0,
|
||||
}
|
||||
}
|
||||
|
||||
# --- CLI ---
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Graph query engine for knowledge store")
|
||||
parser.add_argument("command", choices=["neighbors", "path", "subgraph", "stats"])
|
||||
parser.add_argument("from_id", nargs="?", help="Starting fact ID")
|
||||
parser.add_argument("to_id", nargs="?", help="Target fact ID (for path query)")
|
||||
parser.add_argument("--knowledge-dir", default="knowledge", help="Knowledge directory")
|
||||
parser.add_argument("--depth", type=int, default=2, help="Depth for subgraph query")
|
||||
parser.add_argument("--max-hops", type=int, default=10, help="Max hops for path query")
|
||||
args = parser.parse_args()
|
||||
|
||||
start = time.time()
|
||||
knowledge_dir = Path(args.knowledge_dir)
|
||||
index = load_index(knowledge_dir)
|
||||
facts = index.get("facts", [])
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
|
||||
result = None
|
||||
if args.command == "neighbors":
|
||||
if not args.from_id:
|
||||
print("ERROR: neighbors requires <fact_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_neighbors(args.from_id, adj, id_to_fact)
|
||||
elif args.command == "path":
|
||||
if not args.from_id or not args.to_id:
|
||||
print("ERROR: path requires <from_id> <to_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_path(args.from_id, args.to_id, adj, max_hops=args.max_hops)
|
||||
elif args.command == "subgraph":
|
||||
if not args.from_id:
|
||||
print("ERROR: subgraph requires <fact_id>", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
result = query_subgraph(args.from_id, adj, id_to_fact, depth=args.depth)
|
||||
elif args.command == "stats":
|
||||
result = query_stats(adj, id_to_fact)
|
||||
|
||||
result["elapsed_ms"] = round((time.time() - start) * 1000, 2)
|
||||
print(json.dumps(result, indent=2))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,165 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tests for scripts/graph_query.py — Graph Query Engine.
|
||||
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from graph_query import load_index, build_adjacency, query_neighbors, query_path, query_subgraph, query_stats
|
||||
|
||||
|
||||
def make_index(facts: list[dict], tmp_dir: Path) -> Path:
|
||||
index = {
|
||||
"version": 1,
|
||||
"last_updated": "2026-04-13T20:00:00Z",
|
||||
"total_facts": len(facts),
|
||||
"facts": facts,
|
||||
}
|
||||
path = tmp_dir / "index.json"
|
||||
with open(path, "w") as f:
|
||||
json.dump(index, f)
|
||||
return path
|
||||
|
||||
|
||||
def test_neighbors():
|
||||
"""Neighbor query returns directly connected facts."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "category": "fact", "related": ["b", "c"]},
|
||||
{"id": "b", "fact": "B", "category": "fact", "related": ["a"]},
|
||||
{"id": "c", "fact": "C", "category": "fact", "related": ["a"]},
|
||||
{"id": "d", "fact": "D", "category": "fact", "related": []},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_neighbors("a", adj, id_to_fact)
|
||||
neighbor_ids = {n["id"] for n in result["neighbors"]}
|
||||
assert neighbor_ids == {"b", "c"}, f"Expected b,c got {neighbor_ids}"
|
||||
assert result["count"] == 2
|
||||
print("PASS: neighbors")
|
||||
|
||||
|
||||
def test_path_found():
|
||||
"""Path query finds shortest path."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b", "d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_path("a", "d", adj)
|
||||
assert result["path"] == ["a", "b", "c", "d"], f"Got path {result['path']}"
|
||||
assert result["length"] == 3
|
||||
print("PASS: path_found")
|
||||
|
||||
|
||||
def test_path_not_found():
|
||||
"""Path query returns error when no path exists."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a"]},
|
||||
{"id": "c", "fact": "C", "related": ["d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_path("a", "c", adj, max_hops=5)
|
||||
assert result["path"] is None
|
||||
assert "error" in result
|
||||
print("PASS: path_not_found")
|
||||
|
||||
|
||||
def test_subgraph_extraction():
|
||||
"""Subgraph extraction returns nodes within depth."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b", "c"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "d"]},
|
||||
{"id": "c", "fact": "C", "related": ["a"]},
|
||||
{"id": "d", "fact": "D", "related": ["b", "e"]},
|
||||
{"id": "e", "fact": "E", "related": ["d"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_subgraph("a", adj, id_to_fact, depth=1)
|
||||
node_ids = {n["id"] for n in result["nodes"]}
|
||||
assert node_ids == {"a", "b", "c"}, f"Got {node_ids}"
|
||||
assert result["node_count"] == 3
|
||||
print("PASS: subgraph_depth1")
|
||||
|
||||
|
||||
def test_subgraph_depth2():
|
||||
"""Depth-2 subgraph includes further nodes."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b", "d"]},
|
||||
{"id": "d", "fact": "D", "related": ["c"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_subgraph("a", adj, id_to_fact, depth=2)
|
||||
node_ids = {n["id"] for n in result["nodes"]}
|
||||
assert node_ids == {"a", "b", "c"}, f"Got {node_ids}"
|
||||
print("PASS: subgraph_depth2")
|
||||
|
||||
|
||||
def test_stats():
|
||||
"""Statistics query returns graph metrics."""
|
||||
facts = [
|
||||
{"id": "a", "fact": "A", "related": ["b"]},
|
||||
{"id": "b", "fact": "B", "related": ["a", "c"]},
|
||||
{"id": "c", "fact": "C", "related": ["b"]},
|
||||
]
|
||||
adj, id_to_fact = build_adjacency(facts)
|
||||
result = query_stats(adj, id_to_fact)
|
||||
assert result["statistics"]["total_facts"] == 3
|
||||
assert result["statistics"]["total_edges"] == 2 # undirected double-counted /2
|
||||
assert result["statistics"]["average_degree"] > 0
|
||||
print("PASS: stats")
|
||||
|
||||
|
||||
def test_cli_integration():
|
||||
"""CLI produces valid JSON with correct query types."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
import subprocess as sp
|
||||
tmp_dir = Path(tmp)
|
||||
facts = [
|
||||
{"id": "x", "fact": "X", "related": ["y"]},
|
||||
{"id": "y", "fact": "Y", "related": ["x", "z"]},
|
||||
{"id": "z", "fact": "Z", "related": ["y"]},
|
||||
]
|
||||
index_path = make_index(facts, tmp_dir)
|
||||
knowledge_dir = index_path.parent
|
||||
script_path = Path(__file__).resolve().parent / "graph_query.py"
|
||||
|
||||
result = sp.run(
|
||||
[sys.executable, str(script_path), "neighbors", "x", "--knowledge-dir", str(knowledge_dir)],
|
||||
capture_output=True, text=True, cwd=str(tmp_dir)
|
||||
)
|
||||
assert result.returncode == 0, f"neighbors failed: {result.stderr}"
|
||||
out = json.loads(result.stdout)
|
||||
assert out["query"] == "neighbors"
|
||||
assert out["fact_id"] == "x"
|
||||
assert out["count"] == 1
|
||||
|
||||
result = sp.run(
|
||||
[sys.executable, str(script_path), "path", "x", "z", "--knowledge-dir", str(knowledge_dir)],
|
||||
capture_output=True, text=True, cwd=str(tmp_dir)
|
||||
)
|
||||
assert result.returncode == 0, f"path failed: {result.stderr}"
|
||||
out = json.loads(result.stdout)
|
||||
assert out["path"] == ["x", "y", "z"]
|
||||
|
||||
print("PASS: cli_integration")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_neighbors()
|
||||
test_path_found()
|
||||
test_path_not_found()
|
||||
test_subgraph_extraction()
|
||||
test_subgraph_depth2()
|
||||
test_stats()
|
||||
test_cli_integration()
|
||||
print("\nAll graph_query tests passed!")
|
||||
Reference in New Issue
Block a user