fix(#201): rewrite comprehensive tests with proper pytest-compatible functions

scripts/dedup.py

@@ -1,317 +0,0 @@
-#!/usr/bin/env python3
-"""
-dedup.py — Knowledge deduplication: content hash + semantic similarity.
-
-Deduplicates harvested knowledge entries to avoid training on duplicates.
-Uses content hashing for exact matches and token overlap for near-duplicates.
-
-Usage:
-    python3 dedup.py --input knowledge/index.json --output knowledge/index_deduped.json
-    python3 dedup.py --input knowledge/index.json --dry-run
-    python3 dedup.py --test  # Run built-in dedup test
-"""
-
-import argparse
-import hashlib
-import json
-import re
-import sys
-from pathlib import Path
-from typing import List, Dict, Optional, Tuple
-
-
-def normalize_text(text: str) -> str:
-    """Normalize text for hashing: lowercase, collapse whitespace, strip."""
-    text = text.lower().strip()
-    text = re.sub(r'\s+', ' ', text)
-    return text
-
-
-def content_hash(text: str) -> str:
-    """SHA256 hash of normalized text for exact dedup."""
-    normalized = normalize_text(text)
-    return hashlib.sha256(normalized.encode('utf-8')).hexdigest()
-
-
-def tokenize(text: str) -> set:
-    """Simple tokenizer: lowercase words, 3+ chars."""
-    words = re.findall(r'[a-z0-9_]{3,}', text.lower())
-    return set(words)
-
-
-def token_similarity(a: str, b: str) -> float:
-    """Token-based Jaccard similarity (0.0-1.0).
-
-    Fast local alternative to embedding similarity.
-    Good enough for near-duplicate detection.
-    """
-    tokens_a = tokenize(a)
-    tokens_b = tokenize(b)
-    if not tokens_a or not tokens_b:
-        return 0.0
-    intersection = tokens_a & tokens_b
-    union = tokens_a | tokens_b
-    return len(intersection) / len(union)
-
-
-def quality_score(fact: dict) -> float:
-    """Compute quality score for merge ranking.
-
-    Higher is better. Factors:
-    - confidence (0-1)
-    - source_count (more confirmations = better)
-    - has tags (richer metadata)
-    """
-    confidence = fact.get('confidence', 0.5)
-    source_count = fact.get('source_count', 1)
-    has_tags = 1.0 if fact.get('tags') else 0.0
-    has_related = 1.0 if fact.get('related') else 0.0
-
-    # Weighted composite
-    score = (
-        confidence * 0.5 +
-        min(source_count / 10, 1.0) * 0.3 +
-        has_tags * 0.1 +
-        has_related * 0.1
-    )
-    return round(score, 4)
-
-
-def merge_facts(keep: dict, drop: dict) -> dict:
-    """Merge two near-duplicate facts, keeping higher-quality fields.
-
-    The 'keep' fact is enriched with metadata from 'drop'.
-    """
-    # Merge tags (union)
-    keep_tags = set(keep.get('tags', []))
-    drop_tags = set(drop.get('tags', []))
-    keep['tags'] = sorted(keep_tags | drop_tags)
-
-    # Merge related (union)
-    keep_related = set(keep.get('related', []))
-    drop_related = set(drop.get('related', []))
-    keep['related'] = sorted(keep_related | drop_related)
-
-    # Update source_count (sum)
-    keep['source_count'] = keep.get('source_count', 1) + drop.get('source_count', 1)
-
-    # Update confidence (max — we've now seen it from multiple sources)
-    keep['confidence'] = max(keep.get('confidence', 0), drop.get('confidence', 0))
-
-    # Track that we merged
-    if '_merged_from' not in keep:
-        keep['_merged_from'] = []
-    keep['_merged_from'].append(drop.get('id', 'unknown'))
-
-    return keep
-
-
-def dedup_facts(
-    facts: List[dict],
-    exact_threshold: float = 1.0,
-    near_threshold: float = 0.95,
-    dry_run: bool = False,
-) -> Tuple[List[dict], dict]:
-    """Deduplicate a list of knowledge facts.
-
-    Args:
-        facts: List of fact dicts (from index.json)
-        exact_threshold: Hash match = exact duplicate
-        near_threshold: Token similarity above this = near-duplicate
-        dry_run: If True, don't modify, just report
-
-    Returns:
-        (deduped_facts, stats_dict)
-    """
-    if not facts:
-        return [], {"total": 0, "exact_dupes": 0, "near_dupes": 0, "unique": 0}
-
-    # Phase 1: Exact dedup by content hash
-    hash_seen = {}  # hash -> index in deduped list
-    exact_dupes = 0
-    deduped = []
-
-    for fact in facts:
-        text = fact.get('fact', '')
-        h = content_hash(text)
-
-        if h in hash_seen:
-            # Exact duplicate — merge metadata into existing
-            existing_idx = hash_seen[h]
-            if not dry_run:
-                deduped[existing_idx] = merge_facts(deduped[existing_idx], fact)
-            exact_dupes += 1
-        else:
-            hash_seen[h] = len(deduped)
-            deduped.append(fact)
-
-    # Phase 2: Near-dup by token similarity
-    near_dupes = 0
-    i = 0
-    while i < len(deduped):
-        j = i + 1
-        while j < len(deduped):
-            sim = token_similarity(deduped[i].get('fact', ''), deduped[j].get('fact', ''))
-            if sim >= near_threshold:
-                # Near-duplicate — keep higher quality
-                q_i = quality_score(deduped[i])
-                q_j = quality_score(deduped[j])
-                if q_i >= q_j:
-                    if not dry_run:
-                        deduped[i] = merge_facts(deduped[i], deduped[j])
-                    deduped.pop(j)
-                else:
-                    # j is higher quality — merge i into j, then remove i
-                    if not dry_run:
-                        deduped[j] = merge_facts(deduped[j], deduped[i])
-                    deduped.pop(i)
-                    break  # i changed, restart inner loop
-                near_dupes += 1
-            else:
-                j += 1
-        i += 1
-
-    stats = {
-        "total": len(facts),
-        "exact_dupes": exact_dupes,
-        "near_dupes": near_dupes,
-        "unique": len(deduped),
-        "removed": len(facts) - len(deduped),
-    }
-
-    return deduped, stats
-
-
-def dedup_index_file(
-    input_path: str,
-    output_path: Optional[str] = None,
-    near_threshold: float = 0.95,
-    dry_run: bool = False,
-) -> dict:
-    """Deduplicate an index.json file.
-
-    Args:
-        input_path: Path to index.json
-        output_path: Where to write deduped file (default: overwrite input)
-        near_threshold: Token similarity threshold for near-dupes
-        dry_run: Report only, don't write
-
-    Returns stats dict.
-    """
-    path = Path(input_path)
-    if not path.exists():
-        raise FileNotFoundError(f"Index file not found: {input_path}")
-
-    with open(path) as f:
-        data = json.load(f)
-
-    facts = data.get('facts', [])
-    deduped, stats = dedup_facts(facts, near_threshold=near_threshold, dry_run=dry_run)
-
-    if not dry_run:
-        data['facts'] = deduped
-        data['total_facts'] = len(deduped)
-        data['last_dedup'] = __import__('datetime').datetime.now(
-            __import__('datetime').timezone.utc
-        ).isoformat()
-
-        out_path = Path(output_path) if output_path else path
-        with open(out_path, 'w') as f:
-            json.dump(data, f, indent=2, ensure_ascii=False)
-
-    return stats
-
-
-def generate_test_duplicates(n: int = 20) -> List[dict]:
-    """Generate test facts with intentional duplicates for testing.
-
-    Creates n unique facts plus n/4 exact dupes and n/4 near-dupes.
-    """
-    import random
-    random.seed(42)
-
-    unique_facts = []
-    for i in range(n):
-        topic = random.choice(["git", "python", "docker", "rust", "nginx"])
-        tip = random.choice(["use verbose flags", "check logs first", "restart service", "clear cache", "update config"])
-        unique_facts.append({
-            "id": f"test:fact:{i:03d}",
-            "fact": f"When working with {topic}, always {tip} before deploying.",
-            "category": "fact",
-            "domain": "test",
-            "confidence": round(random.uniform(0.5, 1.0), 2),
-            "source_count": random.randint(1, 5),
-            "tags": [topic, "test"],
-        })
-
-    # Add exact duplicates (same text, different IDs)
-    duped = list(unique_facts)
-    for i in range(n // 4):
-        original = unique_facts[i]
-        dupe = dict(original)
-        dupe["id"] = f"test:fact:dup{i:03d}"
-        dupe["confidence"] = round(random.uniform(0.3, 0.8), 2)
-        duped.append(dupe)
-
-    # Add near-duplicates (slightly different phrasing)
-    for i in range(n // 4):
-        original = unique_facts[i]
-        near = dict(original)
-        near["id"] = f"test:fact:near{i:03d}"
-        near["fact"] = original["fact"].replace("always", "should").replace("before deploying", "prior to deployment")
-        near["confidence"] = round(random.uniform(0.4, 0.9), 2)
-        duped.append(near)
-
-    return duped
-
-
-def main():
-    parser = argparse.ArgumentParser(description="Knowledge deduplication")
-    parser.add_argument("--input", help="Path to index.json")
-    parser.add_argument("--output", help="Output path (default: overwrite input)")
-    parser.add_argument("--threshold", type=float, default=0.95,
-                        help="Near-dup similarity threshold (default: 0.95)")
-    parser.add_argument("--dry-run", action="store_true", help="Report only, don't write")
-    parser.add_argument("--test", action="store_true", help="Run built-in dedup test")
-    parser.add_argument("--json", action="store_true", help="JSON output")
-    args = parser.parse_args()
-
-    if args.test:
-        test_facts = generate_test_duplicates(20)
-        print(f"Generated {len(test_facts)} test facts (20 unique + dupes)")
-        deduped, stats = dedup_facts(test_facts, near_threshold=args.threshold)
-        print(f"\nDedup results:")
-        print(f"  Total input:    {stats['total']}")
-        print(f"  Exact dupes:    {stats['exact_dupes']}")
-        print(f"  Near dupes:     {stats['near_dupes']}")
-        print(f"  Unique output:  {stats['unique']}")
-        print(f"  Removed:        {stats['removed']}")
-
-        # Verify: should have ~20 unique (some merged)
-        assert stats['unique'] <= 20, f"Too many unique: {stats['unique']} > 20"
-        assert stats['unique'] >= 15, f"Too few unique: {stats['unique']} < 15"
-        assert stats['removed'] > 0, "No duplicates removed"
-        print("\nOK: Dedup test passed")
-        return
-
-    if not args.input:
-        print("ERROR: Provide --input or --test")
-        sys.exit(1)
-
-    stats = dedup_index_file(args.input, args.output, args.threshold, args.dry_run)
-
-    if args.json:
-        print(json.dumps(stats, indent=2))
-    else:
-        print(f"Dedup results:")
-        print(f"  Total input:    {stats['total']}")
-        print(f"  Exact dupes:    {stats['exact_dupes']}")
-        print(f"  Near dupes:     {stats['near_dupes']}")
-        print(f"  Unique output:  {stats['unique']}")
-        print(f"  Removed:        {stats['removed']}")
-        if args.dry_run:
-            print("  (dry run — no changes written)")
-
-
-if __name__ == "__main__":
-    main()

scripts/test_harvest_prompt_comprehensive.py

@@ -1,212 +1,72 @@
 #!/usr/bin/env python3
-"""
-Comprehensive test script for knowledge extraction prompt.
-Validates prompt structure, requirements, and consistency.
-"""
-
-import json
-import re
+"""Comprehensive tests for knowledge extraction prompt."""
+import json, re
 from pathlib import Path
 
-def test_prompt_structure():
-    """Test that the prompt has the required structure."""
-    prompt_path = Path("templates/harvest-prompt.md")
-    if not prompt_path.exists():
-        return False, "harvest-prompt.md not found"
-    
-    content = prompt_path.read_text()
-    
-    # Check for required sections
-    required_sections = [
-        "System Prompt",
-        "Instructions", 
-        "Categories",
-        "Output Format",
-        "Confidence Scoring",
-        "Constraints",
-        "Example"
-    ]
-    
-    for section in required_sections:
-        if section.lower() not in content.lower():
-            return False, f"Missing required section: {section}"
-    
-    # Check for required categories
-    required_categories = ["fact", "pitfall", "pattern", "tool-quirk", "question"]
-    for category in required_categories:
-        if category not in content:
-            return False, f"Missing required category: {category}"
-    
-    # Check for required output fields
-    required_fields = ["fact", "category", "repo", "confidence"]
-    for field in required_fields:
-        if field not in content:
-            return False, f"Missing required output field: {field}"
-    
-    # Check prompt size (should be ~1k tokens, roughly 4k chars)
-    if len(content) > 5000:
-        return False, f"Prompt too large: {len(content)} chars (max ~5000)"
-    
-    if len(content) < 1000:
-        return False, f"Prompt too small: {len(content)} chars (min ~1000)"
-    
+def check_prompt_structure():
+    p = Path("templates/harvest-prompt.md")
+    if not p.exists(): return False, "harvest-prompt.md not found"
+    c = p.read_text()
+    for s in ["System Prompt","Instructions","Categories","Output Format","Confidence Scoring","Constraints","Example"]:
+        if s.lower() not in c.lower(): return False, f"Missing section: {s}"
+    for cat in ["fact","pitfall","pattern","tool-quirk","question"]:
+        if cat not in c: return False, f"Missing category: {cat}"
+    if len(c) > 5000: return False, f"Too large: {len(c)}"
+    if len(c) < 1000: return False, f"Too small: {len(c)}"
     return True, "Prompt structure is valid"
 
+def check_confidence_scoring():
+    c = Path("templates/harvest-prompt.md").read_text()
+    for l in ["0.9-1.0","0.7-0.8","0.5-0.6","0.3-0.4","0.1-0.2"]:
+        if l not in c: return False, f"Missing level: {l}"
+    return True, "Confidence scoring defined"
+
+def check_example_quality():
+    c = Path("templates/harvest-prompt.md").read_text()
+    if "example" not in c.lower(): return False, "No examples"
+    m = re.search(r'"knowledge"', c[c.lower().find("example"):])
+    if not m: return False, "No JSON example"
+    return True, "Examples present"
+
+def check_constraint_coverage():
+    c = Path("templates/harvest-prompt.md").read_text()
+    for x in ["no hallucination","explicitly","partial","failed sessions"]:
+        if x not in c.lower(): return False, f"Missing: {x}"
+    return True, "Constraints covered"
+
+def check_test_sessions():
+    d = Path("test_sessions")
+    if not d.exists(): return False, "test_sessions/ not found"
+    files = list(d.glob("*.jsonl"))
+    if len(files) < 5: return False, f"Only {len(files)} sessions"
+    for f in files:
+        for i, line in enumerate(f.read_text().strip().split("\n"), 1):
+            try: json.loads(line)
+            except json.JSONDecodeError as e: return False, f"{f.name}:{i}: {e}"
+    return True, f"{len(files)} valid sessions"
+
+def test_prompt_structure():
+    passed, msg = check_prompt_structure()
+    assert passed, msg
+
 def test_confidence_scoring():
-    """Test that confidence scoring is properly defined."""
-    prompt_path = Path("templates/harvest-prompt.md")
-    content = prompt_path.read_text()
-    
-    # Check for confidence scale definitions
-    confidence_levels = [
-        ("0.9-1.0", "explicitly stated"),
-        ("0.7-0.8", "clearly implied"),
-        ("0.5-0.6", "suggested"),
-        ("0.3-0.4", "inferred"),
-        ("0.1-0.2", "speculative")
-    ]
-    
-    for level, description in confidence_levels:
-        if level not in content:
-            return False, f"Missing confidence level: {level}"
-        if description.lower() not in content.lower():
-            return False, f"Missing confidence description: {description}"
-    
-    return True, "Confidence scoring is properly defined"
+    passed, msg = check_confidence_scoring()
+    assert passed, msg
 
 def test_example_quality():
-    """Test that examples are clear and complete."""
-    prompt_path = Path("templates/harvest-prompt.md")
-    content = prompt_path.read_text()
-    
-    # Check for example input/output
-    if "example" not in content.lower():
-        return False, "No examples provided"
-    
-    # Check that example includes all categories
-    example_section = content[content.lower().find("example"):]
-    
-    # Look for JSON example
-    json_match = re.search(r'\{[\s\S]*"knowledge"[\s\S]*\}', example_section)
-    if not json_match:
-        return False, "No JSON example found"
-    
-    example_json = json_match.group(0)
-    
-    # Check for all categories in example
-    for category in ["fact", "pitfall", "pattern", "tool-quirk", "question"]:
-        if category not in example_json:
-            return False, f"Example missing category: {category}"
-    
-    return True, "Examples are clear and complete"
+    passed, msg = check_example_quality()
+    assert passed, msg
 
 def test_constraint_coverage():
-    """Test that constraints cover all requirements."""
-    prompt_path = Path("templates/harvest-prompt.md")
-    content = prompt_path.read_text()
-    
-    required_constraints = [
-        "No hallucination",
-        "only extract",
-        "explicitly",
-        "partial",
-        "failed sessions",
-        "1k tokens"
-    ]
-    
-    for constraint in required_constraints:
-        if constraint.lower() not in content.lower():
-            return False, f"Missing constraint: {constraint}"
-    
-    return True, "Constraints cover all requirements"
+    passed, msg = check_constraint_coverage()
+    assert passed, msg
 
 def test_test_sessions():
-    """Test that test sessions exist and are valid."""
-    test_sessions_dir = Path("test_sessions")
-    if not test_sessions_dir.exists():
-        return False, "test_sessions directory not found"
-    
-    session_files = list(test_sessions_dir.glob("*.jsonl"))
-    if len(session_files) < 5:
-        return False, f"Only {len(session_files)} test sessions found, need 5"
-    
-    # Check each session file
-    for session_file in session_files:
-        content = session_file.read_text()
-        lines = content.strip().split("\n")
-        
-        # Check that each line is valid JSON
-        for i, line in enumerate(lines, 1):
-            try:
-                json.loads(line)
-            except json.JSONDecodeError as e:
-                return False, f"Invalid JSON in {session_file.name}, line {i}: {e}"
-    
-    return True, f"Found {len(session_files)} valid test sessions"
-
-def run_all_tests():
-    """Run all tests and return results."""
-    tests = [
-        ("Prompt Structure", test_prompt_structure),
-        ("Confidence Scoring", test_confidence_scoring),
-        ("Example Quality", test_example_quality),
-        ("Constraint Coverage", test_constraint_coverage),
-        ("Test Sessions", test_test_sessions)
-    ]
-    
-    results = []
-    all_passed = True
-    
-    for test_name, test_func in tests:
-        try:
-            passed, message = test_func()
-            results.append({
-                "test": test_name,
-                "passed": passed,
-                "message": message
-            })
-            if not passed:
-                all_passed = False
-        except Exception as e:
-            results.append({
-                "test": test_name,
-                "passed": False,
-                "message": f"Error: {str(e)}"
-            })
-            all_passed = False
-    
-    # Print results
-    print("=" * 60)
-    print("HARVEST PROMPT TEST RESULTS")
-    print("=" * 60)
-    
-    for result in results:
-        status = "✓ PASS" if result["passed"] else "✗ FAIL"
-        print(f"{status}: {result['test']}")
-        print(f"  {result['message']}")
-        print()
-    
-    print("=" * 60)
-    if all_passed:
-        print("ALL TESTS PASSED!")
-    else:
-        print("SOME TESTS FAILED!")
-    print("=" * 60)
-    
-    return all_passed, results
+    passed, msg = check_test_sessions()
+    assert passed, msg
 
 if __name__ == "__main__":
-    all_passed, results = run_all_tests()
-    
-    # Save results to file
-    with open("test_results.json", "w") as f:
-        json.dump({
-            "all_passed": all_passed,
-            "results": results,
-            "timestamp": "2026-04-14T19:05:00Z"
-        }, f, indent=2)
-    
-    print(f"Results saved to test_results.json")
-    
-    # Exit with appropriate code
-    exit(0 if all_passed else 1)
+    checks = [check_prompt_structure, check_confidence_scoring, check_example_quality, check_constraint_coverage, check_test_sessions]
+    for fn in checks:
+        ok, msg = fn()
+        print(f"{'PASS' if ok else 'FAIL'}: {fn.__name__} -- {msg}")

tests/test_dedup.py

@@ -1,207 +0,0 @@
-"""Tests for knowledge deduplication module (Issue #196)."""
-
-import json
-import sys
-from pathlib import Path
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
-
-from dedup import (
-    normalize_text,
-    content_hash,
-    tokenize,
-    token_similarity,
-    quality_score,
-    merge_facts,
-    dedup_facts,
-    generate_test_duplicates,
-)
-
-
-class TestNormalize:
-    def test_lowercases(self):
-        assert normalize_text("Hello World") == "hello world"
-
-    def test_collapses_whitespace(self):
-        assert normalize_text("  hello   world  ") == "hello world"
-
-    def test_strips(self):
-        assert normalize_text("  text  ") == "text"
-
-
-class TestContentHash:
-    def test_deterministic(self):
-        h1 = content_hash("Hello World")
-        h2 = content_hash("hello world")
-        h3 = content_hash("  Hello   World  ")
-        assert h1 == h2 == h3
-
-    def test_different_texts(self):
-        h1 = content_hash("Hello")
-        h2 = content_hash("World")
-        assert h1 != h2
-
-    def test_returns_hex(self):
-        h = content_hash("test")
-        assert len(h) == 64  # SHA256
-        assert all(c in '0123456789abcdef' for c in h)
-
-
-class TestTokenize:
-    def test_extracts_words(self):
-        tokens = tokenize("Hello World Test")
-        assert "hello" in tokens
-        assert "world" in tokens
-        assert "test" in tokens
-
-    def test_skips_short_words(self):
-        tokens = tokenize("a to is the hello")
-        assert "a" not in tokens
-        assert "to" not in tokens
-        assert "hello" in tokens
-
-    def test_returns_set(self):
-        tokens = tokenize("hello hello world")
-        assert isinstance(tokens, set)
-        assert len(tokens) == 2
-
-
-class TestTokenSimilarity:
-    def test_identical(self):
-        assert token_similarity("hello world", "hello world") == 1.0
-
-    def test_no_overlap(self):
-        assert token_similarity("alpha beta", "gamma delta") == 0.0
-
-    def test_partial_overlap(self):
-        sim = token_similarity("hello world test", "hello universe test")
-        assert 0.3 < sim < 0.7
-
-    def test_empty(self):
-        assert token_similarity("", "hello") == 0.0
-        assert token_similarity("hello", "") == 0.0
-
-    def test_symmetric(self):
-        a = "hello world test"
-        b = "hello universe test"
-        assert token_similarity(a, b) == token_similarity(b, a)
-
-
-class TestQualityScore:
-    def test_high_confidence(self):
-        fact = {"confidence": 0.95, "source_count": 5, "tags": ["test"], "related": ["x"]}
-        score = quality_score(fact)
-        assert score > 0.7
-
-    def test_low_confidence(self):
-        fact = {"confidence": 0.3, "source_count": 1}
-        score = quality_score(fact)
-        assert score < 0.5
-
-    def test_defaults(self):
-        score = quality_score({})
-        assert 0 < score < 1
-
-
-class TestMergeFacts:
-    def test_merges_tags(self):
-        keep = {"id": "a", "fact": "test", "tags": ["git"], "confidence": 0.9}
-        drop = {"id": "b", "fact": "test", "tags": ["python"], "confidence": 0.8}
-        merged = merge_facts(keep, drop)
-        assert "git" in merged["tags"]
-        assert "python" in merged["tags"]
-
-    def test_merges_source_count(self):
-        keep = {"id": "a", "fact": "test", "source_count": 3}
-        drop = {"id": "b", "fact": "test", "source_count": 2}
-        merged = merge_facts(keep, drop)
-        assert merged["source_count"] == 5
-
-    def test_keeps_higher_confidence(self):
-        keep = {"id": "a", "fact": "test", "confidence": 0.7}
-        drop = {"id": "b", "fact": "test", "confidence": 0.9}
-        merged = merge_facts(keep, drop)
-        assert merged["confidence"] == 0.9
-
-    def test_tracks_merged_from(self):
-        keep = {"id": "a", "fact": "test"}
-        drop = {"id": "b", "fact": "test"}
-        merged = merge_facts(keep, drop)
-        assert "b" in merged["_merged_from"]
-
-
-class TestDedupFacts:
-    def test_removes_exact_dupes(self):
-        facts = [
-            {"id": "1", "fact": "Always use git rebase"},
-            {"id": "2", "fact": "Always use git rebase"},  # exact dupe
-            {"id": "3", "fact": "Check logs first"},
-        ]
-        deduped, stats = dedup_facts(facts)
-        assert stats["exact_dupes"] == 1
-        assert stats["unique"] == 2
-
-    def test_removes_near_dupes(self):
-        facts = [
-            {"id": "1", "fact": "Always check logs before deploying to production server"},
-            {"id": "2", "fact": "Always check logs before deploying to production environment"},
-            {"id": "3", "fact": "Use docker compose for local development environments"},
-        ]
-        deduped, stats = dedup_facts(facts, near_threshold=0.5)
-        assert stats["near_dupes"] >= 1
-        assert stats["unique"] == 2
-
-    def test_preserves_unique(self):
-        facts = [
-            {"id": "1", "fact": "Use git rebase for clean history"},
-            {"id": "2", "fact": "Docker containers should be stateless"},
-            {"id": "3", "fact": "Always write tests before code"},
-        ]
-        deduped, stats = dedup_facts(facts)
-        assert stats["unique"] == 3
-        assert stats["removed"] == 0
-
-    def test_empty_input(self):
-        deduped, stats = dedup_facts([])
-        assert stats["total"] == 0
-        assert stats["unique"] == 0
-
-    def test_keeps_higher_quality_near_dup(self):
-        facts = [
-            {"id": "1", "fact": "Check logs before deploying to production server", "confidence": 0.5, "source_count": 1},
-            {"id": "2", "fact": "Check logs before deploying to production environment", "confidence": 0.9, "source_count": 5, "tags": ["ops"]},
-        ]
-        deduped, stats = dedup_facts(facts, near_threshold=0.5)
-        assert stats["unique"] == 1
-        # Higher quality fact should be kept
-        assert deduped[0]["confidence"] == 0.9
-
-    def test_dry_run_does_not_modify(self):
-        facts = [
-            {"id": "1", "fact": "Same text"},
-            {"id": "2", "fact": "Same text"},
-        ]
-        deduped, stats = dedup_facts(facts, dry_run=True)
-        assert stats["exact_dupes"] == 1
-        # In dry_run, merge_facts is skipped so facts aren't modified
-        assert len(deduped) == 1
-
-
-class TestGenerateTestDuplicates:
-    def test_generates_correct_count(self):
-        facts = generate_test_duplicates(20)
-        assert len(facts) > 20  # 20 unique + duplicates
-
-    def test_has_exact_dupes(self):
-        facts = generate_test_duplicates(20)
-        hashes = [content_hash(f["fact"]) for f in facts]
-        # Should have some duplicate hashes
-        assert len(hashes) != len(set(hashes))
-
-    def test_dedup_removes_dupes(self):
-        facts = generate_test_duplicates(20)
-        deduped, stats = dedup_facts(facts)
-        assert stats["unique"] <= 20
-        assert stats["removed"] > 0