feat(sdlc-rename): скрипт переключения живой установки scripts/cutover.sh

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-24 12:03:50 +04:00
parent 20edf9243a
commit 7ac525699a
3 changed files with 480 additions and 0 deletions

328
scripts/cutover.py Normal file
View File

@ -0,0 +1,328 @@
#!/usr/bin/env python3
"""Переключение живой установки хаба jira-hub -> sdlc-hub (одноразовый скрипт, stdlib, Python >= 3.9).
Без --apply только печатает проверки и шаги, ничего не меняя. С --apply выполняет шаги 2-7 спеки sdlc-rename
по порядку и останавливается на первой ошибке.
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import socket
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, List, Tuple
OLD_LABEL = "ru.aepif.jira-hub"
NEW_LABEL = "ru.aepif.sdlc-hub"
SUFFIX = ".pre-sdlc"
def _home() -> Path:
return Path(os.environ.get("HOME") or os.path.expanduser("~"))
@dataclass
class Config:
bridge: Path
data_root: Path
launch_agents: Path
logs: Path
remote: str = "https://git.aepif.ru/aep/sdlc-hub.git"
projects: List[Path] = field(default_factory=list)
port: int = 8790
@classmethod
def from_env(cls) -> "Config":
e = os.environ
h = _home()
projects = e.get("CUTOVER_PROJECTS")
return cls(
bridge=Path(e.get("CUTOVER_BRIDGE") or h / "projects" / "jira-bridge"),
data_root=Path(e.get("CUTOVER_DATA_ROOT") or h / ".local" / "share"),
launch_agents=Path(e.get("CUTOVER_LAUNCH_AGENTS") or h / "Library" / "LaunchAgents"),
logs=Path(e.get("CUTOVER_LOGS") or h / "Library" / "Logs"),
remote=e.get("CUTOVER_REMOTE") or "https://git.aepif.ru/aep/sdlc-hub.git",
projects=[Path(p) for p in projects.split(":") if p] if projects is not None
else [h / "projects" / "awp-ca", h / "projects" / "stocks_drrk"],
port=int(e.get("CUTOVER_PORT") or 8790),
)
@property
def old_checkout(self) -> Path:
return self.bridge / "jira-hub"
@property
def new_checkout(self) -> Path:
return self.bridge / "sdlc-hub"
def _git(repo: Path, *args: str) -> subprocess.CompletedProcess:
env = dict(os.environ, GIT_OPTIONAL_LOCKS="0")
return subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, env=env)
def preflight(cfg: Config) -> List[str]:
"""Проверки шага 1. Только чтение."""
problems: List[str] = []
old = cfg.old_checkout
if not old.is_dir():
problems.append("старый checkout %s не найден" % old)
else:
st = _git(old, "status", "--porcelain")
if st.returncode != 0:
problems.append("git status в %s не выполнился: %s" % (old, st.stderr.strip()))
elif st.stdout.strip():
problems.append("checkout %s не чист (есть незакоммиченные изменения)" % old)
br = _git(old, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
if br != "main":
problems.append("checkout %s не на ветке main (сейчас: %s)" % (old, br or "?"))
head = _git(old, "rev-parse", "HEAD").stdout.strip()
# ls-remote вместо fetch: проверка остаётся строго чтением (fetch пишет FETCH_HEAD)
lr = _git(old, "ls-remote", "origin", "refs/heads/main")
remote_head = lr.stdout.split()[0] if lr.returncode == 0 and lr.stdout.split() else ""
if not remote_head:
problems.append("не удалось получить origin/main из %s" % old)
elif remote_head != head:
problems.append("HEAD не совпадает с origin/main: сделайте git pull --ff-only в %s" % old)
lr = subprocess.run(["git", "ls-remote", cfg.remote], capture_output=True, text=True)
if lr.returncode != 0:
problems.append("репозиторий %s недоступен: переименуйте aep/jira-hub -> aep/sdlc-hub в Gitea (%s)"
% (cfg.remote, lr.stderr.strip().splitlines()[-1] if lr.stderr.strip() else "нет ответа"))
for p in (cfg.data_root / "sdlc-hub", cfg.new_checkout):
if p.exists():
problems.append("каталог %s уже существует" % p)
if not (cfg.data_root / "jira-hub").exists():
problems.append("каталог данных %s не найден" % (cfg.data_root / "jira-hub"))
return problems
def rewrite_mcp_json(path: Path, checkout: Path) -> bool:
data = json.loads(path.read_text(encoding="utf-8"))
servers = data.get("mcpServers")
if not isinstance(servers, dict):
return False
names = {"jira-hub": ("sdlc-jira", "jira"), "jira-hub-wiki": ("sdlc-wiki", "wiki")}
changed = False
new: dict = {}
for key, val in servers.items():
if key not in names:
new[key] = val
continue
new_key, module = names[key]
args = list(val.get("args", []))
tail: List[str] = []
for marker in ("mcp", "jira-hub"):
if marker in args:
tail = args[args.index(marker) + 1:]
break
if tail and tail[0] in ("jira", "wiki"):
tail = tail[1:]
val = dict(val)
val["args"] = ["run", "--project", str(checkout), "sdlc-hub", "mcp", module] + tail
new[new_key] = val
changed = True
if changed:
data["mcpServers"] = new
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
return changed
_REPLACEMENTS = [
("mcp__jira-hub-wiki__", "mcp__sdlc-wiki__"),
("mcp__jira-hub__", "mcp__sdlc-jira__"),
("aep/jira-hub", "aep/sdlc-hub"),
('"jira-hub-wiki"', '"sdlc-wiki"'),
('"jira-hub"', '"sdlc-jira"'),
("MCP `jira-hub`", "MCP `sdlc-jira`"),
("MCP jira-hub", "MCP sdlc-jira"),
]
def rewrite_text(path: Path) -> List[str]:
"""Замены имён серверов и инструментов; возвращает строки с остатками `jira-hub`."""
try:
text = path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
return []
out = text
for a, b in _REPLACEMENTS:
out = out.replace(a, b)
if out != text:
path.write_text(out, encoding="utf-8")
return [ln.strip() for ln in out.splitlines() if "jira-hub" in ln]
def backup(path: Path) -> Path:
dst = path.with_name(path.name + SUFFIX)
shutil.copy2(str(path), str(dst))
return dst
def plist_text(checkout: Path, log: Path, uv: str) -> str:
args = "".join("<string>%s</string>" % a for a in [uv, "run", "--project", str(checkout), "sdlc-hub", "serve"])
return (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">\n'
'<plist version="1.0"><dict>\n'
" <key>Label</key><string>%s</string>\n"
" <key>ProgramArguments</key><array>%s</array>\n"
" <key>EnvironmentVariables</key><dict>\n"
" <key>PATH</key><string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string>\n"
" <key>PYTHONUNBUFFERED</key><string>1</string>\n"
" </dict>\n"
" <key>RunAtLoad</key><true/>\n"
" <key>KeepAlive</key><true/>\n"
" <key>StandardOutPath</key><string>%s</string>\n"
" <key>StandardErrorPath</key><string>%s</string>\n"
"</dict></plist>\n" % (NEW_LABEL, args, log, log)
)
def _run(*cmd: str, cwd: Path = None, check: bool = True) -> subprocess.CompletedProcess:
return subprocess.run(list(cmd), cwd=str(cwd) if cwd else None, check=check)
def _domain() -> str:
return "gui/%d" % os.getuid()
def _project_files(project: Path) -> List[Path]:
files: List[Path] = []
claude = project / ".claude"
files += sorted(claude.glob("settings*.json")) if claude.is_dir() else []
if (project / "CLAUDE.md").is_file():
files.append(project / "CLAUDE.md")
skills = claude / "skills"
if skills.is_dir():
files += [p for p in sorted(skills.rglob("*")) if p.is_file() and not p.name.endswith(SUFFIX)]
return files
def steps(cfg: Config, leftovers: List[str] = None, backups: List[Path] = None) -> List[Tuple[str, Callable[[], None]]]:
leftovers = leftovers if leftovers is not None else []
backups = backups if backups is not None else []
new_co = cfg.new_checkout
old_plist = cfg.launch_agents / (OLD_LABEL + ".plist")
new_plist = cfg.launch_agents / (NEW_LABEL + ".plist")
log = cfg.logs / "sdlc-hub.log"
uv = shutil.which("uv") or "/opt/homebrew/bin/uv"
def stop() -> None:
_run("launchctl", "bootout", "%s/%s" % (_domain(), OLD_LABEL), check=False) # не загружен - не ошибка
if old_plist.exists():
old_plist.unlink()
def move_data() -> None:
shutil.move(str(cfg.data_root / "jira-hub"), str(cfg.data_root / "sdlc-hub"))
def move_checkout() -> None:
shutil.move(str(cfg.old_checkout), str(new_co))
wts = sorted(p for p in (new_co / ".wt").glob("*") if p.is_dir()) if (new_co / ".wt").is_dir() else []
_run("git", "-C", str(new_co), "worktree", "repair", *[str(p) for p in wts])
_run("git", "-C", str(new_co), "remote", "set-url", "origin", cfg.remote)
shutil.rmtree(str(new_co / ".venv"), ignore_errors=True)
_run("uv", "sync", cwd=new_co)
_run("make", "web", cwd=new_co)
def start() -> None:
cfg.launch_agents.mkdir(parents=True, exist_ok=True)
cfg.logs.mkdir(parents=True, exist_ok=True)
new_plist.write_text(plist_text(new_co, log, uv), encoding="utf-8")
_run("launchctl", "bootstrap", _domain(), str(new_plist))
deadline = time.time() + 30
while time.time() < deadline:
try:
socket.create_connection(("127.0.0.1", cfg.port), timeout=1).close()
return
except OSError:
time.sleep(1)
raise RuntimeError("хаб не ответил на порту %d за 30 с (лог %s)" % (cfg.port, log))
def consumers() -> None:
for project in cfg.projects:
mcp = project / ".mcp.json"
if mcp.is_file():
text = mcp.read_text(encoding="utf-8")
if "jira-hub" in text:
backups.append(backup(mcp))
rewrite_mcp_json(mcp, new_co)
leftovers.extend("%s: %s" % (mcp, ln.strip())
for ln in mcp.read_text(encoding="utf-8").splitlines() if "jira-hub" in ln)
for f in _project_files(project):
try:
if "jira-hub" not in f.read_text(encoding="utf-8"):
continue
except (UnicodeDecodeError, OSError):
continue
backups.append(backup(f))
leftovers.extend("%s: %s" % (f, ln) for ln in rewrite_text(f))
def user_scope() -> None:
_run("claude", "mcp", "remove", "-s", "user", "jira-hub")
_run("claude", "mcp", "add", "-s", "user", "sdlc-jira", "--",
"uv", "run", "--project", str(new_co), "sdlc-hub", "mcp", "jira")
return [
("2. остановить launchd %s и удалить %s" % (OLD_LABEL, old_plist), stop),
("3. перенести данные %s -> %s" % (cfg.data_root / "jira-hub", cfg.data_root / "sdlc-hub"), move_data),
("4. перенести checkout %s -> %s, worktree repair, origin -> %s, uv sync, make web"
% (cfg.old_checkout, new_co, cfg.remote), move_checkout),
("5. поставить и запустить %s (лог %s), дождаться порта %d" % (NEW_LABEL, log, cfg.port), start),
("6. править проекты %s (.mcp.json, .claude/settings*.json, CLAUDE.md, .claude/skills; копии *%s)"
% (", ".join(str(p) for p in cfg.projects), SUFFIX), consumers),
("7. user scope Claude: remove jira-hub, add sdlc-jira", user_scope),
]
def main(argv: List[str]) -> int:
ap = argparse.ArgumentParser(description="Переключение живого хаба jira-hub -> sdlc-hub")
ap.add_argument("--apply", action="store_true", help="выполнить (по умолчанию сухой прогон)")
args = ap.parse_args(argv)
cfg = Config.from_env()
problems = preflight(cfg)
print("1. проверки:")
if problems:
for p in problems:
print(" ПРОБЛЕМА: " + p)
else:
print(" ок")
leftovers: List[str] = []
backups: List[Path] = []
plan = steps(cfg, leftovers, backups)
if not args.apply:
print("Сухой прогон, ничего не изменено. Шаги:")
for desc, _ in plan:
print(" " + desc)
return 1 if problems else 0
if problems:
print("Остановлено: исправьте проблемы шага 1, ничего не изменено.")
return 1
for desc, fn in plan:
print("-> " + desc)
try:
fn()
except Exception as e: # noqa: BLE001
print("остановлено на шаге %s: %s" % (desc.split(".")[0], e))
return 1
print(" готово")
print("\nОсталось владельцу:")
print(" - /mcp reconnect в открытых сессиях Claude")
print(" - codegraph init в %s и в каждом worktree (.wt/*)" % cfg.new_checkout)
print(" - проверить правки и удалить копии *%s:" % SUFFIX)
for b in backups:
print(" " + str(b))
if leftovers:
print(" - проверить руками (остались упоминания jira-hub):")
for ln in leftovers:
print(" " + ln)
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))

2
scripts/cutover.sh Executable file
View File

@ -0,0 +1,2 @@
#!/bin/bash
exec python3 "$(dirname "$0")/cutover.py" "$@"

150
tests/test_cutover.py Normal file
View File

@ -0,0 +1,150 @@
"""Скрипт переключения живой установки: только сухой прогон, отказы 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))
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()) == []