tests/unit/test_attribute.py - master - miker/git-ai-py

tests/unit/test_attribute.py
564 строки23 KB
Mike S
Initial commit
08 авг 2026, 16:43
08 авг 2026, 16:43491f864
"""Тесты атрибуции коммита.

Проверяемых утверждений три, и все они продуктовые:

1. Строка агента остаётся строкой агента, даже если между правкой и коммитом
   её переписал форматтер.
2. Ненайденная строка становится `human` только при непрерывном наблюдении
   всего промежутка, в который она могла появиться (A2.8).
3. Merge-коммит не приписывает себе объём влитой ветки (A2.6).
"""

from __future__ import annotations

import os
import subprocess
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

from git_ai_metrics.adapters.git.gitcmd import GitContext, context
from git_ai_metrics.adapters.git.gitcmd import run as git_run
from git_ai_metrics.provenance.attribute import (
    AttributionError,
    CommitAttribution,
    CommitKind,
    Coverage,
    attribute_commit,
)
from git_ai_metrics.provenance.coverage import CoverageLog
from git_ai_metrics.provenance.linemap import (
    Attribution,
    Confidence,
    FileLineMap,
    Origin,
    Tag,
)
from git_ai_metrics.provenance.worklog import Worklog
from tests.conftest import git, init_repo

AGENT_SESSION = "sess-1"
AGENT_MODEL = "qwen3-coder-plus"


def agent_edit(at: datetime) -> Attribution:
    return Attribution(
        origin=Origin.AI_AUTHORED,
        observed_at=at,
        session_id=AGENT_SESSION,
        tool_use_id="tool-1",
        model=AGENT_MODEL,
    )


def baseline(at: datetime) -> Attribution:
    """Снимок состояния файла в коммите: происхождение строк ещё не известно."""
    return Attribution(origin=Origin.UNKNOWN, observed_at=at, confidence=Confidence.DEGRADED)


#: Начало отсчёта тестовых часов. Время коммитов задаётся явно: без этого два
#: подряд идущих коммита попадают в одну секунду, промежуток между ними
#: вырождается в точку, и проверить разрыв покрытия становится нечем.
CLOCK = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)


@dataclass
class Bench:
    """Репозиторий с рабочим состоянием демона рядом и с управляемыми часами."""

    repo: Path
    ctx: GitContext
    worklog: Worklog
    coverage: CoverageLog
    now: datetime = field(default=CLOCK)

    def write(self, name: str, content: str) -> None:
        target = self.repo / name
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content, encoding="utf-8")

    def commit(self, message: str = "change", *, after: float = 3600) -> str:
        self.now += timedelta(seconds=after)
        git(self.repo, "add", "-A")
        stamp = self.now.isoformat()
        subprocess.run(
            ["git", "commit", "--quiet", "-m", message],
            cwd=self.repo,
            env={**os.environ, "GIT_AUTHOR_DATE": stamp, "GIT_COMMITTER_DATE": stamp},
            capture_output=True,
            text=True,
            check=True,
        )
        return self.head()

    def head(self) -> str:
        return git(self.repo, "rev-parse", "HEAD").strip()

    def time_of(self, rev: str) -> datetime:
        return datetime.fromisoformat(git(self.repo, "show", "-s", "--format=%cI", rev).strip())

    def snapshot(self, base: str, name: str, linemap: FileLineMap) -> None:
        self.worklog.save(base, self.repo / name, linemap)

    def cover(self, start: datetime, end: datetime) -> None:
        """Записать непрерывное наблюдение демона на отрезке."""
        self.coverage.start(start)
        moment = start
        while moment < end:
            moment = min(moment + timedelta(seconds=30), end)
            self.coverage.beat(moment)

    def attribute(self, sha: str | None = None) -> CommitAttribution:
        return attribute_commit(
            self.ctx,
            sha or self.head(),
            worklog=self.worklog,
            coverage=self.coverage,
        )


