sdlc-hub/tests/test_release.py

import hashlib
import json
import os
import stat
from pathlib import Path

import pytest

from sdlc_hub.core import release


def agent_file(tmp_path, version, body="print('x')\n", name="agent.py"):
    p = tmp_path / name
    p.write_text(f'#!/usr/bin/env python3\n"""агент"""\nAGENT_VERSION = {version}\n{body}', encoding="utf-8")
    return p


def test_publish_writes_file_and_manifest(tmp_path):
    rd = tmp_path / "release"
    src = agent_file(tmp_path, 6)
    m = release.publish(rd, src)
    assert m["version"] == 6 and m["size"] == src.stat().st_size and len(m["sha256"]) == 64
    assert (rd / "agent.py").read_bytes() == src.read_bytes()
    assert json.loads((rd / "manifest.json").read_text(encoding="utf-8")) == m
    assert stat.S_IMODE(os.stat(rd).st_mode) == 0o700
    assert release.current(rd) == {"version": 6, "sha256": m["sha256"], "published_at": m["published_at"]}
    loaded = release.load(rd)
    assert loaded == {"version": 6, "sha256": m["sha256"], "content": src.read_text(encoding="utf-8")}
    assert not [p.name for p in rd.iterdir() if p.name.startswith(".")]      # временных файлов не осталось


def test_publish_only_up(tmp_path):
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 6))
    before = (rd / "manifest.json").read_bytes()
    with pytest.raises(release.PublishError, match="опубликована v6, новая должна быть больше"):
        release.publish(rd, agent_file(tmp_path, 6, body="print('y')\n"))
    with pytest.raises(release.PublishError, match="опубликована v6"):
        release.publish(rd, agent_file(tmp_path, 5))
    assert (rd / "manifest.json").read_bytes() == before
    assert release.publish(rd, agent_file(tmp_path, 7))["version"] == 7


@pytest.mark.parametrize("text", ["print(1)\n", "AGENT_VERSION = 5\nAGENT_VERSION = 6\n", "AGENT_VERSION = 'v5'\n",
                                  "    AGENT_VERSION = 5\n"])
def test_version_line_required_once(tmp_path, text):
    p = tmp_path / "a.py"
    p.write_text(text, encoding="utf-8")
    with pytest.raises(release.PublishError, match="AGENT_VERSION"):
        release.publish(tmp_path / "release", p)
    assert not (tmp_path / "release" / "manifest.json").exists()


def test_publish_rejects_non_utf8_and_big(tmp_path):
    p = tmp_path / "a.py"
    p.write_bytes(b"AGENT_VERSION = 5\n\xff\xfe")
    with pytest.raises(release.PublishError, match="UTF-8"):
        release.publish(tmp_path / "release", p)
    p.write_bytes(b"AGENT_VERSION = 5\n" + b"#" * (1024 * 1024))
    with pytest.raises(release.PublishError, match="1 МБ"):
        release.publish(tmp_path / "release", p)
    with pytest.raises(release.PublishError, match="не прочитан"):
        release.publish(tmp_path / "release", tmp_path / "нет.py")


def test_current_none_without_release(tmp_path):
    assert release.current(tmp_path / "release") is None
    assert release.load(tmp_path / "release") is None


def test_current_broken_manifest(tmp_path):
    rd = tmp_path / "release"
    rd.mkdir()
    (rd / "manifest.json").write_text('{"version": 6, "sha', encoding="utf-8")
    assert release.current(rd) is None
    (rd / "manifest.json").write_text('{"version": "шесть", "sha256": "x"}', encoding="utf-8")
    assert release.current(rd) is None


def test_load_rejects_sha_mismatch(tmp_path):
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 6))
    (rd / "agent.py").write_text("AGENT_VERSION = 6\nподменён\n", encoding="utf-8")
    assert release.current(rd)["version"] == 6
    assert release.load(rd) is None


def test_hint():
    rel = {"version": 6, "sha256": "s", "published_at": "t"}
    online5 = {"agent_version": "5", "online": True}
    assert release.hint(rel, online5, None, None) == "агент запущен без run.py — обновится только вручную"
    assert release.hint(rel, online5, None, 1) is None
    assert release.hint(rel, {"agent_version": "5", "online": False}, None, None) is None
    assert release.hint(rel, {"agent_version": "6", "online": True}, None, None) is None
    assert release.hint(None, online5, None, None) is None
    rb = {"state": "rolled_back", "from_version": 6, "to_version": 7, "error": "код 1 через 3 с"}
    assert release.hint(rel, {"agent_version": "6", "online": True}, rb, 1) == "откат с v7 на v6: код 1 через 3 с"
    rj = {"state": "rejected", "from_version": 5, "to_version": 6, "error": "sha256 не сходится"}
    assert release.hint(rel, online5, rj, 1) == "v6 отвергнута: sha256 не сходится"
    assert release.hint(rel, online5, {"state": "ok", "from_version": 5, "to_version": 6}, 1) is None


