utun2/tools/chat_integration_test.py

#!/usr/bin/env python3
"""
chat_integration_test.py — integration test: 2 chat nodes over UDP, sync messages.

Запускает два процесса utun в фоне, создаёт канал, пишет 4 сообщения,
подключает второй узел по invite-ссылке, проверяет members и messages.

Usage:
    python3 tools/chat_integration_test.py
"""

import asyncio
import json
import os
import socket
import sys
import tempfile
import time

sys.path.insert(0, os.path.dirname(__file__))
from chat_client import ChatClient, ChatClientError

UTUN_BIN = os.path.join(os.path.dirname(__file__), "..", "src", "utun")
READY_TIMEOUT = 3.0       # max wait for utun ready
SYNC_TIMEOUT = 8.0        # max wait for join + sync
REQUEST_TIMEOUT = 5.0     # per-request timeout
TOTAL_TIMEOUT = 25.0      # overall test timeout


def find_free_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


def write_config(path, etcp_port, ctrl_port, db_subdir):
    content = f"""[global]
db_path={db_subdir}

[server: srv]
addr=127.0.0.1:{etcp_port}
type=public

[chatserver]
db_path={db_subdir}
headless_control_bind=127.0.0.1:{ctrl_port}

[allowed_keys]
allow_all=1
"""
    with open(path, "w") as f:
        f.write(content)


async def kill_proc(proc, label):
    if proc is None or proc.returncode is not None:
        return
    try:
        proc.terminate()
        try:
            await asyncio.wait_for(proc.wait(), timeout=3.0)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
    except ProcessLookupError:
        pass


async def wait_ready(cli, timeout=READY_TIMEOUT):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            await cli.ping()
            return True
        except ChatClientError:
            await asyncio.sleep(0.1)
    return False


def check(name, expr, detail=""):
    if not expr:
        detail = f" ({detail})" if detail else ""
        raise AssertionError(f"FAIL: {name}{detail}")
    print(f"  OK: {name}")


