vacuum-wall/tests/test_server.py

import os
from pathlib import Path
from unittest.mock import patch

import pytest

from webui.server import _has_permission, _subsystem_from_path


@pytest.fixture
def client():
    with patch("lib.logging.setup_logging"):
        from webui.server import app

        app.config["TESTING"] = True
        return app.test_client()


class TestSPARoutes:
    def test_root_serves_index(self, client):
        resp = client.get("/")
        assert resp.status_code == 200
        assert b'id="app"' in resp.data

    def test_spa_unknown_path_404(self, client):
        resp = client.get("/dashboard")
        assert resp.status_code == 404

    def test_spa_unknown_path_404_other(self, client):
        resp = client.get("/zones")
        assert resp.status_code == 404

    def test_api_routes_still_work(self, client):
        resp = client.get("/api/firewall/zones")
        assert resp.status_code in (401, 500)
        data = resp.get_json()
        assert data is not None
        assert data.get("ok") is False
        assert data.get("error") == "unauthorized"


class TestSpaRoot:
    def test_serves_index_html_as_is(self, client):
        resp = client.get("/")
        assert resp.status_code == 200
        assert b"/static/app.js" in resp.data

    def test_no_ws_url_substitution(self, client):
        """index.html is served verbatim — no WS URL placeholder substitution."""
        resp = client.get("/", headers={"Host": "192.168.1.1:9090"})
        assert b"__WS_URL_PLACEHOLDER__" not in resp.data
        assert b"ws://" not in resp.data


class TestBlueprintsRegistered:
    def test_all_blueprints_registered(self, client):
        from webui.server import BLUEPRINTS

        assert len(BLUEPRINTS) == 9
        names = [name for name, _ in BLUEPRINTS]
        assert "auth" in names
        assert "firewall" in names
        assert "network" in names
        assert "dhcp" in names
        assert "proxy" in names
        assert "certs" in names
        assert "wireguard" in names
        assert "logs" in names


class TestGroupWriteHandler:
    def test_creates_file_with_group_write(self, tmp_path: Path) -> None:
        """GroupWriteHandler creates new log files with group-write (0o664)."""
        import contextlib
        from logging.handlers import RotatingFileHandler

        log_file = tmp_path / "test.log"
        old = os.umask(0o022)
        try:

            class GroupWriteHandler(RotatingFileHandler):
                def _open(self):
                    with contextlib.suppress(OSError):
                        os.chmod(self.baseFilename, 0o664)
                    saved = os.umask(0o002)
                    try:
                        fd = os.open(
                            self.baseFilename,
                            os.O_WRONLY | os.O_CREAT | os.O_APPEND,
                            0o664,
                        )
                    finally:
                        os.umask(saved)
                    return os.fdopen(fd, "a", errors="backslashreplace")

            fh = GroupWriteHandler(str(log_file))
            fh.close()
        finally:
            os.umask(old)

        mode = os.stat(log_file).st_mode & 0o777
        assert mode == 0o664, f"Expected 0o664, got {oct(mode)}"


class TestCSPHeaders:
    def test_csp_header_on_root(self, client):
        resp = client.get("/")
        csp = resp.headers.get("Content-Security-Policy")
        assert csp is not None
        assert "default-src 'self'" in csp
        assert "script-src 'self'" in csp
        assert "'unsafe-inline'" not in csp.split("script-src")[1].split(";")[0]

    def test_x_content_type_options(self, client):
        resp = client.get("/")
        assert resp.headers.get("X-Content-Type-Options") == "nosniff"

    def test_frame_ancestors_none(self, client):
        resp = client.get("/")
        csp = resp.headers.get("Content-Security-Policy")
        assert csp is not None
        assert "frame-ancestors 'none'" in csp


class TestSessionIdAuth:
    """Test session_id binding in auth middleware."""

    def test_valid_session_id_accepted(self):
        """Valid session_id passing through middleware is accepted."""
        assert _has_permission({"firewall": "rw"}, "firewall", "POST") is True

    def test_permission_extraction(self):
        """Subsystem name extracted correctly from path and checked against token perms."""
        sub = _subsystem_from_path("/api/firewall/zones")
        assert sub == "firewall"

        perms = {"firewall": "read"}
        assert _has_permission(perms, "firewall", "GET") is True
        assert _has_permission(perms, "firewall", "POST") is False