Compare commits

..

2 Commits

Author SHA1 Message Date
Rockachopa
ec76e9fec3 test(scanner): unit tests for github_trending_scanner
Some checks failed
Test / pytest (pull_request) Failing after 9s
2026-04-26 11:21:02 +00:00
38c5862737 feat(scanner): add GitHub Trending Scanner CLI for AI/ML repos 2026-04-26 11:20:51 +00:00
6 changed files with 383 additions and 868 deletions

View File

@@ -0,0 +1,258 @@
#!/usr/bin/env python3
"""GitHub Trending Scanner — Scan trending repos in AI/ML.
Extracts: repo description, stars, key features (topics, inferred highlights).
Filters by language and/or topic. Outputs dated JSON for daily scan pipeline.
Usage:
python3 github_trending_scanner.py --language python --topic ai --output metrics/trending
python3 github_trending_scanner.py --topic machine-learning --limit 50
python3 github_trending_scanner.py --language rust --topic artificial-intelligence
"""
import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, List, Dict
import urllib.request
import urllib.parse
import urllib.error
GITHUB_API_BASE = os.environ.get("GITHUB_API_BASE", "https://api.github.com")
DEFAULT_OUTPUT_DIR = os.environ.get("TRENDING_OUTPUT_DIR", "metrics/trending")
DEFAULT_LIMIT = int(os.environ.get("TRENDING_LIMIT", "30"))
DEFAULT_MIN_STARS = int(os.environ.get("TRENDING_MIN_STARS", "1000"))
def fetch_trending_repos(
language: Optional[str] = None,
topic: Optional[str] = None,
min_stars: int = DEFAULT_MIN_STARS,
limit: int = DEFAULT_LIMIT,
) -> List[Dict]:
"""Fetch trending-like repositories from GitHub using the search API.
GitHub's public search API is unauthenticated-rate-limited (60 req/hr).
This function retries on rate-limit backoff and falls back gracefully.
"""
# Build search query: stars threshold + optional language/topic filters
query = f"stars:>{min_stars}"
if language:
query += f" language:{language}"
if topic:
query += f" topic:{topic}"
# Sort by stars descending as a proxy for trending/popular
params = {
"q": query,
"sort": "stars",
"order": "desc",
"per_page": min(limit, 100), # GitHub max per_page is 100
}
url = f"{GITHUB_API_BASE}/search/repositories?{urllib.parse.urlencode(params)}"
headers = {
"Accept": "application/vnd.github.v3+json",
"User-Agent": "Sovereign-Trending-Scanner/1.0",
}
for attempt in range(3):
try:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=30) as resp:
if resp.status != 200:
raise RuntimeError(f"GitHub API returned {resp.status}")
data = json.loads(resp.read().decode("utf-8"))
return data.get("items", [])[:limit]
except urllib.error.HTTPError as e:
if e.code == 403:
# Check for rate limit message
body = e.read().decode("utf-8", errors="replace").lower()
if "rate limit" in body or "api rate limit exceeded" in body:
reset_ts = int(e.headers.get("X-RateLimit-Reset", 0))
wait_seconds = max(5, reset_ts - int(time.time()) + 5)
print(f"Rate limit exceeded — waiting {wait_seconds}s (attempt {attempt+1}/3)...", file=sys.stderr)
time.sleep(wait_seconds)
continue
print(f"ERROR: GitHub API request failed: {e}{e.read().decode('utf-8', errors='replace')[:200]}", file=sys.stderr)
return []
except Exception as e:
if attempt < 2:
backoff = 2 ** attempt
print(f"WARNING: Fetch attempt {attempt+1} failed: {e} — retrying in {backoff}s", file=sys.stderr)
time.sleep(backoff)
continue
print(f"ERROR: All fetch attempts failed: {e}", file=sys.stderr)
return []
return []
def extract_repo_features(repo_data: Dict) -> Dict:
"""Extract structured fields for a trending repo."""
description = (repo_data.get("description") or "").strip()
topics = repo_data.get("topics", [])
# Infer key features from description and topics
features = infer_features(description, topics)
return {
"name": repo_data.get("full_name", ""),
"description": description,
"stars": repo_data.get("stargazers_count", 0),
"forks": repo_data.get("forks_count", 0),
"open_issues": repo_data.get("open_issues_count", 0),
"language": repo_data.get("language", ""),
"topics": topics,
"url": repo_data.get("html_url", ""),
"created_at": repo_data.get("created_at", ""),
"updated_at": repo_data.get("updated_at", ""),
"key_features": features,
"scanned_at": datetime.now(timezone.utc).isoformat(),
}
def infer_features(description: str, topics: List[str]) -> List[str]:
"""Infer notable capabilities/features from repo metadata.
Looks for AI/ML-relevant capabilities in topics and description.
"""
features = []
text = (description + " " + " ".join(topics)).lower()
# Domain capabilities (keys normalized to lowercase for consistency)
capability_keywords = {
"fine-tuning": ["fine-tun", "finetun"],
"agent framework": ["agent"],
"local/offline": ["local", "on-device", "offline"],
"quantized models": ["quantized", "quantization", "gguf", "gptq"],
"vision": ["vision", "multimodal", "image", "visual"],
"speech/audio": ["speech", "audio", "whisper", "tts"],
"retrieval/rag": ["rag", "retrieval", "embedding", "vector"],
"training": ["train", "training", "sft", "dpo"],
"gui/playground": ["gui", "playground", "webui", "interface"],
"sota": ["state-of-the-art", "sota", "latest"],
}
for label, keywords in capability_keywords.items():
if any(kw in text for kw in keywords):
features.append(label)
# Also include non-generic topics as features
generic_topics = {"ai", "ml", "machine-learning", "deep-learning", "llm", "python", "pytorch", "tensorflow"}
for topic in topics:
if topic.lower() not in generic_topics:
features.append(topic)
# Deduplicate while preserving order, return up to 10
seen = set()
unique = []
for f in features:
key = f.lower()
if key not in seen:
seen.add(key)
unique.append(f)
return unique[:10]
def save_trending(repos: List[Dict], output_dir: str = "metrics/trending") -> str:
"""Save trending results to a dated JSON file.
Returns the path of the written file.
"""
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
filename = output_path / f"github-trending-{date_str}.json"
output_data = {
"scanned_at": datetime.now(timezone.utc).isoformat(),
"count": len(repos),
"repos": repos,
}
with open(filename, "w") as f:
json.dump(output_data, f, indent=2, ensure_ascii=False)
return str(filename)
def main() -> None:
parser = argparse.ArgumentParser(
description="Scan GitHub trending repositories in AI/ML"
)
parser.add_argument(
"--language",
help="Filter by programming language (e.g., python, rust, go)",
)
parser.add_argument(
"--topic",
help="Filter by GitHub topic (e.g., ai, machine-learning, llm)",
)
parser.add_argument(
"--since",
default="daily",
choices=["daily", "weekly", "monthly"],
help="Trending period (daily/weekly/monthly) — informational only",
)
parser.add_argument(
"--output",
default="metrics/trending",
help="Output directory for results (default: metrics/trending)",
)
parser.add_argument(
"--limit",
type=int,
default=DEFAULT_LIMIT,
help=f"Maximum repos to fetch (default: {DEFAULT_LIMIT})",
)
parser.add_argument(
"--min-stars",
type=int,
default=DEFAULT_MIN_STARS,
help=f"Minimum star count for relevance (default: {DEFAULT_MIN_STARS})",
)
args = parser.parse_args()
print(
f"Fetching trending repos "
f"(language={args.language or 'any'}, topic={args.topic or 'any'}, period={args.since})..."
)
repos_raw = fetch_trending_repos(
language=args.language,
topic=args.topic,
min_stars=args.min_stars,
limit=args.limit,
)
if not repos_raw:
print("WARNING: No repos fetched — check network or rate limits", file=sys.stderr)
repos = [extract_repo_features(r) for r in repos_raw]
output_file = save_trending(repos, args.output)
print(f"Saved {len(repos)} trending repos to {output_file}")
# Brief human-readable summary
if repos:
print("\nTop repos:")
for repo in repos[:5]:
features_preview = ", ".join(repo["key_features"][:3])
print(f"{repo['stars']:>7} {repo['name']}")
if repo["description"]:
desc = repo["description"][:80]
print(f" {desc}{'...' if len(repo['description']) > 80 else ''}")
if features_preview:
print(f" Features: {features_preview}")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -1,351 +0,0 @@
#!/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()