@pytest.fixture
def bench(tmp_path: Path) -> Bench:
    # Начальный коммит делается через Bench, чтобы и он получил время с
    # тестовых часов: иначе родитель оказался бы «позже» своих потомков.
    repo = init_repo(tmp_path / "repo", initial_commit=False)
    ctx = context(repo)
    bench = Bench(
        repo=repo,
        ctx=ctx,
        worklog=Worklog.for_context(ctx),
        coverage=CoverageLog(path=ctx.git_dir / "git-ai-metrics" / "coverage.jsonl"),
    )
    bench.write("README.md", "initial\n")
    bench.commit("initial")
    return bench


class TestFromWorklog:
    def test_agent_lines_keep_their_origin(self, bench: Bench) -> None:
        base = bench.head()
        observed = bench.time_of(base)
        bench.snapshot(
            base,
            "app.py",
            FileLineMap.from_content("def run():\n    return 1\n", agent_edit(observed)),
        )
        bench.write("app.py", "def run():\n    return 1\n")
        bench.commit()

        result = bench.attribute()

        (file,) = result.files
        assert file.path == "app.py"
        assert file.counts_by_origin()[Origin.AI_AUTHORED] == 2
        assert result.coverage is Coverage.FULL
        assert result.sessions == (AGENT_SESSION,)
        assert result.models == (AGENT_MODEL,)

    def test_session_and_tool_survive(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("x = 1\n", agent_edit(bench.time_of(base)))
        )
        bench.write("a.py", "x = 1\n")
        bench.commit()

        (line,) = bench.attribute().files[0].lines

        assert line.provenance.tool_use_id == "tool-1"
        assert line.provenance.confidence is Confidence.OBSERVED
        assert line.matched is True

    def test_partial_staging_keeps_matching_lines(self, bench: Bench) -> None:
        # Карта знает более позднее состояние файла, чем то, что закоммичено:
        # агент правил файл дальше уже после `git add`. Совпавшие строки обязаны
        # сохранить атрибуцию, а не потеряться из-за расхождения длины.
        base = bench.head()
        observed = bench.time_of(base)
        bench.snapshot(
            base,
            "a.py",
            FileLineMap.from_content("one\ntwo\nthree\nfour\n", agent_edit(observed)),
        )
        bench.write("a.py", "one\ntwo\n")
        bench.commit()

        result = bench.attribute()

        assert result.counts_by_origin()[Origin.AI_AUTHORED] == 2
        assert result.coverage is Coverage.FULL

    def test_formatter_rewrite_keeps_agent_origin(self, bench: Bench) -> None:
        # Классический A2.1: агент написал строки, pre-commit форматтер сменил
        # отступ. Без второй ступени поиска весь вклад агента обнулился бы.
        base = bench.head()
        bench.snapshot(
            base,
            "a.py",
            FileLineMap.from_content(
                "def run():\n        return 1\n", agent_edit(bench.time_of(base))
            ),
        )
        bench.write("a.py", "def run():\n    return 1\n")
        bench.commit()

        result = bench.attribute()

        (file,) = result.files
        assert file.counts_by_origin()[Origin.AI_AUTHORED] == 2
        assert file.counts_by_tag()[Tag.FORMATTER_TOUCHED] == 1
        assert result.counts_by_tag()[Tag.FORMATTER_TOUCHED] == 1

    def test_repeated_lines_are_consumed_once(self, bench: Bench) -> None:
        # В карте одна закрывающая скобка агента, в коммите три. Без
        # расходования все три стали бы строками агента, и вклад вырос бы втрое.
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("}\n", agent_edit(bench.time_of(base)))
        )
        bench.write("a.py", "}\n}\n}\n")
        bench.commit()

        counts = bench.attribute().counts_by_origin()

        assert counts[Origin.AI_AUTHORED] == 1
        assert counts[Origin.AI_AUTHORED] + counts[Origin.UNKNOWN] == 3

    def test_map_found_under_previous_name_after_rename(self, bench: Bench) -> None:
        # `git mv` обгоняет демона: карта ещё лежит под старым именем. Не
        # заглянуть туда значило бы потерять атрибуцию из-за одного лишь mv.
        bench.write("old.py", "alpha\nbeta\n")
        base = bench.commit("add")
        bench.snapshot(
            base,
            "old.py",
            FileLineMap.from_content("alpha\nbeta\ngamma\n", agent_edit(bench.time_of(base))),
        )
        git(bench.repo, "mv", "old.py", "new.py")
        bench.write("new.py", "alpha\nbeta\ngamma\n")
        bench.commit("rename")

        (file,) = bench.attribute().files

        assert file.path == "new.py"
        assert file.old_path == "old.py"
        assert file.is_rename is True
        # Неизменившиеся строки в дифф не попадают — атрибутируется только та,
        # что этот коммит действительно добавил.
        assert file.counts_by_origin()[Origin.AI_AUTHORED] == 1

    def test_corrupt_map_degrades_to_unknown(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("x\n", agent_edit(bench.time_of(base)))
        )
        for stored in (bench.ctx.git_dir / "git-ai-metrics" / "worklog" / base).glob("*.json"):
            stored.write_text("{ не json", encoding="utf-8")
        bench.write("a.py", "x\n")
        bench.commit()

        result = bench.attribute()

        assert result.counts_by_origin()[Origin.UNKNOWN] == 1
        assert result.coverage is Coverage.NONE


