Update control move
This commit is contained in:
Даниил Грабарь
+25
-23
@@ -4,7 +4,7 @@ import argparse
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from cobot import process, ui
|
||||
from cobot import process
|
||||
from cobot.ui import done
|
||||
from cobot.process import StepProgress
|
||||
|
||||
@@ -18,7 +18,7 @@ def _git(*args: str) -> str:
|
||||
return subprocess.check_output(["git", *args], cwd=_PROJECT_DIR, text=True).strip()
|
||||
|
||||
|
||||
def _update() -> None:
|
||||
def _update() -> int:
|
||||
"""Fetch the current branch, show incoming commits, pull, and reinstall the cobot CLI.
|
||||
Progress: fetch (0-30 %), pull (30-80 %), reinstall (80-100 %).
|
||||
Получает текущую ветку, показывает входящие коммиты, делает pull и переустанавливает CLI.
|
||||
@@ -32,42 +32,44 @@ def _update() -> None:
|
||||
|
||||
p.set(0, "Получение с удалённого репозитория...")
|
||||
rc = process.stream(["git", "fetch", "origin"], cwd=str(_PROJECT_DIR), on_line=p.log)
|
||||
if rc not in (0, -9, -15):
|
||||
done(False, "git fetch завершился с ошибкой")
|
||||
return
|
||||
if rc != 0:
|
||||
done(False, f"git fetch завершился с кодом {rc}")
|
||||
return 1
|
||||
p.set(30)
|
||||
|
||||
behind = _git("rev-list", f"HEAD..origin/{branch}", "--count")
|
||||
if behind == "0":
|
||||
p.set(100, "Уже актуально")
|
||||
done(True, "Уже актуальная версия")
|
||||
return
|
||||
p.log("Исходники уже актуальны; проверка установки CLI...")
|
||||
else:
|
||||
p.raw(f"\n[bold]{behind} новых коммит(ов):[/bold]")
|
||||
for line in _git("log", f"HEAD..origin/{branch}", "--oneline").splitlines():
|
||||
p.log(line)
|
||||
|
||||
p.raw(f"\n[bold]{behind} новых коммит(ов):[/bold]")
|
||||
for line in _git("log", f"HEAD..origin/{branch}", "--oneline").splitlines():
|
||||
p.log(line)
|
||||
|
||||
p.set(30, "Применение изменений...")
|
||||
rc = process.stream(["git", "pull", "origin", branch],
|
||||
cwd=str(_PROJECT_DIR), on_line=p.log)
|
||||
if rc not in (0, -9, -15):
|
||||
done(False, "git pull завершился с ошибкой")
|
||||
return
|
||||
p.set(30, "Применение изменений...")
|
||||
rc = process.stream(["git", "pull", "origin", branch],
|
||||
cwd=str(_PROJECT_DIR), on_line=p.log)
|
||||
if rc != 0:
|
||||
done(False, f"git pull завершился с кодом {rc}")
|
||||
return 1
|
||||
p.set(80)
|
||||
|
||||
p.set(80, "Переустановка cobot CLI...")
|
||||
p.raw("\n[cyan]▸[/cyan] Переустановка cobot CLI...")
|
||||
rc = process.stream(["uv", "tool", "install", "--editable", str(_PROJECT_DIR)],
|
||||
on_line=p.log)
|
||||
if rc in (0, -9, -15):
|
||||
if rc == 0:
|
||||
p.raw("[green]✓[/green] cobot переустановлен")
|
||||
p.set(100, "Готово")
|
||||
else:
|
||||
p.raw("[yellow]Предупреждение:[/yellow] переустановка не удалась")
|
||||
p.set(100, "Готово")
|
||||
except subprocess.CalledProcessError as exc:
|
||||
ok = False
|
||||
fail_msg = (f"Установка CLI завершилась с кодом {rc}. "
|
||||
"Исходники обновлены, но установка не завершена. "
|
||||
"Сохраните вывод ошибки выше и повторите cobot update.")
|
||||
except (subprocess.CalledProcessError, OSError) as exc:
|
||||
ok, fail_msg = False, str(exc)
|
||||
|
||||
done(ok, "Проект обновлён" if ok else fail_msg)
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
def register(subparsers: argparse._SubParsersAction) -> None:
|
||||
@@ -76,4 +78,4 @@ def register(subparsers: argparse._SubParsersAction) -> None:
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> None:
|
||||
_update()
|
||||
raise SystemExit(_update())
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Update recovery checks; external git and installer operations are isolated."""
|
||||
|
||||
import io
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from cobot.commands import update
|
||||
|
||||
|
||||
class UpdateTests(unittest.TestCase):
|
||||
def run_update(self, codes, behind="1"):
|
||||
output = io.StringIO()
|
||||
commands = []
|
||||
|
||||
def stream(command, **kwargs):
|
||||
commands.append(command)
|
||||
result = next(codes, 0)
|
||||
if isinstance(result, Exception):
|
||||
raise result
|
||||
return result
|
||||
|
||||
def git(*args):
|
||||
return {"rev-parse": "controller", "rev-list": behind,
|
||||
"log": "abc123 Update CLI"}[args[0]]
|
||||
|
||||
console = Console(file=output, color_system=None)
|
||||
with patch.object(update, "_git", side_effect=git), \
|
||||
patch.object(update.process, "stream", side_effect=stream), \
|
||||
patch("cobot.process.console", console), \
|
||||
patch("cobot.ui.console", console):
|
||||
code = update._update()
|
||||
return code, output.getvalue(), commands
|
||||
|
||||
def test_install_failure_reports_failure(self):
|
||||
for install_code in (1, -9, -15):
|
||||
with self.subTest(install_code=install_code):
|
||||
code, output, _ = self.run_update(iter([0, 0, install_code]))
|
||||
self.assertEqual(code, 1)
|
||||
self.assertNotIn("Проект обновлён", output)
|
||||
self.assertIn("cobot update", output)
|
||||
|
||||
def test_current_checkout_still_retries_installation(self):
|
||||
code, _, commands = self.run_update(iter([0, 0]), behind="0")
|
||||
self.assertEqual([command[:2] for command in commands],
|
||||
[["git", "fetch"], ["uv", "tool"]])
|
||||
self.assertEqual(code, 0)
|
||||
|
||||
def test_killed_git_stops_before_next_step(self):
|
||||
for codes, expected_count in [([-9], 1), ([0, -15], 2)]:
|
||||
with self.subTest(codes=codes):
|
||||
code, output, commands = self.run_update(iter(codes))
|
||||
self.assertNotEqual(code, 0)
|
||||
self.assertEqual(len(commands), expected_count)
|
||||
self.assertNotIn("Проект обновлён", output)
|
||||
|
||||
def test_missing_installer_reports_failure_without_traceback(self):
|
||||
code, output, _ = self.run_update(
|
||||
iter([0, 0, FileNotFoundError("uv not found")]))
|
||||
self.assertNotEqual(code, 0)
|
||||
self.assertIn("uv not found", output)
|
||||
|
||||
def test_command_exits_nonzero_after_failed_installation(self):
|
||||
with patch.object(update, "_update", return_value=1):
|
||||
with self.assertRaises(SystemExit) as result:
|
||||
update.run(None)
|
||||
self.assertEqual(result.exception.code, 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user