Compare commits
1 Commits
step35/104
...
step35/199
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86eb1c9a50 |
@@ -1,176 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Doc Freshness Checker — Issue #104
|
||||
|
||||
Compare docs to code. Flag docs that reference removed functions or outdated APIs.
|
||||
|
||||
Usage:
|
||||
python3 scripts/doc_freshness.py [--root .] [--docs-dir .] [--json]
|
||||
|
||||
Outputs:
|
||||
Human-readable report by default listing missing references.
|
||||
JSON output with --json for machine consumption.
|
||||
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import ast
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Set, List, Tuple, Dict, Any
|
||||
|
||||
|
||||
def collect_python_symbols(repo_root: str) -> Set[str]:
|
||||
"""Collect all top-level function and class names from Python files."""
|
||||
symbols: Set[str] = set()
|
||||
for root, dirs, files in os.walk(repo_root):
|
||||
# Skip irrelevant dirs
|
||||
dirs[:] = [d for d in dirs if d not in ['.git', '__pycache__', '.venv', 'venv', 'node_modules']]
|
||||
for fname in files:
|
||||
if fname.endswith('.py'):
|
||||
path = os.path.join(root, fname)
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
tree = ast.parse(f.read())
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||||
symbols.add(node.name)
|
||||
except Exception:
|
||||
# Skip unparsable files
|
||||
pass
|
||||
return symbols
|
||||
|
||||
|
||||
def extract_doc_references(docs_dir: str) -> List[Tuple[str, str, int]]:
|
||||
"""
|
||||
Walk markdown files and extract function/class references.
|
||||
|
||||
Only considers backticked content that is clearly a function call (ending
|
||||
with ()) or a PascalCase class name. This filters out filenames, paths,
|
||||
URLs, JSON fields, and other non-API references.
|
||||
"""
|
||||
refs: List[Tuple[str, str, int]] = []
|
||||
backtick_pat = re.compile(r'`([^`]+)`')
|
||||
func_pat = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
|
||||
class_pat = re.compile(r'^[A-Z][a-zA-Z0-9_]*$')
|
||||
|
||||
for root, dirs, files in os.walk(docs_dir):
|
||||
dirs[:] = [d for d in dirs if d != '.git']
|
||||
for fname in files:
|
||||
if not fname.endswith('.md'):
|
||||
continue
|
||||
path = os.path.join(root, fname)
|
||||
rel_path = os.path.relpath(path, docs_dir)
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as fh:
|
||||
for lineno, line in enumerate(fh, 1):
|
||||
for m in backtick_pat.finditer(line):
|
||||
raw = m.group(1).strip()
|
||||
# Function call: ends with ()
|
||||
if raw.endswith('()'):
|
||||
name = raw[:-2].strip()
|
||||
if func_pat.fullmatch(name):
|
||||
refs.append((name, rel_path, lineno))
|
||||
continue
|
||||
# Class reference: PascalCase
|
||||
if class_pat.fullmatch(raw):
|
||||
refs.append((raw, rel_path, lineno))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return refs
|
||||
|
||||
|
||||
def check_doc_freshness(repo_root: str, docs_dir: str) -> Dict[str, Any]:
|
||||
"""Run the full check and return structured results."""
|
||||
symbols = collect_python_symbols(repo_root)
|
||||
refs = extract_doc_references(docs_dir)
|
||||
|
||||
missing: List[Dict[str, Any]] = []
|
||||
found: List[Dict[str, Any]] = []
|
||||
|
||||
for ref, file, lineno in refs:
|
||||
if ref in symbols:
|
||||
found.append({"reference": ref, "file": file, "line": lineno})
|
||||
else:
|
||||
missing.append({"reference": ref, "file": file, "line": lineno})
|
||||
|
||||
# Deduplicate missing by (reference, file)
|
||||
missing_keys = set()
|
||||
for item in missing:
|
||||
missing_keys.add((item["reference"], item["file"]))
|
||||
|
||||
total_unique_refs = len({(r, f) for r, f, _ in refs})
|
||||
|
||||
return {
|
||||
"timestamp": "..", # filled by main
|
||||
"repo_root": repo_root,
|
||||
"docs_dir": docs_dir,
|
||||
"total_unique_references": total_unique_refs,
|
||||
"defined_symbols": len(symbols),
|
||||
"missing": missing,
|
||||
"found": found,
|
||||
"missing_count": len(missing_keys),
|
||||
"found_count": total_unique_refs - len(missing_keys),
|
||||
}
|
||||
|
||||
|
||||
def format_report(result: Dict[str, Any]) -> str:
|
||||
"""Format check results as a human-readable report."""
|
||||
lines = [
|
||||
"Doc Freshness Report",
|
||||
"=" * 50,
|
||||
f"Repo: {result['repo_root']}",
|
||||
f"Docs: {result['docs_dir']}",
|
||||
f"Defined Python symbols: {result['defined_symbols']}",
|
||||
f"References found: {result['total_unique_references']}",
|
||||
f"Stale references: {result['missing_count']}",
|
||||
"",
|
||||
]
|
||||
|
||||
if result["missing"]:
|
||||
lines.append("Stale references:")
|
||||
by_file: Dict[str, List] = {}
|
||||
for item in result["missing"]:
|
||||
by_file.setdefault(item["file"], []).append(item)
|
||||
for fname in sorted(by_file):
|
||||
lines.append(f"\n {fname}:")
|
||||
for item in by_file[fname]:
|
||||
lines.append(f" line {item['line']}: {item['reference']}")
|
||||
else:
|
||||
lines.append("All references are current.")
|
||||
|
||||
lines.append("")
|
||||
lines.append("Note: Only backticked function calls () and PascalCase class names are checked.")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Doc Freshness Checker — compare docs to code")
|
||||
parser.add_argument("--root", default=".", help="Repository root (code location)")
|
||||
parser.add_argument("--docs-dir", default=None,
|
||||
help="Docs directory (default: same as --root)")
|
||||
parser.add_argument("--json", action="store_true", help="Machine-readable output")
|
||||
args = parser.parse_args()
|
||||
|
||||
docs_dir = args.docs_dir or args.root
|
||||
|
||||
result = check_doc_freshness(args.root, docs_dir)
|
||||
result["timestamp"] = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
if args.json:
|
||||
print(json.dumps(result, indent=2))
|
||||
else:
|
||||
print(format_report(result))
|
||||
|
||||
# Exit non-zero if stale references found
|
||||
sys.exit(1 if result["missing_count"] > 0 else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
255
scripts/knowledge_to_training_pairs.py
Normal file
255
scripts/knowledge_to_training_pairs.py
Normal file
@@ -0,0 +1,255 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
knowledge_to_training_pairs.py — Convert quality-gated knowledge entries into training pairs.
|
||||
|
||||
Reads knowledge/index.json (or a custom JSONL of entries), applies quality filters,
|
||||
and emits terse→rich training pairs in JSONL format for model fine-tuning.
|
||||
|
||||
Usage:
|
||||
python3 scripts/knowledge_to_training_pairs.py \
|
||||
--input knowledge/index.json \
|
||||
--output training_pairs.jsonl \
|
||||
--min-confidence 0.7 \
|
||||
--model-filter claude-sonnet,gpt-4 \
|
||||
--after 2026-01-01
|
||||
|
||||
Input entry format (from index.json facts):
|
||||
{
|
||||
"id": "hermes-agent:pitfall:001",
|
||||
"fact": "deploy-crons.py leaves jobs in mixed model format",
|
||||
"category": "pitfall",
|
||||
"domain": "hermes-agent",
|
||||
"confidence": 0.95,
|
||||
...
|
||||
}
|
||||
|
||||
Output training pair format:
|
||||
{
|
||||
"terse": "How do I handle deploy-crons.py mixed model format?",
|
||||
"rich": "deploy-crons.py leaves jobs in mixed model format.",
|
||||
"domain": "hermes-agent",
|
||||
"source_confidence": 0.95,
|
||||
"source_model": "unknown"
|
||||
}
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def fact_to_terse(fact: str, category: str, domain: str) -> str:
|
||||
"""
|
||||
Derive a short user query from a knowledge fact.
|
||||
|
||||
Strategy:
|
||||
- Pitfalls → "How do I avoid/handle/fix <fact excerpt>?"
|
||||
- Patterns → "What's the recommended way to <pattern core>?"
|
||||
- Tool quirks → "How does <tool> behave in <context>?"
|
||||
- Facts → "What should I know about <fact excerpt>?"
|
||||
- Questions → "What is the answer to: <fact>?"
|
||||
"""
|
||||
fact_lower = fact.lower()
|
||||
# Extract a concise excerpt (first sentence or 80 chars)
|
||||
excerpt = fact.split('. ')[0] if '. ' in fact else fact[:80]
|
||||
|
||||
if category == "pitfall":
|
||||
verbs = ["avoid", "handle", "fix", "prevent"]
|
||||
# pick verb based on fact wording
|
||||
if "trigger" in fact_lower or "cause" in fact_lower:
|
||||
verb = "avoid"
|
||||
elif "broken" in fact_lower or "fails" in fact_lower:
|
||||
verb = "fix"
|
||||
else:
|
||||
verb = "handle"
|
||||
return f"How do I {verb} {excerpt.rstrip('.')}?"
|
||||
elif category == "pattern":
|
||||
return f"What's the recommended way to {excerpt.rstrip('.')}?"
|
||||
elif category == "tool-quirk":
|
||||
# Try to extract tool name
|
||||
tool = fact.split()[0] if fact.split() else domain
|
||||
return f"How does {tool} behave in this context?"
|
||||
elif category == "question":
|
||||
return f"What is the answer to: {excerpt}?"
|
||||
else: # fact or unknown
|
||||
return f"What should I know about {excerpt.rstrip('.')}?"
|
||||
|
||||
|
||||
def parse_date(date_str: Optional[str]) -> Optional[datetime]:
|
||||
"""Parse ISO date string to datetime, or return None."""
|
||||
if not date_str:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(date_str.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def load_knowledge_index(path: str) -> list[dict]:
|
||||
"""Load knowledge facts from index.json (or plain JSONL of entries)."""
|
||||
p = Path(path)
|
||||
if not p.exists():
|
||||
print(f"ERROR: Knowledge input not found: {path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
with open(p) as f:
|
||||
data = json.load(f)
|
||||
|
||||
# index.json format: {"facts": [...], ...}
|
||||
if isinstance(data, dict) and "facts" in data:
|
||||
return data["facts"]
|
||||
# JSONL format: one entry per line
|
||||
if isinstance(data, list):
|
||||
return data
|
||||
# Plain file with JSON array
|
||||
print(f"ERROR: Unrecognized input format in {path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def filter_entries(entries: list[dict],
|
||||
min_confidence: float = 0.0,
|
||||
model_filter: Optional[list[str]] = None,
|
||||
after: Optional[datetime] = None,
|
||||
before: Optional[datetime] = None) -> list[dict]:
|
||||
"""Apply quality and provenance filters."""
|
||||
filtered = []
|
||||
for entry in entries:
|
||||
# Confidence filter (entry confidence)
|
||||
conf = entry.get("confidence", 0.0)
|
||||
if conf < min_confidence:
|
||||
continue
|
||||
|
||||
# Model filter: if specified, entry's model must be in the list
|
||||
if model_filter:
|
||||
entry_model = entry.get("model", entry.get("provenance", {}).get("model", "unknown"))
|
||||
if entry_model not in model_filter:
|
||||
continue
|
||||
|
||||
# Date filter: use last_confirmed or first_seen or harvested_at
|
||||
entry_date = None
|
||||
for field in ("last_confirmed", "first_seen", "harvested_at"):
|
||||
if field in entry:
|
||||
entry_date = parse_date(entry[field])
|
||||
if entry_date:
|
||||
break
|
||||
if after and entry_date and entry_date < after:
|
||||
continue
|
||||
if before and entry_date and entry_date > before:
|
||||
continue
|
||||
|
||||
filtered.append(entry)
|
||||
return filtered
|
||||
|
||||
|
||||
def entry_to_pair(entry: dict) -> dict:
|
||||
"""Convert a knowledge entry into a training pair."""
|
||||
fact = entry.get("fact", "").strip()
|
||||
if not fact:
|
||||
return None
|
||||
|
||||
category = entry.get("category", "fact")
|
||||
domain = entry.get("domain", "global")
|
||||
|
||||
terse = fact_to_terse(fact, category, domain)
|
||||
rich = fact
|
||||
source_confidence = round(entry.get("confidence", 0.0), 4)
|
||||
source_model = entry.get("model", entry.get("provenance", {}).get("model", "unknown"))
|
||||
|
||||
return {
|
||||
"terse": terse,
|
||||
"rich": rich,
|
||||
"domain": domain,
|
||||
"source_confidence": source_confidence,
|
||||
"source_model": source_model,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Knowledge entries → training pairs")
|
||||
parser.add_argument("--input", "-i", default="knowledge/index.json",
|
||||
help="Input knowledge index or JSONL (default: knowledge/index.json)")
|
||||
parser.add_argument("--output", "-o", default="training_pairs.jsonl",
|
||||
help="Output JSONL file")
|
||||
parser.add_argument("--min-confidence", type=float, default=0.5,
|
||||
help="Minimum entry confidence to include (0.0-1.0, default: 0.5)")
|
||||
parser.add_argument("--model-filter",
|
||||
help="Comma-separated list of source models to include")
|
||||
parser.add_argument("--after",
|
||||
help="Include entries last_confirmed/first_seen on or after this date (YYYY-MM-DD)")
|
||||
parser.add_argument("--before",
|
||||
help="Include entries last_confirmed/first_seen on or before this date (YYYY-MM-DD)")
|
||||
parser.add_argument("--dry-run", action="store_true",
|
||||
help="Print sample pairs and stats without writing")
|
||||
args = parser.parse_args()
|
||||
|
||||
# Load
|
||||
entries = load_knowledge_index(args.input)
|
||||
print(f"Loaded {len(entries)} entries from {args.input}", file=sys.stderr)
|
||||
|
||||
# Parse filters
|
||||
model_list = args.model_filter.split(",") if args.model_filter else None
|
||||
after_dt = parse_date(args.after) if args.after else None
|
||||
before_dt = parse_date(args.before) if args.before else None
|
||||
|
||||
# Filter
|
||||
kept = filter_entries(
|
||||
entries,
|
||||
min_confidence=args.min_confidence,
|
||||
model_filter=model_list,
|
||||
after=after_dt,
|
||||
before=before_dt,
|
||||
)
|
||||
print(f"After filtering: {len(kept)} / {len(entries)} entries", file=sys.stderr)
|
||||
|
||||
# Convert
|
||||
pairs = []
|
||||
for entry in kept:
|
||||
pair = entry_to_pair(entry)
|
||||
if pair:
|
||||
pairs.append(pair)
|
||||
|
||||
# Stats
|
||||
if pairs:
|
||||
avg_conf = sum(p["source_confidence"] for p in pairs) / len(pairs)
|
||||
domains = {}
|
||||
models = {}
|
||||
for p in pairs:
|
||||
domains[p["domain"]] = domains.get(p["domain"], 0) + 1
|
||||
models[p["source_model"]] = models.get(p["source_model"], 0) + 1
|
||||
else:
|
||||
avg_conf = 0.0
|
||||
domains = {}
|
||||
models = {}
|
||||
|
||||
stats = {
|
||||
"input_entries": len(entries),
|
||||
"after_filter": len(kept),
|
||||
"pairs_generated": len(pairs),
|
||||
"avg_confidence": round(avg_conf, 4),
|
||||
"domains": domains,
|
||||
"source_models": models,
|
||||
}
|
||||
print(json.dumps(stats, indent=2), file=sys.stderr)
|
||||
|
||||
if args.dry_run:
|
||||
print("\nSample pairs:", file=sys.stderr)
|
||||
for p in pairs[:3]:
|
||||
print(json.dumps(p, ensure_ascii=False), file=sys.stderr)
|
||||
return
|
||||
|
||||
# Write JSONL
|
||||
out_path = Path(args.output)
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
for pair in pairs:
|
||||
f.write(json.dumps(pair, ensure_ascii=False) + "\n")
|
||||
|
||||
print(f"\nWrote {len(pairs)} training pairs to {out_path}", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,89 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for scripts/doc_freshness.py — Issue #104."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
|
||||
|
||||
import doc_freshness as df
|
||||
|
||||
|
||||
def test_collect_python_symbols():
|
||||
"""Should collect function and class names from Python files."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Create a simple Python file
|
||||
py_path = os.path.join(tmpdir, "sample.py")
|
||||
with open(py_path, "w") as f:
|
||||
f.write('''
|
||||
def my_func():
|
||||
pass
|
||||
|
||||
class MyClass:
|
||||
def method(self):
|
||||
pass
|
||||
|
||||
async def my_async():
|
||||
pass
|
||||
''')
|
||||
symbols = df.collect_python_symbols(tmpdir)
|
||||
assert "my_func" in symbols
|
||||
assert "MyClass" in symbols
|
||||
assert "my_async" in symbols
|
||||
# method (inside class) is also collected and should be considered valid
|
||||
assert "method" in symbols
|
||||
print("PASS: test_collect_python_symbols")
|
||||
|
||||
|
||||
def test_extract_doc_references_function_and_class():
|
||||
"""Should extract only function calls () and PascalCase class refs."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
docs = os.path.join(tmpdir, "docs")
|
||||
os.makedirs(docs)
|
||||
md_path = os.path.join(docs, "test.md")
|
||||
with open(md_path, "w") as f:
|
||||
f.write('''
|
||||
# Test
|
||||
|
||||
`call_this()` is a function.
|
||||
`SomeClass` is a class.
|
||||
`not_a_function` (lowercase, no parens) should be ignored.
|
||||
`filename.py` should be ignored.
|
||||
`https://example.com` ignored.
|
||||
''')
|
||||
refs = df.extract_doc_references(docs)
|
||||
names = [r[0] for r in refs]
|
||||
assert "call_this" in names
|
||||
assert "SomeClass" in names
|
||||
assert "not_a_function" not in names
|
||||
assert "filename" not in names # filename.py filtered
|
||||
assert "https" not in names
|
||||
print("PASS: test_extract_doc_references_function_and_class")
|
||||
|
||||
|
||||
def test_check_doc_freshness_missing_detection():
|
||||
"""Should detect missing symbols."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Code with one function
|
||||
code_dir = os.path.join(tmpdir, "code")
|
||||
os.makedirs(code_dir)
|
||||
with open(os.path.join(code_dir, "a.py"), "w") as f:
|
||||
f.write("def existing_func(): pass\n")
|
||||
# Docs reference existing_func and missing_func
|
||||
docs_dir = os.path.join(tmpdir, "docs")
|
||||
os.makedirs(docs_dir)
|
||||
with open(os.path.join(docs_dir, "readme.md"), "w") as f:
|
||||
f.write("`existing_func()` and `missing_func()` are mentioned.")
|
||||
result = df.check_doc_freshness(code_dir, docs_dir)
|
||||
assert result["missing_count"] == 1
|
||||
assert result["found_count"] == 1
|
||||
print("PASS: test_check_doc_freshness_missing_detection")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_collect_python_symbols()
|
||||
test_extract_doc_references_function_and_class()
|
||||
test_check_doc_freshness_missing_detection()
|
||||
print("All tests passed!")
|
||||
174
tests/test_knowledge_to_training_pairs.py
Normal file
174
tests/test_knowledge_to_training_pairs.py
Normal file
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Smoke tests for knowledge_to_training_pairs.py
|
||||
|
||||
Tests:
|
||||
- Output is valid JSONL
|
||||
- Each line has required fields (terse, rich, domain, source_confidence, source_model)
|
||||
- Confidence values are in [0,1]
|
||||
- Terse is non-empty and reasonably short (< 200 chars)
|
||||
- Rich matches the original fact
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# Add scripts dir to path for imports
|
||||
SCRIPT_DIR = Path(__file__).parent.parent / "scripts"
|
||||
sys.path.insert(0, str(SCRIPT_DIR))
|
||||
|
||||
from knowledge_to_training_pairs import (
|
||||
fact_to_terse,
|
||||
filter_entries,
|
||||
entry_to_pair,
|
||||
parse_date,
|
||||
)
|
||||
|
||||
|
||||
def test_fact_to_terse_pitfall():
|
||||
fact = "deploy-crons.py leaves jobs in mixed model format"
|
||||
category = "pitfall"
|
||||
domain = "hermes-agent"
|
||||
terse = fact_to_terse(fact, category, domain)
|
||||
assert terse.startswith("How do I")
|
||||
assert "?" in terse
|
||||
assert len(terse) < 150
|
||||
print("PASS: test_fact_to_terse_pitfall")
|
||||
|
||||
|
||||
def test_fact_to_terse_fact():
|
||||
fact = "Python is a high-level programming language"
|
||||
terse = fact_to_terse(fact, "fact", "global")
|
||||
assert terse.startswith("What should I know about")
|
||||
assert "?" in terse
|
||||
print("PASS: test_fact_to_terse_fact")
|
||||
|
||||
|
||||
def test_fact_to_terse_pattern():
|
||||
fact = "Use sparse checkout for large repos"
|
||||
terse = fact_to_terse(fact, "pattern", "devops")
|
||||
assert "recommended way" in terse or "best way" in terse
|
||||
print("PASS: test_fact_to_terse_pattern")
|
||||
|
||||
|
||||
def test_entry_to_pair_structure():
|
||||
entry = {
|
||||
"id": "test:001",
|
||||
"fact": "Test fact text.",
|
||||
"category": "fact",
|
||||
"domain": "test-domain",
|
||||
"confidence": 0.85,
|
||||
"model": "test-model",
|
||||
}
|
||||
pair = entry_to_pair(entry)
|
||||
assert pair is not None
|
||||
assert "terse" in pair
|
||||
assert "rich" in pair
|
||||
assert "domain" in pair
|
||||
assert "source_confidence" in pair
|
||||
assert "source_model" in pair
|
||||
assert pair["rich"] == "Test fact text."
|
||||
assert pair["domain"] == "test-domain"
|
||||
assert 0.0 <= pair["source_confidence"] <= 1.0
|
||||
print("PASS: test_entry_to_pair_structure")
|
||||
|
||||
|
||||
def test_filter_by_confidence():
|
||||
entries = [
|
||||
{"fact": "A", "confidence": 0.9},
|
||||
{"fact": "B", "confidence": 0.4},
|
||||
{"fact": "C", "confidence": 0.6},
|
||||
]
|
||||
filtered = filter_entries(entries, min_confidence=0.5)
|
||||
assert len(filtered) == 2
|
||||
assert all(e["confidence"] >= 0.5 for e in filtered)
|
||||
print("PASS: test_filter_by_confidence")
|
||||
|
||||
|
||||
def test_filter_by_model():
|
||||
entries = [
|
||||
{"fact": "A", "model": "claude-sonnet"},
|
||||
{"fact": "B", "model": "gpt-4"},
|
||||
{"fact": "C", "model": "unknown"},
|
||||
]
|
||||
filtered = filter_entries(entries, model_filter=["claude-sonnet", "gpt-4"])
|
||||
assert len(filtered) == 2
|
||||
assert all(e["model"] in ("claude-sonnet", "gpt-4") for e in filtered)
|
||||
print("PASS: test_filter_by_model")
|
||||
|
||||
|
||||
def test_filter_by_date():
|
||||
entries = [
|
||||
{"fact": "A", "last_confirmed": "2026-04-10"},
|
||||
{"fact": "B", "last_confirmed": "2026-03-01"},
|
||||
{"fact": "C", "first_seen": "2026-04-15"},
|
||||
]
|
||||
after_dt = parse_date("2026-04-01")
|
||||
filtered = filter_entries(entries, after=after_dt)
|
||||
assert len(filtered) == 2
|
||||
print("PASS: test_filter_by_date")
|
||||
|
||||
|
||||
def test_end_to_end_jsonl_output():
|
||||
"""Integration test: run the script and verify JSONL validity."""
|
||||
import subprocess
|
||||
|
||||
repo_dir = SCRIPT_DIR.parent
|
||||
result = subprocess.run(
|
||||
["python3", "scripts/knowledge_to_training_pairs.py", "--dry-run"],
|
||||
capture_output=True, text=True, cwd=repo_dir
|
||||
)
|
||||
assert result.returncode == 0
|
||||
stderr = result.stderr.strip()
|
||||
|
||||
# The stats JSON object is at the top of stderr. Find its bounds via brace matching.
|
||||
start = stderr.find('{')
|
||||
assert start >= 0, "Stats JSON not found in stderr"
|
||||
stderr_sub = stderr[start:]
|
||||
|
||||
depth = 0
|
||||
end = 0
|
||||
for i, ch in enumerate(stderr_sub):
|
||||
if ch == '{':
|
||||
depth += 1
|
||||
elif ch == '}':
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
end = i + 1
|
||||
break
|
||||
assert end > 0, "Unterminated JSON in stderr"
|
||||
|
||||
stats = json.loads(stderr_sub[:end])
|
||||
assert stats["input_entries"] > 0
|
||||
assert stats["pairs_generated"] > 0
|
||||
print("PASS: test_end_to_end_jsonl_output")
|
||||
|
||||
|
||||
def test_terse_length_constraint():
|
||||
"""Terse should be reasonably short for training."""
|
||||
# Sample facts from actual knowledge
|
||||
test_facts = [
|
||||
("deploy-crons.py leaves jobs in mixed model format", "pitfall", "hermes-agent"),
|
||||
("Cron jobs with blank fallback_model fields trigger warnings", "pitfall", "hermes-agent"),
|
||||
("Use the Gitea REST API when clone times out", "pattern", "devops"),
|
||||
]
|
||||
for fact, cat, domain in test_facts:
|
||||
terse = fact_to_terse(fact, cat, domain)
|
||||
assert len(terse) < 200, f"Terse too long ({len(terse)}): {terse}"
|
||||
print("PASS: test_terse_length_constraint")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_fact_to_terse_pitfall()
|
||||
test_fact_to_terse_fact()
|
||||
test_fact_to_terse_pattern()
|
||||
test_entry_to_pair_structure()
|
||||
test_filter_by_confidence()
|
||||
test_filter_by_model()
|
||||
test_filter_by_date()
|
||||
test_end_to_end_jsonl_output()
|
||||
test_terse_length_constraint()
|
||||
print("\nAll smoke tests passed.")
|
||||
Reference in New Issue
Block a user