dima/chatballs
1
0 Fork 0
mirror of https://github.com/dartdavros/chatballs.git synced 2026-10-18 22:59:02 +03:00
Code Issues Packages Projects Releases Wiki Activity
Files
8a8486b676e09b68d03ffad70535c05fb714c6e2
chatballs/tests/cli/test_custocrm_cli.py
T
AndreyandClaude Fable 5.1 1a61780dec 🚚 chore!: переименование инфраструктуры CustoCRM/hub → Chatballs с миграцией 
- каталог code/custocrm → code/chatballs; CLI custocrm → chatballs; пакеты
  hub_platform → chatballs, hub_backend → chatballs_backend (app labels прежние)
- переменные CUS_* и CUSTOCRM_* → CHATBALLS_*; образы chatballs-*; compose-проект
  и база chatballs (были edevs_hub); роли Postgres chatballs_* (были custocrm_*);
  схема RLS chatballs и GUC chatballs.organization_id; cookie chatballs_*
- CI: APP_DIR code/chatballs, DEPLOY_ROOT /opt/chatballs
- deploy/migrate/rename-to-chatballs.sh — миграция существующей установки без
  потери данных: остановка старого проекта, .env (с резервной копией),
  переименование суперпользователя initdb через временную роль, остальных ролей,
  базы, схемы и функции RLS; проверено на локальном стеке
- dev-стек хранит Postgres в bind-mount data/postgres, как prod (именованный том
  compose.dev был устаревшим снимком и вводил в заблуждение)
- снятие демо удаляет объекты, созданные поверх демо-данных тестировавшим
  (звонки по демо-диалогу), вместо падения на PROTECT
- реальные домены *.custocrm.ru и идентификаторы документов не тронуты

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-05 22:21:12 +03:00

242 lines
8.3 KiB
Python
Raw Blame History

