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

tests/unit/test_coverage.py
384 строки14 KB
Mike S
Initial commit
08 авг 2026, 16:43
08 авг 2026, 16:43491f864
"""Тесты окон покрытия (A2.8).

Проверяемое утверждение одно и оно продуктовое: остановка демона обязана давать
``unknown``, а не ``human``. Всё остальное здесь — способы это сломать.
"""

from __future__ import annotations

import json
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

from git_ai_metrics.provenance.coverage import DEFAULT_TTL_SEC, CoverageLog, Window
from git_ai_metrics.provenance.linemap import Origin
from git_ai_metrics.util.atomicio import MAX_ATOMIC_LINE_BYTES

T0 = datetime(2026, 8, 8, 12, 0, tzinfo=UTC)


def at(seconds: float) -> datetime:
    return T0 + timedelta(seconds=seconds)


@pytest.fixture
def log(tmp_path: Path) -> CoverageLog:
    return CoverageLog(path=tmp_path / "coverage.jsonl", ttl_sec=60.0)


class TestWindows:
    def test_no_file_means_no_coverage(self, log: CoverageLog) -> None:
        report = log.read()

        assert report.windows == ()
        assert report.covers(T0) is False

    def test_single_start_is_an_instant(self, log: CoverageLog) -> None:
        log.start(at(0))

        # Одна отметка — наблюдение ровно в один момент, не отрезок.
        assert log.read().windows == (Window(start=at(0), end=at(0)),)

    def test_beats_extend_one_window(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        log.beat(at(100))

        assert log.read().windows == (Window(start=at(0), end=at(100)),)

    def test_gap_larger_than_ttl_splits(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        # Демон убит: следующая отметка приходит только через десять минут.
        log.beat(at(650))
        log.beat(at(700))

        assert log.read().windows == (
            Window(start=at(0), end=at(50)),
            Window(start=at(650), end=at(700)),
        )

    def test_gap_exactly_ttl_does_not_split(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(60))

        assert len(log.read().windows) == 1

    def test_gap_just_over_ttl_splits(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(60.001))

        assert len(log.read().windows) == 2

    def test_restart_always_breaks_window(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(10))
        # Перезапуск через секунду: разрыв меньше ttl, но демон всё равно
        # был мёртв, и заклеивать этот промежуток нельзя.
        log.start(at(11))
        log.beat(at(20))

        assert log.read().windows == (
            Window(start=at(0), end=at(10)),
            Window(start=at(11), end=at(20)),
        )

    def test_stop_closes_window(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(30))
        log.stop(at(40))
        log.start(at(500))
        log.beat(at(520))

        assert log.read().windows == (
            Window(start=at(0), end=at(40)),
            Window(start=at(500), end=at(520)),
        )

    def test_stop_without_start_is_ignored(self, log: CoverageLog) -> None:
        log.stop(at(0))

        assert log.read().windows == ()

    def test_window_end_never_exceeds_last_observation(self, log: CoverageLog) -> None:
        # Сердцевина A2.8: окно не имеет права тянуться до «последняя отметка
        # плюс ttl». Обещание жить ещё минуту наблюдением не является.
        log.start(at(0))
        log.beat(at(30))

        window = log.read().windows[0]

        assert window.end == at(30)
        assert log.covers(at(31)) is False

    def test_out_of_order_records_are_sorted(self, log: CoverageLog) -> None:
        log.beat(at(100))
        log.start(at(0))
        log.beat(at(50))

        assert log.read().windows == (Window(start=at(0), end=at(100)),)

    def test_duration_and_total(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(30))
        log.stop(at(30))
        log.start(at(1000))
        log.stop(at(1010))

        report = log.read()

        assert report.windows[0].duration == timedelta(seconds=30)
        assert report.total == timedelta(seconds=40)


class TestClassification:
    def test_inside_window_is_human(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        log.beat(at(100))

        assert log.origin_for(at(75)) is Origin.HUMAN

    def test_outside_window_is_unknown(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))

        assert log.origin_for(at(500)) is Origin.UNKNOWN

    def test_before_first_start_is_unknown(self, log: CoverageLog) -> None:
        log.start(at(100))
        log.beat(at(150))

        assert log.origin_for(at(50)) is Origin.UNKNOWN

    def test_daemon_never_ran_is_unknown(self, log: CoverageLog) -> None:
        # Ровно тот случай, ради которого модуль существует: без журнала всё
        # обязано быть unknown, иначе вклад человека завышается на 100%.
        assert log.origin_for(T0) is Origin.UNKNOWN

    def test_gap_between_windows_is_unknown(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        log.stop(at(100))
        log.start(at(1000))
        log.beat(at(1050))
        log.stop(at(1100))

        assert log.origin_for(at(500)) is Origin.UNKNOWN
        assert log.origin_for(at(50)) is Origin.HUMAN
        assert log.origin_for(at(1050)) is Origin.HUMAN

    def test_boundaries_are_inclusive(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        log.stop(at(100))

        assert log.covers(at(0)) is True
        assert log.covers(at(100)) is True

    def test_range_inside_one_window(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(50))
        log.beat(at(100))

        assert log.covers_range(at(10), at(90)) is True
        assert log.origin_for_range(at(10), at(90)) is Origin.HUMAN

    def test_range_spanning_a_gap(self, log: CoverageLog) -> None:
        # Отрезок задевает оба окна и разрыв между ними: правку могли сделать
        # именно в разрыве, поэтому покрытым он не считается.
        log.start(at(0))
        log.stop(at(0))
        log.start(at(600))
        log.beat(at(650))

        assert log.covers_range(at(0), at(650)) is False
        assert log.origin_for_range(at(0), at(650)) is Origin.UNKNOWN

    def test_inverted_range_is_not_covered(self, log: CoverageLog) -> None:
        # Расхождение часов или переписанная история: промежутка не было.
        log.start(at(0))
        log.beat(at(600))

        assert log.covers_range(at(500), at(100)) is False

    def test_stalled_daemon_does_not_claim_its_silence(self, log: CoverageLog) -> None:
        # Демон был жив (успел записать остановку), но час не подавал признаков
        # жизни: замороженный цикл событий, SIGSTOP, зависший ввод-вывод. Живой
        # процесс не значит наблюдающий, поэтому тишина остаётся unknown.
        log.start(at(0))
        log.stop(at(3600))

        assert log.read().windows == (Window(start=at(0), end=at(0)),)
        assert log.origin_for(at(1800)) is Origin.UNKNOWN


class TestMalformed:
    def test_torn_line_is_counted_not_fatal(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.path.write_text(log.path.read_text() + '{"e":"beat","at":"2026-08-0\n')
        log.beat(at(30))

        report = log.read()

        # Оборванная строка не имеет права утащить за собой весь журнал.
        assert report.malformed == 1
        assert report.windows == (Window(start=at(0), end=at(30)),)

    @pytest.mark.parametrize(
        "line",
        [
            "не json",
            "[1,2,3]",
            '"строка"',
            "null",
            '{"e":"beat"}',
            '{"at":"2026-08-08T12:00:00+00:00"}',
            '{"e":"нет такого","at":"2026-08-08T12:00:00+00:00"}',
            '{"e":"beat","at":"не дата"}',
            '{"e":"beat","at":"2026-08-08T12:00:00+00:00","ttl":-1}',
            '{"e":"beat","at":"2026-08-08T12:00:00+00:00","ttl":"строка"}',
            '{"e":"window","at":"2026-08-08T12:00:00+00:00"}',
            '{"e":"window","at":"2026-08-08T12:00:00+00:00","to":"2026-08-08T11:00:00+00:00"}',
        ],
    )
    def test_bad_lines_never_raise(self, log: CoverageLog, line: str) -> None:
        log.start(at(0))
        log.path.write_text(log.path.read_text() + line + "\n")

        report = log.read()

        assert report.malformed == 1
        assert len(report.windows) == 1

    def test_blank_lines_are_not_malformed(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.path.write_text(log.path.read_text() + "\n\n   \n")

        assert log.read().malformed == 0

    def test_unreadable_file_yields_empty(self, tmp_path: Path) -> None:
        # Каталог вместо файла: read_text даёт OSError, а вызов идёт из
        # обработчика события и падать не имеет права.
        directory = tmp_path / "coverage.jsonl"
        directory.mkdir()

        assert CoverageLog(path=directory).read().windows == ()


class TestAtomicity:
    def test_records_fit_atomic_write(self, log: CoverageLog) -> None:
        # Строка длиннее PIPE_BUF перестаёт записываться атомарно, и два
        # процесса дадут мешанину вместо журнала (A4.2).
        log.start(at(0))
        log.beat(at(1))
        log.stop(at(2))
        log.compact()

        for line in log.path.read_text().splitlines():
            assert len(line.encode("utf-8")) + 1 <= MAX_ATOMIC_LINE_BYTES

    def test_append_does_not_rewrite_file(self, log: CoverageLog) -> None:
        log.start(at(0))
        first = log.path.read_text()

        log.beat(at(10))

        assert log.path.read_text().startswith(first)


class TestCompact:
    def test_result_is_unchanged(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(30))
        log.stop(at(40))
        log.start(at(1000))
        log.beat(at(1030))
        before = log.read().windows

        log.compact()

        assert log.read().windows == before

    def test_file_shrinks(self, log: CoverageLog) -> None:
        log.start(at(0))
        for second in range(1, 200):
            log.beat(at(second))
        big = log.path.stat().st_size

        count = log.compact()

        assert count == 1
        assert log.path.stat().st_size < big / 20

    def test_compact_of_empty_log(self, log: CoverageLog) -> None:
        assert log.compact() == 0
        assert log.read().windows == ()

    def test_compacted_window_can_be_extended(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.beat(at(30))
        log.compact()

        log.start(at(60))
        log.beat(at(90))

        assert log.read().windows == (
            Window(start=at(0), end=at(30)),
            Window(start=at(60), end=at(90)),
        )

    def test_window_record_closes_a_live_window(self, log: CoverageLog) -> None:
        # Свёрнутое окно может оказаться позже живых отметок: компакция и
        # дозапись идут из разных процессов, а записи сортируются по времени.
        log.start(at(0))
        log.beat(at(30))
        log.path.write_text(
            log.path.read_text()
            + json.dumps({"e": "window", "at": at(500).isoformat(), "to": at(600).isoformat()})
            + "\n"
        )

        assert log.read().windows == (
            Window(start=at(0), end=at(30)),
            Window(start=at(500), end=at(600)),
        )

    def test_compacted_record_is_well_formed(self, log: CoverageLog) -> None:
        log.start(at(0))
        log.stop(at(30))
        log.compact()

        payload = json.loads(log.path.read_text().strip())

        assert payload == {"e": "window", "at": at(0).isoformat(), "to": at(30).isoformat()}


class TestDefaults:
    def test_default_ttl_is_used_when_absent_in_record(self, tmp_path: Path) -> None:
        path = tmp_path / "coverage.jsonl"
        path.write_text(
            f'{{"e":"start","at":"{at(0).isoformat()}"}}\n'
            f'{{"e":"beat","at":"{at(DEFAULT_TTL_SEC).isoformat()}"}}\n'
        )

        assert len(CoverageLog(path=path).read().windows) == 1

    def test_writer_ttl_travels_with_the_record(self, tmp_path: Path) -> None:
        # Период задаёт писатель, и читатель обязан пользоваться записанным
        # значением, а не своим: иначе смена настройки переписывает прошлое.
        path = tmp_path / "coverage.jsonl"
        CoverageLog(path=path, ttl_sec=600.0).start(at(0))
        CoverageLog(path=path, ttl_sec=600.0).beat(at(300))

        assert len(CoverageLog(path=path, ttl_sec=1.0).read().windows) == 1

    def test_now_is_used_when_moment_omitted(self, log: CoverageLog) -> None:
        before = datetime.now(UTC)

        log.start()

        window = log.read().windows[0]
        assert before <= window.start <= datetime.now(UTC)