async def main():
    etcp_a = find_free_port()
    etcp_b = find_free_port()
    ctrl_a = find_free_port()
    ctrl_b = find_free_port()

    print(f"ports: etcp={etcp_a},{etcp_b}  ctrl={ctrl_a},{ctrl_b}")

    tmpdir = tempfile.mkdtemp(prefix="utun_chat_test_")
    db_a = os.path.join(tmpdir, "db_a")
    db_b = os.path.join(tmpdir, "db_b")
    os.makedirs(db_a, exist_ok=True)
    os.makedirs(db_b, exist_ok=True)

    config_a = os.path.join(tmpdir, "a.conf")
    config_b = os.path.join(tmpdir, "b.conf")
    write_config(config_a, etcp_a, ctrl_a, db_a)
    write_config(config_b, etcp_b, ctrl_b, db_b)

    proc_a = None
    proc_b = None

    try:
        # ── Start utun processes ──
        print("\n--- Starting utun ---")
        log_a = os.path.join(tmpdir, "utun_a.log")
        log_b = os.path.join(tmpdir, "utun_b.log")
        pid_a = os.path.join(tmpdir, "utun_a.pid")
        pid_b = os.path.join(tmpdir, "utun_b.pid")

        proc_a = await asyncio.create_subprocess_exec(
            UTUN_BIN, "-f", "-p", pid_a, "-l", log_a, "-c", config_a,
            stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
        )
        proc_b = await asyncio.create_subprocess_exec(
            UTUN_BIN, "-f", "-p", pid_b, "-l", log_b, "-c", config_b,
            stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
        )

        print(f"  proc_a pid={proc_a.pid}  proc_b pid={proc_b.pid}")

        await asyncio.sleep(0.5)

        # ── Open connections ──
        print("\n--- Connecting to headless control ---")
        async with ChatClient(port=ctrl_a, timeout=REQUEST_TIMEOUT) as cli_a:
            if not await wait_ready(cli_a):
                raise RuntimeError(f"node A not ready after {READY_TIMEOUT}s")
            print(f"  node A ready")

            async with ChatClient(port=ctrl_b, timeout=REQUEST_TIMEOUT) as cli_b:
                if not await wait_ready(cli_b):
                    raise RuntimeError(f"node B not ready after {READY_TIMEOUT}s")
                print(f"  node B ready")

                # ── Create channel ──
                print("\n--- Create channel ---")
                await cli_a.create_channel("TestGroup")
                channels = await cli_a.channels()
                check("channel created", isinstance(channels, list) and len(channels) == 1,
                      f"channels={json.dumps(channels)}")
                ch_id = str(channels[0]["id"])
                print(f"  channel_id={ch_id}  name={channels[0]['name']}")

                # ── Write 4 messages ──
                print("\n--- Write messages ---")
                for i in range(4):
                    await cli_a.send(ch_id, f"Message {i+1}")

                msgs_a = await cli_a.messages(ch_id, count=10)
                check("4 messages on A", isinstance(msgs_a, list) and len(msgs_a) == 4,
                      f"got {len(msgs_a) if isinstance(msgs_a, list) else '?'}")
                for i, m in enumerate(msgs_a):
                    print(f"  [#{i+1}] ts={m.get('ts','?')} author={m.get('author_id','?')}")

                # ── Invite ──
                print("\n--- Invite link ---")
                invite = await cli_a.invite(ch_id)
                link = invite.get("link", "")
                check("invite link", isinstance(link, str) and link.startswith("utun://"),
                      f"link={'...' + link[-20:] if link else 'MISSING'}")
                print(f"  link={link[:80]}...")

                # ── Connect node B ──
                print("\n--- Connect node B ---")
                conn = await cli_b.connect_channel(link)
                check("connect accepted", isinstance(conn, dict) and conn.get("connecting"),
                      f"resp={json.dumps(conn)}")
                print(f"  connecting={conn.get('connecting')}")

                # Wait for join + sync
                print(f"\n--- Wait sync (max {SYNC_TIMEOUT}s) ---")
                deadline = time.monotonic() + SYNC_TIMEOUT
                member_count = 0
                while time.monotonic() < deadline:
                    try:
                        members = await cli_b.members(ch_id)
                        member_count = len(members) if isinstance(members, list) else 0
                        if member_count >= 2:
                            break
                    except ChatClientError:
                        pass
                    await asyncio.sleep(0.15)
                check("2 members on B", member_count >= 2,
                      f"got {member_count} members after {SYNC_TIMEOUT}s")

                # Small pause to let db_sync fully settle
                await asyncio.sleep(0.5)

                # ── Verify members on B ──
                members = await cli_b.members(ch_id)
                print(f"\n--- Members on B ({len(members)}) ---")
                for m in members:
                    connected = "✓" if m.get("connected") else "✗"
                    online = "✓" if m.get("online") else "✗"
                    print(f"  {m['node_id']} name={m.get('name','?')} online={online} connected={connected}")
                check("owner online", any(m.get("online") for m in members),
                      f"members={json.dumps(members)}")
                check("owner connected", any(m.get("connected") for m in members),
                      f"members={json.dumps(members)}")

                # ── Verify messages on B ──
                if proc_b.returncode is not None:
                    raise RuntimeError(f"node B died with code {proc_b.returncode}")
                print(f"\n--- Messages on B ---")
                try:
                    msgs_b = await cli_b.messages(ch_id, count=10)
                except ChatClientError as e:
                    print(f"  messages error: {e}")
                    print(f"  proc_b.returncode={proc_b.returncode}")
                    raise
                check("4 messages on B", isinstance(msgs_b, list) and len(msgs_b) == 4,
                      f"got {len(msgs_b) if isinstance(msgs_b, list) else '?'}")
                for i, m in enumerate(msgs_b):
                    print(f"  [#{i+1}] ts={m.get('ts','?')} author={m.get('author_id','?')}")

                # ── Verify members on A ──
                members_a = await cli_a.members(ch_id)
                print(f"\n--- Members on A ({len(members_a)}) ---")
                for m in members_a:
                    print(f"  {m['node_id']} name={m.get('name','?')} online={m.get('online')} connected={m.get('connected')}")
                check("2 members on A", isinstance(members_a, list) and len(members_a) >= 2,
                      f"got {len(members_a) if isinstance(members_a, list) else '?'}")

        print("\n=== TEST PASSED ===")
        return 0

    except Exception as e:
        print(f"\n=== TEST FAILED: {e} ===", file=sys.stderr)
        import traceback
        traceback.print_exc()
        return 1

    finally:
        print("\n--- Cleanup ---")
        await kill_proc(proc_a, "proc_a")
        await kill_proc(proc_b, "proc_b")

        for f in [config_a, config_b]:
            try: os.unlink(f)
            except OSError: pass
        try: os.rmdir(db_a)
        except OSError: pass
        try: os.rmdir(db_b)
        except OSError: pass
        try: os.rmdir(tmpdir)
        except OSError: pass

        print(f"  temp dir cleaned: {tmpdir}")


if __name__ == "__main__":
    async def _run():
        try:
            return await asyncio.wait_for(main(), timeout=TOTAL_TIMEOUT)
        except asyncio.TimeoutError:
            print(f"\n=== TEST FAILED: total timeout {TOTAL_TIMEOUT}s ===", file=sys.stderr)
            return 1
    sys.exit(asyncio.run(_run()))