Files
RepoPulse-Public/backend/app/signals.py
T

495 lines
19 KiB
Python

from __future__ import annotations
import json
import re
import tomllib
import xml.etree.ElementTree as ET
from collections import defaultdict
from pathlib import Path
from .models import DependencySignals, Finding, Severity, TestSignals
LOCKFILES = {
"package-lock.json",
"pnpm-lock.yaml",
"yarn.lock",
"bun.lock",
"bun.lockb",
"uv.lock",
"poetry.lock",
"Pipfile.lock",
"Cargo.lock",
"go.sum",
}
MANIFEST_MAX_BYTES = 2_000_000
COVERAGE_NAMES = {
"coverage.xml",
"lcov.info",
"cobertura.xml",
".coverage",
"jacoco.xml",
}
COVERAGE_NAMES_CASEFOLD = {item.casefold() for item in COVERAGE_NAMES}
def _python_dependency_name(declaration: str) -> str:
return re.split(r"[<>=!~ @;\[]", declaration, maxsplit=1)[0].strip()
def _python_constraint_kind(declaration: str) -> str:
"""Classify the useful subset of Python version constraints without penalising bounded ranges."""
value = declaration.split(";", maxsplit=1)[0].strip().lower()
if " @ " in value or "==" in value:
return "pinned"
if "~=" in value or "^" in value:
return "bounded"
has_lower = ">" in value
has_upper = "<" in value
if has_lower and has_upper:
return "bounded"
return "unbounded"
def _python_dependency_finding(declaration: str, relative: str) -> Finding | None:
if _python_constraint_kind(declaration) != "unbounded":
return None
return Finding(
code="dependency.python-unbounded",
category="Dependencies",
severity=Severity.WARNING,
message=f"Python dependency has no bounded version range: {declaration.split(';')[0]}",
path=relative,
deduction=3,
)
def detect_test_signals(
relative_paths: list[str],
source_files: int,
test_files: int,
text_by_path: dict[str, str],
) -> TestSignals:
frameworks: set[str] = set()
coverage: list[str] = []
ci: list[str] = []
ci_validation: list[str] = []
ci_build: list[str] = []
ci_deploy_only: list[str] = []
coverage_evidence: list[str] = []
for path in relative_paths:
normalized = path.replace("\\", "/")
name = Path(normalized).name
name_lower = name.casefold()
lower = normalized.lower()
if name_lower in COVERAGE_NAMES_CASEFOLD or "/coverage/" in f"/{lower}":
coverage.append(normalized)
if lower.startswith((".github/workflows/", ".gitea/workflows/")) or name_lower in {
".gitlab-ci.yml",
"jenkinsfile",
"azure-pipelines.yml",
}:
ci.append(normalized)
if (name_lower.startswith("test_") or "/tests/" in f"/{lower}") and name_lower.endswith(".py"):
frameworks.add("pytest/unittest")
if (".test." in name_lower or ".spec." in name_lower) and name_lower.endswith(
(".ts", ".tsx", ".js", ".jsx")
):
frameworks.add("JavaScript test runner")
if name_lower.endswith("_test.go"):
frameworks.add("go test")
if name_lower.endswith(("tests.cs", "test.cs")):
frameworks.add(".NET test")
for path, text in text_by_path.items():
name = Path(path).name
normalized = path.replace("\\", "/")
lowered_text = text.casefold()
if normalized in ci:
validates = any(
token in lowered_text
for token in ("pytest", "vitest", "jest", "npm test", "pnpm test", "go test", "dotnet test", "ruff", "eslint", "typecheck", "py_compile")
)
builds = any(
token in lowered_text
for token in ("npm run build", "pnpm build", "docker build", "vite build", "cargo build", "go build", "dotnet build")
)
deploys = any(token in lowered_text for token in ("deploy", "release", "publish", "kubectl apply"))
if validates:
ci_validation.append(normalized)
if builds:
ci_build.append(normalized)
if deploys and not validates:
ci_deploy_only.append(normalized)
if any(token in lowered_text for token in ("--cov", "coverage", "lcov", "codecov")):
coverage_evidence.append(normalized)
if name.casefold() in COVERAGE_NAMES_CASEFOLD:
coverage_evidence.append(normalized)
if name.casefold() == "package.json":
try:
data = json.loads(text)
except (TypeError, json.JSONDecodeError):
continue
if not isinstance(data, dict):
continue
deps: dict[object, object] = {}
for section_name in (
"dependencies",
"devDependencies",
"peerDependencies",
"optionalDependencies",
):
section = data.get(section_name, {})
if isinstance(section, dict):
deps.update(section)
if "vitest" in deps:
frameworks.add("Vitest")
if "jest" in deps:
frameworks.add("Jest")
if "@playwright/test" in deps:
frameworks.add("Playwright")
if "cypress" in deps:
frameworks.add("Cypress")
elif name.casefold() == "pyproject.toml":
lowered = text.lower()
if "pytest" in lowered:
frameworks.add("pytest")
ratio = round((test_files / source_files) * 100, 2) if source_files else None
return TestSignals(
frameworks=sorted(frameworks),
coverage_artifacts=sorted(set(coverage))[:30],
ci_workflows=sorted(set(ci))[:30],
ci_validation_workflows=sorted(set(ci_validation))[:30],
ci_build_workflows=sorted(set(ci_build))[:30],
ci_deploy_only_workflows=sorted(set(ci_deploy_only))[:30],
coverage_evidence=sorted(set(coverage_evidence))[:30],
test_source_ratio=ratio,
)
def analyze_dependencies(
root: Path,
files_by_name: dict[str, list[Path]],
) -> tuple[DependencySignals, list[Finding]]:
findings: list[Finding] = []
manifests: list[str] = []
lockfiles: list[str] = []
dependency_origins: dict[str, set[str]] = defaultdict(set)
declared = 0
unbounded = 0
unpinned = 0
for lock_name in sorted(LOCKFILES):
for path in files_by_name.get(lock_name, []):
lockfiles.append(str(path.relative_to(root)).replace("\\", "/"))
for package_json in files_by_name.get("package.json", []):
relative = str(package_json.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
if package_json.stat().st_size > MANIFEST_MAX_BYTES:
raise ValueError("package.json is too large to parse")
data = json.loads(package_json.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise TypeError("package.json root must be an object")
except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError):
findings.append(
Finding(
code="dependency.invalid-package-json",
category="Dependencies",
severity=Severity.WARNING,
message="package.json could not be parsed safely",
path=relative,
deduction=8,
)
)
continue
deps: dict[str, object] = {}
invalid_sections = False
for section_name in ("dependencies", "devDependencies", "peerDependencies", "optionalDependencies"):
section = data.get(section_name, {})
if section is None:
continue
if not isinstance(section, dict):
invalid_sections = True
continue
deps.update(section)
if invalid_sections:
findings.append(
Finding(
code="dependency.invalid-package-json",
category="Dependencies",
severity=Severity.WARNING,
message="One or more package dependency sections are not objects",
path=relative,
deduction=8,
)
)
declared += len(deps)
for name, version in deps.items():
dependency_name = str(name)
dependency_origins[dependency_name.casefold()].add(relative)
raw = str(version).strip().lower()
if raw in {"*", "latest", "x"} and not raw.startswith("workspace:"):
unbounded += 1
findings.append(
Finding(
code="dependency.unbounded",
category="Dependencies",
severity=Severity.WARNING,
message=f"Dependency {dependency_name} uses an unbounded version",
path=relative,
deduction=5,
)
)
parent = package_json.parent
package_roots = [parent, *[ancestor for ancestor in parent.parents if ancestor == root or root in ancestor.parents]]
if not any(
(package_root / lock).exists()
for package_root in package_roots
for lock in ("package-lock.json", "pnpm-lock.yaml", "yarn.lock", "bun.lock", "bun.lockb")
):
findings.append(Finding(code="dependency.no-lockfile", category="Dependencies", severity=Severity.WARNING, message="JavaScript package has no recognized lockfile", path=relative, deduction=8))
requirement_paths = [
path
for name, values in files_by_name.items()
if name.casefold().startswith("requirements") and name.casefold().endswith(".txt")
for path in values
]
included_requirements: list[str] = []
processed_requirements: set[Path] = set()
pending_requirements = list(requirement_paths)
while pending_requirements:
requirements = pending_requirements.pop(0)
try:
requirements = requirements.resolve(strict=True)
except OSError:
continue
if requirements in processed_requirements or requirements != root and root not in requirements.parents:
continue
processed_requirements.add(requirements)
relative = str(requirements.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
lines = requirements.read_text(encoding="utf-8").splitlines()
except (OSError, UnicodeDecodeError):
findings.append(
Finding(
code="dependency.unreadable-requirements",
category="Dependencies",
severity=Severity.WARNING,
message="Python requirements file could not be read",
path=relative,
deduction=8,
)
)
continue
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
if stripped.startswith(("-r ", "--requirement ")):
include = stripped.split(maxsplit=1)[1].strip()
candidate = (requirements.parent / include).resolve()
if candidate == root or root in candidate.parents:
pending_requirements.append(candidate)
included_requirements.append(str(candidate.relative_to(root)).replace("\\", "/"))
continue
if stripped.startswith("-"):
continue
declared += 1
name = _python_dependency_name(stripped)
if name:
dependency_origins[name.casefold()].add(relative)
finding = _python_dependency_finding(stripped, relative)
if finding:
unbounded += 1
findings.append(finding)
for pyproject in files_by_name.get("pyproject.toml", []):
relative = str(pyproject.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
if pyproject.stat().st_size > MANIFEST_MAX_BYTES:
raise ValueError("pyproject.toml is too large to parse")
data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise TypeError("pyproject.toml root must be a table")
except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError, TypeError):
findings.append(
Finding(
code="dependency.invalid-pyproject",
category="Dependencies",
severity=Severity.WARNING,
message="pyproject.toml could not be parsed safely",
path=relative,
deduction=8,
)
)
continue
project = data.get("project", {})
if not isinstance(project, dict):
findings.append(
Finding(
code="dependency.invalid-pyproject",
category="Dependencies",
severity=Severity.WARNING,
message="pyproject.toml project section is not a table",
path=relative,
deduction=8,
)
)
project = {}
project_deps = project.get("dependencies", [])
if isinstance(project_deps, list):
declared += len(project_deps)
for dependency in project_deps:
declaration = str(dependency)
name = _python_dependency_name(declaration)
if name:
dependency_origins[name.casefold()].add(relative)
finding = _python_dependency_finding(declaration, relative)
if finding:
unbounded += 1
findings.append(finding)
elif project_deps is not None:
findings.append(
Finding(
code="dependency.invalid-pyproject",
category="Dependencies",
severity=Severity.WARNING,
message="pyproject.toml project.dependencies is not a list",
path=relative,
deduction=8,
)
)
optional_deps = project.get("optional-dependencies", {})
if isinstance(optional_deps, dict):
for dependencies in optional_deps.values():
if not isinstance(dependencies, list):
continue
declared += len(dependencies)
for dependency in dependencies:
declaration = str(dependency)
name = _python_dependency_name(declaration)
if name:
dependency_origins[name.casefold()].add(relative)
finding = _python_dependency_finding(declaration, relative)
if finding:
unbounded += 1
findings.append(finding)
elif optional_deps is not None:
findings.append(
Finding(
code="dependency.invalid-pyproject",
category="Dependencies",
severity=Severity.WARNING,
message="pyproject.toml project.optional-dependencies is not a table",
path=relative,
deduction=8,
)
)
for go_mod in files_by_name.get("go.mod", []):
relative = str(go_mod.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
text = go_mod.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
findings.append(
Finding(
code="dependency.unreadable-go-mod",
category="Dependencies",
severity=Severity.WARNING,
message="go.mod could not be read",
path=relative,
deduction=8,
)
)
continue
matches = list(re.finditer(r"(?m)^\s*([\w./-]+)\s+v\d[^\s]*", text))
if "require" in text and not matches:
findings.append(
Finding(
code="dependency.invalid-go-mod",
category="Dependencies",
severity=Severity.WARNING,
message="go.mod contains no parseable versioned dependencies",
path=relative,
deduction=8,
)
)
for match in matches:
declared += 1
dependency_origins[match.group(1).casefold()].add(relative)
for cargo in files_by_name.get("Cargo.toml", []):
relative = str(cargo.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
data = tomllib.loads(cargo.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError):
findings.append(
Finding(
code="dependency.invalid-cargo-toml",
category="Dependencies",
severity=Severity.WARNING,
message="Cargo.toml could not be parsed safely",
path=relative,
deduction=8,
)
)
continue
cargo_deps = data.get("dependencies", {})
if isinstance(cargo_deps, dict):
declared += len(cargo_deps)
for name in cargo_deps:
dependency_origins[str(name).casefold()].add(relative)
for pom in files_by_name.get("pom.xml", []):
relative = str(pom.relative_to(root)).replace("\\", "/")
manifests.append(relative)
try:
xml_root = ET.parse(pom).getroot()
except (OSError, UnicodeError, ET.ParseError):
findings.append(
Finding(
code="dependency.invalid-pom",
category="Dependencies",
severity=Severity.WARNING,
message="pom.xml could not be parsed safely",
path=relative,
deduction=8,
)
)
continue
dependencies = [element for element in xml_root.iter() if element.tag.split("}")[-1] == "dependency"]
declared += len(dependencies)
for dependency in dependencies:
artifact = next((child.text for child in dependency if child.tag.split("}")[-1] == "artifactId"), None)
if artifact:
dependency_origins[artifact.casefold()].add(relative)
duplicates = sorted(name for name, origins in dependency_origins.items() if len(origins) > 1)
for name in duplicates[:25]:
origins = sorted(dependency_origins[name])
findings.append(Finding(code="dependency.duplicate-manifest", category="Dependencies", severity=Severity.INFO, message=f"Dependency {name} is declared across multiple manifests", path=origins[0], deduction=1))
return (
DependencySignals(
manifests=sorted(set(manifests)),
lockfiles=sorted(set(lockfiles)),
declared_dependencies=declared,
unbounded_dependencies=unbounded,
unpinned_dependencies=unpinned,
duplicate_dependencies=duplicates[:50],
included_requirements=sorted(set(included_requirements))[:50],
),
findings,
)