"""Тесты chatballs CLI (этап 1): deploy workflow ordering, validation, lock, doctor, status.
Запускают реальный bash-скрипт против замоканного docker/flock (см. conftest.py).
Покрывают: SPEC-HUB-0019 §11 (lock), §12 (doctor), §17 (deploy workflow),
§29 (observability). Не требуют Docker daemon.
"""
from __future__ import annotations
import hashlib
import subprocess
def _run(env, *args):
"""Вызывает chatballs через bash с окружением из fake_env."""
return subprocess.run(
[env.bash, str(env.chatballs), *args],
env=env.make_env(),
cwd=str(env.instance),
capture_output=True,
text=True,
)
def _rewrite_checksums(env):
lines = []
release_files = (
path for path in env.release.rglob("*") if path.is_file() and path.name != "checksums.txt"
)
for path in sorted(release_files):
digest = hashlib.sha256(path.read_bytes()).hexdigest()
relative = path.relative_to(env.release).as_posix()
lines.append(f"{digest} ./{relative}\n")
(env.release / "checksums.txt").write_text("".join(lines), encoding="utf-8")
def _log_lines(env):
if not env.log.exists():
return []
return [line for line in env.log.read_text().splitlines() if line.strip()]
def _index_of(log, fragment):
return next(index for index, line in enumerate(log) if fragment in line)
# ---------------------------------------------------------------------------
# deploy
# ---------------------------------------------------------------------------
def test_deploy_success_orders_canonical_workflow(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode == 0, r.stderr
log = _log_lines(fake_env)
joined = "\n".join(log)
# Канонический порядок (ADR-HUB-0028 §workflow).
idx_pull = _index_of(log, " pull")
idx_infra = _index_of(log, " up -d postgres redis")
idx_init = _index_of(log, "run --rm init")
idx_app = _index_of(
log,
" up -d backend-app backend-platform backend-admin worker frontend gateway",
)
idx_exec = _index_of(log, "exec -T backend-app")
assert idx_pull < idx_infra < idx_init < idx_app < idx_exec, joined
# applied_release записан.
assert (fake_env.instance / "state" / "applied_release").exists()
assert not (fake_env.instance / "compose.yaml").exists()
def test_deploy_includes_coturn_when_calls_profile_active(fake_env):
fake_env.write_env(
COMPOSE_PROFILES="calls",
CHATBALLS_WEB_LISTENING_IP="203.0.113.10",
CHATBALLS_CALL_TURN_SECRET="turn-secret",
CHATBALLS_CALL_TURN_REALM="turn.hub.test",
CHATBALLS_TURN_EXTERNAL_IP="203.0.113.11",
CHATBALLS_TURN_LISTENING_IP="203.0.113.11",
)
fake_env.install_docker()
fake_env.install_flock(held=False)
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode == 0, r.stderr
joined = "\n".join(_log_lines(fake_env))
assert (
"up -d backend-app backend-platform backend-admin worker frontend gateway coturn"
in joined
)
def test_deploy_rejects_shared_web_and_turn_ip(fake_env):
fake_env.write_env(
COMPOSE_PROFILES="calls",
CHATBALLS_WEB_LISTENING_IP="203.0.113.10",
CHATBALLS_CALL_TURN_SECRET="turn-secret",
CHATBALLS_CALL_TURN_REALM="turn.hub.test",
CHATBALLS_TURN_EXTERNAL_IP="203.0.113.10",
CHATBALLS_TURN_LISTENING_IP="203.0.113.10",
)
fake_env.install_docker()
fake_env.install_flock(held=False)
result = _run(fake_env, "deploy", "--non-interactive")
assert result.returncode != 0
assert "different public IP" in result.stderr
assert " pull" not in "\n".join(_log_lines(fake_env))
def test_deploy_fails_when_release_env_missing(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
# Удаляем release.env — релиз не активирован.
(fake_env.release / "release.env").unlink()
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode != 0
joined = "\n".join(_log_lines(fake_env))
assert " up -d " not in joined # до запуска контейнеров не дошло
def test_deploy_fails_when_release_checksum_is_invalid(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
(fake_env.release / "Caddyfile").write_text("tampered\n", encoding="utf-8")
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode != 0
assert "checksum" in r.stderr.lower()
assert " pull" not in "\n".join(_log_lines(fake_env))
def test_deploy_fails_when_release_image_is_not_digest_pinned(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
release_env = fake_env.release / "release.env"
release_env.write_text(
release_env.read_text().replace(
"registry.test/backend:1.0.0@sha256:" + "a" * 64,
"registry.test/backend:latest",
)
)
_rewrite_checksums(fake_env)
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode != 0
assert "immutable" in r.stderr.lower()
assert " pull" not in "\n".join(_log_lines(fake_env))
def test_deploy_fails_on_lock_held(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=True) # блокировка занята
r = _run(fake_env, "deploy", "--non-interactive")
assert r.returncode == 3, r.stderr
assert "lock" in r.stderr.lower() or "lock" in r.stdout.lower()
joined = "\n".join(_log_lines(fake_env))
assert " up -d " not in joined # destructive операция не началась
# ---------------------------------------------------------------------------
# doctor
# ---------------------------------------------------------------------------
def test_doctor_passes_on_valid_env(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
r = _run(fake_env, "doctor")
assert r.returncode == 0, r.stderr
def test_doctor_fails_on_default_secret_and_missing_password(fake_env):
fake_env.write_env(
CHATBALLS_SECRET_KEY="change-me-long-random-secret", # default -> должно провалиться
POSTGRES_PASSWORD="", # пусто -> должно провалиться
)
fake_env.install_docker()
fake_env.install_flock(held=False)
r = _run(fake_env, "doctor")
assert r.returncode == 1
assert "POSTGRES_PASSWORD" in r.stderr
assert "CHATBALLS_SECRET_KEY" in r.stderr
def test_doctor_fails_when_acme_email_missing(fake_env):
fake_env.write_env(CHATBALLS_ACME_EMAIL="")
fake_env.install_docker()
fake_env.install_flock(held=False)
result = _run(fake_env, "doctor")
assert result.returncode == 1
assert "CHATBALLS_ACME_EMAIL" in result.stderr
def test_doctor_fails_when_surface_domains_match(fake_env):
fake_env.write_env(CHATBALLS_PLATFORM_DOMAIN="app.test")
fake_env.install_docker()
fake_env.install_flock(held=False)
result = _run(fake_env, "doctor")
assert result.returncode == 1
assert "must be distinct" in result.stderr
# ---------------------------------------------------------------------------
# status
# ---------------------------------------------------------------------------
def test_status_json_reports_target_version(fake_env):
fake_env.write_env()
fake_env.install_docker()
fake_env.install_flock(held=False)
r = _run(fake_env, "status", "--json")
assert r.returncode == 0, r.stderr
out = r.stdout.strip()
assert out.startswith("{") and '"target_version"' in out
assert "1.0.0-test" in out
Reference in new issue View Git Blame Copy Permalink