def test_hint_none_when_no_agent():
    rel = {"version": 6, "sha256": "s", "published_at": "t"}
    assert release.hint(rel, {"agent_version": None, "online": False}, None, None) is None


# --- hub-hygiene-d: публикация при сбое записи, проверка манифеста (решения 1–6) ---------------------------------

def _fail_replace_into(monkeypatch, *names):
    """os.replace в файл с одним из имён names падает «диск полон»; остальные переименования работают."""
    real = os.replace

    def fake(src, dst, *a, **kw):
        if Path(dst).name in names:
            raise OSError(28, "No space left on device")
        return real(src, dst, *a, **kw)

    monkeypatch.setattr(release.os, "replace", fake)


def _listing(rd):
    return {p.name: p.read_bytes() for p in rd.iterdir()}


def test_publish_success_leaves_no_prev(tmp_path):
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 5))
    release.publish(rd, agent_file(tmp_path, 6))
    assert sorted(p.name for p in rd.iterdir()) == ["agent.py", "manifest.json"]
    assert release.current(rd)["version"] == 6


def test_manifest_write_failure_keeps_previous(tmp_path, monkeypatch):
    """M-1 #25, спека «Диск полон при записи манифеста»: прежние v8 agent.py и манифест, ни .tmp, ни agent.py.prev."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 8))
    before = _listing(rd)
    _fail_replace_into(monkeypatch, "manifest.json")
    with pytest.raises(release.PublishError, match=r"^не удалось записать публикацию: .*No space left on device"):
        release.publish(rd, agent_file(tmp_path, 9, body="print('v9')\n"))
    monkeypatch.undo()
    assert _listing(rd) == before
    assert release.load(rd)["version"] == 8


def test_agent_write_failure_keeps_previous(tmp_path, monkeypatch):
    """M-1, M-6 #25: сбой fsync при записи agent.py — временный файл удалён, прежний agent.py на месте."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 8))
    before = _listing(rd)

    def no_space(fd):
        raise OSError(28, "No space left on device")

    monkeypatch.setattr(release.os, "fsync", no_space)
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, agent_file(tmp_path, 9))
    monkeypatch.undo()
    assert _listing(rd) == before
    assert release.load(rd)["version"] == 8


def test_first_publish_failure_leaves_nothing(tmp_path, monkeypatch):
    """Review Focus 3: прежней публикации нет — после сбоя release/ пуст, а не с agent.py без манифеста."""
    rd = tmp_path / "release"
    _fail_replace_into(monkeypatch, "manifest.json")
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, agent_file(tmp_path, 1))
    monkeypatch.undo()
    assert list(rd.iterdir()) == [] and release.current(rd) is None


def test_stale_prev_removed_by_next_publish(tmp_path):
    """Review Focus 1: agent.py.prev, оставленный упавшим процессом, убирает следующая успешная публикация."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 6))
    (rd / "agent.py").rename(rd / "agent.py.prev")    # процесс упал между переименованием и записью
    assert release.load(rd) is None
    release.publish(rd, agent_file(tmp_path, 7))
    assert sorted(p.name for p in rd.iterdir()) == ["agent.py", "manifest.json"]
    assert release.load(rd)["version"] == 7


def test_restore_failure_is_logged_not_raised(tmp_path, monkeypatch, caplog):
    """Review Focus 2: и возврат agent.py.prev не удался — PublishError с текстом, причина возврата в логе."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 8))
    before = _listing(rd)
    new_src = agent_file(tmp_path, 9)
    _fail_replace_into(monkeypatch, "manifest.json")
    real = release.os.replace

    def fake(src, dst, *a, **kw):
        if Path(src).name == "agent.py.prev":
            raise OSError(13, "Permission denied")
        return real(src, dst, *a, **kw)

    monkeypatch.setattr(release.os, "replace", fake)
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, new_src)
    assert "прежняя публикация не восстановлена" in caplog.text
    monkeypatch.undo()
    after = _listing(rd)    # задокументированное состояние: новый agent.py, старый agent.py.prev, старый манифест
    assert sorted(after) == ["agent.py", "agent.py.prev", "manifest.json"]
    assert after["agent.py"] == new_src.read_bytes()
    assert after["agent.py.prev"] == before["agent.py"]
    assert after["manifest.json"] == before["manifest.json"]