class TestCoverageFallback:
    def test_watched_whole_window_is_human(self, bench: Bench) -> None:
        base = bench.head()
        bench.write("hand.py", "written by hand\n")
        sha = bench.commit()
        bench.cover(
            bench.time_of(base) - timedelta(seconds=5),
            bench.time_of(sha) + timedelta(seconds=5),
        )

        (line,) = bench.attribute(sha).files[0].lines

        assert line.provenance.origin is Origin.HUMAN
        assert line.provenance.confidence is Confidence.INFERRED
        assert line.matched is False

    def test_daemon_never_ran_is_unknown(self, bench: Bench) -> None:
        bench.write("hand.py", "written by hand\n")
        sha = bench.commit()

        (line,) = bench.attribute(sha).files[0].lines

        assert line.provenance.origin is Origin.UNKNOWN
        assert line.provenance.confidence is Confidence.DEGRADED

    def test_gap_in_window_is_unknown(self, bench: Bench) -> None:
        # Демон наблюдал начало и конец промежутка, но в середине лежал —
        # именно тогда правку и мог сделать агент.
        base = bench.head()
        bench.write("hand.py", "line\n")
        sha = bench.commit()
        start, end = bench.time_of(base), bench.time_of(sha)
        bench.cover(start - timedelta(seconds=30), start + timedelta(seconds=30))
        bench.cover(end - timedelta(seconds=30), end + timedelta(seconds=30))

        assert bench.attribute(sha).counts_by_origin()[Origin.UNKNOWN] == 1

    def test_window_starts_at_last_observation_of_the_file(self, bench: Bench) -> None:
        # Ветка отошла от старого коммита час назад, а файл наблюдался минуту
        # назад: проверять надо эту минуту, а не всю жизнь ветки. Покрытие
        # начинается позже родительского коммита — по нему ответ был бы unknown.
        base = bench.head()
        bench.write("a.py", "old\nfresh\n")
        sha = bench.commit()
        observed = bench.time_of(sha) - timedelta(seconds=60)
        bench.snapshot(base, "a.py", FileLineMap.from_content("old\n", baseline(observed)))
        bench.cover(observed - timedelta(seconds=1), bench.time_of(sha) + timedelta(seconds=1))

        counts = bench.attribute(sha).counts_by_origin()

        assert counts[Origin.HUMAN] == 1
        assert counts[Origin.UNKNOWN] == 1

    def test_coverage_is_partial_when_some_lines_matched(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("known\n", agent_edit(bench.time_of(base)))
        )
        bench.write("a.py", "known\nunknown\n")
        bench.commit()

        assert bench.attribute().coverage is Coverage.PARTIAL


