#!/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)