@pytest.mark.parametrize("manifest", [
    {"version": 6.9, "sha256": "a" * 64}, {"version": True, "sha256": "a" * 64}, {"version": "7", "sha256": "a" * 64},
    {"version": 0, "sha256": "a" * 64}, {"version": 6, "sha256": "a" * 63}, {"version": 6, "sha256": "A" * 64},
    {"version": 6}, [6, "a" * 64], None,
])
def test_invalid_manifest_means_no_release(tmp_path, manifest):
    """M-7 #25, решение 4: version — целое (не bool) ≥ 1, sha256 — 64 символа [0-9a-f], корень — объект."""
    rd = tmp_path / "release"
    rd.mkdir()
    (rd / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
    assert release.current(rd) is None and release.load(rd) is None


def test_invalid_manifest_warns_once_per_new_error(tmp_path, monkeypatch, caplog):
    """M-3 #25, решение 5: одно предупреждение на каждый новый текст ошибки; годный манифест сбрасывает память."""
    monkeypatch.setattr(release, "_warned", {})
    rd = tmp_path / "release"
    rd.mkdir()
    manifest = rd / "manifest.json"
    manifest.write_text('{"version": 6.9, "sha256": "' + "a" * 64 + '"}', encoding="utf-8")
    for _ in range(100):
        assert release.current(rd) is None
    warnings = lambda: [r for r in caplog.records if "манифест" in r.getMessage()]
    assert len(warnings()) == 1 and "6.9" in warnings()[0].getMessage()
    manifest.write_text('{"version": 6, "sha256": "short"}', encoding="utf-8")      # другая ошибка — ещё одно
    release.current(rd)
    release.current(rd)
    assert len(warnings()) == 2
    manifest.write_text('{"version": 6, "sha256": "' + "a" * 64 + '"}', encoding="utf-8")   # годный сбрасывает память
    assert release.current(rd)["version"] == 6
    manifest.write_text('{"version": 6, "sha256": "short"}', encoding="utf-8")
    release.current(rd)
    assert len(warnings()) == 3


def test_load_non_utf8_with_matching_sha_is_none(tmp_path, caplog):
    """M-5 #25, решение 6: sha256 сошёлся, но файл не в UTF-8 — публикации нет, ошибка в логе."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 6))
    data = b"AGENT_VERSION = 6\n\xff\xfe\n"
    (rd / "agent.py").write_bytes(data)
    m = json.loads((rd / "manifest.json").read_text(encoding="utf-8"))
    m["sha256"] = hashlib.sha256(data).hexdigest()
    (rd / "manifest.json").write_text(json.dumps(m), encoding="utf-8")
    assert release.current(rd)["version"] == 6
    assert release.load(rd) is None
    assert "agent.py не в UTF-8" in caplog.text


def test_move_failure_keeps_previous_agent(tmp_path, monkeypatch):
    """T2-I1: сбой переименования agent.py -> agent.py.prev при существующем agent.py — прежний agent.py не трогаем."""
    rd = tmp_path / "release"
    release.publish(rd, agent_file(tmp_path, 8))
    before = _listing(rd)
    real = release.os.replace

    def fake(src, dst, *a, **kw):
        if Path(dst).name == "agent.py.prev":
            raise OSError(13, "Permission denied")
        return real(src, dst, *a, **kw)

    monkeypatch.setattr(release.os, "replace", fake)
    with pytest.raises(release.PublishError, match=r"^не удалось записать публикацию: "):
        release.publish(rd, agent_file(tmp_path, 9))
    monkeypatch.undo()
    assert _listing(rd) == before
    assert release.load(rd)["version"] == 8


def test_hint_unknown_launcher_until_reported():
    """M-2 #25, решение 7: пока после запуска хаба не было claim, «без run.py» не утверждается; откат — по update_state."""
    rel = {"version": 10, "sha256": "s", "published_at": "t"}
    online9 = {"agent_version": "9", "online": True}
    assert release.hint(rel, online9, None, None, reported=False) is None
    assert release.hint(rel, online9, None, None, reported=True) == "агент запущен без run.py — обновится только вручную"
    rb = {"state": "rolled_back", "from_version": 9, "to_version": 10, "error": "код 1"}
    assert release.hint(rel, online9, rb, None, reported=False) == "откат с v10 на v9: код 1"


def test_invalid_manifest_warning_not_repeated_across_alternating_dirs(tmp_path, monkeypatch, caplog):
    """M-1: два агента — негодный манифест одного и отсутствующий у другого не сбрасывают память друг друга."""
    monkeypatch.setattr(release, "_warned", {})
    bad, none = tmp_path / "jira", tmp_path / "wiki"
    bad.mkdir()
    none.mkdir()
    (bad / "manifest.json").write_text('{"version": 0, "sha256": "x"}', encoding="utf-8")
    for _ in range(10):
        assert release.current(bad) is None
        assert release.current(none) is None
    assert len([r for r in caplog.records if "манифест" in r.getMessage()]) == 1


# --- пакет C: восстановление прерванной публикации (design hub-backend-fixes, решение 9; #39 T2.2, M-6) ----------

def _no_space(monkeypatch):
    def no_space(fd):
        raise OSError(28, "No space left on device")

    monkeypatch.setattr(release.os, "fsync", no_space)


def test_recover_missing_agent_restores_prev(tmp_path, monkeypatch, caplog):
    rd = tmp_path / "release"
    v5 = agent_file(tmp_path, 5).read_bytes()
    release.publish(rd, agent_file(tmp_path, 5))
    os.replace(rd / "agent.py", rd / "agent.py.prev")          # процесс упал между переименованием и записью
    _no_space(monkeypatch)
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, agent_file(tmp_path, 6, name="new.py"))
    monkeypatch.undo()
    assert (rd / "agent.py").read_bytes() == v5
    assert not (rd / "agent.py.prev").exists()
    assert release.load(rd)["version"] == 5
    assert "agent.py.prev" in caplog.text


