Stitch505
97b3623915
- discover/docker.py: handle comma-separated compose_file in labels
- discover/network.py: replace os.getlogin() with robust user detection
- target.py: add lsb_release fallback via hostnamectl, guard None compose_file,
use container name (not cached CID) for docker logs
- main.py: call reset_state(mode='target') before target mode,
improve EOF handling info message
- source.py: remove redundant set_stage('DONE') inside transfer_offer
- transfer.py: fix stage naming for resume after transfer
- add dry_run.py for local logic validation
137 lines
4.4 KiB
Python
137 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
dry_run.py — Локальная проверка логики без Docker/SSH
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
# Добавляем корень проекта
|
|
_PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
if _PROJECT_ROOT not in sys.path:
|
|
sys.path.insert(0, _PROJECT_ROOT)
|
|
|
|
errors = []
|
|
|
|
def check(msg, ok):
|
|
if not ok:
|
|
errors.append(msg)
|
|
print(f" FAIL: {msg}")
|
|
else:
|
|
print(f" OK: {msg}")
|
|
|
|
print("=== DRY-RUN: Проверка импортов ===")
|
|
|
|
try:
|
|
from core import state
|
|
check("state модуль импортируется", True)
|
|
except Exception as e:
|
|
check(f"state импорт: {e}", False)
|
|
|
|
try:
|
|
from core.fsm import FSM
|
|
check("FSM класс доступен", True)
|
|
except Exception as e:
|
|
check(f"FSM импорт: {e}", False)
|
|
|
|
try:
|
|
from core.color import menu, prompt
|
|
check("color модуль доступен", True)
|
|
except Exception as e:
|
|
check(f"color импорт: {e}", False)
|
|
|
|
try:
|
|
from discover.docker import discover_docker, get_container_pid
|
|
check("discover.docker доступен", True)
|
|
except Exception as e:
|
|
check(f"discover.docker: {e}", False)
|
|
|
|
try:
|
|
from discover.nginx import discover_nginx
|
|
check("discover.nginx доступен", True)
|
|
except Exception as e:
|
|
check(f"discover.nginx: {e}", False)
|
|
|
|
try:
|
|
from transfer.transfer import do_transfer
|
|
check("transfer.transfer доступен", True)
|
|
except Exception as e:
|
|
check(f"transfer.transfer: {e}", False)
|
|
|
|
try:
|
|
from transfer.ssh import list_private_keys
|
|
check("transfer.ssh доступен", True)
|
|
except Exception as e:
|
|
check(f"transfer.ssh: {e}", False)
|
|
|
|
try:
|
|
from manifest.manifest import build_manifest, save_manifest
|
|
check("manifest доступен", True)
|
|
except Exception as e:
|
|
check(f"manifest: {e}", False)
|
|
|
|
print("\n=== DRY-RUN: Проверка FSM ===")
|
|
# Проверим что SOURCE_STEPS содержит все нужные шаги
|
|
fsm_s = FSM(mode="source")
|
|
check("SOURCE_STEPS содержит TRANSFER", "TRANSFER" in fsm_s.steps)
|
|
check("SOURCE_STEPS содержит DONE", "DONE" in fsm_s.steps)
|
|
|
|
fsm_t = FSM(mode="target")
|
|
check("TARGET_STEPS содержит DONE", "DONE" in fsm_t.steps)
|
|
|
|
print("\n=== DRY-RUN: Проверка state.json ===")
|
|
state.load_state()
|
|
check("State загружается без ошибок", True)
|
|
state.set_stage("INIT", mode=None)
|
|
check("State сохраняется", True)
|
|
|
|
print("\n=== DRY-RUN: Проверка reset_state ===")
|
|
state.mark_completed("SOURCE_DISCOVER")
|
|
state.set_stage("SOURCE_PACK", archive_path="/tmp/test.tar.gz")
|
|
state.reset_state(mode="source")
|
|
st = state.load_state()
|
|
check("reset_state очищает completed_steps", len(st.get("completed_steps", [])) == 0)
|
|
check("reset_stage после reset = INIT", st.get("stage") == "INIT")
|
|
|
|
print("\n=== DRY-RUN: Проверка manifest ===")
|
|
manifest = build_manifest(
|
|
docker_data={"container_name":"test","image":"img","status":"running","compose_file":"/tmp/compose.yml","env_file":None,"mounts":[],"ports":{},"networks":[],"host_config":{},"labels":{}},
|
|
nginx_data=[],
|
|
sidecars=[],
|
|
host_network={},
|
|
systemd_units=[],
|
|
cron_jobs=[],
|
|
extra_hints=[],
|
|
)
|
|
check("Manifest содержит service.name", manifest["service"]["name"] == "test")
|
|
check("Manifest содержит docker.compose_file", manifest["docker"]["compose_file"] == "/tmp/compose.yml")
|
|
|
|
print("\n=== DRY-RUN: Проверка transfer без host ===")
|
|
st = state.load_state()
|
|
state.save_state({**st, "archive_path":None, "target_host":None, "target_user":None, "target_port":22})
|
|
try:
|
|
do_transfer()
|
|
check("do_transfer без archive_path — бросил RuntimeError", False)
|
|
except RuntimeError as e:
|
|
check("do_transfer без archive_path — RuntimeError", "Архив не найден" in str(e))
|
|
except Exception as e:
|
|
check(f"do_transfer без archive_path — неожиданная ошибка: {e}", False)
|
|
|
|
# Проверка ssh key discovery (без файловой системы)
|
|
try:
|
|
keys = list_private_keys()
|
|
check("list_private_keys не падает", True)
|
|
except Exception as e:
|
|
check(f"list_private_keys: {e}", False)
|
|
|
|
print("\n=== DRY-RUN: ИТОГ ===")
|
|
if errors:
|
|
print(f"FAIL: {len(errors)} проверок не прошло:")
|
|
for e in errors:
|
|
print(f" - {e}")
|
|
sys.exit(1)
|
|
else:
|
|
print("Все проверки прошли успешно!")
|
|
sys.exit(0)
|