Compare commits
5 Commits
step35/98-
...
step35/108
| Author | SHA1 | Date | |
|---|---|---|---|
|
9e2514723f | ||
|
|
c04227b03b | ||
|
|
a7b14c4b2b | ||
|
|
a46d465d2d | ||
|
|
4b5a675355 |
472
docs/API.md
472
docs/API.md
@@ -1,472 +0,0 @@
|
||||
# Compounding Intelligence — Scripts API Reference
|
||||
|
||||
*Generated: 2026-04-26 11:02 UTC*
|
||||
|
||||
This document auto-documents the public API surface of all scripts
|
||||
in `scripts/`. Each section covers one script: module purpose,
|
||||
public functions, and their signatures.
|
||||
|
||||
---
|
||||
|
||||
## `scripts/api_doc_generator.py`
|
||||
|
||||
API Doc Generator — Issue #98
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `extract_functions_from_ast` | `extract_functions_from_ast(tree, file_rel)` | Extract public function names, signatures, and first-line doc summaries. |
|
||||
| `parse_module` | `parse_module(filepath)` | Parse a Python file and return its module-level docstring and public functions. |
|
||||
| `scan_scripts_dir` | `scan_scripts_dir(scripts_dir)` | Scan all .py files in scripts/ and extract API info. |
|
||||
| `render_markdown` | `render_markdown(modules)` | Generate full docs/API.md content from the scanned modules. |
|
||||
| `render_json` | `render_json(modules)` | Emit machine-readable JSON version of the API reference. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/automation_opportunity_finder.py`
|
||||
|
||||
Automation Opportunity Finder — Scan fleet for manual processes that could be automated.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `analyze_cron_jobs` | `analyze_cron_jobs(hermes_home)` | Analyze cron job definitions for automation gaps. |
|
||||
| `analyze_documents` | `analyze_documents(root_dirs)` | Scan documentation for manual step patterns. |
|
||||
| `analyze_scripts` | `analyze_scripts(root_dirs)` | Detect repeated command sequences in scripts. |
|
||||
| `analyze_session_transcripts` | `analyze_session_transcripts(session_dirs)` | Find repeated tool-call patterns in session transcripts. |
|
||||
| `analyze_shell_history` | `analyze_shell_history(root_dirs)` | Find repeated shell commands from history files. |
|
||||
| `deduplicate_proposals` | `deduplicate_proposals(proposals)` | Remove duplicate proposals based on title similarity. |
|
||||
| `rank_proposals` | `rank_proposals(proposals)` | Sort proposals by impact * confidence (highest first). |
|
||||
| `format_text_report` | `format_text_report(proposals)` | Format proposals as human-readable text. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/bootstrapper.py`
|
||||
|
||||
Bootstrapper — assemble pre-session context from knowledge store.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `load_index` | `load_index(index_path)` | Load and validate the knowledge index. |
|
||||
| `filter_facts` | `filter_facts(facts, repo, agent, include_global)` | Filter facts by repo, agent, and global scope. |
|
||||
| `sort_facts` | `sort_facts(facts)` | Sort facts by: confidence (desc), then category priority, then fact text. |
|
||||
| `load_repo_knowledge` | `load_repo_knowledge(repo)` | Load per-repo knowledge markdown if it exists. |
|
||||
| `load_agent_knowledge` | `load_agent_knowledge(agent)` | Load per-agent knowledge markdown if it exists. |
|
||||
| `load_global_knowledge` | `load_global_knowledge()` | Load all global knowledge markdown files. |
|
||||
| `render_facts_section` | `render_facts_section(facts, category, label)` | Render a section of facts for a single category. |
|
||||
| `estimate_tokens` | `estimate_tokens(text)` | Rough token estimate. |
|
||||
| `truncate_to_tokens` | `truncate_to_tokens(text, max_tokens)` | Truncate text to approximately max_tokens, cutting at line boundaries. |
|
||||
| `build_bootstrap_context` | `build_bootstrap_context(repo, agent, include_global, max_tokens, index_path)` | Build the full bootstrap context block. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/dead_code_detector.py`
|
||||
|
||||
Dead Code Detector for Python Codebases
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `is_safe_unused` | `is_safe_unused(name, filepath)` | Check if an unused name is expected to be unused. |
|
||||
| `get_git_blame` | `get_git_blame(filepath, lineno)` | Get last author of a line via git blame. |
|
||||
| `analyze_file` | `analyze_file(filepath)` | Analyze a single Python file for dead code. |
|
||||
| `scan_repo` | `scan_repo(repo_path, exclude_patterns)` | Scan an entire repo for dead code. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/dedup.py`
|
||||
|
||||
dedup.py — Knowledge deduplication: content hash + semantic similarity.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `normalize_text` | `normalize_text(text)` | Normalize text for hashing: lowercase, collapse whitespace, strip. |
|
||||
| `content_hash` | `content_hash(text)` | SHA256 hash of normalized text for exact dedup. |
|
||||
| `tokenize` | `tokenize(text)` | Simple tokenizer: lowercase words, 3+ chars. |
|
||||
| `token_similarity` | `token_similarity(a, b)` | Token-based Jaccard similarity (0.0-1.0). |
|
||||
| `quality_score` | `quality_score(fact)` | Compute quality score for merge ranking. |
|
||||
| `merge_facts` | `merge_facts(keep, drop)` | Merge two near-duplicate facts, keeping higher-quality fields. |
|
||||
| `dedup_facts` | `dedup_facts(facts, exact_threshold, near_threshold, dry_run)` | Deduplicate a list of knowledge facts. |
|
||||
| `dedup_index_file` | `dedup_index_file(input_path, output_path, near_threshold, dry_run)` | Deduplicate an index.json file. |
|
||||
| `generate_test_duplicates` | `generate_test_duplicates(n)` | Generate test facts with intentional duplicates for testing. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/dependency_graph.py`
|
||||
|
||||
Cross-Repo Dependency Graph Builder
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `normalize_repo_name` | `normalize_repo_name(name)` | Normalize a repo name for comparison. |
|
||||
| `scan_file_for_deps` | `scan_file_for_deps(filepath, content, own_repo)` | Scan a file's content for references to other repos. |
|
||||
| `scan_repo` | `scan_repo(repo_path, repo_name)` | Scan a repo directory for dependencies. |
|
||||
| `detect_cycles` | `detect_cycles(graph)` | Detect circular dependencies using DFS. |
|
||||
| `to_dot` | `to_dot(graph)` | Generate DOT format output. |
|
||||
| `to_mermaid` | `to_mermaid(graph)` | Generate Mermaid format output. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/diff_analyzer.py`
|
||||
|
||||
Diff Analyzer — Parse unified diffs and categorize every change.
|
||||
|
||||
*(no public functions — script runs as `main()` only)*
|
||||
|
||||
## `scripts/freshness.py`
|
||||
|
||||
Knowledge Freshness Cron — Detect stale entries from code changes (Issue #200)
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `compute_file_hash` | `compute_file_hash(filepath)` | Compute SHA-256 hash of a file. Returns None if file doesn't exist. |
|
||||
| `get_git_file_changes` | `get_git_file_changes(repo_path, days)` | Get files changed in git in the last N days. |
|
||||
| `load_knowledge_entries` | `load_knowledge_entries(knowledge_dir)` | Load knowledge entries from YAML files in the knowledge directory. |
|
||||
| `check_freshness` | `check_freshness(knowledge_dir, repo_root, days)` | Check freshness of knowledge entries against recent code changes. |
|
||||
| `update_stale_hashes` | `update_stale_hashes(knowledge_dir, repo_root)` | Update hashes for stale entries. Returns count of updated entries. |
|
||||
| `format_report` | `format_report(result, max_items)` | Format freshness check results as a human-readable report. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/gitea_issue_parser.py`
|
||||
|
||||
Gitea Issue Body Parser — Extract structured data from markdown issue bodies.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `parse_issue_body` | `parse_issue_body(body, title, labels)` | Parse a Gitea issue markdown body into structured JSON. |
|
||||
| `fetch_issue_from_url` | `fetch_issue_from_url(url)` | Fetch an issue from a Gitea API URL and parse it. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/harvester.py`
|
||||
|
||||
harvester.py — Extract durable knowledge from Hermes session transcripts.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `find_api_key` | `find_api_key()` | Find API key from common locations. |
|
||||
| `load_extraction_prompt` | `load_extraction_prompt()` | Load the extraction prompt template. |
|
||||
| `call_llm` | `call_llm(prompt, transcript, api_base, api_key, model)` | Call the LLM API to extract knowledge from a transcript. |
|
||||
| `parse_extraction_response` | `parse_extraction_response(content)` | Parse the LLM response to extract knowledge items. |
|
||||
| `load_existing_knowledge` | `load_existing_knowledge(knowledge_dir)` | Load the existing knowledge index. |
|
||||
| `fact_fingerprint` | `fact_fingerprint(fact)` | Generate a deduplication fingerprint for a fact. |
|
||||
| `deduplicate` | `deduplicate(new_facts, existing, similarity_threshold)` | Remove duplicate facts from new_facts that already exist in the knowledge store. |
|
||||
| `validate_fact` | `validate_fact(fact)` | Validate a single knowledge item has required fields. |
|
||||
| `write_knowledge` | `write_knowledge(index, new_facts, knowledge_dir, source_session)` | Write new facts to the knowledge store. |
|
||||
| `harvest_session` | `harvest_session(session_path, knowledge_dir, api_base, api_key, model, dry_run, min_confidence)` | Harvest knowledge from a single session. |
|
||||
| `batch_harvest` | `batch_harvest(sessions_dir, knowledge_dir, api_base, api_key, model, since, limit, dry_run)` | Harvest knowledge from multiple sessions in batch. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/improvement_proposals.py`
|
||||
|
||||
Improvement Proposal Generator for compounding-intelligence.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `analyze_sessions` | `analyze_sessions(sessions)` | Analyze session data to find waste patterns. |
|
||||
| `generate_proposals` | `generate_proposals(patterns, hourly_rate, implementation_overhead)` | Generate improvement proposals from waste patterns. |
|
||||
| `format_proposals_markdown` | `format_proposals_markdown(proposals, patterns, generated_at)` | Format proposals as a markdown document. |
|
||||
| `format_proposals_json` | `format_proposals_json(proposals)` | Format proposals as JSON. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/knowledge_gap_identifier.py`
|
||||
|
||||
Knowledge Gap Identifier — Pipeline 10.7
|
||||
|
||||
*(no public functions — script runs as `main()` only)*
|
||||
|
||||
## `scripts/knowledge_staleness_check.py`
|
||||
|
||||
Knowledge Store Staleness Detector — Detect stale knowledge entries by comparing source file hashes.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `compute_file_hash` | `compute_file_hash(filepath)` | Compute SHA-256 hash of a file. Returns None if file doesn't exist. |
|
||||
| `check_staleness` | `check_staleness(index_path, repo_root)` | Check all entries in knowledge index for staleness. |
|
||||
| `fix_hashes` | `fix_hashes(index_path, repo_root)` | Add hashes to entries missing them. Returns count of fixed entries. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/perf_bottleneck_finder.py`
|
||||
|
||||
Performance Bottleneck Finder — Identify slow tests, builds, and CI steps.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `find_slow_tests_pytest` | `find_slow_tests_pytest(repo_path)` | Run pytest --durations and parse slow tests. |
|
||||
| `find_slow_tests_by_scan` | `find_slow_tests_by_scan(repo_path)` | Scan test files for patterns that indicate slow tests. |
|
||||
| `analyze_build_artifacts` | `analyze_build_artifacts(repo_path)` | Find large build artifacts that slow down builds. |
|
||||
| `analyze_makefile_targets` | `analyze_makefile_targets(repo_path)` | Analyze Makefile for potentially slow targets. |
|
||||
| `analyze_github_actions` | `analyze_github_actions(repo_path)` | Analyze GitHub Actions workflow files for inefficiencies. |
|
||||
| `analyze_gitea_ci` | `analyze_gitea_ci(repo_path)` | Analyze Gitea/Drone CI config files. |
|
||||
| `find_slow_imports` | `find_slow_imports(repo_path)` | Find Python files with heavy import chains. |
|
||||
| `severity_sort_key` | `severity_sort_key(b)` | Sort by severity then duration. |
|
||||
| `generate_report` | `generate_report(repo_path)` | Run all analyses and generate a performance report. |
|
||||
| `format_markdown` | `format_markdown(report)` | Format report as markdown. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/priority_rebalancer.py`
|
||||
|
||||
Priority Rebalancer — Re-evaluate issue priorities based on accumulated data.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `collect_knowledge_signals` | `collect_knowledge_signals(knowledge_dir)` | Analyze knowledge store for coverage gaps and staleness. |
|
||||
| `collect_staleness_signals` | `collect_staleness_signals(scripts_dir, knowledge_dir)` | Run staleness checker if available. |
|
||||
| `collect_metrics_signals` | `collect_metrics_signals(metrics_dir)` | Analyze metrics directory for pipeline health. |
|
||||
| `extract_priority` | `extract_priority(labels)` | Extract priority level from issue labels. |
|
||||
| `compute_issue_score` | `compute_issue_score(issue, repo, signals, now)` | Compute priority score for a single issue. |
|
||||
| `generate_report` | `generate_report(scores, signals, org, repos_scanned)` | Generate the full priority report. |
|
||||
| `generate_markdown_report` | `generate_markdown_report(report)` | Generate human-readable markdown report. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/refactoring_opportunity_finder.py`
|
||||
|
||||
Finds refactoring opportunities in codebases
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `compute_file_complexity` | `compute_file_complexity(filepath)` | Compute cyclomatic complexity for a Python file. |
|
||||
| `calculate_refactoring_score` | `calculate_refactoring_score(metrics)` | Calculate a refactoring priority score (0-100) based on file metrics. |
|
||||
| `scan_directory` | `scan_directory(directory, extensions)` | Scan directory for source files. |
|
||||
| `generate_proposals` | `generate_proposals(directory, min_score)` | Generate refactoring proposals by analyzing source files. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/sampler.py`
|
||||
|
||||
sampler.py — Score and rank sessions by harvest value.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `scan_session_fast` | `scan_session_fast(path)` | Extract scoring metadata from a session without parsing the full JSONL. |
|
||||
| `parse_session_timestamp` | `parse_session_timestamp(filename)` | Parse timestamp from session filename. |
|
||||
| `score_session` | `score_session(meta, now, seen_repos)` | Score a session for harvest value. Returns (score, breakdown). |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/session_metadata.py`
|
||||
|
||||
session_metadata.py - Extract structured metadata from Hermes session transcripts.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `extract_session_metadata` | `extract_session_metadata(file_path)` | Extract structured metadata from a Hermes session JSONL transcript. |
|
||||
| `process_session_directory` | `process_session_directory(directory_path, output_file)` | Process all JSONL files in a directory. |
|
||||
| `main` | `main()` | CLI entry point. |
|
||||
|
||||
## `scripts/session_pair_harvester.py`
|
||||
|
||||
Session Transcript → Training Pair Harvester
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `compute_hash` | `compute_hash(text)` | Content hash for deduplication. |
|
||||
| `extract_pairs_from_session` | `extract_pairs_from_session(session_data, min_ratio, min_response_words)` | Extract terse→rich pairs from a single session object. |
|
||||
| `extract_from_jsonl_file` | `extract_from_jsonl_file(filepath, **kwargs)` | Extract pairs from a session JSONL file. |
|
||||
| `deduplicate_pairs` | `deduplicate_pairs(pairs)` | Remove duplicate pairs across files. |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
## `scripts/session_reader.py`
|
||||
|
||||
session_reader.py — Parse Hermes session JSONL transcripts.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `read_session` | `read_session(path)` | Read a session JSONL file and return all messages as a list. |
|
||||
| `read_session_iter` | `read_session_iter(path)` | Iterate over session messages without loading all into memory. |
|
||||
| `extract_conversation` | `extract_conversation(messages)` | Extract user/assistant conversation turns, skipping tool-only messages. |
|
||||
| `truncate_for_context` | `truncate_for_context(messages, head, tail)` | Truncate long sessions: keep first N + last N messages. |
|
||||
| `messages_to_text` | `messages_to_text(messages)` | Convert message list to plain text for LLM consumption. |
|
||||
| `get_session_metadata` | `get_session_metadata(path)` | Extract metadata from a session file (first message often has config info). |
|
||||
|
||||
## `scripts/test_automation_opportunity_finder.py`
|
||||
|
||||
Tests for scripts/automation_opportunity_finder.py — 8 tests.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_analyze_cron_jobs_no_file` | `test_analyze_cron_jobs_no_file()` | Returns empty list when no cron jobs file exists. |
|
||||
| `test_analyze_cron_jobs_disabled` | `test_analyze_cron_jobs_disabled()` | Detects disabled cron jobs. |
|
||||
| `test_analyze_cron_jobs_errors` | `test_analyze_cron_jobs_errors()` | Detects cron jobs with error status. |
|
||||
| `test_analyze_documents_finds_todos` | `test_analyze_documents_finds_todos()` | Detects TODO markers in documents. |
|
||||
| `test_analyze_scripts_repeated_commands` | `test_analyze_scripts_repeated_commands()` | Detects repeated shell commands across scripts. |
|
||||
| `test_analyze_session_transcripts` | `test_analyze_session_transcripts()` | Detects repeated tool-call sequences. |
|
||||
| `test_deduplicate_proposals` | `test_deduplicate_proposals()` | Deduplicates proposals with similar titles. |
|
||||
| `test_rank_proposals` | `test_rank_proposals()` | Ranks proposals by impact * confidence. |
|
||||
|
||||
## `scripts/test_bootstrapper.py`
|
||||
|
||||
Tests for bootstrapper.py — context assembly from knowledge store.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `make_index` | `make_index(facts, tmp_dir)` | Create a temporary index.json with given facts. |
|
||||
| `test_empty_index` | `test_empty_index()` | Empty knowledge store produces graceful output. |
|
||||
| `test_filter_by_repo` | `test_filter_by_repo()` | Filter facts by repository. |
|
||||
| `test_filter_by_agent` | `test_filter_by_agent()` | Filter facts by agent type. |
|
||||
| `test_no_global_flag` | `test_no_global_flag()` | Excluding global facts works. |
|
||||
| `test_sort_by_confidence` | `test_sort_by_confidence()` | Facts sort by confidence descending. |
|
||||
| `test_sort_pitfalls_first` | `test_sort_pitfalls_first()` | Pitfalls sort before facts at same confidence. |
|
||||
| `test_truncate_to_tokens` | `test_truncate_to_tokens()` | Truncation cuts at line boundary. |
|
||||
| `test_estimate_tokens` | `test_estimate_tokens()` | Token estimation is reasonable. |
|
||||
| `test_build_full_context` | `test_build_full_context()` | Full context with facts renders correctly. |
|
||||
| `test_max_tokens_respected` | `test_max_tokens_respected()` | Output respects max_tokens limit. |
|
||||
| `test_missing_index_graceful` | `test_missing_index_graceful()` | Missing index.json doesn't crash. |
|
||||
|
||||
## `scripts/test_diff_analyzer.py`
|
||||
|
||||
Tests for scripts/diff_analyzer.py — 10 tests.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_empty` | `test_empty()` | - |
|
||||
| `test_addition` | `test_addition()` | - |
|
||||
| `test_deletion` | `test_deletion()` | - |
|
||||
| `test_modification` | `test_modification()` | - |
|
||||
| `test_rename` | `test_rename()` | - |
|
||||
| `test_multiple_files` | `test_multiple_files()` | - |
|
||||
| `test_binary` | `test_binary()` | - |
|
||||
| `test_to_dict` | `test_to_dict()` | - |
|
||||
| `test_context_only` | `test_context_only()` | - |
|
||||
| `test_multi_hunk` | `test_multi_hunk()` | - |
|
||||
| `run_all` | `run_all()` | - |
|
||||
|
||||
## `scripts/test_gitea_issue_parser.py`
|
||||
|
||||
Tests for scripts/gitea_issue_parser.py
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_basic_parsing` | `test_basic_parsing()` | - |
|
||||
| `test_numbered_criteria` | `test_numbered_criteria()` | - |
|
||||
| `test_epic_ref_from_body` | `test_epic_ref_from_body()` | - |
|
||||
| `test_empty_body` | `test_empty_body()` | - |
|
||||
| `test_no_sections` | `test_no_sections()` | - |
|
||||
| `test_multiple_sections` | `test_multiple_sections()` | - |
|
||||
| `run_all` | `run_all()` | - |
|
||||
|
||||
## `scripts/test_harvest_prompt.py`
|
||||
|
||||
Test harness for knowledge extraction prompt.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `validate_knowledge_item` | `validate_knowledge_item(item, idx)` | Validate a single knowledge item. Returns list of errors. |
|
||||
| `validate_extraction` | `validate_extraction(data)` | Validate a full extraction result. Returns (is_valid, errors, warnings). |
|
||||
| `validate_transcript_coverage` | `validate_transcript_coverage(data, transcript)` | Check that extracted facts are actually supported by the transcript. |
|
||||
| `run_tests` | `run_tests()` | Run the built-in test suite. |
|
||||
| `validate_file` | `validate_file(filepath)` | Validate an existing extraction JSON file. |
|
||||
|
||||
## `scripts/test_harvest_prompt_comprehensive.py`
|
||||
|
||||
Comprehensive tests for knowledge extraction prompt.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `check_prompt_structure` | `check_prompt_structure()` | - |
|
||||
| `check_confidence_scoring` | `check_confidence_scoring()` | - |
|
||||
| `check_example_quality` | `check_example_quality()` | - |
|
||||
| `check_constraint_coverage` | `check_constraint_coverage()` | - |
|
||||
| `check_test_sessions` | `check_test_sessions()` | - |
|
||||
| `test_prompt_structure` | `test_prompt_structure()` | - |
|
||||
| `test_confidence_scoring` | `test_confidence_scoring()` | - |
|
||||
| `test_example_quality` | `test_example_quality()` | - |
|
||||
| `test_constraint_coverage` | `test_constraint_coverage()` | - |
|
||||
| `test_test_sessions` | `test_test_sessions()` | - |
|
||||
|
||||
## `scripts/test_harvester_pipeline.py`
|
||||
|
||||
Smoke test for harvester pipeline — verifies the full chain:
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_session_reader` | `test_session_reader()` | Test that session_reader parses JSONL correctly. |
|
||||
| `test_validate_fact` | `test_validate_fact()` | Test fact validation. |
|
||||
| `test_deduplicate` | `test_deduplicate()` | Test deduplication. |
|
||||
| `test_knowledge_store_roundtrip` | `test_knowledge_store_roundtrip()` | Test loading and writing knowledge index. |
|
||||
| `test_full_chain_no_llm` | `test_full_chain_no_llm()` | Test the full pipeline minus the LLM call. |
|
||||
|
||||
## `scripts/test_improvement_proposals.py`
|
||||
|
||||
Tests for scripts/improvement_proposals.py — 15 tests.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_empty_sessions` | `test_empty_sessions()` | - |
|
||||
| `test_no_patterns_on_clean_sessions` | `test_no_patterns_on_clean_sessions()` | - |
|
||||
| `test_repeated_error_detection` | `test_repeated_error_detection()` | Same error across 3+ sessions triggers pattern. |
|
||||
| `test_repeated_error_threshold` | `test_repeated_error_threshold()` | 2 occurrences should NOT trigger (threshold is 3). |
|
||||
| `test_slow_tool_detection` | `test_slow_tool_detection()` | Tool with avg latency > 5000ms across 5+ calls. |
|
||||
| `test_fast_tool_not_flagged` | `test_fast_tool_not_flagged()` | Tool under 5000ms avg should not trigger. |
|
||||
| `test_failed_retry_detection` | `test_failed_retry_detection()` | 3+ consecutive calls to same tool triggers retry pattern. |
|
||||
| `test_manual_process_detection` | `test_manual_process_detection()` | 10+ tool calls with <= 3 unique tools. |
|
||||
| `test_generate_proposals_from_patterns` | `test_generate_proposals_from_patterns()` | Proposals generated from waste patterns. |
|
||||
| `test_proposal_roi_positive` | `test_proposal_roi_positive()` | ROI weeks should be a positive number for recoverable time. |
|
||||
| `test_proposals_sorted_by_impact` | `test_proposals_sorted_by_impact()` | Proposals should be sorted by monthly hours saved (descending). |
|
||||
| `test_format_markdown` | `test_format_markdown()` | Markdown output should contain expected sections. |
|
||||
| `test_format_json` | `test_format_json()` | JSON output should be valid and parseable. |
|
||||
| `test_normalize_error` | `test_normalize_error()` | Error normalization should remove paths and hashes. |
|
||||
| `test_cli_integration` | `test_cli_integration()` | End-to-end test: write input JSON, run script, check output. |
|
||||
| `run_all` | `run_all()` | - |
|
||||
|
||||
## `scripts/test_knowledge_staleness.py`
|
||||
|
||||
Tests for scripts/knowledge_staleness_check.py — 8 tests.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_fresh_entry` | `test_fresh_entry()` | - |
|
||||
| `test_stale_entry` | `test_stale_entry()` | - |
|
||||
| `test_missing_source` | `test_missing_source()` | - |
|
||||
| `test_no_hash` | `test_no_hash()` | - |
|
||||
| `test_no_source_field` | `test_no_source_field()` | - |
|
||||
| `test_fix_hashes` | `test_fix_hashes()` | - |
|
||||
| `test_empty_index` | `test_empty_index()` | - |
|
||||
| `test_compute_hash_nonexistent` | `test_compute_hash_nonexistent()` | - |
|
||||
| `run_all` | `run_all()` | - |
|
||||
|
||||
## `scripts/test_priority_rebalancer.py`
|
||||
|
||||
Tests for Priority Rebalancer
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test` | `test(name)` | - |
|
||||
| `assert_eq` | `assert_eq(a, b, msg)` | - |
|
||||
| `assert_true` | `assert_true(v, msg)` | - |
|
||||
| `assert_false` | `assert_false(v, msg)` | - |
|
||||
| `make_issue` | `make_issue(**kwargs)` | - |
|
||||
|
||||
## `scripts/test_refactoring_opportunity_finder.py`
|
||||
|
||||
Tests for scripts/refactoring_opportunity_finder.py — 10 tests.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_complexity_simple_function` | `test_complexity_simple_function()` | Simple function should have low complexity. |
|
||||
| `test_complexity_with_conditionals` | `test_complexity_with_conditionals()` | Function with if/else should have higher complexity. |
|
||||
| `test_complexity_with_loops` | `test_complexity_with_loops()` | Function with loops should increase complexity. |
|
||||
| `test_complexity_with_class` | `test_complexity_with_class()` | Class with methods should count both. |
|
||||
| `test_complexity_syntax_error` | `test_complexity_syntax_error()` | File with syntax error should return zeros. |
|
||||
| `test_refactoring_score_high_complexity` | `test_refactoring_score_high_complexity()` | High complexity should give high score. |
|
||||
| `test_refactoring_score_low_complexity` | `test_refactoring_score_low_complexity()` | Low complexity should give lower score. |
|
||||
| `test_refactoring_score_high_churn` | `test_refactoring_score_high_churn()` | High churn should increase score. |
|
||||
| `test_refactoring_score_no_coverage` | `test_refactoring_score_no_coverage()` | No coverage data should assume medium risk. |
|
||||
| `test_refactoring_score_large_file` | `test_refactoring_score_large_file()` | Large files should score higher. |
|
||||
| `run_all` | `run_all()` | - |
|
||||
|
||||
## `scripts/test_session_pair_harvester.py`
|
||||
|
||||
Tests for session_pair_harvester.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `test_basic_extraction` | `test_basic_extraction()` | - |
|
||||
| `test_filters_short_responses` | `test_filters_short_responses()` | - |
|
||||
| `test_skips_tool_results` | `test_skips_tool_results()` | - |
|
||||
| `test_deduplication` | `test_deduplication()` | - |
|
||||
| `test_ratio_filter` | `test_ratio_filter()` | - |
|
||||
|
||||
## `scripts/validate_knowledge.py`
|
||||
|
||||
Validate knowledge files and index.json against the schema.
|
||||
|
||||
| Function | Signature | Description |
|
||||
|----------|-----------|-------------|
|
||||
| `validate_fact` | `validate_fact(fact, src)` | - |
|
||||
| `main` | `main()` | - |
|
||||
|
||||
|
||||
---
|
||||
|
||||
**Total scripts documented:** 33
|
||||
|
||||
*Generated by `scripts/api_doc_generator.py` (Issue #98)*
|
||||
@@ -1,219 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
API Doc Generator — Issue #98
|
||||
|
||||
Scans all Python modules in `scripts/`, extracts their public API surface
|
||||
(module docstring + public function signatures + first-line doc summaries),
|
||||
and produces a single markdown reference document at `docs/API.md`.
|
||||
|
||||
Usage:
|
||||
python3 scripts/api_doc_generator.py # Write docs/API.md
|
||||
python3 scripts/api_doc_generator.py --check # Verify docs/API.md is up-to-date
|
||||
python3 scripts/api_doc_generator.py --json # Emit JSON for downstream tooling
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import TypedDict, List, Optional
|
||||
|
||||
|
||||
# ─── Paths ────────────────────────────────────────────────────────────────────
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
REPO_ROOT = SCRIPT_DIR.parent
|
||||
SCRIPTS_DIR = REPO_ROOT / "scripts"
|
||||
DOCS_DIR = REPO_ROOT / "docs"
|
||||
OUTPUT_PATH = DOCS_DIR / "API.md"
|
||||
|
||||
|
||||
# ─── Data structures ───────────────────────────────────────────────────────────
|
||||
class FunctionInfo(TypedDict):
|
||||
name: str
|
||||
signature: str
|
||||
summary: str
|
||||
|
||||
|
||||
class ModuleInfo(TypedDict):
|
||||
path: str # relative to repo root, e.g. "scripts/harvester.py"
|
||||
docstring: str
|
||||
functions: List[FunctionInfo]
|
||||
|
||||
|
||||
# ─── AST extraction ────────────────────────────────────────────────────────────
|
||||
def extract_functions_from_ast(tree: ast.AST, file_rel: str) -> List[FunctionInfo]:
|
||||
"""Extract public function names, signatures, and first-line doc summaries."""
|
||||
funcs: list[FunctionInfo] = []
|
||||
|
||||
for node in ast.iter_child_nodes(tree):
|
||||
if not isinstance(node, ast.FunctionDef):
|
||||
continue
|
||||
# Skip private functions
|
||||
if node.name.startswith("_"):
|
||||
continue
|
||||
|
||||
# Build signature: arg1, arg2=default, *args, **kwargs
|
||||
args = []
|
||||
for arg in node.args.args:
|
||||
args.append(arg.arg)
|
||||
if node.args.vararg:
|
||||
args.append(f"*{node.args.vararg.arg}")
|
||||
if node.args.kwarg:
|
||||
args.append(f"**{node.args.kwarg.arg}")
|
||||
|
||||
# Get first line of docstring
|
||||
summary = ""
|
||||
if (node.body and isinstance(node.body[0], ast.Expr) and
|
||||
isinstance(node.body[0].value, ast.Constant) and
|
||||
isinstance(node.body[0].value.value, str)):
|
||||
raw = node.body[0].value.value.strip()
|
||||
summary = raw.split("\n")[0].strip()
|
||||
if len(summary) > 100:
|
||||
summary = summary[:97] + "..."
|
||||
|
||||
funcs.append({
|
||||
"name": node.name,
|
||||
"signature": ", ".join(args),
|
||||
"summary": summary,
|
||||
})
|
||||
|
||||
return funcs
|
||||
|
||||
|
||||
def parse_module(filepath: Path) -> Optional[ModuleInfo]:
|
||||
"""Parse a Python file and return its module-level docstring and public functions."""
|
||||
try:
|
||||
with open(filepath, "r", encoding="utf-8") as f:
|
||||
source = f.read()
|
||||
tree = ast.parse(source, filename=str(filepath))
|
||||
except Exception as e:
|
||||
print(f"WARNING: Could not parse {filepath}: {e}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
# Module docstring
|
||||
module_doc = ast.get_docstring(tree) or ""
|
||||
module_doc = module_doc.strip().split("\n")[0] # first line only
|
||||
|
||||
# Public functions
|
||||
functions = extract_functions_from_ast(tree, filepath.name)
|
||||
|
||||
rel = filepath.relative_to(REPO_ROOT)
|
||||
return {
|
||||
"path": str(rel),
|
||||
"docstring": module_doc,
|
||||
"functions": functions,
|
||||
}
|
||||
|
||||
|
||||
# ─── Scanning ──────────────────────────────────────────────────────────────────
|
||||
def scan_scripts_dir(scripts_dir: Path) -> List[ModuleInfo]:
|
||||
"""Scan all .py files in scripts/ and extract API info."""
|
||||
modules: list[ModuleInfo] = []
|
||||
for pyfile in sorted(scripts_dir.glob("*.py")):
|
||||
info = parse_module(pyfile)
|
||||
if info is not None:
|
||||
modules.append(info)
|
||||
return modules
|
||||
|
||||
|
||||
# ─── Markdown rendering ─────────────────────────────────────────────────────────
|
||||
def render_markdown(modules: List[ModuleInfo]) -> str:
|
||||
"""Generate full docs/API.md content from the scanned modules."""
|
||||
lines = [
|
||||
"# Compounding Intelligence — Scripts API Reference",
|
||||
"",
|
||||
f"*Generated: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}*",
|
||||
"",
|
||||
"This document auto-documents the public API surface of all scripts",
|
||||
"in `scripts/`. Each section covers one script: module purpose,",
|
||||
"public functions, and their signatures.",
|
||||
"",
|
||||
"---",
|
||||
"",
|
||||
]
|
||||
|
||||
for mod in modules:
|
||||
rel = mod["path"]
|
||||
name = Path(rel).stem # e.g. harvester
|
||||
lines.append(f"## `{rel}`")
|
||||
lines.append("")
|
||||
if mod["docstring"]:
|
||||
lines.append(mod["docstring"])
|
||||
lines.append("")
|
||||
|
||||
if mod["functions"]:
|
||||
lines.append("| Function | Signature | Description |")
|
||||
lines.append("|----------|-----------|-------------|")
|
||||
for fn in mod["functions"]:
|
||||
sig = fn["name"] + "(" + fn["signature"] + ")"
|
||||
desc = fn["summary"] or "-"
|
||||
lines.append(f"| `{fn['name']}` | `{sig}` | {desc} |")
|
||||
lines.append("")
|
||||
else:
|
||||
lines.append("*(no public functions — script runs as `main()` only)*")
|
||||
lines.append("")
|
||||
|
||||
lines.extend([
|
||||
"",
|
||||
"---",
|
||||
"",
|
||||
f"**Total scripts documented:** {len(modules)}",
|
||||
"",
|
||||
"*Generated by `scripts/api_doc_generator.py` (Issue #98)*",
|
||||
])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ─── JSON output (optional, for automation) ───────────────────────────────────
|
||||
def render_json(modules: List[ModuleInfo]) -> str:
|
||||
"""Emit machine-readable JSON version of the API reference."""
|
||||
import json
|
||||
payload = {
|
||||
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||||
"generator": "scripts/api_doc_generator.py",
|
||||
"repo": "Timmy_Foundation/compounding-intelligence",
|
||||
"modules": modules,
|
||||
}
|
||||
return json.dumps(payload, indent=2)
|
||||
|
||||
|
||||
# ─── Main ──────────────────────────────────────────────────────────────────────
|
||||
def main() -> int:
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="Generate API docs for scripts/")
|
||||
parser.add_argument("--check", action="store_true",
|
||||
help="Exit 1 if docs/API.md is out-of-date")
|
||||
parser.add_argument("--json", action="store_true",
|
||||
help="Emit JSON to stdout instead of writing markdown")
|
||||
args = parser.parse_args()
|
||||
|
||||
modules = scan_scripts_dir(SCRIPTS_DIR)
|
||||
modules.sort(key=lambda m: m["path"])
|
||||
|
||||
if args.json:
|
||||
print(render_json(modules))
|
||||
return 0
|
||||
|
||||
md = render_markdown(modules)
|
||||
|
||||
if args.check:
|
||||
if OUTPUT_PATH.exists():
|
||||
existing = OUTPUT_PATH.read_text(encoding="utf-8")
|
||||
if existing == md:
|
||||
print("✅ docs/API.md is up-to-date")
|
||||
return 0
|
||||
print("❌ docs/API.md is missing or out-of-date — regenerate with "
|
||||
"`python3 scripts/api_doc_generator.py`", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
DOCS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
OUTPUT_PATH.write_text(md, encoding="utf-8")
|
||||
print(f"✅ Wrote {OUTPUT_PATH} ({len(modules)} modules documented)")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
351
scripts/pr_complexity_scorer.py
Normal file
351
scripts/pr_complexity_scorer.py
Normal file
@@ -0,0 +1,351 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PR Complexity Scorer - Estimate review effort for PRs.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, asdict
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
GITEA_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
||||
|
||||
DEPENDENCY_FILES = {
|
||||
"requirements.txt", "pyproject.toml", "setup.py", "setup.cfg",
|
||||
"Pipfile", "poetry.lock", "package.json", "yarn.lock", "Gemfile",
|
||||
"go.mod", "Cargo.toml", "pom.xml", "build.gradle"
|
||||
}
|
||||
|
||||
TEST_PATTERNS = [
|
||||
r"tests?/.*\.py$", r".*_test\.py$", r"test_.*\.py$",
|
||||
r"spec/.*\.rb$", r".*_spec\.rb$",
|
||||
r"__tests__/", r".*\.test\.(js|ts|jsx|tsx)$"
|
||||
]
|
||||
|
||||
WEIGHT_FILES = 0.25
|
||||
WEIGHT_LINES = 0.25
|
||||
WEIGHT_DEPS = 0.30
|
||||
WEIGHT_TEST_COV = 0.20
|
||||
|
||||
SMALL_FILES = 5
|
||||
MEDIUM_FILES = 20
|
||||
LARGE_FILES = 50
|
||||
|
||||
SMALL_LINES = 100
|
||||
MEDIUM_LINES = 500
|
||||
LARGE_LINES = 2000
|
||||
|
||||
TIME_PER_POINT = {1: 5, 2: 10, 3: 15, 4: 20, 5: 25, 6: 30, 7: 45, 8: 60, 9: 90, 10: 120}
|
||||
|
||||
|
||||
@dataclass
|
||||
class PRComplexity:
|
||||
pr_number: int
|
||||
title: str
|
||||
files_changed: int
|
||||
additions: int
|
||||
deletions: int
|
||||
has_dependency_changes: bool
|
||||
test_coverage_delta: Optional[int]
|
||||
score: int
|
||||
estimated_minutes: int
|
||||
reasons: List[str]
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
def __init__(self, token: str):
|
||||
self.token = token
|
||||
self.base_url = GITEA_BASE.rstrip("/")
|
||||
|
||||
def _request(self, path: str, params: Dict = None) -> Any:
|
||||
url = f"{self.base_url}{path}"
|
||||
if params:
|
||||
qs = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
|
||||
url += f"?{qs}"
|
||||
|
||||
req = urllib.request.Request(url)
|
||||
req.add_header("Authorization", f"token {self.token}")
|
||||
req.add_header("Content-Type", "application/json")
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
except urllib.error.HTTPError as e:
|
||||
print(f"API error {e.code}: {e.read().decode()[:200]}", file=sys.stderr)
|
||||
return None
|
||||
except urllib.error.URLError as e:
|
||||
print(f"Network error: {e}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
def get_open_prs(self, org: str, repo: str) -> List[Dict]:
|
||||
prs = []
|
||||
page = 1
|
||||
while True:
|
||||
batch = self._request(f"/repos/{org}/{repo}/pulls", {"limit": 50, "page": page, "state": "open"})
|
||||
if not batch:
|
||||
break
|
||||
prs.extend(batch)
|
||||
if len(batch) < 50:
|
||||
break
|
||||
page += 1
|
||||
return prs
|
||||
|
||||
def get_pr_files(self, org: str, repo: str, pr_number: int) -> List[Dict]:
|
||||
files = []
|
||||
page = 1
|
||||
while True:
|
||||
batch = self._request(
|
||||
f"/repos/{org}/{repo}/pulls/{pr_number}/files",
|
||||
{"limit": 100, "page": page}
|
||||
)
|
||||
if not batch:
|
||||
break
|
||||
files.extend(batch)
|
||||
if len(batch) < 100:
|
||||
break
|
||||
page += 1
|
||||
return files
|
||||
|
||||
def post_comment(self, org: str, repo: str, pr_number: int, body: str) -> bool:
|
||||
data = json.dumps({"body": body}).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
f"{self.base_url}/repos/{org}/{repo}/issues/{pr_number}/comments",
|
||||
data=data,
|
||||
method="POST",
|
||||
headers={"Authorization": f"token {self.token}", "Content-Type": "application/json"}
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return resp.status in (200, 201)
|
||||
except urllib.error.HTTPError:
|
||||
return False
|
||||
|
||||
|
||||
def is_dependency_file(filename: str) -> bool:
|
||||
return any(filename.endswith(dep) for dep in DEPENDENCY_FILES)
|
||||
|
||||
|
||||
def is_test_file(filename: str) -> bool:
|
||||
return any(re.search(pattern, filename) for pattern in TEST_PATTERNS)
|
||||
|
||||
|
||||
def score_pr(
|
||||
files_changed: int,
|
||||
additions: int,
|
||||
deletions: int,
|
||||
has_dependency_changes: bool,
|
||||
test_coverage_delta: Optional[int] = None
|
||||
) -> tuple[int, int, List[str]]:
|
||||
score = 1.0
|
||||
reasons = []
|
||||
|
||||
# Files changed
|
||||
if files_changed <= SMALL_FILES:
|
||||
fscore = 1.0
|
||||
reasons.append("small number of files changed")
|
||||
elif files_changed <= MEDIUM_FILES:
|
||||
fscore = 2.0
|
||||
reasons.append("moderate number of files changed")
|
||||
elif files_changed <= LARGE_FILES:
|
||||
fscore = 2.5
|
||||
reasons.append("large number of files changed")
|
||||
else:
|
||||
fscore = 3.0
|
||||
reasons.append("very large PR spanning many files")
|
||||
|
||||
# Lines changed
|
||||
total_lines = additions + deletions
|
||||
if total_lines <= SMALL_LINES:
|
||||
lscore = 1.0
|
||||
reasons.append("small change size")
|
||||
elif total_lines <= MEDIUM_LINES:
|
||||
lscore = 2.0
|
||||
reasons.append("moderate change size")
|
||||
elif total_lines <= LARGE_LINES:
|
||||
lscore = 3.0
|
||||
reasons.append("large change size")
|
||||
else:
|
||||
lscore = 4.0
|
||||
reasons.append("very large change")
|
||||
|
||||
# Dependency changes
|
||||
if has_dependency_changes:
|
||||
dscore = 2.5
|
||||
reasons.append("dependency changes (architectural impact)")
|
||||
else:
|
||||
dscore = 0.0
|
||||
|
||||
# Test coverage delta
|
||||
tscore = 0.0
|
||||
if test_coverage_delta is not None:
|
||||
if test_coverage_delta > 0:
|
||||
reasons.append(f"test additions (+{test_coverage_delta} test files)")
|
||||
tscore = -min(2.0, test_coverage_delta / 2.0)
|
||||
elif test_coverage_delta < 0:
|
||||
reasons.append(f"test removals ({abs(test_coverage_delta)} test files)")
|
||||
tscore = min(2.0, abs(test_coverage_delta) * 0.5)
|
||||
else:
|
||||
reasons.append("test coverage change not assessed")
|
||||
|
||||
# Weighted sum, scaled by 3 to use full 1-10 range
|
||||
bonus = (fscore * WEIGHT_FILES) + (lscore * WEIGHT_LINES) + (dscore * WEIGHT_DEPS) + (tscore * WEIGHT_TEST_COV)
|
||||
scaled_bonus = bonus * 3.0
|
||||
score = 1.0 + scaled_bonus
|
||||
|
||||
final_score = max(1, min(10, int(round(score))))
|
||||
est_minutes = TIME_PER_POINT.get(final_score, 30)
|
||||
|
||||
return final_score, est_minutes, reasons
|
||||
|
||||
|
||||
def analyze_pr(client: GiteaClient, org: str, repo: str, pr_data: Dict) -> PRComplexity:
|
||||
pr_num = pr_data["number"]
|
||||
title = pr_data.get("title", "")
|
||||
files = client.get_pr_files(org, repo, pr_num)
|
||||
|
||||
additions = sum(f.get("additions", 0) for f in files)
|
||||
deletions = sum(f.get("deletions", 0) for f in files)
|
||||
filenames = [f.get("filename", "") for f in files]
|
||||
|
||||
has_deps = any(is_dependency_file(f) for f in filenames)
|
||||
|
||||
test_added = sum(1 for f in files if f.get("status") == "added" and is_test_file(f.get("filename", "")))
|
||||
test_removed = sum(1 for f in files if f.get("status") == "removed" and is_test_file(f.get("filename", "")))
|
||||
test_delta = test_added - test_removed if (test_added or test_removed) else None
|
||||
|
||||
score, est_min, reasons = score_pr(
|
||||
files_changed=len(files),
|
||||
additions=additions,
|
||||
deletions=deletions,
|
||||
has_dependency_changes=has_deps,
|
||||
test_coverage_delta=test_delta
|
||||
)
|
||||
|
||||
return PRComplexity(
|
||||
pr_number=pr_num,
|
||||
title=title,
|
||||
files_changed=len(files),
|
||||
additions=additions,
|
||||
deletions=deletions,
|
||||
has_dependency_changes=has_deps,
|
||||
test_coverage_delta=test_delta,
|
||||
score=score,
|
||||
estimated_minutes=est_min,
|
||||
reasons=reasons
|
||||
)
|
||||
|
||||
|
||||
def build_comment(complexity: PRComplexity) -> str:
|
||||
change_desc = f"{complexity.files_changed} files, +{complexity.additions}/-{complexity.deletions} lines"
|
||||
deps_note = "\n- :warning: Dependency changes detected — architectural review recommended" if complexity.has_dependency_changes else ""
|
||||
test_note = ""
|
||||
if complexity.test_coverage_delta is not None:
|
||||
if complexity.test_coverage_delta > 0:
|
||||
test_note = f"\n- :+1: {complexity.test_coverage_delta} test file(s) added"
|
||||
elif complexity.test_coverage_delta < 0:
|
||||
test_note = f"\n- :warning: {abs(complexity.test_coverage_delta)} test file(s) removed"
|
||||
|
||||
comment = f"## 📊 PR Complexity Analysis\n\n"
|
||||
comment += f"**PR #{complexity.pr_number}: {complexity.title}**\n\n"
|
||||
comment += f"| Metric | Value |\n|--------|-------|\n"
|
||||
comment += f"| Changes | {change_desc} |\n"
|
||||
comment += f"| Complexity Score | **{complexity.score}/10** |\n"
|
||||
comment += f"| Estimated Review Time | ~{complexity.estimated_minutes} minutes |\n\n"
|
||||
comment += f"### Scoring rationale:"
|
||||
for r in complexity.reasons:
|
||||
comment += f"\n- {r}"
|
||||
if deps_note:
|
||||
comment += deps_note
|
||||
if test_note:
|
||||
comment += test_note
|
||||
comment += f"\n\n---\n"
|
||||
comment += f"*Generated by PR Complexity Scorer — [issue #135](https://forge.alexanderwhitestone.com/Timmy_Foundation/compounding-intelligence/issues/135)*"
|
||||
return comment
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="PR Complexity Scorer")
|
||||
parser.add_argument("--org", default="Timmy_Foundation")
|
||||
parser.add_argument("--repo", default="compounding-intelligence")
|
||||
parser.add_argument("--token", default=os.environ.get("GITEA_TOKEN") or os.path.expanduser("~/.config/gitea/token"))
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
parser.add_argument("--apply", action="store_true")
|
||||
parser.add_argument("--output", default="metrics/pr_complexity.json")
|
||||
args = parser.parse_args()
|
||||
|
||||
token_path = args.token
|
||||
if os.path.exists(token_path):
|
||||
with open(token_path) as f:
|
||||
token = f.read().strip()
|
||||
else:
|
||||
token = args.token
|
||||
|
||||
if not token:
|
||||
print("ERROR: No Gitea token provided", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
client = GiteaClient(token)
|
||||
|
||||
print(f"Fetching open PRs for {args.org}/{args.repo}...")
|
||||
prs = client.get_open_prs(args.org, args.repo)
|
||||
if not prs:
|
||||
print("No open PRs found.")
|
||||
sys.exit(0)
|
||||
|
||||
print(f"Found {len(prs)} open PR(s). Analyzing...")
|
||||
|
||||
results = []
|
||||
Path(args.output).parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for pr in prs:
|
||||
pr_num = pr["number"]
|
||||
title = pr.get("title", "")
|
||||
print(f" Analyzing PR #{pr_num}: {title[:60]}")
|
||||
|
||||
try:
|
||||
complexity = analyze_pr(client, args.org, args.repo, pr)
|
||||
results.append(complexity.to_dict())
|
||||
|
||||
comment = build_comment(complexity)
|
||||
|
||||
if args.dry_run:
|
||||
print(f" → Score: {complexity.score}/10, Est: {complexity.estimated_minutes}min [DRY-RUN]")
|
||||
elif args.apply:
|
||||
success = client.post_comment(args.org, args.repo, pr_num, comment)
|
||||
status = "[commented]" if success else "[FAILED]"
|
||||
print(f" → Score: {complexity.score}/10, Est: {complexity.estimated_minutes}min {status}")
|
||||
else:
|
||||
print(f" → Score: {complexity.score}/10, Est: {complexity.estimated_minutes}min [no action]")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ERROR analyzing PR #{pr_num}: {e}", file=sys.stderr)
|
||||
|
||||
with open(args.output, "w") as f:
|
||||
json.dump({
|
||||
"org": args.org,
|
||||
"repo": args.repo,
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"pr_count": len(results),
|
||||
"results": results
|
||||
}, f, indent=2)
|
||||
|
||||
if results:
|
||||
scores = [r["score"] for r in results]
|
||||
print(f"\nResults saved to {args.output}")
|
||||
print(f"Summary: {len(results)} PRs, scores range {min(scores):.0f}-{max(scores):.0f}")
|
||||
else:
|
||||
print("\nNo results to save.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
170
scripts/test_pr_complexity_scorer.py
Normal file
170
scripts/test_pr_complexity_scorer.py
Normal file
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tests for PR Complexity Scorer — unit tests for the scoring logic.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
from pr_complexity_scorer import (
|
||||
score_pr,
|
||||
is_dependency_file,
|
||||
is_test_file,
|
||||
TIME_PER_POINT,
|
||||
SMALL_FILES,
|
||||
MEDIUM_FILES,
|
||||
LARGE_FILES,
|
||||
SMALL_LINES,
|
||||
MEDIUM_LINES,
|
||||
LARGE_LINES,
|
||||
)
|
||||
|
||||
PASS = 0
|
||||
FAIL = 0
|
||||
|
||||
def test(name):
|
||||
def decorator(fn):
|
||||
global PASS, FAIL
|
||||
try:
|
||||
fn()
|
||||
PASS += 1
|
||||
print(f" [PASS] {name}")
|
||||
except AssertionError as e:
|
||||
FAIL += 1
|
||||
print(f" [FAIL] {name}: {e}")
|
||||
except Exception as e:
|
||||
FAIL += 1
|
||||
print(f" [FAIL] {name}: Unexpected error: {e}")
|
||||
return decorator
|
||||
|
||||
def assert_eq(a, b, msg=""):
|
||||
if a != b:
|
||||
raise AssertionError(f"{msg} expected {b!r}, got {a!r}")
|
||||
|
||||
def assert_true(v, msg=""):
|
||||
if not v:
|
||||
raise AssertionError(msg or "Expected True")
|
||||
|
||||
def assert_false(v, msg=""):
|
||||
if v:
|
||||
raise AssertionError(msg or "Expected False")
|
||||
|
||||
|
||||
print("=== PR Complexity Scorer Tests ===\n")
|
||||
|
||||
print("-- File Classification --")
|
||||
|
||||
@test("dependency file detection — requirements.txt")
|
||||
def _():
|
||||
assert_true(is_dependency_file("requirements.txt"))
|
||||
assert_true(is_dependency_file("src/requirements.txt"))
|
||||
assert_false(is_dependency_file("requirements_test.txt"))
|
||||
|
||||
@test("dependency file detection — pyproject.toml")
|
||||
def _():
|
||||
assert_true(is_dependency_file("pyproject.toml"))
|
||||
assert_false(is_dependency_file("myproject.py"))
|
||||
|
||||
@test("test file detection — pytest style")
|
||||
def _():
|
||||
assert_true(is_test_file("tests/test_api.py"))
|
||||
assert_true(is_test_file("test_module.py"))
|
||||
assert_true(is_test_file("src/module_test.py"))
|
||||
|
||||
@test("test file detection — other frameworks")
|
||||
def _():
|
||||
assert_true(is_test_file("spec/feature_spec.rb"))
|
||||
assert_true(is_test_file("__tests__/component.test.js"))
|
||||
assert_false(is_test_file("testfixtures/helper.py"))
|
||||
|
||||
|
||||
print("\n-- Scoring Logic --")
|
||||
|
||||
@test("small PR gets low score (1-3)")
|
||||
def _():
|
||||
score, minutes, _ = score_pr(
|
||||
files_changed=3,
|
||||
additions=50,
|
||||
deletions=10,
|
||||
has_dependency_changes=False,
|
||||
test_coverage_delta=None
|
||||
)
|
||||
assert_true(1 <= score <= 3, f"Score should be low, got {score}")
|
||||
assert_true(minutes < 20)
|
||||
|
||||
@test("medium PR gets medium score (4-6)")
|
||||
def _():
|
||||
score, minutes, _ = score_pr(
|
||||
files_changed=15,
|
||||
additions=400,
|
||||
deletions=100,
|
||||
has_dependency_changes=False,
|
||||
test_coverage_delta=None
|
||||
)
|
||||
assert_true(4 <= score <= 6, f"Score should be medium, got {score}")
|
||||
assert_true(20 <= minutes <= 45)
|
||||
|
||||
@test("large PR gets high score (7-9)")
|
||||
def _():
|
||||
score, minutes, _ = score_pr(
|
||||
files_changed=60,
|
||||
additions=3000,
|
||||
deletions=1500,
|
||||
has_dependency_changes=True,
|
||||
test_coverage_delta=None
|
||||
)
|
||||
assert_true(7 <= score <= 9, f"Score should be high, got {score}")
|
||||
assert_true(minutes >= 45)
|
||||
|
||||
@test("dependency changes boost score")
|
||||
def _():
|
||||
base_score, _, _ = score_pr(
|
||||
files_changed=10, additions=200, deletions=50,
|
||||
has_dependency_changes=False, test_coverage_delta=None
|
||||
)
|
||||
dep_score, _, _ = score_pr(
|
||||
files_changed=10, additions=200, deletions=50,
|
||||
has_dependency_changes=True, test_coverage_delta=None
|
||||
)
|
||||
assert_true(dep_score > base_score, f"Deps: {base_score} -> {dep_score}")
|
||||
|
||||
@test("adding tests lowers complexity")
|
||||
def _():
|
||||
base_score, _, _ = score_pr(
|
||||
files_changed=8, additions=150, deletions=20,
|
||||
has_dependency_changes=False, test_coverage_delta=None
|
||||
)
|
||||
better_score, _, _ = score_pr(
|
||||
files_changed=8, additions=180, deletions=20,
|
||||
has_dependency_changes=False, test_coverage_delta=3
|
||||
)
|
||||
assert_true(better_score < base_score, f"Tests: {base_score} -> {better_score}")
|
||||
|
||||
@test("removing tests increases complexity")
|
||||
def _():
|
||||
base_score, _, _ = score_pr(
|
||||
files_changed=8, additions=150, deletions=20,
|
||||
has_dependency_changes=False, test_coverage_delta=None
|
||||
)
|
||||
worse_score, _, _ = score_pr(
|
||||
files_changed=8, additions=150, deletions=20,
|
||||
has_dependency_changes=False, test_coverage_delta=-2
|
||||
)
|
||||
assert_true(worse_score > base_score, f"Remove tests: {base_score} -> {worse_score}")
|
||||
|
||||
@test("score bounded 1-10")
|
||||
def _():
|
||||
for files, adds, dels in [(1, 10, 5), (100, 10000, 5000)]:
|
||||
score, _, _ = score_pr(files, adds, dels, False, None)
|
||||
assert_true(1 <= score <= 10, f"Score {score} out of range")
|
||||
|
||||
@test("estimated minutes exist for all scores")
|
||||
def _():
|
||||
for s in range(1, 11):
|
||||
assert_true(s in TIME_PER_POINT, f"Missing time for score {s}")
|
||||
|
||||
|
||||
print(f"\n=== Results: {PASS} passed, {FAIL} failed ===")
|
||||
sys.exit(0 if FAIL == 0 else 1)
|
||||
470
scripts/vulnerability_scanner.py
Normal file
470
scripts/vulnerability_scanner.py
Normal file
@@ -0,0 +1,470 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
vulnerability_scanner.py — Check Python dependencies against CVE databases (Issue #108)
|
||||
|
||||
Scans requirements.txt (or any pip-compatible dependency file) and queries
|
||||
the Open Source Vulnerability (OSV) database for known security issues.
|
||||
|
||||
OSV API: https://api.osv.dev/v1/query (free, no auth, PyPI ecosystem supported)
|
||||
|
||||
Output:
|
||||
- Human-readable summary on stdout
|
||||
- JSON report with full vulnerability details
|
||||
- Exit code: 0 if no vulnerabilities found, 1 if critical/high found, 2 otherwise
|
||||
|
||||
Usage:
|
||||
python3 scripts/vulnerability_scanner.py
|
||||
python3 scripts/vulnerability_scanner.py --deps requirements.txt --output json
|
||||
python3 scripts/vulnerability_scanner.py --min-severity high
|
||||
python3 scripts/vulnerability_scanner.py --deps requirements.txt --report-format markdown
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
# --- Configuration ---
|
||||
|
||||
OSV_API_URL = "https://api.osv.dev/v1/query"
|
||||
DEFAULT_REQUIREMENTS_PATH = "requirements.txt"
|
||||
SEVERITY_LEVELS = ["critical", "high", "medium", "low", "unknown"]
|
||||
|
||||
# Map OSV severities to our buckets
|
||||
CVSS_SEVERITY_MAP = {
|
||||
"CRITICAL": "critical",
|
||||
"HIGH": "high",
|
||||
"MEDIUM": "medium",
|
||||
"LOW": "low",
|
||||
"NONE": "none",
|
||||
}
|
||||
|
||||
# --- Data Structures ---
|
||||
|
||||
|
||||
@dataclass
|
||||
class Vulnerability:
|
||||
"""A single vulnerability finding."""
|
||||
package: str
|
||||
version: str
|
||||
vuln_id: str
|
||||
severity: str
|
||||
cvss_score: Optional[float]
|
||||
summary: str
|
||||
details_url: str
|
||||
fixed_versions: List[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScanResult:
|
||||
"""Results from a vulnerability scan."""
|
||||
scanned_packages: int
|
||||
vulnerabilities: List[Vulnerability]
|
||||
errors: List[Tuple[str, str]] # (package, error_message)
|
||||
|
||||
|
||||
# --- Requirement Parsing ---
|
||||
|
||||
|
||||
def parse_requirements_file(path: str) -> Dict[str, str]:
|
||||
"""
|
||||
Parse a requirements.txt file into {package_name: version_spec}.
|
||||
|
||||
Handles:
|
||||
- pkg==1.2.3
|
||||
- pkg>=1.0.0
|
||||
- pkg[extra]==1.2.3
|
||||
- -e/--editable entries (skipped)
|
||||
- -r inclusions (recursive, limited depth)
|
||||
- comments and blank lines
|
||||
"""
|
||||
packages = {}
|
||||
processed_includes = set()
|
||||
|
||||
def parse_line(line: str, filename: str, depth: int = 0) -> None:
|
||||
if depth > 3:
|
||||
print(f"WARNING: Max include depth exceeded in {filename}", file=sys.stderr)
|
||||
return
|
||||
|
||||
line = line.strip()
|
||||
if not line or line.startswith('#'):
|
||||
return
|
||||
|
||||
# Handle -r or --requirement includes
|
||||
if line.startswith('-r ') or line.startswith('--requirement '):
|
||||
if depth >= 3:
|
||||
return
|
||||
include_path = line.split(None, 1)[1].strip()
|
||||
# Resolve relative to current file's directory
|
||||
base_dir = os.path.dirname(os.path.abspath(filename))
|
||||
full_path = os.path.join(base_dir, include_path)
|
||||
if full_path not in processed_includes:
|
||||
processed_includes.add(full_path)
|
||||
try:
|
||||
with open(full_path, 'r', encoding='utf-8') as f:
|
||||
for incl_line in f:
|
||||
parse_line(incl_line, full_path, depth + 1)
|
||||
except FileNotFoundError:
|
||||
print(f"WARNING: Could not read included file: {full_path}", file=sys.stderr)
|
||||
return
|
||||
|
||||
# Skip editable installs and other flags
|
||||
if line.startswith('-e ') or line.startswith('--editable ') or line.startswith('-'):
|
||||
return
|
||||
|
||||
# Extract package name and version spec
|
||||
# Handles: pkg==1.2.3, pkg>=1.0, pkg[extra]==1.2.3, pkg ~= 1.0
|
||||
# Strip inline comment first
|
||||
line = line.split('#', 1)[0].strip()
|
||||
if not line:
|
||||
return
|
||||
|
||||
# Skip editable installs and other option lines
|
||||
if line.startswith('-e ') or line.startswith('--editable ') or (line.startswith('-') and not re.match(r'^[a-zA-Z0-9]', line[1:])):
|
||||
return
|
||||
|
||||
# Extract package name: leading identifier before any extras or version spec
|
||||
pkg_match = re.match(r'^([a-zA-Z0-9]([a-zA-Z0-9._-]*[a-zA-Z0-9])?)', line)
|
||||
if not pkg_match:
|
||||
return
|
||||
pkg_name = pkg_match.group(1).lower()
|
||||
|
||||
# Strip extras [extra] from remainder
|
||||
remainder = line[pkg_match.end():]
|
||||
remainder = re.sub(r'\[.*?\]', '', remainder)
|
||||
|
||||
# Extract version comparison
|
||||
version = ""
|
||||
ver_match = re.search(r'(===|==|~=|>=|<=|!=)\s*([^\s;]+)', remainder)
|
||||
if ver_match:
|
||||
version = ver_match.group(1) + ver_match.group(2)
|
||||
|
||||
packages[pkg_name] = version
|
||||
|
||||
# Read and parse the file
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
for line in f:
|
||||
parse_line(line, path, 0)
|
||||
except FileNotFoundError:
|
||||
print(f"ERROR: Requirements file not found: {path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
return packages
|
||||
|
||||
|
||||
# --- OSV API Queries ---
|
||||
|
||||
|
||||
def query_osv(package: str, version: str) -> List[dict]:
|
||||
"""
|
||||
Query the OSV API for vulnerabilities affecting a specific package version.
|
||||
|
||||
Returns list of vulnerability dicts (raw API response) or empty list on error.
|
||||
"""
|
||||
# Normalize version spec for OSV query
|
||||
# OSV expects a specific version, not a range. We query for the exact version
|
||||
# if available, otherwise we query without version to get all vulns for the package
|
||||
# and let the caller filter.
|
||||
query_version = version if re.match(r'^[0-9]', version) else None
|
||||
|
||||
payload = {
|
||||
"package": {
|
||||
"name": package,
|
||||
"ecosystem": "PyPI"
|
||||
}
|
||||
}
|
||||
if query_version:
|
||||
payload["version"] = query_version
|
||||
|
||||
data = json.dumps(payload).encode('utf-8')
|
||||
req = urllib.request.Request(
|
||||
OSV_API_URL,
|
||||
data=data,
|
||||
headers={'Content-Type': 'application/json'},
|
||||
method='POST'
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=15) as response:
|
||||
result = json.loads(response.read().decode('utf-8'))
|
||||
return result.get('vulns', []) + result.get('vulnerabilities', [])
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 404:
|
||||
return [] # No vulnerabilities found
|
||||
print(f"WARNING: OSV query failed for {package}: HTTP {e.code}", file=sys.stderr)
|
||||
except (urllib.error.URLError, json.JSONDecodeError, TimeoutError) as e:
|
||||
print(f"WARNING: OSV query failed for {package}: {e}", file=sys.stderr)
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def parse_osv_vuln(raw_vulns: List[dict], package: str, version_spec: str) -> List[Vulnerability]:
|
||||
"""
|
||||
Parse raw OSV API responses into Vulnerability objects.
|
||||
"""
|
||||
vulns = []
|
||||
for v in raw_vulns:
|
||||
vuln_id = v.get('id', 'UNKNOWN')
|
||||
summary = v.get('summary', 'No summary provided.')
|
||||
|
||||
# Severity from CVSS or ecosystem-specific
|
||||
severity = "unknown"
|
||||
cvss_score = None
|
||||
if 'severity' in v:
|
||||
for sev_info in v['severity']:
|
||||
if sev_info.get('type') == 'CVSS_V3':
|
||||
score = sev_info.get('score', '')
|
||||
if isinstance(score, dict):
|
||||
cvss_score = score.get('baseScore')
|
||||
sev_str = score.get('baseSeverity', '').upper()
|
||||
severity = CVSS_SEVERITY_MAP.get(sev_str, 'unknown')
|
||||
break
|
||||
elif sev_info.get('type') == 'CVSS_V2':
|
||||
# Fallback
|
||||
score = sev_info.get('score', '')
|
||||
if isinstance(score, dict):
|
||||
cvss_score = score.get('baseScore')
|
||||
sev_str = sev_info.get('type', '').upper()
|
||||
severity = "unknown"
|
||||
|
||||
# Affected packages/ranges
|
||||
affected = v.get('affected', [])
|
||||
fixed_versions = []
|
||||
for aff in affected:
|
||||
for r in aff.get('ranges', []):
|
||||
for event in r.get('events', []):
|
||||
if event.get('introduced'):
|
||||
# We have the version, fixed would be in 'fixed' events
|
||||
pass
|
||||
if event.get('fixed'):
|
||||
fixed_versions.append(event['fixed'])
|
||||
|
||||
# Build details URL
|
||||
details_url = f"https://osv.dev/vulnerability/{vuln_id}"
|
||||
|
||||
vuln = Vulnerability(
|
||||
package=package,
|
||||
version=version_spec,
|
||||
vuln_id=vuln_id,
|
||||
severity=severity,
|
||||
cvss_score=cvss_score,
|
||||
summary=summary,
|
||||
details_url=details_url,
|
||||
fixed_versions=list(set(fixed_versions))
|
||||
)
|
||||
vulns.append(vuln)
|
||||
|
||||
return vulns
|
||||
|
||||
|
||||
# --- Filtering & Reporting ---
|
||||
|
||||
|
||||
def filter_by_severity(vulns: List[Vulnerability], min_severity: str) -> List[Vulnerability]:
|
||||
"""Filter vulnerabilities to include only those at or above the given severity."""
|
||||
if min_severity.lower() not in SEVERITY_LEVELS:
|
||||
return vulns # No filtering if invalid
|
||||
|
||||
min_idx = SEVERITY_LEVELS.index(min_severity.lower())
|
||||
filtered = []
|
||||
for v in vulns:
|
||||
sev_idx = SEVERITY_LEVELS.index(v.severity.lower())
|
||||
if sev_idx <= min_idx: # lower index = more severe
|
||||
filtered.append(v)
|
||||
return filtered
|
||||
|
||||
|
||||
def generate_text_report(result: ScanResult, packages: Dict[str, str]) -> str:
|
||||
"""Generate human-readable text report."""
|
||||
lines = []
|
||||
lines.append("=" * 60)
|
||||
lines.append("Vulnerability Scan Report")
|
||||
lines.append("=" * 60)
|
||||
lines.append(f"Packages scanned: {result.scanned_packages}")
|
||||
lines.append(f"Vulnerabilities found: {len(result.vulnerabilities)}")
|
||||
|
||||
if result.errors:
|
||||
lines.append(f"Errors: {len(result.errors)}")
|
||||
|
||||
# Group by severity
|
||||
by_severity: Dict[str, List[Vulnerability]] = {}
|
||||
for v in result.vulnerabilities:
|
||||
by_severity.setdefault(v.severity.upper(), []).append(v)
|
||||
|
||||
for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW", "UNKNOWN"]:
|
||||
vuln_list = by_severity.get(sev, [])
|
||||
if vuln_list:
|
||||
lines.append(f"\n{sev}: {len(vuln_list)}")
|
||||
for v in vuln_list:
|
||||
lines.append(f" [{v.package} {packages.get(v.package, '')}] {v.vuln_id}")
|
||||
lines.append(f" {v.summary[:80]}")
|
||||
if v.cvss_score:
|
||||
lines.append(f" CVSS: {v.cvss_score}")
|
||||
if v.fixed_versions:
|
||||
lines.append(f" Fixed in: {', '.join(v.fixed_versions[:3])}")
|
||||
lines.append(f" {v.details_url}")
|
||||
|
||||
if result.errors:
|
||||
lines.append("\nERRORS:")
|
||||
for pkg, err in result.errors[:10]:
|
||||
lines.append(f" {pkg}: {err}")
|
||||
|
||||
lines.append("\n" + "=" * 60)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def generate_json_report(result: ScanResult, packages: Dict[str, str]) -> str:
|
||||
"""Generate JSON report."""
|
||||
report = {
|
||||
"scanned_packages": result.scanned_packages,
|
||||
"vulnerabilities": [
|
||||
{
|
||||
"package": v.package,
|
||||
"version_spec": packages.get(v.package, v.version),
|
||||
"vulnerability_id": v.vuln_id,
|
||||
"severity": v.severity,
|
||||
"cvss_score": v.cvss_score,
|
||||
"summary": v.summary,
|
||||
"details_url": v.details_url,
|
||||
"fixed_versions": v.fixed_versions,
|
||||
}
|
||||
for v in result.vulnerabilities
|
||||
],
|
||||
"errors": [{"package": p, "error": e} for p, e in result.errors],
|
||||
}
|
||||
return json.dumps(report, indent=2)
|
||||
|
||||
|
||||
# --- Main Orchestration ---
|
||||
|
||||
|
||||
def run_scan(
|
||||
deps_path: str,
|
||||
min_severity: str = "low",
|
||||
query_osv_api: bool = True
|
||||
) -> ScanResult:
|
||||
"""
|
||||
Execute the full vulnerability scan pipeline.
|
||||
|
||||
Args:
|
||||
deps_path: Path to requirements-style file
|
||||
min_severity: Minimum severity to include in results
|
||||
query_osv_api: If False, skip API calls (for testing/dry-run)
|
||||
|
||||
Returns:
|
||||
ScanResult with all findings
|
||||
"""
|
||||
# 1. Parse dependencies
|
||||
packages = parse_requirements_file(deps_path)
|
||||
if not packages:
|
||||
return ScanResult(scanned_packages=0, vulnerabilities=[], errors=[])
|
||||
|
||||
# 2. Query OSV for each package
|
||||
vulnerabilities: List[Vulnerability] = []
|
||||
errors: List[Tuple[str, str]] = []
|
||||
|
||||
for pkg, version_spec in packages.items():
|
||||
if not query_osv_api:
|
||||
continue
|
||||
|
||||
raw_vulns = query_osv(pkg, version_spec or "")
|
||||
if raw_vulns:
|
||||
parsed = parse_osv_vuln(raw_vulns, pkg, version_spec or "")
|
||||
vulnerabilities.extend(parsed)
|
||||
|
||||
# 3. Filter by severity
|
||||
filtered = filter_by_severity(vulnerabilities, min_severity)
|
||||
|
||||
# 4. Build result
|
||||
return ScanResult(
|
||||
scanned_packages=len(packages),
|
||||
vulnerabilities=filtered,
|
||||
errors=errors
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Scan Python dependencies for known vulnerabilities using OSV database"
|
||||
)
|
||||
parser.add_argument(
|
||||
'--deps', '-d',
|
||||
default=DEFAULT_REQUIREMENTS_PATH,
|
||||
help='Path to requirements.txt (default: requirements.txt)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--output', '-o',
|
||||
choices=['text', 'json', 'markdown'],
|
||||
default='text',
|
||||
help='Output format (default: text)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--min-severity',
|
||||
default='low',
|
||||
choices=SEVERITY_LEVELS,
|
||||
help='Minimum severity to report (default: low — report all)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--json',
|
||||
action='store_true',
|
||||
help='Output JSON (shorthand for --output json)'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--quiet', '-q',
|
||||
action='store_true',
|
||||
help='Only print summary, skip detailed vulnerability list'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Update output if --json flag is used
|
||||
if args.json:
|
||||
args.output = 'json'
|
||||
|
||||
# Run the scan
|
||||
result = run_scan(args.deps, args.min_severity, query_osv_api=True)
|
||||
|
||||
# Output
|
||||
if args.output == 'json':
|
||||
print(generate_json_report(result, parse_requirements_file(args.deps)))
|
||||
elif args.output == 'markdown':
|
||||
# Simple markdown table
|
||||
print("# Vulnerability Scan Report\n")
|
||||
print(f"**Packages scanned:** {result.scanned_packages}")
|
||||
print(f"**Vulnerabilities:** {len(result.vulnerabilities)}\n")
|
||||
if result.vulnerabilities:
|
||||
print("| Severity | Package | Version | Vuln ID | Summary |")
|
||||
print("|----------|---------|---------|---------|---------|")
|
||||
for v in result.vulnerabilities:
|
||||
print(f"| {v.severity.upper()} | {v.package} | {v.version} | [{v.vuln_id}]({v.details_url}) | {v.summary[:50]} |")
|
||||
print("\n")
|
||||
else:
|
||||
# text (default)
|
||||
if not args.quiet:
|
||||
print(generate_text_report(result, parse_requirements_file(args.deps)))
|
||||
else:
|
||||
crit = sum(1 for v in result.vulnerabilities if v.severity == 'critical')
|
||||
high = sum(1 for v in result.vulnerabilities if v.severity == 'high')
|
||||
med = sum(1 for v in result.vulnerabilities if v.severity == 'medium')
|
||||
print(f"CRITICAL={crit} HIGH={high} MEDIUM={med} TOTAL={len(result.vulnerabilities)}")
|
||||
|
||||
# Exit code logic: 0 if no vulns at min_severity+, 1 if critical/high found, 2 for other vulns
|
||||
has_critical_high = any(v.severity in ('critical', 'high') for v in result.vulnerabilities)
|
||||
has_other = any(v.severity not in ('critical', 'high') for v in result.vulnerabilities)
|
||||
|
||||
if has_critical_high:
|
||||
return 1
|
||||
elif has_other:
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
0
test_push_marker.txt
Normal file
0
test_push_marker.txt
Normal file
@@ -1,148 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Smoke tests for API Doc Generator — Issue #98
|
||||
|
||||
Validates that the generator runs, produces docs/API.md, and that
|
||||
the generated markdown contains expected sections for the known scripts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Resolve repo root
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPTS_DIR = REPO_ROOT / "scripts"
|
||||
DOCS_DIR = REPO_ROOT / "docs"
|
||||
API_MD = DOCS_DIR / "API.md"
|
||||
GENERATOR = SCRIPTS_DIR / "api_doc_generator.py"
|
||||
|
||||
|
||||
# ─── Generator presence ─────────────────────────────────────────────────────────
|
||||
class TestGeneratorPresence:
|
||||
def test_generator_script_exists(self):
|
||||
assert GENERATOR.exists(), f"Missing: {GENERATOR}"
|
||||
|
||||
def test_generator_is_executable(self):
|
||||
with open(GENERATOR) as f:
|
||||
first = f.readline().strip()
|
||||
assert first.startswith("#!"), "Missing shebang"
|
||||
assert "python" in first.lower()
|
||||
|
||||
|
||||
# ─── API.md generation ──────────────────────────────────────────────────────────
|
||||
class TestAPIDocGeneration:
|
||||
def test_generator_runs_successfully(self):
|
||||
"""Run the generator and verify exit code 0."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(GENERATOR)],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT, timeout=30
|
||||
)
|
||||
assert result.returncode == 0, (
|
||||
f"Generator failed (code {{result.returncode}})\n"
|
||||
f"STDERR: {{result.stderr[:500]}}"
|
||||
)
|
||||
|
||||
def test_api_md_is_created(self):
|
||||
"""docs/API.md must exist after generation."""
|
||||
assert API_MD.exists(), f"Missing output: {API_MD}"
|
||||
|
||||
def test_api_md_is_not_empty(self):
|
||||
"""Generate markdown must have substantial content."""
|
||||
content = API_MD.read_text(encoding="utf-8")
|
||||
assert len(content) > 1000, "API.md is suspiciously small"
|
||||
|
||||
def test_api_md_has_expected_structure(self):
|
||||
"""Top-level headings and table markers must be present."""
|
||||
content = API_MD.read_text(encoding="utf-8")
|
||||
assert "# Compounding Intelligence — Scripts API Reference" in content
|
||||
assert "## `scripts/" in content
|
||||
assert "| Function | Signature | Description |" in content
|
||||
|
||||
def test_api_md_covers_expected_scripts(self):
|
||||
"""At minimum the core scripts should be documented."""
|
||||
content = API_MD.read_text(encoding="utf-8")
|
||||
# Core scripts that must appear
|
||||
core = ["scripts/harvester.py", "scripts/bootstrapper.py",
|
||||
"scripts/session_reader.py", "scripts/dedup.py"]
|
||||
for rel in core:
|
||||
assert f"## `{rel}`" in content, f"Missing section for {rel}"
|
||||
|
||||
def test_api_md_contains_function_names(self):
|
||||
"""Spot-check: known public functions from key modules must appear."""
|
||||
content = API_MD.read_text(encoding="utf-8")
|
||||
checks = [
|
||||
("harvester", "read_session"),
|
||||
("bootstrapper", "load_index"),
|
||||
("session_reader", "extract_conversation"),
|
||||
("dedup", "normalize_text"),
|
||||
]
|
||||
for module_stem, func_name in checks:
|
||||
assert f"| `{func_name}` |" in content, f"Missing function {func_name} from {module_stem}"
|
||||
|
||||
|
||||
# ─── Idempotence / --check ─────────────────────────────────────────────────────
|
||||
class TestIdempotence:
|
||||
def test_check_flag_passes_when_current(self):
|
||||
"""`--check` should exit 0 immediately after generation."""
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(GENERATOR), "--check"],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT, timeout=30
|
||||
)
|
||||
assert result.returncode == 0, (
|
||||
f"--check failed\nSTDOUT: {{result.stdout}}\nSTDERR: {{result.stderr[:200]}}"
|
||||
)
|
||||
|
||||
def test_check_fails_when_api_md_stale(self):
|
||||
"""If docs/API.md is manually altered, --check should detect staleness."""
|
||||
# Generate fresh baseline first
|
||||
subprocess.run([sys.executable, str(GENERATOR)], capture_output=True, cwd=REPO_ROOT, timeout=30)
|
||||
|
||||
# Corrupt API.md slightly (append a line at the end)
|
||||
original = API_MD.read_text(encoding="utf-8")
|
||||
corrupted = original + "\n<!-- corrupted -->\n"
|
||||
API_MD.write_text(corrupted, encoding="utf-8")
|
||||
|
||||
# --check should now fail
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(GENERATOR), "--check"],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT, timeout=30
|
||||
)
|
||||
assert result.returncode != 0, "--check should detect stale API.md"
|
||||
assert "out-of-date" in result.stderr.lower() or "out-of-date" in result.stdout.lower()
|
||||
|
||||
# Restore clean state
|
||||
subprocess.run([sys.executable, str(GENERATOR)], capture_output=True, cwd=REPO_ROOT, timeout=30)
|
||||
assert API_MD.read_text(encoding="utf-8") == original
|
||||
|
||||
# ─── JSON output ────────────────────────────────────────────────────────────────
|
||||
class TestJSONOutput:
|
||||
def test_json_flag_emits_valid_json(self):
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(GENERATOR), "--json"],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT, timeout=30
|
||||
)
|
||||
assert result.returncode == 0
|
||||
import json
|
||||
payload = json.loads(result.stdout)
|
||||
assert "modules" in payload
|
||||
assert len(payload["modules"]) >= 30
|
||||
|
||||
def test_json_has_expected_fields(self):
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(GENERATOR), "--json"],
|
||||
capture_output=True, text=True, cwd=REPO_ROOT, timeout=30
|
||||
)
|
||||
import json
|
||||
payload = json.loads(result.stdout)
|
||||
mod = payload["modules"][0]
|
||||
for key in ("path", "docstring", "functions"):
|
||||
assert key in mod, f"Missing key {{key}} in module payload"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
237
tests/test_vulnerability_scanner.py
Normal file
237
tests/test_vulnerability_scanner.py
Normal file
@@ -0,0 +1,237 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for scripts/vulnerability_scanner.py — 10 tests."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__) or ".", ".."))
|
||||
import importlib.util
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"vulnerability_scanner",
|
||||
os.path.join(os.path.dirname(__file__) or ".", "..", "scripts", "vulnerability_scanner.py"))
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
|
||||
parse_requirements_file = mod.parse_requirements_file
|
||||
query_osv = mod.query_osv
|
||||
parse_osv_vuln = mod.parse_osv_vuln
|
||||
filter_by_severity = mod.filter_by_severity
|
||||
Vulnerability = mod.Vulnerability
|
||||
|
||||
|
||||
# --- Test Data ---
|
||||
|
||||
SAMPLE_OSV_RESPONSE = [
|
||||
{
|
||||
"id": "GHSA-xxxx-xxxx-xxxx",
|
||||
"summary": " Arbitrary code execution in django",
|
||||
"severity": [{"type": "CVSS_V3", "score": {"baseScore": 9.8, "baseSeverity": "CRITICAL"}}],
|
||||
"affected": [{
|
||||
"ranges": [{
|
||||
"events": [
|
||||
{"introduced": "0"},
|
||||
{"fixed": "3.2.14"}
|
||||
]
|
||||
}]
|
||||
}]
|
||||
},
|
||||
{
|
||||
"id": "PYSEC-2024-1234",
|
||||
"summary": " Denial of service in cryptography",
|
||||
"severity": [{"type": "CVSS_V3", "score": {"baseScore": 5.3, "baseSeverity": "MEDIUM"}}],
|
||||
"affected": [{
|
||||
"ranges": [{
|
||||
"events": [
|
||||
{"introduced": "0"},
|
||||
{"fixed": "42.0.0"}
|
||||
]
|
||||
}]
|
||||
}]
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
# --- Tests ---
|
||||
|
||||
|
||||
def test_parse_requirements_simple():
|
||||
"""Should parse a simple requirements file."""
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||||
f.write("django==4.2.0\n")
|
||||
f.write("requests>=2.28.0\n")
|
||||
f.write("click~=8.0\n")
|
||||
f.flush()
|
||||
pkgs = parse_requirements_file(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
assert "django" in pkgs
|
||||
assert pkgs["django"] == "==4.2.0"
|
||||
assert "requests" in pkgs
|
||||
assert pkgs["requests"] == ">=2.28.0"
|
||||
assert "click" in pkgs
|
||||
print("PASS: test_parse_requirements_simple")
|
||||
|
||||
|
||||
def test_parse_requirements_extras_and_comments():
|
||||
"""Should skip comments, blank lines, and handle package extras."""
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||||
f.write("# This is a comment\n")
|
||||
f.write("django[argon2]==4.2.0\n")
|
||||
f.write("\n")
|
||||
f.write(" requests >=2.28.0 # inline comment\n")
|
||||
f.flush()
|
||||
pkgs = parse_requirements_file(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
assert "django" in pkgs
|
||||
assert pkgs["django"] == "==4.2.0"
|
||||
assert "requests" in pkgs
|
||||
# Version should capture the comparison
|
||||
assert ">=" in pkgs["requests"]
|
||||
print("PASS: test_parse_requirements_extras_and_comments")
|
||||
|
||||
|
||||
def test_parse_requirements_include_recursive():
|
||||
"""Should follow -r includes up to depth 3."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Main requirements.txt
|
||||
main = os.path.join(tmpdir, "requirements.txt")
|
||||
with open(main, 'w') as f:
|
||||
f.write("django==4.2.0\n")
|
||||
f.write("-r base.txt\n")
|
||||
|
||||
# base.txt
|
||||
base = os.path.join(tmpdir, "base.txt")
|
||||
with open(base, 'w') as f:
|
||||
f.write("requests>=2.28.0\n")
|
||||
f.write("-r deep.txt\n")
|
||||
|
||||
# deep.txt
|
||||
deep = os.path.join(tmpdir, "deep.txt")
|
||||
with open(deep, 'w') as f:
|
||||
f.write("click~=8.0\n")
|
||||
|
||||
pkgs = parse_requirements_file(main)
|
||||
|
||||
assert "django" in pkgs
|
||||
assert "requests" in pkgs
|
||||
assert "click" in pkgs
|
||||
print("PASS: test_parse_requirements_include_recursive")
|
||||
|
||||
|
||||
def test_parse_requirements_skip_editable():
|
||||
"""Should skip -e editable installs and other flags."""
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
|
||||
f.write("-e git+https://github.com/user/repo.git@branch#egg=package\n")
|
||||
f.write("--index-url https://pypi.org/simple\n")
|
||||
f.write("django==4.2.0\n")
|
||||
f.flush()
|
||||
pkgs = parse_requirements_file(f.name)
|
||||
os.unlink(f.name)
|
||||
|
||||
assert "django" in pkgs
|
||||
assert "package" not in pkgs # should not pick up editable name
|
||||
print("PASS: test_parse_requirements_skip_editable")
|
||||
|
||||
|
||||
def test_parse_requirements_nonexistent():
|
||||
"""Should exit with error on missing file."""
|
||||
with patch('sys.exit') as mock_exit:
|
||||
pkgs = parse_requirements_file("/nonexistent/requirements.txt")
|
||||
mock_exit.assert_called_once_with(1)
|
||||
print("PASS: test_parse_requirements_nonexistent")
|
||||
|
||||
|
||||
def test_filter_by_severity():
|
||||
"""Should filter vulnerabilities by severity threshold."""
|
||||
vulns = [
|
||||
Vulnerability("pkg1", "==1.0", "V1", "critical", 9.8, "summary", "url", []),
|
||||
Vulnerability("pkg2", "==2.0", "V2", "high", 7.5, "summary", "url", []),
|
||||
Vulnerability("pkg3", "==3.0", "V3", "medium", 5.0, "summary", "url", []),
|
||||
Vulnerability("pkg4", "==4.0", "V4", "low", 2.0, "summary", "url", []),
|
||||
]
|
||||
|
||||
# min_severity: low includes all
|
||||
filtered = filter_by_severity(vulns, "low")
|
||||
assert len(filtered) == 4
|
||||
|
||||
# min_severity: medium excludes low
|
||||
filtered = filter_by_severity(vulns, "medium")
|
||||
assert len(filtered) == 3
|
||||
assert all(v.severity in ("critical", "high", "medium") for v in filtered)
|
||||
|
||||
# min_severity: high excludes medium + low
|
||||
filtered = filter_by_severity(vulns, "high")
|
||||
assert len(filtered) == 2
|
||||
|
||||
# min_severity: critical only
|
||||
filtered = filter_by_severity(vulns, "critical")
|
||||
assert len(filtered) == 1
|
||||
|
||||
print("PASS: test_filter_by_severity")
|
||||
|
||||
|
||||
def test_parse_osv_vuln():
|
||||
"""Should parse OSV API response correctly."""
|
||||
parsed = parse_osv_vuln(SAMPLE_OSV_RESPONSE, "django", "==4.2.0")
|
||||
|
||||
assert len(parsed) == 2
|
||||
assert parsed[0].package == "django"
|
||||
assert parsed[0].vuln_id == "GHSA-xxxx-xxxx-xxxx"
|
||||
assert parsed[0].severity == "critical"
|
||||
assert parsed[0].cvss_score == 9.8
|
||||
assert parsed[1].severity == "medium"
|
||||
assert parsed[1].cvss_score == 5.3
|
||||
print("PASS: test_parse_osv_vuln")
|
||||
|
||||
|
||||
def test_parse_osv_vuln_empty():
|
||||
"""Should handle empty OSV response."""
|
||||
parsed = parse_osv_vuln([], "django", "==4.2.0")
|
||||
assert parsed == []
|
||||
print("PASS: test_parse_osv_vuln_empty")
|
||||
|
||||
|
||||
def test_query_osv_network_success():
|
||||
"""Should successfully query OSV API for a real known vulnerable package."""
|
||||
# Query for an old django version that likely has known CVEs
|
||||
# This test actually hits the network — tagged as integration
|
||||
vulns = query_osv("django", "==3.2.0")
|
||||
# We don't assert specific results since vulns change over time
|
||||
# But we assert the function returns a list and doesn't error
|
||||
assert isinstance(vulns, list)
|
||||
print("PASS: test_query_osv_network_success")
|
||||
|
||||
|
||||
def test_query_osv_404_no_vulns():
|
||||
"""OSV returns empty list for packages with no vulns (404-like)."""
|
||||
# Mock a 404 response from OSV API
|
||||
with patch('urllib.request.urlopen') as mock_urlopen:
|
||||
mock_response = MagicMock()
|
||||
mock_response.read.return_value = b'{"vulns": []}'
|
||||
mock_response.__enter__ = lambda self: self
|
||||
mock_response.__exit__ = lambda self, *args: None
|
||||
mock_urlopen.return_value = mock_response
|
||||
|
||||
result = query_osv("nonexistent-package-xyz123", "==1.0.0")
|
||||
assert result == []
|
||||
print("PASS: test_query_osv_404_no_vulns")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Run all tests
|
||||
test_parse_requirements_simple()
|
||||
test_parse_requirements_extras_and_comments()
|
||||
test_parse_requirements_include_recursive()
|
||||
test_parse_requirements_skip_editable()
|
||||
test_parse_requirements_nonexistent()
|
||||
test_filter_by_severity()
|
||||
test_parse_osv_vuln()
|
||||
test_parse_osv_vuln_empty()
|
||||
test_query_osv_network_success()
|
||||
test_query_osv_404_no_vulns()
|
||||
print("\nAll tests passed.")
|
||||
Reference in New Issue
Block a user