99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from .models import RepositoryPolicy
|
|
|
|
POLICY_NAMES = (".repopulse.yml", ".repopulse.yaml")
|
|
MAX_POLICY_BYTES = 64_000
|
|
|
|
|
|
def _scalar(value: str) -> bool | int | str:
|
|
normalized = value.strip().strip('"\'')
|
|
lowered = normalized.casefold()
|
|
if lowered in {"true", "yes", "on"}:
|
|
return True
|
|
if lowered in {"false", "no", "off"}:
|
|
return False
|
|
try:
|
|
return int(normalized)
|
|
except ValueError:
|
|
return normalized
|
|
|
|
|
|
def load_policy(root: Path) -> RepositoryPolicy:
|
|
policy_path = next((root / name for name in POLICY_NAMES if (root / name).is_file()), None)
|
|
if policy_path is None:
|
|
return RepositoryPolicy()
|
|
try:
|
|
if policy_path.stat().st_size > MAX_POLICY_BYTES:
|
|
return RepositoryPolicy()
|
|
lines = policy_path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
except OSError:
|
|
return RepositoryPolicy()
|
|
|
|
values: dict[str, object] = {}
|
|
suppressions: list[str] = []
|
|
in_suppressions = False
|
|
for raw_line in lines:
|
|
line = raw_line.split("#", 1)[0].rstrip()
|
|
if not line.strip():
|
|
continue
|
|
stripped = line.strip()
|
|
if stripped == "suppressions:":
|
|
in_suppressions = True
|
|
continue
|
|
if in_suppressions and stripped.startswith("-"):
|
|
value = stripped[1:].strip().strip('"\'')
|
|
if value:
|
|
suppressions.append(value)
|
|
continue
|
|
in_suppressions = False
|
|
if ":" not in stripped:
|
|
continue
|
|
key, value = stripped.split(":", 1)
|
|
key = key.strip()
|
|
if key in {"auto_scan", "min_health_score", "max_blockers", "require_tests", "require_ci"}:
|
|
values[key] = _scalar(value)
|
|
values["suppressions"] = suppressions
|
|
try:
|
|
return RepositoryPolicy.model_validate(values)
|
|
except ValueError:
|
|
return RepositoryPolicy()
|
|
|
|
|
|
def apply_policy(report, policy: RepositoryPolicy):
|
|
findings = [
|
|
item.model_copy(
|
|
update={
|
|
"suppressed": item.suppressed or item.code in set(policy.suppressions),
|
|
}
|
|
)
|
|
for item in report.findings
|
|
]
|
|
active_findings = [item for item in findings if not item.suppressed]
|
|
scores = {category: 100 for category in report.category_scores}
|
|
for finding in active_findings:
|
|
scores[finding.category] = max(0, scores[finding.category] - finding.deduction)
|
|
health_score = round(sum(scores.values()) / len(scores)) if scores else 100
|
|
from .release_gate import build_release_gate
|
|
|
|
release_gate = build_release_gate(
|
|
health_score=health_score,
|
|
findings=findings,
|
|
metrics=report.metrics,
|
|
tests=report.tests,
|
|
dependencies=report.dependencies,
|
|
policy=policy,
|
|
)
|
|
return report.model_copy(
|
|
update={
|
|
"findings": findings,
|
|
"category_scores": scores,
|
|
"health_score": health_score,
|
|
"release_readiness": release_gate.status,
|
|
"release_gate": release_gate,
|
|
"policy": policy,
|
|
}
|
|
)
|