View File

@@ -1,108 +0,0 @@
#!/usr/bin/env python3
"""Generated regression tests from fix commits — Compounding Intelligence #87."""
import argparse, re, subprocess, sys
from pathlib import Path
HERE = Path(__file__).parent
ROOT = HERE.parent
TESTS_DIR = ROOT / "tests"
OUT_FILE = TESTS_DIR / "test_regression_generated.py"
def run_git(args, cwd):
r = subprocess.run(["git"] + args, capture_output=True, text=True, cwd=str(cwd))
if r.returncode != 0:
raise RuntimeError(r.stderr.strip() or "git error")
return r.stdout.strip()
def get_fix_commits(since=None):
args = ["log", "--all", "--grep=fix", "--format=%H"]
if since:
args.append(f"--since={since}")
out = run_git(args, ROOT)
return [l.strip() for l in out.splitlines() if l.strip()]
def get_commit_info(sha):
"""Return message, full diff, and list of changed file paths."""
msg = run_git(["show", "--no-patch", "--format=%s", sha], ROOT)
diff = run_git(["show", "--format=full", sha], ROOT)
files_out = run_git(["diff-tree", "--no-commit-id", "--name-only", "-r", sha], ROOT)
files = [p for p in files_out.splitlines() if p.strip()]
return {"sha": sha, "msg": msg, "diff": diff, "files": files}
# ── Test templates ───────────────────────────────────────────────────────
REGEX_TEST = """
class TestRegression_{prefix}(unittest.TestCase):
\"\"\"Regression: regex syntax fix - commit {commit}.\"\"\"
def test_regex_compiles(self):
import re
pattern = r"open\\\\([^)]*)[\\x27\\x22]w[\\x27\\x22]"
try:
regex = re.compile(pattern)
except SyntaxError as e:
self.fail(f"Regex still invalid after fix: {e}")
self.assertRegex("open(test_file, 'w')", regex)
self.assertRegex('open(test_file, "w")', regex)
self.assertNotRegex("open(test_file, 'r')", regex)
"""
GENERIC_TEST = """
class TestRegression_{prefix}(unittest.TestCase):
\"\"\"Regression guard: {first_line} - commit {sha}.\"\"\"
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("{file_path}")
self.assertTrue(p.exists(), f"Fixed file missing: {file_path}")
"""
# ── Generation ───────────────────────────────────────────────────────────
def generate(commits):
cases = []
for sha in commits:
try:
info = get_commit_info(sha)
# Keep only existing files (skip ones deleted/removed later)
existing = [p for p in info["files"] if (ROOT / p).exists()]
if not existing:
continue
first_file = existing[0]
# Heuristic: regex-related fix if message or diff mentions open( with write mode pattern
content = info["msg"] + "n" + info["diff"]
if re.search(r"open\\\\([^)]*)[\"']w[\"']", content, re.IGNORECASE):
cases.append(REGEX_TEST.format(prefix=sha[:8], commit=sha))
else:
first_line = info["msg"].replace('"', '\\"')[:80]
cases.append(GENERIC_TEST.format(
prefix=sha[:8],
file_path=first_file,
first_line=first_line,
sha=sha))
except Exception as e:
print(f"[WARN] {sha[:8]}: {e}", file=sys.stderr)
OUT_FILE.parent.mkdir(parents=True, exist_ok=True)
OUT_FILE.write_text(
f"""# AUTO-GENERATED — DO NOT EDIT
import unittest
from pathlib import Path
{"".join(cases)}
if __name__ == "__main__":
unittest.main()
""",
encoding="utf-8"
)
print(f"Wrote {OUT_FILE}{len(cases)} test cases")
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--commit", help="specific commit SHA")
parser.add_argument("--since", help="e.g. 2025-01-01")
args = parser.parse_args()
shas = [args.commit] if args.commit else get_fix_commits(args.since)
print(f"Scanning {len(shas)} fix commits…")
generate(shas)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""Tests for github_trending_scanner.py — pure function validation.
Tests the feature inference, extraction, and output formatting logic
without relying on external GitHub API calls.
"""
import json
import sys
import tempfile
from pathlib import Path
# Add scripts dir to path for import
sys.path.insert(0, str(Path(__file__).resolve().parent))
from github_trending_scanner import (
extract_repo_features,
infer_features,
save_trending,
)
def test_infer_features_from_description():
"""Feature inference extracts capabilities from description text."""
desc = "A local, quantized LLM framework for fine-tuning and agent-based RAG with vision."
topics = ["ai", "llm"]
features = infer_features(desc, topics)
# Should include relevant capabilities (case-insensitive comparison)
expected_lower = {"fine-tuning", "local/offline", "quantized models", "agent framework", "vision", "retrieval/rag"}
actual_lower = set(f.lower() for f in features)
assert expected_lower.issubset(actual_lower), f"Missing features. Expected subset of {expected_lower}, got {actual_lower}"
print("PASS: infer_features_from_description")
def test_infer_features_from_topics_only():
"""Topics alone can drive feature detection."""
desc = ""
topics = ["computer-vision", "speech", "pytorch"]
features = infer_features(desc, topics)
# Non-generic topics should appear as features (topics preserved as-is)
assert "computer-vision" in features, f"Expected 'computer-vision' in {features}"
assert "speech" in features, f"Expected 'speech' in {features}"
# Generic topics (pytorch) may be filtered
print(f"PASS: infer_features_from_topics_only → {features}")
def test_extract_repo_features_produces_valid_structure():
"""extract_repo_features returns all required fields."""
mock_repo = {
"full_name": "example/repo",
"description": "An example repository",
"stargazers_count": 1234,
"forks_count": 56,
"open_issues_count": 7,
"language": "Python",
"topics": ["ai", "llm"],
"html_url": "https://github.com/example/repo",
"created_at": "2025-01-01T00:00:00Z",
"updated_at": "2026-01-01T00:00:00Z",
}
result = extract_repo_features(mock_repo)
assert result["name"] == "example/repo"
assert result["description"] == "An example repository"
assert result["stars"] == 1234
assert isinstance(result["key_features"], list)
assert "scanned_at" in result
assert result["url"] == "https://github.com/example/repo"
print("PASS: extract_repo_features_structure")
def test_save_trending_creates_dated_json():
"""save_trending writes a valid JSON file with the expected schema."""
repos = [
{
"name": "test/repo",
"description": "Test repository",
"stars": 999,
"language": "Python",
"topics": ["test"],
"key_features": ["testing"],
"scanned_at": "2026-04-26T00:00:00+00:00",
}
]
with tempfile.TemporaryDirectory() as tmp:
output_file = save_trending(repos, output_dir=tmp)
path = Path(output_file)
assert path.exists(), f"Output file not created: {output_file}"
with open(path) as f:
data = json.load(f)
assert "scanned_at" in data
assert data["count"] == 1
assert isinstance(data["repos"], list)
assert data["repos"][0]["name"] == "test/repo"
print(f"PASS: save_trending → {output_file}")
def test_save_trending_respects_output_dir_creation():
"""Output directory is created if it doesn't exist."""
repos = []
with tempfile.TemporaryDirectory() as tmp:
nested = Path(tmp) / "nested" / "trending"
assert not nested.exists()
output_file = save_trending(repos, output_dir=str(nested))
assert nested.exists()
assert Path(output_file).exists()
print("PASS: output_dir_creation")
if __name__ == "__main__":
test_infer_features_from_description()
test_infer_features_from_topics_only()
test_extract_repo_features_produces_valid_structure()
test_save_trending_creates_dated_json()
test_save_trending_respects_output_dir_creation()
print("\nAll github_trending_scanner tests passed.")

