209 lines
8.1 KiB
Python
209 lines
8.1 KiB
Python
"""Скрипт переключения живой установки: только сухой прогон, отказы preflight и чистые функции."""
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import plistlib
|
||
import sys
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "cutover.py"
|
||
_spec = importlib.util.spec_from_file_location("cutover", SCRIPT)
|
||
cutover = importlib.util.module_from_spec(_spec)
|
||
sys.modules["cutover"] = cutover
|
||
_spec.loader.exec_module(cutover)
|
||
|
||
GIT_ENV = {
|
||
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t",
|
||
}
|
||
|
||
|
||
def git(*args, cwd):
|
||
subprocess.run(["git", *args], cwd=str(cwd), check=True, capture_output=True, env={**os.environ, **GIT_ENV})
|
||
|
||
|
||
def snapshot(root: Path):
|
||
out = {}
|
||
for p in sorted(root.rglob("*")):
|
||
st = p.stat()
|
||
out[str(p)] = (st.st_mtime_ns, st.st_size)
|
||
return out
|
||
|
||
|
||
@pytest.fixture
|
||
def env(tmp_path, monkeypatch):
|
||
home = tmp_path / "home"
|
||
home.mkdir()
|
||
monkeypatch.setenv("HOME", str(home))
|
||
bridge = tmp_path / "bridge"
|
||
bare = tmp_path / "remote.git"
|
||
bare.mkdir()
|
||
git("init", "--bare", "-b", "main", cwd=bare)
|
||
old = bridge / "jira-hub"
|
||
old.mkdir(parents=True)
|
||
git("init", "-b", "main", cwd=old)
|
||
(old / "a.txt").write_text("x")
|
||
git("add", ".", cwd=old)
|
||
git("commit", "-m", "init", cwd=old)
|
||
git("remote", "add", "origin", str(bare), cwd=old)
|
||
git("push", "-u", "origin", "main", cwd=old)
|
||
data = tmp_path / "share"
|
||
(data / "jira-hub").mkdir(parents=True)
|
||
(data / "jira-hub" / "hub.sqlite").write_text("db")
|
||
proj = tmp_path / "proj"
|
||
proj.mkdir()
|
||
(proj / ".mcp.json").write_text(json.dumps({"mcpServers": {"jira-hub": {"command": "uv", "args": []}}}))
|
||
for k, v in {
|
||
"CUTOVER_BRIDGE": bridge, "CUTOVER_DATA_ROOT": data, "CUTOVER_LAUNCH_AGENTS": tmp_path / "la",
|
||
"CUTOVER_LOGS": tmp_path / "logs", "CUTOVER_REMOTE": bare, "CUTOVER_PROJECTS": proj,
|
||
}.items():
|
||
monkeypatch.setenv(k, str(v))
|
||
monkeypatch.setattr(cutover.shutil, "which", lambda name: "/usr/bin/" + name)
|
||
return {"tmp": tmp_path, "bridge": bridge, "old": old, "data": data, "bare": bare, "proj": proj}
|
||
|
||
|
||
def test_rewrite_mcp_json(tmp_path):
|
||
p = tmp_path / ".mcp.json"
|
||
old = "/old/jira-hub"
|
||
p.write_text(json.dumps({"mcpServers": {
|
||
"jira-hub": {"command": "uv", "args": ["run", "--project", old, "jira-hub", "mcp", "--project", "AWPCA"]},
|
||
"jira-hub-wiki": {"command": "uv", "args": ["run", "--project", old, "jira-hub", "mcp", "wiki", "--project", "AWPCA"]},
|
||
"other": {"command": "x", "args": ["1"]},
|
||
}}))
|
||
co = Path("/new/sdlc-hub")
|
||
assert cutover.rewrite_mcp_json(p, co) is True
|
||
servers = json.loads(p.read_text())["mcpServers"]
|
||
assert list(servers) == ["sdlc-jira", "sdlc-wiki", "other"]
|
||
assert servers["sdlc-jira"]["args"] == ["run", "--project", str(co), "sdlc-hub", "mcp", "jira", "--project", "AWPCA"]
|
||
assert servers["sdlc-wiki"]["args"] == ["run", "--project", str(co), "sdlc-hub", "mcp", "wiki", "--project", "AWPCA"]
|
||
assert servers["other"] == {"command": "x", "args": ["1"]}
|
||
assert cutover.rewrite_mcp_json(p, co) is False
|
||
|
||
|
||
def test_rewrite_text_settings(tmp_path):
|
||
p = tmp_path / "settings.json"
|
||
p.write_text(json.dumps({"enabledMcpjsonServers": ["jira-hub", "jira-hub-wiki"],
|
||
"permissions": {"allow": ["mcp__jira-hub__issue", "mcp__jira-hub-wiki__page"]}}))
|
||
assert cutover.rewrite_text(p) == []
|
||
d = json.loads(p.read_text())
|
||
assert d["enabledMcpjsonServers"] == ["sdlc-jira", "sdlc-wiki"]
|
||
assert d["permissions"]["allow"] == ["mcp__sdlc-jira__issue", "mcp__sdlc-wiki__page"]
|
||
|
||
|
||
def test_rewrite_text_reports_leftovers(tmp_path):
|
||
p = tmp_path / "CLAUDE.md"
|
||
p.write_text("Мост Jira (aep/jira-hub, aep/jira-agent) работает через MCP `jira-hub` (user scope)\nкаталог jira-hub\n")
|
||
left = cutover.rewrite_text(p)
|
||
text = p.read_text()
|
||
assert "aep/sdlc-hub" in text and "MCP `sdlc-jira`" in text
|
||
assert left == ["каталог jira-hub"]
|
||
|
||
|
||
def test_backup(tmp_path):
|
||
p = tmp_path / "f.json"
|
||
p.write_text("a")
|
||
b = cutover.backup(p)
|
||
assert b == tmp_path / "f.json.pre-sdlc" and b.read_text() == "a"
|
||
|
||
|
||
def test_plist_text(tmp_path):
|
||
log = tmp_path / "sdlc-hub.log"
|
||
d = plistlib.loads(cutover.plist_text(tmp_path / "sdlc-hub", log, "/opt/homebrew/bin/uv").encode())
|
||
assert d["Label"] == "ru.aepif.sdlc-hub"
|
||
assert d["ProgramArguments"][-2:] == ["sdlc-hub", "serve"]
|
||
assert d["ProgramArguments"][0] == "/opt/homebrew/bin/uv"
|
||
assert d["StandardOutPath"] == d["StandardErrorPath"] == str(log)
|
||
assert d["KeepAlive"] is True
|
||
|
||
|
||
def test_dry_run_changes_nothing(env, capsys):
|
||
before = snapshot(env["tmp"])
|
||
assert cutover.main([]) == 0
|
||
assert snapshot(env["tmp"]) == before
|
||
assert "2." in capsys.readouterr().out
|
||
|
||
|
||
def test_apply_refuses_without_remote(env, capsys):
|
||
os.environ["CUTOVER_REMOTE"] = str(env["tmp"] / "nope.git")
|
||
before = snapshot(env["tmp"])
|
||
assert cutover.main(["--apply"]) == 1
|
||
assert "aep/sdlc-hub" in capsys.readouterr().out
|
||
assert snapshot(env["tmp"]) == before
|
||
|
||
|
||
def test_apply_refuses_when_target_exists(env, capsys):
|
||
(env["data"] / "sdlc-hub").mkdir()
|
||
before = snapshot(env["tmp"])
|
||
assert cutover.main(["--apply"]) == 1
|
||
assert "уже существует" in capsys.readouterr().out
|
||
assert snapshot(env["tmp"]) == before
|
||
|
||
|
||
def test_dirty_checkout_refused(env):
|
||
(env["old"] / "dirty.txt").write_text("x")
|
||
probs = cutover.preflight(cutover.Config.from_env())
|
||
assert any("не чист" in s for s in probs)
|
||
|
||
|
||
def test_preflight_clean(env):
|
||
assert cutover.preflight(cutover.Config.from_env()) == []
|
||
|
||
|
||
def test_preflight_reports_missing_tool(env, monkeypatch):
|
||
monkeypatch.setattr(cutover.shutil, "which", lambda name: None if name in ("claude", "make") else "/x/" + name)
|
||
probs = cutover.preflight(cutover.Config.from_env())
|
||
assert any("claude" in s for s in probs) and any("make" in s for s in probs)
|
||
assert not any("uv " in s for s in probs)
|
||
|
||
|
||
class _Fake:
|
||
def __init__(self, rc_by_cmd=None):
|
||
self.calls = []
|
||
self.rc_by_cmd = rc_by_cmd or {}
|
||
|
||
def __call__(self, cmd, **kw):
|
||
self.calls.append((list(cmd), kw))
|
||
rc = self.rc_by_cmd.get(cmd[2] if len(cmd) > 2 else "", 0)
|
||
r = subprocess.CompletedProcess(cmd, rc)
|
||
if kw.get("check") and rc:
|
||
raise subprocess.CalledProcessError(rc, cmd)
|
||
return r
|
||
|
||
|
||
def test_step7_add_before_remove_and_remove_not_fatal(env, monkeypatch, capsys):
|
||
fake = _Fake({"remove": 1})
|
||
monkeypatch.setattr(cutover.subprocess, "run", fake)
|
||
cfg = cutover.Config.from_env()
|
||
desc, fn = cutover.steps(cfg)[-1]
|
||
fn()
|
||
cmds = [c for c, _ in fake.calls]
|
||
assert cmds[0][:3] == ["claude", "mcp", "add"] and "sdlc-jira" in cmds[0]
|
||
assert cmds[1][:3] == ["claude", "mcp", "remove"]
|
||
assert fake.calls[1][1].get("check") is False
|
||
assert "ПРЕДУПРЕЖДЕНИЕ" in capsys.readouterr().out
|
||
|
||
|
||
def test_step2_fails_if_port_still_open(env, monkeypatch):
|
||
monkeypatch.setattr(cutover.subprocess, "run", _Fake())
|
||
monkeypatch.setattr(cutover, "_port_open", lambda port: True)
|
||
cfg = cutover.Config.from_env()
|
||
with pytest.raises(RuntimeError, match="порт"):
|
||
cutover.steps(cfg)[0][1]()
|
||
|
||
|
||
def test_failure_prints_state_and_rollback(env, monkeypatch, capsys):
|
||
def boom():
|
||
raise RuntimeError("сломалось")
|
||
|
||
plan = [("2. a", lambda: None), ("3. b", lambda: None), ("4. c", boom)]
|
||
monkeypatch.setattr(cutover, "preflight", lambda cfg: [])
|
||
monkeypatch.setattr(cutover, "steps", lambda cfg, leftovers=None, backups=None: plan)
|
||
assert cutover.main(["--apply"]) == 1
|
||
out = capsys.readouterr().out
|
||
assert "остановлено на шаге 4: сломалось" in out
|
||
assert "2. a" in out and "3. b" in out
|
||
assert "хаб ru.aepif.jira-hub остановлен" in out and "каталог данных перенесён" in out
|
||
assert "Откат" in out and "README" in out
|