omnichannel-configserver-mcp/examples/demo/e2e.py

#!/usr/bin/env python3
"""E2E-тест деплоя omnichannel-configserver-mcp с реальным config-agent.

Поднимает контейнер config-agent, который опрашивает config_server и реально
выполняет задачи через docker-compose, затем гоняет жизненный цикл сервиса
через MCP: set_env -> set_compose -> deploy -> restart -> down.

Требуется:
  - поднятый стенд: docker compose -f examples/demo/docker-compose.yml up -d
  - собранный бинарник: go build -o omnichannel-configserver-mcp .
  - доступ к образу config-agent (registry.devlexicom.ru).

Запуск: python3 examples/demo/e2e.py
"""
import json
import os
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from demo import MCP  # noqa: E402  (общий минимальный MCP-клиент)

AGENT_IMAGE = "registry.devlexicom.ru/omnichannel_platform/config_server/config-agent:v1.1.0"
AGENT_NAME = "omni-e2e-agent"
NETWORK = "omni-mcp-demo_default"
OPT_DIR = "/tmp/omni-e2e/opt"
APP = "demo_e2e"
APP_DIR = os.path.join(OPT_DIR, APP)
HOST_IP = "10.0.0.10"

INITIAL_COMPOSE = """services:
  web:
    image: alpine:3.20
    command: ["sleep", "3600"]
"""
UPDATED_COMPOSE = """services:
  web:
    image: alpine:3.20
    command: ["sleep", "7200"]
"""


def sh(*args, check=True):
    return subprocess.run(args, capture_output=True, text=True, check=check)


def docker(*args, check=True):
    return sh("docker", *args, check=check)


def prepare_app_dir():
    os.makedirs(APP_DIR, exist_ok=True)
    with open(os.path.join(APP_DIR, "docker-compose.yml"), "w") as fh:
        fh.write(INITIAL_COMPOSE)


def start_agent():
    docker("rm", "-f", AGENT_NAME, check=False)
    docker(
        "run", "-d", "--name", AGENT_NAME,
        "--network", NETWORK,
        "-v", "/var/run/docker.sock:/var/run/docker.sock",
        "-v", "/usr/bin/docker:/usr/bin/docker:ro",
        "-v", f"{OPT_DIR}:/opt",
        "-e", "SERVER_URL=http://config-server:5000",
        "-e", "AGENT_SECRET=demo-agent-secret",
        "-e", f"AGENT_REGISTER_IP={HOST_IP}",
        "-e", "TASK_CHECK_INTERVAL=3",
        "-e", "SCAN_INTERVAL=15",
        AGENT_IMAGE,
    )
    print(f"config-agent запущен: {AGENT_NAME}")


def stop_agent():
    docker("rm", "-f", AGENT_NAME, check=False)
    print("config-agent остановлен")


def find_app(mcp, timeout=60):
    deadline = time.time() + timeout
    while time.time() < deadline:
        text, _ = mcp.call("list_applications", {})
        try:
            apps = json.loads(text)
        except ValueError:
            apps = []
        for app in apps:
            if app.get("name") == APP:
                return app["id"]
        time.sleep(2)
    raise RuntimeError(f"агент не зарегистрировал приложение {APP} за {timeout}s")


def wait_task(mcp, task_id, timeout=120):
    deadline = time.time() + timeout
    while time.time() < deadline:
        text, _ = mcp.call("get_task", {"task_id": task_id})
        task = json.loads(text)
        if task["status"] in ("completed", "failed"):
            return task
        time.sleep(2)
    raise RuntimeError(f"задача {task_id} не завершилась за {timeout}s")


def call_json(mcp, tool, args):
    text, is_error = mcp.call(tool, args)
    if is_error:
        raise RuntimeError(f"{tool}: {text}")
    return json.loads(text)


def container_running(app):
    out = docker("ps", "--filter", f"name={app}", "--format", "{{.Names}}").stdout
    return app in out


def wait_file_contains(path, needle, timeout=40):
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            if needle in open(path).read():
                return True
        except OSError:
            pass
        time.sleep(2)
    return False


CHECKS = []


def check(name, ok):
    CHECKS.append((name, ok))
    print(f"  [{'PASS' if ok else 'FAIL'}] {name}")


def main():
    mcp = MCP()
    try:
        app_id = find_app(mcp)
        print(f"Приложение найдено: {APP} (app_id={app_id})")

        # 1. Изменение конфигурации через MCP и доставка агентом.
        call_json(mcp, "set_env", {"app_id": app_id, "filename": ".env", "content": "E2E=1\n"})
        call_json(mcp, "set_compose", {"app_id": app_id, "content": UPDATED_COMPOSE})
        check("set_env/set_compose приняты сервером", True)
        check("compose доставлен агентом на диск",
              wait_file_contains(os.path.join(APP_DIR, "docker-compose.yml"), "7200"))

        # 2. Deploy.
        ack = call_json(mcp, "deploy", {"app_id": app_id, "wait": True})
        task = ack.get("task", {})
        check("deploy завершён (completed)", task.get("status") == "completed")
        check("контейнер demo_e2e поднят", container_running(APP))
        print(f"     вывод deploy: {task.get('output','')[:160].strip()}")

        # 3. Restart.
        ack = call_json(mcp, "restart", {"app_id": app_id, "wait": True})
        check("restart завершён (completed)", ack.get("task", {}).get("status") == "completed")
        check("контейнер жив после restart", container_running(APP))

        # 4. Down.
        ack = call_json(mcp, "down", {"app_id": app_id, "confirm": True, "wait": True})
        check("down завершён (completed)", ack.get("task", {}).get("status") == "completed")
        time.sleep(2)
        check("контейнер остановлен", not container_running(APP))

        # 5. Защита без confirm.
        text, is_error = mcp.call("down", {"app_id": app_id})
        check("down без confirm отклонён", is_error)

    finally:
        mcp.close()
        stop_agent()

    failed = [name for name, ok in CHECKS if not ok]
    print(f"\nE2E: {len(CHECKS) - len(failed)}/{len(CHECKS)} проверок пройдено")
    if failed:
        print("Провалено: " + "; ".join(failed))
        sys.exit(1)


if __name__ == "__main__":
    prepare_app_dir()
    start_agent()
    main()