#!/usr/bin/env python3
"""
Test script for bill amendment checking logic.
Run this locally BEFORE deploying to production to verify:
1. The query returns the expected bills
2. Amendment detection works correctly
3. No errors in the detection logic
Usage:
# Test with local Neo4j
python scripts/test_amendment_check.py
# Test with production DB via tunnel (run ./scripts/dev-tunnel.sh first)
NEO4J_URI=bolt://localhost:7687 python scripts/test_amendment_check.py
# Dry run - just show what bills WOULD be checked (no actual detection)
python scripts/test_amendment_check.py --dry-run
# Test specific bill
python scripts/test_amendment_check.py --bill C-12
# Limit number of bills to check
python scripts/test_amendment_check.py --limit 5
"""
import argparse
import os
import sys
from pathlib import Path
# Fix Windows console encoding for UTF-8
if sys.platform == "win32":
sys.stdout.reconfigure(encoding='utf-8')
# Add packages to path
SCRIPT_DIR = Path(__file__).parent
PIPELINE_DIR = SCRIPT_DIR.parent
sys.path.insert(0, str(PIPELINE_DIR))
from fedmcp_pipeline.utils.neo4j_client import Neo4jClient
from fedmcp_pipeline.utils.config import Config
# Add fedmcp clients
FEDMCP_PATH = PIPELINE_DIR.parent / "fedmcp" / "src"
sys.path.insert(0, str(FEDMCP_PATH))
from fedmcp_pipeline.ingest.bill_amendments import detect_bill_amendments
def get_bills_for_amendment_check(neo4j: Neo4jClient, session_filter: str = None) -> list:
"""
Get bills that would be checked for amendments.
This is the same logic as in lightweight_update.py
"""
# Get current session
if session_filter:
current_session = session_filter
else:
current_session_result = neo4j.run_query("""
MATCH (b:Bill)
RETURN b.session as session
ORDER BY b.session DESC
LIMIT 1
""")
current_session = current_session_result[0]["session"] if current_session_result else "45-1"
print(f"\nš Checking bills for session: {current_session}\n")
result = neo4j.run_query("""
MATCH (b:Bill)
WHERE b.session = $current_session
AND (
// Case 1: Never checked, past second reading (House OR Senate)
(b.has_amendments IS NULL
AND (b.passed_house_second_reading IS NOT NULL
OR b.passed_senate_second_reading IS NOT NULL
OR b.status CONTAINS 'Committee'
OR b.status CONTAINS 'Third'
OR b.status CONTAINS 'Royal Assent'))
OR
// Case 2: Already checked but not at Royal Assent (could get new amendments)
(b.has_amendments IS NOT NULL
AND NOT (b.status CONTAINS 'Royal Assent')
AND (b.last_amendment_check IS NULL
OR b.last_amendment_check < datetime() - duration('P1D')))
OR
// Case 3: Just reached Royal Assent - need final check
(b.status CONTAINS 'Royal Assent'
AND (b.checked_at_royal_assent IS NULL OR b.checked_at_royal_assent = false))
)
RETURN b.number as number,
b.session as session,
b.is_government_bill as is_government_bill,
b.has_amendments as has_amendments,
b.status as status,
b.last_amendment_check as last_check,
b.checked_at_royal_assent as checked_at_ra,
b.total_versions as total_versions
ORDER BY
CASE
WHEN b.has_amendments IS NULL THEN 0
WHEN b.status CONTAINS 'Royal Assent' THEN 1
ELSE 2
END,
b.number
""", {"current_session": current_session})
bills = []
for row in result:
session = row["session"]
try:
parts = session.split("-")
parliament = int(parts[0])
session_num = int(parts[1])
except (ValueError, IndexError):
continue
bills.append({
"number": row["number"],
"session": session,
"parliament": parliament,
"session_number": session_num,
"is_government_bill": row.get("is_government_bill", False),
"status": row.get("status", ""),
"has_amendments": row.get("has_amendments"),
"last_check": row.get("last_check"),
"checked_at_ra": row.get("checked_at_ra"),
"total_versions": row.get("total_versions"),
})
return bills
def categorize_bills(bills: list) -> dict:
"""Categorize bills by why they need checking."""
categories = {
"never_checked": [],
"royal_assent_final": [],
"re_check": [],
}
for bill in bills:
if bill["has_amendments"] is None:
categories["never_checked"].append(bill)
elif "Royal Assent" in (bill["status"] or ""):
categories["royal_assent_final"].append(bill)
else:
categories["re_check"].append(bill)
return categories
def print_bill_summary(bills: list, categories: dict):
"""Print a summary of bills that would be checked."""
print("=" * 70)
print(f"BILLS TO CHECK: {len(bills)} total")
print("=" * 70)
print(f"\nš Never checked (first time): {len(categories['never_checked'])}")
for b in categories["never_checked"][:10]:
print(f" {b['number']:10} - {b['status'][:40] if b['status'] else 'No status'}")
if len(categories["never_checked"]) > 10:
print(f" ... and {len(categories['never_checked']) - 10} more")
print(f"\nš Royal Assent (final check): {len(categories['royal_assent_final'])}")
for b in categories["royal_assent_final"][:10]:
print(f" {b['number']:10} - versions: {b['total_versions'] or '?'}")
if len(categories["royal_assent_final"]) > 10:
print(f" ... and {len(categories['royal_assent_final']) - 10} more")
print(f"\nš Re-check (may have new versions): {len(categories['re_check'])}")
for b in categories["re_check"][:10]:
last = b['last_check']
last_str = str(last)[:10] if last else "never"
print(f" {b['number']:10} - last check: {last_str}, versions: {b['total_versions'] or '?'}")
if len(categories["re_check"]) > 10:
print(f" ... and {len(categories['re_check']) - 10} more")
print()
def test_single_bill(neo4j: Neo4jClient, bill_number: str, session: str = "45-1"):
"""Test amendment detection on a single bill."""
print(f"\nš Testing amendment detection for {bill_number} ({session})...\n")
# Parse session
parts = session.split("-")
parliament = int(parts[0])
session_num = int(parts[1])
# Check if bill is government bill
result = neo4j.run_query("""
MATCH (b:Bill {number: $number, session: $session})
RETURN b.is_government_bill as is_gov, b.status as status
""", {"number": bill_number.upper(), "session": session})
if not result:
print(f"ā Bill {bill_number} not found in session {session}")
return
is_gov = result[0].get("is_gov", False)
status = result[0].get("status", "")
is_royal_assent = "Royal Assent" in status
print(f" Bill: {bill_number}")
print(f" Session: {session}")
print(f" Government bill: {is_gov}")
print(f" Status: {status}")
print(f" Is Royal Assent: {is_royal_assent}")
print()
try:
result = detect_bill_amendments(
neo4j,
parliament=parliament,
session=session_num,
bill_number=bill_number,
is_government=is_gov,
is_royal_assent=is_royal_assent,
)
print("\nš Results:")
print(f" Versions found: {result.get('versions_found', 0)}")
print(f" Has amendments: {result.get('has_amendments', False)}")
print(f" Total diffs: {result.get('total_diffs', 0)}")
diffs_by_type = result.get("diffs_by_type", {})
if diffs_by_type:
print(f" Diffs by type: {diffs_by_type}")
print("\nā
Test completed successfully!")
except Exception as e:
print(f"\nā Error: {e}")
import traceback
traceback.print_exc()
def run_full_test(neo4j: Neo4jClient, limit: int = None, dry_run: bool = False):
"""Run full amendment check test."""
bills = get_bills_for_amendment_check(neo4j)
categories = categorize_bills(bills)
print_bill_summary(bills, categories)
if dry_run:
print("š DRY RUN - No actual amendment detection performed")
return
if limit:
bills = bills[:limit]
print(f"š¢ Limited to {limit} bills for testing\n")
print("=" * 70)
print("RUNNING AMENDMENT DETECTION")
print("=" * 70)
stats = {
"checked": 0,
"with_amendments": 0,
"errors": 0,
}
for bill in bills:
bill_number = bill["number"]
parliament = bill["parliament"]
session_num = bill["session_number"]
is_gov = bill.get("is_government_bill", False)
status = bill.get("status", "")
is_royal_assent = "Royal Assent" in status
print(f"\nš Checking {bill_number}...", end=" ")
try:
result = detect_bill_amendments(
neo4j,
parliament=parliament,
session=session_num,
bill_number=bill_number,
is_government=is_gov,
is_royal_assent=is_royal_assent,
)
stats["checked"] += 1
versions = result.get("versions_found", 0)
has_amendments = result.get("has_amendments", False)
total_diffs = result.get("total_diffs", 0)
if has_amendments:
stats["with_amendments"] += 1
print(f"ā
{versions} versions, {total_diffs} amendments")
else:
print(f"ā
{versions} version(s), no amendments")
except Exception as e:
stats["errors"] += 1
print(f"ā Error: {e}")
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
print(f" Bills checked: {stats['checked']}")
print(f" With amendments: {stats['with_amendments']}")
print(f" Errors: {stats['errors']}")
print()
def main():
parser = argparse.ArgumentParser(description="Test bill amendment checking logic")
parser.add_argument("--dry-run", action="store_true", help="Just show what would be checked")
parser.add_argument("--bill", type=str, help="Test specific bill (e.g., C-12)")
parser.add_argument("--session", type=str, default="45-1", help="Session (default: 45-1)")
parser.add_argument("--limit", type=int, help="Limit number of bills to check")
args = parser.parse_args()
# Load config
config = Config()
print("\n" + "=" * 70)
print("BILL AMENDMENT CHECK TEST")
print("=" * 70)
print(f"Neo4j URI: {config.neo4j_uri}")
print(f"Session: {args.session}")
# Connect to Neo4j
neo4j = Neo4jClient(
uri=config.neo4j_uri,
user=config.neo4j_user,
password=config.neo4j_password
)
try:
if args.bill:
test_single_bill(neo4j, args.bill, args.session)
else:
run_full_test(neo4j, limit=args.limit, dry_run=args.dry_run)
finally:
neo4j.close()
if __name__ == "__main__":
main()
