chatballs/tests/cli/test_custocrm_cli.py

"""Тесты 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")

    # newline="" обязателен: на Windows write_text заменил бы перевод строки

    # на CRLF, и sha256sum получил бы имена файлов с лишним символом на конце.

    (env.release / "checksums.txt").write_text(

        "".join(lines), encoding="utf-8", newline=""

    )


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.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.set_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.set_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.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.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.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.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.install_docker()

    fake_env.install_flock(held=False)


    r = _run(fake_env, "doctor")

    assert r.returncode == 0, r.stderr


# ---------------------------------------------------------------------------

# status

# ---------------------------------------------------------------------------


def test_status_json_reports_target_version(fake_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