test_log_streaming.py

#!/usr/bin/env python3
"""
Dry-run tests for the log streaming pipeline.

Tests:
1. Ring buffer (ws_manager) - append, batch, tail, overflow
2. WS routes import and endpoint registration
3. REST log endpoint import
4. Dashboard HTML contains log viewer elements
5. Agent log forwarding code (import + flag parsing)
"""

import sys
import os
import json
import asyncio
import importlib
import argparse
import inspect

# Ensure the project is on the path
sys.path.insert(0, os.path.dirname(__file__))

passed = 0
failed = 0


def test(name, condition, detail=""):
    global passed, failed
    if condition:
        print(f"  PASS: {name}")
        passed += 1
    else:
        print(f"  FAIL: {name} -- {detail}")
        failed += 1


# == Test 1: Ring buffer in ConnectionManager ==
print("\n=== Test 1: Ring buffer (ws_manager) ===")

from orchestrator.server.ws_manager import ConnectionManager, LOG_BUFFER_MAX_LINES

mgr = ConnectionManager()
loop = asyncio.new_event_loop()

loop.run_until_complete(mgr.append_log("test-worker-1", "line 1"))
loop.run_until_complete(mgr.append_log("test-worker-1", "line 2"))
loop.run_until_complete(mgr.append_log("test-worker-1", "line 3"))

tail = mgr.get_log_tail("test-worker-1", n=10)
test("append_log stores lines", len(tail) == 3, f"got {len(tail)}")
test("lines in order", tail == ["line 1", "line 2", "line 3"], f"got {tail}")

loop.run_until_complete(
    mgr.append_log_batch("test-worker-1", ["batch 1", "batch 2", "batch 3"])
)
tail = mgr.get_log_tail("test-worker-1", n=10)
test("batch append", len(tail) == 6, f"got {len(tail)}")
test("batch order", tail[-3:] == ["batch 1", "batch 2", "batch 3"])

tail2 = mgr.get_log_tail("test-worker-1", n=2)
test("tail limit", len(tail2) == 2, f"got {len(tail2)}")
test("tail returns last N", tail2 == ["batch 2", "batch 3"])

empty = mgr.get_log_tail("nonexistent-worker")
test("empty worker returns []", empty == [])

for i in range(LOG_BUFFER_MAX_LINES + 50):
    loop.run_until_complete(mgr.append_log("overflow-worker", f"line {i}"))
tail_overflow = mgr.get_log_tail("overflow-worker", n=LOG_BUFFER_MAX_LINES)
test(
    f"ring buffer caps at {LOG_BUFFER_MAX_LINES}",
    len(tail_overflow) == LOG_BUFFER_MAX_LINES,
    f"got {len(tail_overflow)}",
)
expected_first = f"line {LOG_BUFFER_MAX_LINES + 50 - LOG_BUFFER_MAX_LINES}"
test(
    "oldest lines dropped",
    tail_overflow[0] == expected_first,
    f"first line: {tail_overflow[0]}, expected: {expected_first}",
)

test("no subscribers initially", mgr.log_subscriber_count("test-worker-1") == 0)
loop.close()


# == Test 2: WS routes ==
print("\n=== Test 2: WS routes ===")

from orchestrator.server.routes.ws import router as ws_router

route_paths = [r.path for r in ws_router.routes]
test("/ws/worker/{worker_id} exists", "/ws/worker/{worker_id}" in route_paths)
test("/ws/dashboard exists", "/ws/dashboard" in route_paths)
test("/ws/logs/{worker_id} exists", "/ws/logs/{worker_id}" in route_paths)

from orchestrator.server.routes import ws as ws_module
test("ws_log_subscriber function exists", hasattr(ws_module, "ws_log_subscriber"))


# == Test 3: REST log endpoint ==
print("\n=== Test 3: REST log endpoint ===")

from orchestrator.server.routes.workers import router as workers_router

worker_route_paths = [r.path for r in workers_router.routes]
test(
    "logs endpoint exists",
    any("/logs" in p for p in worker_route_paths),
    f"routes: {worker_route_paths}",
)
test(
    "ssh-config endpoint exists",
    any("/ssh-config" in p for p in worker_route_paths),
    f"routes: {worker_route_paths}",
)


# == Test 4: Dashboard HTML ==
print("\n=== Test 4: Dashboard HTML ===")

from orchestrator.server.app import _render_dashboard

html = _render_dashboard()
test("log-overlay div present", "logOverlay" in html)
test("logContent div present", "logContent" in html)
test("openLogViewer function", "function openLogViewer" in html)
test("closeLogViewer function", "function closeLogViewer" in html)
test("View Logs button in worker cards", "view-logs-btn" in html)
test("ws/logs/ WS connection URL", "ws/logs/" in html)
test("fetchLogsRest fallback", "fetchLogsRest" in html)
test("Escape key handler", "Escape" in html)
test("Auto-scroll toggle", "toggleAutoScroll" in html)
test("log-line-err class (error highlighting)", "log-line-err" in html)


# == Test 5: Agent log forwarding ==
print("\n=== Test 5: Agent log forwarding ===")

from orchestrator.worker.agent import WorkerAgent, LOG_FORWARD_BATCH_SIZE

test("LOG_FORWARD_BATCH_SIZE defined", LOG_FORWARD_BATCH_SIZE > 0)
test("_start_log_forwarder method", hasattr(WorkerAgent, "_start_log_forwarder"))
test("_log_forwarder_loop method", hasattr(WorkerAgent, "_log_forwarder_loop"))
test("_send_log_batch method", hasattr(WorkerAgent, "_send_log_batch"))

parser = argparse.ArgumentParser()
parser.add_argument("--server-url", type=str, required=True)
parser.add_argument("--worker-id", type=str, required=True)
parser.add_argument("--forward-logs", type=str, default=None, metavar="LOG_FILE")
args = parser.parse_args([
    "--server-url", "http://localhost:8080",
    "--worker-id", "test-worker",
    "--forward-logs", "/root/worker.log",
])
test("--forward-logs parsed", args.forward_logs == "/root/worker.log")

sig = inspect.signature(WorkerAgent.run)
test("run() has forward_logs param", "forward_logs" in sig.parameters)


# == Summary ==
print(f"\n{'='*50}")
print(f"Results: {passed} passed, {failed} failed, {passed + failed} total")
if failed == 0:
    print("ALL TESTS PASSED")
else:
    print("SOME TESTS FAILED")
    sys.exit(1)