class TestCommitShapes:
    def test_root_commit_is_never_human(self, bench: Bench, tmp_path: Path) -> None:
        # Импорт готового дерева не является построчной работой человека, даже
        # если демон в этот момент работал.
        repo = init_repo(tmp_path / "root", initial_commit=False)
        ctx = context(repo)
        coverage = CoverageLog(path=ctx.git_dir / "coverage.jsonl")
        (repo / "vendored.py").write_text("a\nb\nc\n", encoding="utf-8")
        git(repo, "add", "-A")
        git(repo, "commit", "--quiet", "-m", "import")
        sha = git(repo, "rev-parse", "HEAD").strip()
        moment = datetime.fromisoformat(git(repo, "show", "-s", "--format=%cI", sha).strip())
        coverage.start(moment - timedelta(hours=1))
        coverage.beat(moment + timedelta(hours=1))

        result = attribute_commit(ctx, sha, worklog=Worklog.for_context(ctx), coverage=coverage)

        assert result.kind is CommitKind.ROOT
        assert result.parents == ()
        assert result.counts_by_origin()[Origin.UNKNOWN] == 3

    def test_merge_does_not_claim_the_merged_branch(self, bench: Bench) -> None:
        # Сердцевина A2.6: дифф merge к первому родителю — это весь объём ветки.
        base = bench.head()
        bench.write("shared.txt", "one\ntwo\n")
        bench.commit("mainline")
        git(bench.repo, "checkout", "--quiet", "-b", "side", base)
        bench.write("side.txt", "\n".join(f"line {n}" for n in range(50)) + "\n")
        bench.commit("side")
        git(bench.repo, "checkout", "--quiet", "main")
        git(bench.repo, "merge", "--quiet", "--no-ff", "-m", "merge", "side")

        result = bench.attribute()

        assert result.kind is CommitKind.MERGE
        assert result.coverage is Coverage.MERGE
        assert result.excluded_by_default is True
        assert result.total_added == 0

    def test_merge_attributes_conflict_resolution(self, bench: Bench) -> None:
        base = bench.head()
        bench.write("c.txt", "alpha\n")
        bench.commit("mainline")
        git(bench.repo, "checkout", "--quiet", "-b", "side", base)
        bench.write("c.txt", "beta\n")
        bench.commit("side")
        git(bench.repo, "checkout", "--quiet", "main")
        # Конфликт: разрешаем третьим вариантом, которого нет ни у одного родителя.
        merge = subprocess.run(
            ["git", "merge", "--no-ff", "-m", "merge", "side"],
            cwd=bench.repo,
            capture_output=True,
            text=True,
            check=False,
        )
        assert merge.returncode != 0
        bench.write("c.txt", "resolved\n")
        git(bench.repo, "add", "-A")
        git(bench.repo, "commit", "--quiet", "--no-edit")

        result = bench.attribute()

        assert result.kind is CommitKind.MERGE
        assert result.total_added == 1
        assert result.files[0].lines[0].provenance.origin is Origin.UNKNOWN

    def test_deleted_file_has_no_added_lines(self, bench: Bench) -> None:
        bench.write("gone.txt", "a\nb\n")
        bench.commit("add")
        (bench.repo / "gone.txt").unlink()
        bench.commit("remove")

        result = bench.attribute()

        (file,) = result.files
        assert file.added == 0
        assert file.deleted == 2
        assert result.total_deleted == 2
        assert result.coverage is Coverage.FULL

    def test_binary_file_is_flagged(self, bench: Bench) -> None:
        (bench.repo / "blob.bin").write_bytes(bytes(range(32)))
        bench.commit("binary")

        result = bench.attribute()

        assert result.binary_files == 1
        assert result.total_added == 0

    def test_mode_change_only_is_skipped(self, bench: Bench) -> None:
        bench.write("s.sh", "echo hi\n")
        bench.commit("add")
        (bench.repo / "s.sh").chmod(0o755)
        bench.commit("chmod")

        result = bench.attribute()

        assert result.total_added == 0

    def test_unreadable_commit_raises(self, bench: Bench) -> None:
        with pytest.raises(AttributionError):
            bench.attribute("0" * 40)


