mirror of
https://github.com/usestrix/strix.git
synced 2026-09-24 00:51:20 +00:00
Simplify Git blame attribution to existing reporting instructions
This commit is contained in:
bearsyankees
parent
8c19d98705
commit
5569beef9b
6 changed files with 124 additions and 781 deletions
|
|
@ -1,104 +0,0 @@
|
|||
"""Best-effort history lookups against an existing local Git checkout."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BlameInfo:
|
||||
author_name: str
|
||||
author_email: str
|
||||
commit_sha: str
|
||||
commit_timestamp: str
|
||||
commit_summary: str
|
||||
|
||||
|
||||
def _parse_blame(output: str) -> BlameInfo | None:
|
||||
lines = output.splitlines()
|
||||
if not lines or "\x00" in output:
|
||||
return None
|
||||
header = lines[0].split()
|
||||
if not header or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", header[0]):
|
||||
return None
|
||||
commit_sha = header[0]
|
||||
if not commit_sha.strip("0"):
|
||||
# Git uses an all-zero object ID for a line with uncommitted changes.
|
||||
return None
|
||||
metadata: dict[str, str] = {}
|
||||
for raw_line in lines[1:]:
|
||||
if raw_line.startswith("\t"):
|
||||
break
|
||||
key, _, value = raw_line.partition(" ")
|
||||
metadata[key] = value
|
||||
try:
|
||||
timestamp = datetime.fromtimestamp(int(metadata["committer-time"]), tz=UTC).isoformat()
|
||||
return BlameInfo(
|
||||
author_name=metadata["author"],
|
||||
author_email=metadata["author-mail"].removeprefix("<").removesuffix(">"),
|
||||
commit_sha=commit_sha,
|
||||
commit_timestamp=timestamp,
|
||||
commit_summary=metadata["summary"],
|
||||
)
|
||||
except (KeyError, ValueError, OverflowError, OSError):
|
||||
return None
|
||||
|
||||
|
||||
def blame_line(
|
||||
repository: Path, file_path: object, line: int, *, timeout: float = 3.0
|
||||
) -> BlameInfo | None:
|
||||
"""Attribute one working-tree line, or return None when history is unavailable.
|
||||
|
||||
Paths are relative to ``repository``. Git follows committed renames itself;
|
||||
missing paths, binary files and uncommitted lines are intentionally omitted.
|
||||
Lookups never fetch history and are bounded so enrichment stays optional.
|
||||
Callers may supply a shorter timeout to share a budget across lookups.
|
||||
"""
|
||||
if (
|
||||
type(line) is not int
|
||||
or line < 1
|
||||
or not isinstance(file_path, str)
|
||||
or not file_path
|
||||
or timeout <= 0
|
||||
):
|
||||
return None
|
||||
try:
|
||||
relative = Path(file_path)
|
||||
if relative.is_absolute() or ".." in relative.parts:
|
||||
return None
|
||||
root = repository.resolve(strict=True)
|
||||
path = (root / relative).resolve(strict=True)
|
||||
if not path.is_relative_to(root) or not path.is_file():
|
||||
return None
|
||||
with path.open("rb") as source:
|
||||
if b"\x00" in source.read(8192):
|
||||
return None
|
||||
result = subprocess.run( # noqa: S603
|
||||
[ # noqa: S607
|
||||
"git",
|
||||
"-C",
|
||||
str(root),
|
||||
"blame",
|
||||
"--line-porcelain",
|
||||
"--no-textconv",
|
||||
"-L",
|
||||
f"{line},{line}",
|
||||
"--",
|
||||
file_path,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
check=False,
|
||||
timeout=min(timeout, 3.0),
|
||||
env={**os.environ, "GIT_NO_LAZY_FETCH": "1"},
|
||||
)
|
||||
except (OSError, ValueError, RuntimeError, subprocess.SubprocessError):
|
||||
return None
|
||||
return _parse_blame(result.stdout) if result.returncode == 0 else None
|
||||
|
|
@ -1,125 +0,0 @@
|
|||
"""Optional repository history inside the existing technical analysis text."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
from time import monotonic
|
||||
from typing import Any
|
||||
|
||||
from strix.core.repository_history import blame_line
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
_START = "<!-- strix:repository-history -->"
|
||||
_END = "<!-- /strix:repository-history -->"
|
||||
_ENRICHMENT_TIMEOUT = 3.0
|
||||
|
||||
|
||||
def _text(value: str) -> str:
|
||||
"""Keep repository-authored metadata inert in Markdown and HTML renderers."""
|
||||
return re.sub(r"([\\`*_\[\]])", r"\\\1", html.escape(" ".join(value.split())))
|
||||
|
||||
|
||||
def _repositories(run_record: dict[str, Any], target: str | None) -> list[Path]:
|
||||
sources: list[dict[str, Any]] = run_record.get("local_sources") or []
|
||||
roots = {Path(s["source_path"]).resolve() for s in sources if s.get("source_path")}
|
||||
matched: set[Path] = set()
|
||||
for source in sources:
|
||||
if target and target in (
|
||||
source.get("source_path"),
|
||||
source.get("workspace_subdir"),
|
||||
f"/workspace/{source.get('workspace_subdir')}",
|
||||
):
|
||||
matched.add(Path(source["source_path"]).resolve())
|
||||
targets: list[dict[str, Any]] = run_record.get("targets_info") or []
|
||||
for entry in targets:
|
||||
details: dict[str, Any] = entry.get("details") or {}
|
||||
path = details.get("cloned_repo_path") or details.get("target_path")
|
||||
if entry.get("type") not in {"repository", "local_code"} or not path:
|
||||
continue
|
||||
root = Path(path).resolve()
|
||||
roots.add(root)
|
||||
if target and target in (
|
||||
entry.get("original"),
|
||||
details.get("target_repo"),
|
||||
details.get("target_path"),
|
||||
details.get("workspace_subdir"),
|
||||
f"/workspace/{details.get('workspace_subdir')}",
|
||||
):
|
||||
matched.add(root)
|
||||
# File existence cannot identify the repository for an unmatched target.
|
||||
# Fall back only when the scan itself has exactly one possible checkout.
|
||||
selected = matched or roots
|
||||
return sorted(selected) if len(selected) == 1 else []
|
||||
|
||||
|
||||
def _location_line(location: dict[str, Any]) -> int | None:
|
||||
start = location.get("start_line")
|
||||
if type(start) is not int or start < 1:
|
||||
return None
|
||||
primary = location.get("primary_line")
|
||||
end = location.get("end_line", start)
|
||||
if type(primary) is int and type(end) is int and start <= primary <= end:
|
||||
return primary
|
||||
return start
|
||||
|
||||
|
||||
def enrich_report(report: dict[str, Any], run_record: dict[str, Any]) -> None:
|
||||
"""Refresh blame for the current locations, never failing a report operation.
|
||||
|
||||
No new checkout is made. Multiple repositories require a uniquely matching
|
||||
target. A shared time budget bounds all lookups in a report. Markers let
|
||||
revisions and resumed scans replace only the generated part of the analysis.
|
||||
"""
|
||||
try:
|
||||
deadline = monotonic() + _ENRICHMENT_TIMEOUT
|
||||
analysis = re.sub(
|
||||
re.escape(_START) + r".*?" + re.escape(_END),
|
||||
"",
|
||||
report.get("technical_analysis") or "",
|
||||
flags=re.DOTALL,
|
||||
).rstrip()
|
||||
if "technical_analysis" in report:
|
||||
report["technical_analysis"] = analysis
|
||||
locations: list[dict[str, Any]] = report.get("code_locations") or []
|
||||
if not locations:
|
||||
return
|
||||
roots = _repositories(run_record, report.get("target"))
|
||||
if not roots:
|
||||
return
|
||||
entries: list[str] = []
|
||||
seen: set[tuple[str, int]] = set()
|
||||
for location in locations:
|
||||
if monotonic() >= deadline:
|
||||
break
|
||||
file_path = location.get("file")
|
||||
line = _location_line(location)
|
||||
if not isinstance(file_path, str) or line is None or (file_path, line) in seen:
|
||||
continue
|
||||
seen.add((file_path, line))
|
||||
candidates = [root for root in roots if (root / file_path).is_file()]
|
||||
if len(candidates) != 1:
|
||||
continue
|
||||
remaining = deadline - monotonic()
|
||||
if remaining <= 0:
|
||||
break
|
||||
info = blame_line(candidates[0], file_path, line, timeout=remaining)
|
||||
if info is None:
|
||||
continue
|
||||
entries.append(
|
||||
f"**{_text(file_path)}:{line}**\n\n"
|
||||
f"- Author: {_text(info.author_name)} ({_text(info.author_email)})\n"
|
||||
f"- Commit: {_text(info.commit_sha)}\n"
|
||||
f"- Commit timestamp (UTC): {_text(info.commit_timestamp)}\n"
|
||||
f"- Commit summary: {_text(info.commit_summary)}"
|
||||
)
|
||||
if entries:
|
||||
block = "\n\n".join(entries)
|
||||
report["technical_analysis"] = (
|
||||
f"{analysis}\n\n{_START}\n### Last modified by\n\n{block}\n{_END}"
|
||||
).lstrip()
|
||||
except Exception: # noqa: BLE001 - history must never prevent issue creation or reporting.
|
||||
logger.debug("Repository history enrichment failed (non-fatal)", exc_info=True)
|
||||
|
|
@ -14,7 +14,6 @@ from strix.config import codex
|
|||
from strix.config.loader import load_settings
|
||||
from strix.core.paths import run_dir_for, runtime_state_dir
|
||||
from strix.report.coverage import write_coverage
|
||||
from strix.report.history import enrich_report
|
||||
from strix.report.pricing import resolve_litellm_model
|
||||
from strix.report.sarif import write_sarif
|
||||
from strix.report.writer import (
|
||||
|
|
@ -408,7 +407,6 @@ class ReportState:
|
|||
if agent_name:
|
||||
report["agent_name"] = agent_name
|
||||
|
||||
enrich_report(report, self.run_record)
|
||||
if self.vulnerability_found_callback:
|
||||
self.vulnerability_found_callback(report)
|
||||
|
||||
|
|
@ -498,9 +496,6 @@ class ReportState:
|
|||
revised["update_history"] = history
|
||||
revised["updated_at"] = entry["timestamp"]
|
||||
|
||||
if {"technical_analysis", "code_locations", "target"} & changed.keys():
|
||||
enrich_report(revised, self.run_record)
|
||||
|
||||
# Persistence must accept the revision before local state changes. A
|
||||
# failed callback leaves the old evidence intact and the update retryable.
|
||||
if self.vulnerability_updated_callback:
|
||||
|
|
|
|||
|
|
@ -43,7 +43,6 @@ _CODE_LOCATION_FIELDS = (
|
|||
"file",
|
||||
"start_line",
|
||||
"end_line",
|
||||
"primary_line",
|
||||
"snippet",
|
||||
"label",
|
||||
"fix_before",
|
||||
|
|
@ -74,9 +73,7 @@ def _normalize_code_locations(
|
|||
if field not in loc or loc[field] is None:
|
||||
continue
|
||||
value = loc[field]
|
||||
if field in ("start_line", "end_line", "primary_line"):
|
||||
if isinstance(value, (bool, float)):
|
||||
continue
|
||||
if field in ("start_line", "end_line"):
|
||||
try:
|
||||
normalized[field] = int(value)
|
||||
except (TypeError, ValueError):
|
||||
|
|
@ -1020,6 +1017,27 @@ async def create_vulnerability_report(
|
|||
for the full rules around ``fix_before`` / ``fix_after``,
|
||||
multi-part fixes, and informational-vs-actionable entries.
|
||||
|
||||
**Local Git attribution (best effort)**: for a finding with a repository
|
||||
file and valid line number, use ``exec_command`` in the existing full-clone
|
||||
checkout. Identify the affected repository first; if ambiguous, skip
|
||||
attribution. Blame the primary vulnerable line when known, otherwise the
|
||||
``start_line`` of the primary code location. Quote paths and use a short
|
||||
timeout, for example::
|
||||
|
||||
GIT_NO_LAZY_FETCH=1 timeout 3s git -C REPO blame --line-porcelain \\
|
||||
--no-textconv -L LINE,LINE -- FILE
|
||||
|
||||
Add a **Last modified by** subsection to ``technical_analysis`` with the
|
||||
file/line, author name (``author``), author email (``author-mail``), commit
|
||||
SHA (first field), commit timestamp (``committer-time``, rendered in UTC),
|
||||
and commit summary (``summary``), where available. Use only observed Git
|
||||
output; never invent attribution or imply the author introduced the flaw.
|
||||
Skip all-zero SHAs (uncommitted lines), missing/invalid line numbers,
|
||||
missing or renamed paths you cannot resolve, binary/generated files without
|
||||
useful history, unavailable Git metadata, and command errors/timeouts.
|
||||
Do not clone, fetch, or retry just for attribution; omit it and file the
|
||||
finding normally when unavailable. It must never block scanning or reporting.
|
||||
|
||||
**CVSS breakdown** is an object with all 8 metrics (each a single
|
||||
uppercase letter):
|
||||
|
||||
|
|
@ -1126,9 +1144,8 @@ async def create_vulnerability_report(
|
|||
but unverified follow-on risks separate; do not use them to
|
||||
set CVSS metrics.
|
||||
target: Affected URL / domain / repository.
|
||||
technical_analysis: The mechanism and root cause. When local repository
|
||||
history is available, the report automatically appends "Last modified
|
||||
by" details for the code locations. Do not invent attribution.
|
||||
technical_analysis: The mechanism and root cause, including "Last modified
|
||||
by" details when verified using local Git attribution as described above.
|
||||
poc_description: Step-by-step reproduction (steps only, no code).
|
||||
poc_script_code: Working PoC (Python preferred).
|
||||
remediation_steps: Specific, actionable fix (prose, no code).
|
||||
|
|
@ -1226,9 +1243,6 @@ async def create_vulnerability_report(
|
|||
- ``end_line`` (REQUIRED): 1-based; ``>= start_line``.
|
||||
Only equal to ``start_line`` when the block truly is one
|
||||
line.
|
||||
- ``primary_line`` (optional): the primary vulnerable line within
|
||||
this range. Local Git blame uses this line, or ``start_line``
|
||||
when omitted or outside the range. History enrichment is best-effort.
|
||||
- ``snippet`` (optional): verbatim source at this range.
|
||||
- ``label`` (optional): short role description; especially
|
||||
important for multi-part fixes.
|
||||
|
|
@ -1449,6 +1463,8 @@ async def update_vulnerability_report(
|
|||
``severity_change_conditions`` with a new ``cvss_breakdown``.
|
||||
- ``code_locations`` replaces the whole list. A location carrying
|
||||
``fix_after`` needs ``fix_verification``.
|
||||
- When changing a target or code location, refresh or remove any "Last modified
|
||||
by" attribution in ``technical_analysis``; do not retain stale Git details.
|
||||
|
||||
The report keeps its id, its original author, and its filing time. The
|
||||
revision is recorded in the report as update history, so state the
|
||||
|
|
|
|||
|
|
@ -1,10 +1,12 @@
|
|||
"""Repository history through the reporting tool, callbacks, and persisted artifacts."""
|
||||
"""Smoke-test the prompted Git command and agent-supplied report details."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
|
@ -13,10 +15,7 @@ from typing import TYPE_CHECKING, Any
|
|||
import pytest
|
||||
from agents.tool_context import ToolContext
|
||||
|
||||
from strix.core.repository_history import BlameInfo
|
||||
from strix.interface.scan_setup import build_targets_info, prepare_run
|
||||
from strix.report import history
|
||||
from strix.report.history import enrich_report
|
||||
from strix.report.state import ReportState, set_global_report_state
|
||||
from strix.tools.reporting.tool import create_vulnerability_report, update_vulnerability_report
|
||||
|
||||
|
|
@ -26,7 +25,6 @@ if TYPE_CHECKING:
|
|||
|
||||
|
||||
_ANALYSIS = "The query interpolates attacker-controlled input."
|
||||
_TARGET = "https://example.test/team/application.git"
|
||||
_FIRST_AUTHOR = "Alice Original"
|
||||
_LATEST_AUTHOR = "Bea Reviewer"
|
||||
_LOCAL_AUTHOR = "Carol Local"
|
||||
|
|
@ -79,42 +77,6 @@ def _seed_repository(origin: Path) -> dict[str, str]:
|
|||
return {"first": first_sha, "latest": _git(origin, "rev-parse", "HEAD")}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def history_run(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> Iterator[tuple[ReportState, Path, dict[str, str]]]:
|
||||
"""Use an existing full clone, including an older author for the range start."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
origin = tmp_path / "origin"
|
||||
commits = _seed_repository(origin)
|
||||
clone = tmp_path / "application"
|
||||
_git(tmp_path, "clone", "--quiet", "--no-hardlinks", str(origin), str(clone))
|
||||
assert _git(clone, "rev-parse", "--is-shallow-repository") == "false"
|
||||
|
||||
state = ReportState(run_name="history-run")
|
||||
state.set_scan_config(
|
||||
{
|
||||
"targets": [
|
||||
{
|
||||
"type": "repository",
|
||||
"original": _TARGET,
|
||||
"details": {
|
||||
"target_repo": _TARGET,
|
||||
"cloned_repo_path": str(clone),
|
||||
"workspace_subdir": "application",
|
||||
},
|
||||
}
|
||||
],
|
||||
"local_sources": [{"source_path": str(clone), "workspace_subdir": "application"}],
|
||||
}
|
||||
)
|
||||
set_global_report_state(state)
|
||||
try:
|
||||
yield state, clone, commits
|
||||
finally:
|
||||
set_global_report_state(None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scan_setup_run(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
|
|
@ -172,14 +134,14 @@ def scan_setup_run(
|
|||
|
||||
|
||||
async def _create(
|
||||
location: dict[str, Any], *additional_locations: dict[str, Any], target: str = _TARGET
|
||||
location: dict[str, Any], *, target: str, technical_analysis: str
|
||||
) -> dict[str, Any]:
|
||||
arguments = {
|
||||
"title": "SQL injection in the query handler",
|
||||
"description": "The query handler executes unsanitized input.",
|
||||
"impact": "An anonymous caller can access other users' records.",
|
||||
"target": target,
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"technical_analysis": technical_analysis,
|
||||
"poc_description": "Submit a quote in the query parameter.",
|
||||
"poc_script_code": "GET /query?q='",
|
||||
"remediation_steps": "Use parameterized queries.",
|
||||
|
|
@ -190,7 +152,7 @@ async def _create(
|
|||
"severity_change_conditions": "A read-only database role would limit the impact.",
|
||||
"fix_effort": "low",
|
||||
"cvss_breakdown": _CVSS,
|
||||
"code_locations": [location, *additional_locations],
|
||||
"code_locations": [location],
|
||||
}
|
||||
context = ToolContext(
|
||||
context={"agent_id": "root"},
|
||||
|
|
@ -222,296 +184,112 @@ async def _update(report_id: str, **fields: Any) -> dict[str, Any]:
|
|||
return result
|
||||
|
||||
|
||||
@pytest.mark.parametrize("primary_line", [2, None, -1, 3, "invalid", True])
|
||||
async def test_tool_enriches_callbacks_and_artifacts_from_full_clone(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], primary_line: Any
|
||||
) -> None:
|
||||
state, _clone, commits = history_run
|
||||
callbacks: list[dict[str, Any]] = []
|
||||
state.vulnerability_found_callback = lambda report: callbacks.append(dict(report))
|
||||
location = {"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": primary_line}
|
||||
|
||||
result = await _create(location)
|
||||
|
||||
assert result["success"] is True
|
||||
report = state.vulnerability_reports[0]
|
||||
analysis = report["technical_analysis"]
|
||||
expected_author = _LATEST_AUTHOR if primary_line == 2 else _FIRST_AUTHOR
|
||||
expected_sha = commits["latest"] if primary_line == 2 else commits["first"]
|
||||
expected_summary = "Execute the query" if primary_line == 2 else "Add query handler"
|
||||
assert analysis.startswith(_ANALYSIS)
|
||||
assert analysis.count("Last modified by") == 1
|
||||
for value in (expected_author, "author@example.test", expected_sha, expected_summary):
|
||||
assert value in analysis
|
||||
assert "2024-01-02T03:04:05+00:00" in analysis
|
||||
assert callbacks == [report]
|
||||
assert "blame" not in report
|
||||
assert "author_name" not in report
|
||||
saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert saved == [report]
|
||||
markdown = (state.get_run_dir() / "vulnerabilities" / f"{report['id']}.md").read_text()
|
||||
technical_section = markdown.split("## Technical Analysis", 1)[1].split(
|
||||
"## Proof of Concept", 1
|
||||
)[0]
|
||||
assert analysis in technical_section
|
||||
|
||||
|
||||
async def test_resume_and_updates_refresh_history_without_stale_or_duplicate_details(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]],
|
||||
) -> None:
|
||||
state, _clone, commits = history_run
|
||||
assert (await _create({"file": "app.py", "start_line": 1, "end_line": 2}))["success"]
|
||||
resumed = ReportState(run_name=state.run_name)
|
||||
resumed.hydrate_from_run_dir()
|
||||
set_global_report_state(resumed)
|
||||
callbacks: list[dict[str, Any]] = []
|
||||
resumed.vulnerability_updated_callback = lambda report: callbacks.append(dict(report))
|
||||
report_id = resumed.vulnerability_reports[0]["id"]
|
||||
assert commits["first"] in resumed.vulnerability_reports[0]["technical_analysis"]
|
||||
|
||||
result = await _update(
|
||||
report_id,
|
||||
code_locations=[{"file": "app.py", "start_line": 1, "end_line": 2, "primary_line": 2}],
|
||||
)
|
||||
assert result["success"] is True
|
||||
analysis = resumed.vulnerability_reports[0]["technical_analysis"]
|
||||
assert commits["latest"] in analysis
|
||||
assert commits["first"] not in analysis
|
||||
result = await _update(report_id, technical_analysis="Revised root cause.\n\n" + analysis)
|
||||
assert result["success"] is True
|
||||
analysis = resumed.vulnerability_reports[0]["technical_analysis"]
|
||||
assert analysis.count("Last modified by") == 1
|
||||
assert analysis.startswith("Revised root cause.")
|
||||
|
||||
result = await _update(
|
||||
report_id, code_locations=[{"file": "deleted.py", "start_line": 1, "end_line": 2}]
|
||||
)
|
||||
assert result["success"] is True
|
||||
final_report = resumed.vulnerability_reports[0]
|
||||
assert "Last modified by" not in final_report["technical_analysis"]
|
||||
assert callbacks[-1] == final_report
|
||||
saved = json.loads((resumed.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert saved == [final_report]
|
||||
markdown = (resumed.get_run_dir() / "vulnerabilities" / f"{report_id}.md").read_text()
|
||||
assert "Last modified by" not in markdown
|
||||
def test_reporting_tools_expose_optional_git_attribution_guidance() -> None:
|
||||
description = create_vulnerability_report.description
|
||||
for instruction in (
|
||||
"exec_command",
|
||||
"existing full-clone",
|
||||
"primary vulnerable line",
|
||||
"start_line",
|
||||
"Last modified by",
|
||||
"technical_analysis",
|
||||
"author-mail",
|
||||
"committer-time",
|
||||
"summary",
|
||||
"all-zero SHAs",
|
||||
"errors/timeouts",
|
||||
"never block",
|
||||
):
|
||||
assert instruction in description
|
||||
assert "refresh or remove" in update_vulnerability_report.description
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"location",
|
||||
[
|
||||
{"file": "deleted.py", "start_line": 1, "end_line": 1},
|
||||
{"file": "app.py", "start_line": 999, "end_line": 999},
|
||||
{"file": "app.py", "start_line": True, "end_line": 1},
|
||||
{"file": "app.py", "start_line": 1.5, "end_line": 2},
|
||||
{"file": "app.py", "end_line": 1},
|
||||
],
|
||||
"case", ["primary", "start", "local", "missing", "out_of_range", "uncommitted"]
|
||||
)
|
||||
async def test_unavailable_history_does_not_block_reporting(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], location: dict[str, Any]
|
||||
) -> None:
|
||||
state, _clone, _commits = history_run
|
||||
assert (await _create(location))["success"] is True
|
||||
assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS
|
||||
assert (state.get_run_dir() / "vulnerabilities.json").exists()
|
||||
|
||||
|
||||
async def test_unexpected_history_failure_does_not_block_reporting(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
state, _clone, _commits = history_run
|
||||
|
||||
def broken_blame(*_args: Any, **_kwargs: Any) -> None:
|
||||
raise RuntimeError("history lookup failed")
|
||||
|
||||
monkeypatch.setattr("strix.report.history.blame_line", broken_blame)
|
||||
assert (await _create({"file": "app.py", "start_line": 1, "end_line": 1}))["success"] is True
|
||||
assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS
|
||||
assert (state.get_run_dir() / "vulnerabilities" / "vuln-0001.md").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target_alias", [_TARGET, "application", "/workspace/application"])
|
||||
def test_ambiguous_repository_requires_matching_target_and_target_updates_refresh_history(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target_alias: str
|
||||
) -> None:
|
||||
state, clone, commits = history_run
|
||||
other = tmp_path / "other"
|
||||
_git(tmp_path, "clone", "--quiet", str(clone), str(other))
|
||||
(other / "app.py").write_text("query = 'other'\nresult = execute(query)\n", encoding="utf-8")
|
||||
_git(other, "commit", "--quiet", "-am", "Other repository change", author="Other Author")
|
||||
state.run_record["local_sources"].append(
|
||||
{"source_path": str(other), "workspace_subdir": "other"}
|
||||
)
|
||||
report = {
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
|
||||
}
|
||||
enrich_report(report, state.run_record)
|
||||
assert report["technical_analysis"] == _ANALYSIS
|
||||
|
||||
report["target"] = target_alias
|
||||
enrich_report(report, state.run_record)
|
||||
assert commits["first"] in report["technical_analysis"]
|
||||
report["target"] = "other"
|
||||
enrich_report(report, state.run_record)
|
||||
assert "Other Author" in report["technical_analysis"]
|
||||
assert commits["first"] not in report["technical_analysis"]
|
||||
assert report["technical_analysis"].count("Last modified by") == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target", [None, "https://example.test/query", "query handler"])
|
||||
def test_unmatched_target_cannot_select_repository_by_unique_file(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str | None
|
||||
) -> None:
|
||||
state, _clone, _commits = history_run
|
||||
other = tmp_path / "other"
|
||||
other.mkdir()
|
||||
state.run_record["local_sources"].append(
|
||||
{"source_path": str(other), "workspace_subdir": "other"}
|
||||
)
|
||||
report = {
|
||||
"target": target,
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
|
||||
}
|
||||
|
||||
enrich_report(report, state.run_record)
|
||||
|
||||
assert report["technical_analysis"] == _ANALYSIS
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target", ["application", "/workspace/application"])
|
||||
def test_colliding_workspace_alias_cannot_select_repository_by_unique_file(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], tmp_path: Path, target: str
|
||||
) -> None:
|
||||
state, _clone, _commits = history_run
|
||||
other = tmp_path / "other"
|
||||
other.mkdir()
|
||||
state.run_record["local_sources"].append(
|
||||
{"source_path": str(other), "workspace_subdir": "application"}
|
||||
)
|
||||
report = {
|
||||
"target": target,
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
|
||||
}
|
||||
|
||||
enrich_report(report, state.run_record)
|
||||
|
||||
assert report["technical_analysis"] == _ANALYSIS
|
||||
|
||||
|
||||
def test_unmatched_target_can_use_only_configured_repository(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]],
|
||||
) -> None:
|
||||
state, _clone, commits = history_run
|
||||
report = {
|
||||
"target": "https://example.test/query",
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"code_locations": [{"file": "app.py", "start_line": 1, "end_line": 1}],
|
||||
}
|
||||
|
||||
enrich_report(report, state.run_record)
|
||||
|
||||
assert commits["first"] in report["technical_analysis"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("alias", ["local", "repository", "workspace", "unmatched"])
|
||||
async def test_cli_scan_setup_wires_local_and_cloned_targets_into_history(
|
||||
scan_setup_run: tuple[ReportState, dict[str, str], dict[str, str]], alias: str
|
||||
async def test_prompted_blame_command_to_report_artifacts(
|
||||
scan_setup_run: tuple[ReportState, dict[str, str], dict[str, str]], case: str
|
||||
) -> None:
|
||||
# Exercise real CLI setup and Git, then supply the analysis as an agent would.
|
||||
# This is a tool/report smoke test, not an LLM compliance test.
|
||||
state, targets, commits = scan_setup_run
|
||||
target = {
|
||||
"local": targets["local"],
|
||||
"repository": targets["repository"],
|
||||
"workspace": "/workspace/origin",
|
||||
"unmatched": "https://example.test/service",
|
||||
}[alias]
|
||||
local = case == "local"
|
||||
target = targets["local" if local else "repository"]
|
||||
checkout = Path(state.run_record["local_sources"][0 if local else 1]["source_path"])
|
||||
assert _git(checkout, "rev-parse", "--is-shallow-repository") == "false"
|
||||
file_path = "missing.py" if case == "missing" else "app.py"
|
||||
line = 999 if case == "out_of_range" else 2 if case == "primary" else 1
|
||||
if case == "uncommitted":
|
||||
(checkout / "app.py").write_text("uncommitted change\nresult = query\n", encoding="utf-8")
|
||||
|
||||
result = await _create({"file": "app.py", "start_line": 1, "end_line": 1}, target=target)
|
||||
|
||||
assert result["success"] is True
|
||||
analysis = state.vulnerability_reports[0]["technical_analysis"]
|
||||
if alias == "local":
|
||||
assert _LOCAL_AUTHOR in analysis
|
||||
assert commits["local"] in analysis
|
||||
elif alias in {"repository", "workspace"}:
|
||||
assert _FIRST_AUTHOR in analysis
|
||||
assert commits["first"] in analysis
|
||||
assert commits["local"] not in analysis
|
||||
else:
|
||||
assert analysis == _ANALYSIS
|
||||
saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert saved == state.vulnerability_reports
|
||||
|
||||
|
||||
@pytest.mark.parametrize("first_lookup_succeeds", [True, False])
|
||||
async def test_history_budget_is_shared_and_partial_results_are_persisted(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
first_lookup_succeeds: bool,
|
||||
) -> None:
|
||||
state, _clone, commits = history_run
|
||||
clock = [0.0]
|
||||
timeouts: list[float] = []
|
||||
info = BlameInfo(
|
||||
author_name=_FIRST_AUTHOR,
|
||||
author_email="author@example.test",
|
||||
commit_sha=commits["first"],
|
||||
commit_timestamp="2024-01-02T03:04:05+00:00",
|
||||
commit_summary="Add query handler",
|
||||
# Execute the exact example exposed to the model, with quoted real paths.
|
||||
match = re.search(
|
||||
r"GIT_NO_LAZY_FETCH=1 timeout 3s git.*?-- FILE",
|
||||
create_vulnerability_report.description,
|
||||
re.DOTALL,
|
||||
)
|
||||
assert match is not None
|
||||
command = (
|
||||
match.group()
|
||||
.replace("REPO", shlex.quote(str(checkout)))
|
||||
.replace("LINE", str(line))
|
||||
.replace("FILE", shlex.quote(file_path))
|
||||
)
|
||||
output = subprocess.run( # noqa: S603
|
||||
["bash", "-c", command], # noqa: S607
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
check=False,
|
||||
)
|
||||
analysis = _ANALYSIS
|
||||
has_history = case in {"primary", "start", "local"}
|
||||
if has_history:
|
||||
assert output.returncode == 0
|
||||
sha = commits["local" if local else "latest" if case == "primary" else "first"]
|
||||
author = _LOCAL_AUTHOR if local else _LATEST_AUTHOR if case == "primary" else _FIRST_AUTHOR
|
||||
assert output.stdout.split()[0] == sha
|
||||
assert f"author {author}\n" in output.stdout
|
||||
assert "author-mail <author@example.test>" in output.stdout
|
||||
assert "committer-time 1704164645" in output.stdout
|
||||
summary = next(
|
||||
s.removeprefix("summary ")
|
||||
for s in output.stdout.splitlines()
|
||||
if s.startswith("summary ")
|
||||
)
|
||||
analysis += (
|
||||
f"\n\n### Last modified by\n\n{file_path}:{line} — {author} "
|
||||
f"(author@example.test); commit {sha}; "
|
||||
f"2024-01-02T03:04:05+00:00; {summary}"
|
||||
)
|
||||
elif case == "uncommitted":
|
||||
assert output.stdout.split()[0] == "0" * 40
|
||||
else:
|
||||
assert output.returncode != 0
|
||||
|
||||
def slow_blame(*_args: Any, timeout: float) -> BlameInfo | None:
|
||||
timeouts.append(timeout)
|
||||
clock[0] += min(1.0 if len(timeouts) == 1 else 3.0, timeout)
|
||||
return info if first_lookup_succeeds and len(timeouts) == 1 else None
|
||||
|
||||
monkeypatch.setattr(history, "monotonic", lambda: clock[0])
|
||||
monkeypatch.setattr(history, "blame_line", slow_blame)
|
||||
|
||||
callbacks: list[dict[str, Any]] = []
|
||||
state.vulnerability_found_callback = lambda report: callbacks.append(dict(report))
|
||||
result = await _create(
|
||||
{"file": "app.py", "start_line": 1, "end_line": 1},
|
||||
{"file": "app.py", "start_line": 2, "end_line": 2},
|
||||
{"file": "app.py", "start_line": 999, "end_line": 999},
|
||||
{"file": file_path, "start_line": 1, "end_line": 2},
|
||||
target=target,
|
||||
technical_analysis=analysis,
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert timeouts == pytest.approx([3.0, 2.0])
|
||||
assert clock[0] == pytest.approx(3.0)
|
||||
report = state.vulnerability_reports[0]
|
||||
analysis = report["technical_analysis"]
|
||||
if first_lookup_succeeds:
|
||||
assert commits["first"] in analysis
|
||||
assert "**app.py:1**" in analysis
|
||||
assert "**app.py:2**" not in analysis
|
||||
else:
|
||||
assert analysis == _ANALYSIS
|
||||
assert callbacks == [report]
|
||||
assert report["technical_analysis"] == analysis
|
||||
saved = json.loads((state.get_run_dir() / "vulnerabilities.json").read_text())
|
||||
assert saved == [report]
|
||||
markdown = (state.get_run_dir() / "vulnerabilities" / f"{report['id']}.md").read_text()
|
||||
assert analysis in markdown
|
||||
assert (
|
||||
analysis in markdown.split("## Technical Analysis", 1)[1].split("## Proof of Concept", 1)[0]
|
||||
)
|
||||
|
||||
|
||||
def test_invalid_locations_still_consume_shared_history_budget(
|
||||
history_run: tuple[ReportState, Path, dict[str, str]], monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
state, _clone, _commits = history_run
|
||||
clock = [0.0]
|
||||
location_line = history._location_line
|
||||
|
||||
def slow_location_line(location: dict[str, Any]) -> int | None:
|
||||
clock[0] += 0.5
|
||||
return location_line(location)
|
||||
|
||||
monkeypatch.setattr(history, "monotonic", lambda: clock[0])
|
||||
monkeypatch.setattr(history, "_location_line", slow_location_line)
|
||||
report = {
|
||||
"target": _TARGET,
|
||||
"technical_analysis": _ANALYSIS,
|
||||
"code_locations": [{"file": "app.py"} for _ in range(20)],
|
||||
}
|
||||
|
||||
enrich_report(report, state.run_record)
|
||||
|
||||
assert clock[0] <= 3.0
|
||||
assert report["technical_analysis"] == _ANALYSIS
|
||||
if has_history:
|
||||
revised = await _update(
|
||||
report["id"],
|
||||
target="https://example.test/new-target",
|
||||
technical_analysis=_ANALYSIS,
|
||||
)
|
||||
assert revised["success"] is True
|
||||
assert state.vulnerability_reports[0]["technical_analysis"] == _ANALYSIS
|
||||
|
|
|
|||
|
|
@ -1,217 +0,0 @@
|
|||
"""Local history lookup coverage using real Git history and failure injection."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.core.repository_history import blame_line
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _git(repository: Path, *args: str, env: dict[str, str] | None = None) -> str:
|
||||
return subprocess.run( # noqa: S603
|
||||
["git", "-C", str(repository), *args], # noqa: S607
|
||||
env={**os.environ, **(env or {})},
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
|
||||
monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
|
||||
root = tmp_path / "repository"
|
||||
root.mkdir()
|
||||
_git(root, "init")
|
||||
_git(root, "config", "user.name", "First Author")
|
||||
_git(root, "config", "user.email", "first@example.test")
|
||||
(root / "vulnerable.py").write_text("first line\nvulnerable line\nlast line\n")
|
||||
_git(root, "add", "vulnerable.py")
|
||||
_git(
|
||||
root,
|
||||
"commit",
|
||||
"-m",
|
||||
"Original source",
|
||||
env={
|
||||
"GIT_AUTHOR_DATE": "2020-01-01T00:00:00Z",
|
||||
"GIT_COMMITTER_DATE": "2020-01-02T00:00:00Z",
|
||||
},
|
||||
)
|
||||
(root / "vulnerable.py").write_text("first line\nchanged vulnerable line\nlast line\n")
|
||||
_git(root, "add", "vulnerable.py")
|
||||
_git(
|
||||
root,
|
||||
"commit",
|
||||
"-m",
|
||||
"Change vulnerable line\n\nCommit body.",
|
||||
env={
|
||||
"GIT_AUTHOR_NAME": "Second Author",
|
||||
"GIT_AUTHOR_EMAIL": "second@example.test",
|
||||
"GIT_AUTHOR_DATE": "2021-02-03T04:05:06Z",
|
||||
"GIT_COMMITTER_DATE": "2021-02-04T05:06:07Z",
|
||||
},
|
||||
)
|
||||
return root
|
||||
|
||||
|
||||
def test_blame_attributes_requested_line_and_commit_timestamp(repository: Path) -> None:
|
||||
result = blame_line(repository, "vulnerable.py", 2)
|
||||
|
||||
assert result is not None
|
||||
assert result.author_name == "Second Author"
|
||||
assert result.author_email == "second@example.test"
|
||||
assert result.commit_sha == _git(repository, "rev-parse", "HEAD")
|
||||
assert result.commit_timestamp == "2021-02-04T05:06:07+00:00"
|
||||
assert result.commit_summary == "Change vulnerable line"
|
||||
unchanged = blame_line(repository, "vulnerable.py", 1)
|
||||
assert unchanged is not None
|
||||
assert unchanged.author_name == "First Author"
|
||||
assert unchanged.commit_sha == _git(repository, "rev-parse", "HEAD~1")
|
||||
|
||||
|
||||
def test_committed_rename_follows_history_and_missing_old_path_is_skipped(repository: Path) -> None:
|
||||
original_sha = _git(repository, "rev-parse", "HEAD")
|
||||
_git(repository, "mv", "vulnerable.py", "renamed.py")
|
||||
_git(repository, "commit", "-m", "Rename source")
|
||||
|
||||
result = blame_line(repository, "renamed.py", 2)
|
||||
assert result is not None
|
||||
assert result.commit_sha == original_sha
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
|
||||
def test_uncommitted_line_is_skipped_but_unchanged_line_retains_history(repository: Path) -> None:
|
||||
(repository / "vulnerable.py").write_text("first line\nuncommitted change\nlast line\n")
|
||||
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
assert blame_line(repository, "vulnerable.py", 1) is not None
|
||||
_git(repository, "add", "vulnerable.py")
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line", [None, 0, -1, True, "2", 1.5, 500])
|
||||
def test_invalid_missing_or_out_of_bounds_lines_are_skipped(repository: Path, line: Any) -> None:
|
||||
assert blame_line(repository, "vulnerable.py", line) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"file_path", ["", "missing.py", "../vulnerable.py", "/etc/passwd", "bad\x00"]
|
||||
)
|
||||
def test_invalid_or_missing_paths_are_skipped(repository: Path, file_path: str) -> None:
|
||||
assert blame_line(repository, file_path, 1) is None
|
||||
|
||||
|
||||
def test_binary_untracked_and_non_repository_files_are_skipped(
|
||||
repository: Path, tmp_path: Path
|
||||
) -> None:
|
||||
(repository / "binary.dat").write_bytes(b"binary\x00data\n")
|
||||
_git(repository, "add", "binary.dat")
|
||||
_git(repository, "commit", "-m", "Add binary")
|
||||
(repository / "generated.py").write_text("generated source\n")
|
||||
(tmp_path / "plain.py").write_text("not a repository\n")
|
||||
|
||||
assert blame_line(repository, "binary.dat", 1) is None
|
||||
assert blame_line(repository, "generated.py", 1) is None
|
||||
assert blame_line(tmp_path, "plain.py", 1) is None
|
||||
assert blame_line(tmp_path / "absent", "plain.py", 1) is None
|
||||
|
||||
|
||||
def test_tracked_generated_text_can_be_attributed(repository: Path) -> None:
|
||||
(repository / "generated.py").write_text("# Generated file\nvalue = 1\n")
|
||||
_git(repository, "add", "generated.py")
|
||||
_git(repository, "commit", "-m", "Generate source")
|
||||
|
||||
result = blame_line(repository, "generated.py", 2)
|
||||
assert result is not None
|
||||
assert result.commit_summary == "Generate source"
|
||||
|
||||
|
||||
def test_existing_linked_worktree_uses_its_local_history(repository: Path, tmp_path: Path) -> None:
|
||||
worktree = tmp_path / "worktree"
|
||||
_git(repository, "worktree", "add", "--detach", str(worktree), "HEAD")
|
||||
|
||||
result = blame_line(worktree, "vulnerable.py", 2)
|
||||
assert result is not None
|
||||
assert result.commit_sha == _git(repository, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
def test_symlink_cannot_escape_repository(repository: Path, tmp_path: Path) -> None:
|
||||
outside = tmp_path / "outside.py"
|
||||
outside.write_text("outside source\n")
|
||||
(repository / "escape.py").symlink_to(outside)
|
||||
|
||||
with patch("strix.core.repository_history.subprocess.run") as run:
|
||||
assert blame_line(repository, "escape.py", 1) is None
|
||||
run.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"failure",
|
||||
[
|
||||
FileNotFoundError("git missing"),
|
||||
PermissionError("denied"),
|
||||
subprocess.TimeoutExpired("git", 3),
|
||||
],
|
||||
)
|
||||
def test_git_failures_do_not_escape(repository: Path, failure: Exception) -> None:
|
||||
with patch("strix.core.repository_history.subprocess.run", side_effect=failure):
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("output", ["", "malformed", "a" * 40 + " 1 1 1\nauthor Someone\n"])
|
||||
def test_malformed_output_is_skipped(repository: Path, output: str) -> None:
|
||||
result = subprocess.CompletedProcess(["git"], 0, stdout=output)
|
||||
with patch("strix.core.repository_history.subprocess.run", return_value=result):
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("timestamp", ["not-a-time", "9999999999999999999999999999999"])
|
||||
def test_invalid_commit_timestamp_is_skipped(repository: Path, timestamp: str) -> None:
|
||||
output = (
|
||||
f"{'a' * 40} 2 2 1\nauthor Name\nauthor-mail <name@example.test>\n"
|
||||
f"committer-time {timestamp}\nsummary Message\n\tcode\n"
|
||||
)
|
||||
result = subprocess.CompletedProcess(["git"], 0, stdout=output)
|
||||
with patch("strix.core.repository_history.subprocess.run", return_value=result):
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
|
||||
def test_nonzero_exit_is_skipped_and_git_is_local_bounded(repository: Path) -> None:
|
||||
result = subprocess.CompletedProcess(["git"], 128, stdout="", stderr="fatal")
|
||||
with patch("strix.core.repository_history.subprocess.run", return_value=result) as run:
|
||||
assert blame_line(repository, "vulnerable.py", 2) is None
|
||||
|
||||
args, kwargs = run.call_args
|
||||
assert args[0][-4:] == ["-L", "2,2", "--", "vulnerable.py"]
|
||||
assert kwargs["timeout"] <= 3
|
||||
assert kwargs["env"]["GIT_NO_LAZY_FETCH"] == "1"
|
||||
assert "--no-textconv" in args[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("timeout", "expected"), [(0.25, 0.25), (1.5, 1.5), (30, 3)])
|
||||
def test_blame_timeout_respects_remaining_budget_and_per_lookup_limit(
|
||||
repository: Path, timeout: float, expected: float
|
||||
) -> None:
|
||||
result = subprocess.CompletedProcess(["git"], 128, stdout="", stderr="fatal")
|
||||
with patch("strix.core.repository_history.subprocess.run", return_value=result) as run:
|
||||
assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None
|
||||
|
||||
assert run.call_args.kwargs["timeout"] == pytest.approx(expected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("timeout", [0, -1])
|
||||
def test_expired_blame_budget_does_not_start_git(repository: Path, timeout: float) -> None:
|
||||
with patch("strix.core.repository_history.subprocess.run") as run:
|
||||
assert blame_line(repository, "vulnerable.py", 2, timeout=timeout) is None
|
||||
|
||||
run.assert_not_called()
|
||||
Loading…
Add table
Reference in a new issue