def test_recover_partial_agent_keeps_good_prev(tmp_path, monkeypatch):
    rd = tmp_path / "release"
    v5 = agent_file(tmp_path, 5).read_bytes()
    release.publish(rd, agent_file(tmp_path, 5))
    (rd / "agent.py.prev").write_bytes(v5)
    (rd / "agent.py").write_bytes("AGENT_VERSION = 6\n# обрыв".encode())
    _no_space(monkeypatch)
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, agent_file(tmp_path, 6, name="new.py"))
    monkeypatch.undo()
    assert (rd / "agent.py").read_bytes() == v5
    assert not (rd / "agent.py.prev").exists()            # _undo вернул .prev на место agent.py
    assert sorted(p.name for p in rd.iterdir()) == ["agent.py", "manifest.json"]
    assert release.load(rd)["version"] == 5


def test_recover_success_after_crash(tmp_path):
    rd = tmp_path / "release"
    v5 = agent_file(tmp_path, 5).read_bytes()
    release.publish(rd, agent_file(tmp_path, 5))
    (rd / "agent.py.prev").write_bytes(v5)
    (rd / "agent.py").write_bytes("AGENT_VERSION = 6\n# обрыв".encode())
    v6 = agent_file(tmp_path, 6, name="new.py")
    release.publish(rd, v6)
    assert sorted(p.name for p in rd.iterdir()) == ["agent.py", "manifest.json"]
    assert (rd / "agent.py").read_bytes() == v6.read_bytes()
    assert release.load(rd)["version"] == 6


def test_recover_leftover_prev_removed(tmp_path, monkeypatch):
    """agent.py совпадает с манифестом, рядом лишний .prev: _recover удаляет его, даже если публикация потом падает
    до переименования agent.py в .prev."""
    rd = tmp_path / "release"
    v5 = agent_file(tmp_path, 5).read_bytes()
    release.publish(rd, agent_file(tmp_path, 5))
    (rd / "agent.py.prev").write_bytes(v5)
    _fail_replace_into(monkeypatch, "agent.py.prev")
    with pytest.raises(release.PublishError, match="не удалось записать публикацию"):
        release.publish(rd, agent_file(tmp_path, 6, name="new.py"))
    monkeypatch.undo()
    assert sorted(p.name for p in rd.iterdir()) == ["agent.py", "manifest.json"]
    assert release.load(rd)["version"] == 5


def test_recover_unknown_state_untouched(tmp_path, caplog):
    rd = tmp_path / "release"
    rd.mkdir()
    (rd / "agent.py.prev").write_bytes(b"AGENT_VERSION = 3\n")
    release._recover(rd)
    assert _listing(rd) == {"agent.py.prev": b"AGENT_VERSION = 3\n"}
    assert "agent.py.prev" in caplog.text
    assert release.publish(rd, agent_file(tmp_path, 1))["version"] == 1      # первая публикация проходит
    assert release.load(rd)["version"] == 1


def test_recover_oserror_does_not_stop_publish(tmp_path, monkeypatch, caplog):
    rd = tmp_path / "release"
    v5 = agent_file(tmp_path, 5).read_bytes()
    release.publish(rd, agent_file(tmp_path, 5))
    (rd / "agent.py.prev").write_bytes(v5)
    real = Path.unlink

    def fake(self, *a, **kw):
        if self.name == "agent.py.prev" and not getattr(fake, "done", False):
            fake.done = True
            raise OSError(13, "Permission denied")
        return real(self, *a, **kw)

    monkeypatch.setattr(Path, "unlink", fake)
    assert release.publish(rd, agent_file(tmp_path, 6, name="new.py"))["version"] == 6
    assert "Permission denied" in caplog.text