class TestBrokenRepository:
    """Поведение, когда git отвечает не тем, чего от него ждут.

    Ответы git подменяются намеренно: неполный клон, обрезанный вывод и
    нечитаемое время воспроизвести настоящим репозиторием нельзя, а
    обрабатывать их всё равно обязательно — вызов идёт из хука (A3.4).
    """

    def test_truncated_answer_about_commit(
        self, bench: Bench, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(
            "git_ai_metrics.provenance.attribute.run",
            lambda *_args, **_kwargs: subprocess.CompletedProcess([], 0, "", ""),
        )

        with pytest.raises(AttributionError, match="неожиданный ответ"):
            bench.attribute()

    def test_unparseable_commit_time(
        self, bench: Bench, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(
            "git_ai_metrics.provenance.attribute.run",
            lambda *_args, **_kwargs: subprocess.CompletedProcess([], 0, "\nне дата\n", ""),
        )

        with pytest.raises(AttributionError, match="неразбираемое время"):
            bench.attribute()

    @pytest.mark.parametrize(
        ("returncode", "stdout"),
        [
            pytest.param(128, "", id="родителя нет в объектной базе"),
            pytest.param(0, "не дата\n", id="время родителя нечитаемо"),
        ],
    )
    def test_unreadable_parent_leaves_lines_unknown(
        self, bench: Bench, monkeypatch: pytest.MonkeyPatch, returncode: int, stdout: str
    ) -> None:
        # Неполный клон: родитель недостижим, начала промежутка нет. Демон при
        # этом работал — и всё равно ответом обязан быть unknown, а не падение.
        bench.write("a.py", "line\n")
        sha = bench.commit()
        moment = bench.time_of(sha)
        bench.cover(moment - timedelta(hours=2), moment + timedelta(hours=2))
        parent = git(bench.repo, "rev-parse", f"{sha}^").strip()

        def broken(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
            if args[:3] == ["show", "-s", "--format=%cI"] and parent in args:
                return subprocess.CompletedProcess(args, returncode, stdout, "")
            return git_run(args, **kwargs)  # type: ignore[arg-type]

        monkeypatch.setattr("git_ai_metrics.provenance.attribute.run", broken)

        assert bench.attribute(sha).counts_by_origin()[Origin.UNKNOWN] == 1

    def test_symlink_out_of_the_repository(self, bench: Bench, tmp_path: Path) -> None:
        # Символическая ссылка наружу — обычное дело (node_modules, dotfiles).
        # Путь такого файла после разрешения ссылок лежит вне репозитория, и
        # обращение к карте по нему невозможно. Ответ — «ничего не известно».
        outside = tmp_path / "outside.txt"
        outside.write_text("elsewhere\n", encoding="utf-8")
        (bench.repo / "link.txt").symlink_to(outside)
        sha = bench.commit("link")

        result = bench.attribute(sha)

        assert result.counts_by_origin()[Origin.UNKNOWN] == 1
        assert result.coverage is Coverage.NONE


class TestAggregates:
    def test_origins_sum_to_added(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("agent\n", agent_edit(bench.time_of(base)))
        )
        bench.write("a.py", "agent\nother\n")
        bench.write("b.py", "more\n")
        bench.commit()

        result = bench.attribute()

        assert sum(result.counts_by_origin().values()) == result.total_added == 3

    def test_ai_share(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base, "a.py", FileLineMap.from_content("x\ny\n", agent_edit(bench.time_of(base)))
        )
        bench.write("a.py", "x\ny\nz\nw\n")
        bench.commit()

        result = bench.attribute()

        assert result.ai_lines == 2
        assert result.ai_share == pytest.approx(0.5)

    def test_ai_share_is_none_without_lines(self, bench: Bench) -> None:
        bench.write("gone.txt", "a\n")
        bench.commit("add")
        (bench.repo / "gone.txt").unlink()
        bench.commit("remove")

        assert bench.attribute().ai_share is None

    def test_ai_operated_counts_as_ai(self, bench: Bench) -> None:
        base = bench.head()
        bench.snapshot(
            base,
            "gen.txt",
            FileLineMap.from_content(
                "generated\n",
                Attribution(
                    origin=Origin.AI_OPERATED,
                    observed_at=bench.time_of(base),
                    session_id=AGENT_SESSION,
                    confidence=Confidence.INFERRED,
                ),
            ),
        )
        bench.write("gen.txt", "generated\n")
        bench.commit()

        result = bench.attribute()

        assert result.ai_lines == 1
        assert result.counts_by_origin()[Origin.AI_OPERATED] == 1