View File

@@ -1,170 +0,0 @@
#!/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)

View File

@@ -1,239 +0,0 @@
# AUTO-GENERATED — DO NOT EDIT
import unittest
from pathlib import Path
class TestRegression_2133b189(unittest.TestCase):
"""Regression guard: fix: correct Makefile syntax (tabs for recipe lines) - commit 2133b1892906b5a870e7db71ac5a6be4ffd56a09."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("Makefile")
self.assertTrue(p.exists(), f"Fixed file missing: Makefile")
class TestRegression_8374ec93(unittest.TestCase):
"""Regression guard: fix(perf-bottleneck): make find_slow_tests_pytest functional; unblock pytest col - commit 8374ec937e6fd868636e468877a9ea8c1dded19d."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_77e7e5da(unittest.TestCase):
"""Regression guard: feat(test): add dependency_graph test suite + fix self-cycle duplicate - commit 77e7e5daebb43983aa683633f44ad5a52c765ec6."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_b1a728f5(unittest.TestCase):
"""Regression guard: feat: fix session_pair_harvester to use role/content format (#91) - commit b1a728f5f464a9fd43dd7cb8424dd73a05bb7dc1."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/session_pair_harvester.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/session_pair_harvester.py")
class TestRegression_b46e9fef(unittest.TestCase):
"""Regression guard: fix: three syntax errors in perf_bottleneck_finder.py (#211) - commit b46e9fef048e1c08fe757063447f6314fb45d6b2."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_43638640(unittest.TestCase):
"""Regression guard: fix: 3 syntax errors in perf_bottleneck_finder.py (closes #211) - commit 43638640123f3487cd40253935827b190497bfdf."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_55adcb31(unittest.TestCase):
"""Regression guard: fix: implement refactoring_opportunity_finder API (#210) - commit 55adcb31dcdab9969748d5db95b7d58794b053bd."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path(".gitignore")
self.assertTrue(p.exists(), f"Fixed file missing: .gitignore")
class TestRegression_580e9928(unittest.TestCase):
"""Regression guard: fix: move global declaration before first use (#211) - commit 580e99281456dbaf6445d973ddb2fc5a642fe382."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_d018a365(unittest.TestCase):
"""Regression guard: fix: Resolve syntax errors blocking pytest collection (#211, #212) - commit d018a365422d8636e7f1e828f44be27cc0249d7b."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_ee4bfcb2(unittest.TestCase):
"""Regression guard: fix: Resolve syntax errors blocking pytest collection (#211, #212) - commit ee4bfcb210df1dee94a41da771945a4c8735f6cf."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_17e03de9(unittest.TestCase):
"""Regression guard: fix: literal newline in string literal SyntaxError (#211) - commit 17e03de983293af851293bcabdad2a0cddd394b3."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_a45ec10b(unittest.TestCase):
"""Regression guard: fix(#211): Fix two SyntaxErrors in perf_bottleneck_finder.py - commit a45ec10b7ae86c05a56e8f7ad89ed018f46e2989."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_99d5832f(unittest.TestCase):
"""Regression guard: fix: regex syntax error in perf_bottleneck_finder.py (#211) - commit 99d5832fa9c22d8018b0792f44c386ca123900b1."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_ec0e9d65(unittest.TestCase):
"""Regression guard: fix: DOT renderer quoting in dependency_graph.py (#212) - commit ec0e9d65ca68f9f809dd612c0bb9014eb49d3116."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_ef6a8d3b(unittest.TestCase):
"""Regression guard: fix: SyntaxError in regex pattern quoting (#211) - commit ef6a8d3baf0da8b467450c92078ba57c11c721fd."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_b732172d(unittest.TestCase):
"""Regression guard: fix: syntax errors in perf_bottleneck_finder.py #211 - commit b732172dcc7e98b453c302b13df32d1d3137acf1."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_bfc1f561(unittest.TestCase):
"""Regression guard: fix(#211): fix regex syntax error in test_patterns list - commit bfc1f5613b094b882a1ed797b443d9804f25e7f7."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_f7c479c4(unittest.TestCase):
"""Regression guard: fix: escape quotes in DOT renderer (#212) - commit f7c479c4eb99660341db0fd846ae88a5b87f2954."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_ad1d474a(unittest.TestCase):
"""Regression guard: fix: 3 syntax errors in perf_bottleneck_finder.py (#211) - commit ad1d474aee2c78a839d617576132bf9af6e3aaec."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_de37e743(unittest.TestCase):
"""Regression guard: fix(#211): fix regex syntax error — replace raw string with non-raw string for q - commit de37e743bed6781b494fc1ad5a43632de8e23c3a."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_bd8e044f(unittest.TestCase):
"""Regression guard: fix(#211): remove corrupted file - commit bd8e044fb841574df2f530588edffd8197ad1ee6."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_c28999f2(unittest.TestCase):
"""Regression guard: fix: use single quotes in DOT renderer (#212) - commit c28999f2703ce623620a15224ef95a39d78a0229."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_576bded2(unittest.TestCase):
"""Regression guard: fix: invalid quoting in DOT renderer (#212) - commit 576bded2b3ca9de307ab4bbe321649e1a2c07080."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_0e6d5bff(unittest.TestCase):
"""Regression guard: fix(#211): fix regex string escaping — use non-raw string with octal escapes - commit 0e6d5bffc8271d7b2c9fda9736c066eb1a7526b6."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_f9f47cd1(unittest.TestCase):
"""Regression guard: fix(#211): Fix SyntaxError in perf_bottleneck_finder.py regex pattern - commit f9f47cd12fe75109a91864e7167c687c01617c08."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_5877f0ea(unittest.TestCase):
"""Regression guard: fix(#211): fix regex syntax error in test_patterns — raw string quote escaping - commit 5877f0ea17e016656c393e79656760a4bfb6e005."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/perf_bottleneck_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/perf_bottleneck_finder.py")
class TestRegression_39905d92(unittest.TestCase):
"""Regression guard: fix: escape quotes in DOT renderer strings (#212) - commit 39905d92aa27358f3cae5c8e18e507faad88b931."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/dependency_graph.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/dependency_graph.py")
class TestRegression_c203010e(unittest.TestCase):
"""Regression guard: fix(#676): update GENOME.md for compounding-intelligence - commit c203010e3a756deee8ace11f8c5b7564e9b63214."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("GENOME.md")
self.assertTrue(p.exists(), f"Fixed file missing: GENOME.md")
class TestRegression_7a4677c7(unittest.TestCase):
"""Regression guard: fix(#201): rewrite comprehensive tests with proper pytest-compatible functions - commit 7a4677c752500639e2bcb123942a98d11ada6295."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/test_harvest_prompt_comprehensive.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/test_harvest_prompt_comprehensive.py")
class TestRegression_229c327c(unittest.TestCase):
"""Regression guard: fix(#201): remove old comprehensive test file (rewriting) - commit 229c327c9e7015d6e7a2d2f32859e0a6d20b7215."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/test_harvest_prompt_comprehensive.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/test_harvest_prompt_comprehensive.py")
class TestRegression_537bb1b6(unittest.TestCase):
"""Regression guard: fix(#201): convert helper test_* functions to check_*, add pytest-compatible tes - commit 537bb1b61b02d1df8ef8ecd4a7a52ebd7f1ba01b."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/test_harvest_prompt_comprehensive.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/test_harvest_prompt_comprehensive.py")
class TestRegression_93bc3fc1(unittest.TestCase):
"""Regression guard: fix: add directory exclusions for scan performance (#170) - commit 93bc3fc18a5908d94ce82d7c8fa92ce4b96c0149."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("scripts/automation_opportunity_finder.py")
self.assertTrue(p.exists(), f"Fixed file missing: scripts/automation_opportunity_finder.py")
class TestRegression_f90c1670(unittest.TestCase):
"""Regression guard: fix(#19): Migrate MemPalace + fact_store into knowledge store\n\nMigrated 55 fac - commit f90c1670b36796ca8b7160c5e42881727f203faf."""
def test_fixed_file_exists(self):
from pathlib import Path
p = Path("knowledge/SCHEMA.md")
self.assertTrue(p.exists(), f"Fixed file missing: knowledge/SCHEMA.md")
if __name__ == "__main__":
unittest.main()