homelab-codex-ws/services/agent-system/webui/tests/test_web.py

"""Tests for the agent-system webui's actions mirror + mutate proxy.

Covers the fix for the Action Queue panel always showing 0 pending: when
CONTROL_PLANE_URL is set (this webui is a read-only mirror of VPS, e.g. on
PIHA), current_actions() must read the materializer's mirrored
world/actions.json instead of the local (always-empty) ACTIONS_DIR, and
mutate_action() must proxy approve/reject to the VPS control-plane instead
of writing to the local ACTIONS_DIR, which the real executor never reads.
"""
from __future__ import annotations

import json
import sys
import urllib.error
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent.parent))
import web as web_module


class _FakeResponse:
    def __init__(self, status):
        self.status = status

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def _setup_dirs(tmp_path, monkeypatch):
    world = tmp_path / "world"
    actions = tmp_path / "actions"
    world.mkdir()
    actions.mkdir()
    monkeypatch.setattr(web_module, "WORLD_DIR", world)
    monkeypatch.setattr(web_module, "ACTIONS_DIR", actions)
    return world, actions


def test_current_actions_reads_mirror_when_control_plane_url_set(tmp_path, monkeypatch):
    world, actions = _setup_dirs(tmp_path, monkeypatch)
    monkeypatch.setattr(web_module, "CONTROL_PLANE_URL", "http://100.95.58.48:18180")

    mirrored = {"pending": [{"id": "a1", "status": "pending", "type": "redeploy"}],
                "approved": [], "running": [], "completed": [], "failed": [], "rejected": []}
    (world / "actions.json").write_text(json.dumps(mirrored))

    # A stray local pending file must NOT leak into the mirrored result — VPS
    # is the sole source of truth once mirroring is on.
    local_pending = actions / "pending"
    local_pending.mkdir()
    (local_pending / "local-only.json").write_text(json.dumps({"type": "alert_only"}))

    result = web_module.current_actions()

    assert result == mirrored
    assert len(result["pending"]) == 1
    assert result["pending"][0]["id"] == "a1"


def test_current_actions_falls_back_to_local_scan_without_control_plane_url(tmp_path, monkeypatch):
    world, actions = _setup_dirs(tmp_path, monkeypatch)
    monkeypatch.setattr(web_module, "CONTROL_PLANE_URL", "")

    pending_dir = actions / "pending"
    pending_dir.mkdir()
    (pending_dir / "act-1.json").write_text(json.dumps({"type": "redeploy"}))

    result = web_module.current_actions()

    assert len(result["pending"]) == 1
    assert result["pending"][0]["id"] == "act-1"
    assert result["pending"][0]["status"] == "pending"


def test_mutate_action_proxies_to_control_plane_when_set(tmp_path, monkeypatch):
    _setup_dirs(tmp_path, monkeypatch)
    monkeypatch.setattr(web_module, "CONTROL_PLANE_URL", "http://100.95.58.48:18180")

    captured = {}

    def fake_urlopen(req, timeout=10):
        captured["url"] = req.full_url
        captured["body"] = json.loads(req.data.decode("utf-8"))
        return _FakeResponse(200)

    monkeypatch.setattr(web_module.urllib.request, "urlopen", fake_urlopen)

    success, msg = web_module.mutate_action("a1", "approved")

    assert success is True
    assert captured["url"] == "http://100.95.58.48:18180/action/mutate"
    assert captured["body"] == {"id": "a1", "status": "approved"}


def test_mutate_action_proxy_failure_reports_error(tmp_path, monkeypatch):
    _setup_dirs(tmp_path, monkeypatch)
    monkeypatch.setattr(web_module, "CONTROL_PLANE_URL", "http://100.95.58.48:18180")

    def fake_urlopen(req, timeout=10):
        raise urllib.error.URLError("connection refused")

    monkeypatch.setattr(web_module.urllib.request, "urlopen", fake_urlopen)

    success, msg = web_module.mutate_action("a1", "approved")

    assert success is False
    assert "100.95.58.48:18180" in msg


def test_mutate_action_local_when_no_control_plane_url(tmp_path, monkeypatch):
    world, actions = _setup_dirs(tmp_path, monkeypatch)
    monkeypatch.setattr(web_module, "CONTROL_PLANE_URL", "")

    pending_dir = actions / "pending"
    pending_dir.mkdir()
    (pending_dir / "act-1.json").write_text(json.dumps({"status": "pending"}))

    success, msg = web_module.mutate_action("act-1", "approved")

    assert success is True
    assert not (pending_dir / "act-1.json").exists()
    approved_data = json.loads((actions / "approved" / "act-1.json").read_text())
    assert approved_data["status"] == "approved"