ss-tools/backend/tests/test_app_middleware.py

# #region Test.AppModule.Middleware [C:3] [TYPE Module] [SEMANTICS test,app,middleware,hsts,cors,logging]
# @BRIEF Tests for app.py — middleware: HSTS, log_requests, CORS, Session, TraceContext, router registration.
# @RELATION BINDS_TO -> [AppModule]
# @TEST_EDGE: hsts_enabled -> header set when FORCE_HTTPS=true
# @TEST_EDGE: hsts_disabled -> no header when FORCE_HTTPS=false
# @TEST_EDGE: polling_endpoint_no_log -> polling /api/tasks GET does not log
# @TEST_INVARIANT: hsts_gate -> VERIFIED_BY: test_hsts_middleware_enabled, test_hsts_middleware_disabled

from pathlib import Path
import sys
import os

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import Request, HTTPException


# #region _make_mock_request [C:1] [TYPE Function]
def _make_mock_request(method="GET", path="/api/test", client_host="127.0.0.1"):
    req = MagicMock(spec=Request)
    req.method = method
    req.url = MagicMock()
    req.url.path = path
    req.client = MagicMock()
    req.client.host = client_host
    req.query_params = {}
    return req
# #endregion _make_mock_request


class TestHSTSMiddleware:
    """HSTSMiddleware — Strict-Transport-Security header."""

    # #region test_hsts_enabled [C:2] [TYPE Function]
    @patch.dict(os.environ, {"FORCE_HTTPS": "true"}, clear=False)
    @pytest.mark.asyncio
    async def test_hsts_enabled(self):
        from src.app import HSTSMiddleware
        async def call_next(r):
            resp = MagicMock(); resp.headers = {}; return resp
        mw = HSTSMiddleware(MagicMock())
        resp = await mw.dispatch(_make_mock_request(), call_next)
        assert resp.headers.get("Strict-Transport-Security") == "max-age=31536000; includeSubDomains"
    # #endregion test_hsts_enabled

    # #region test_hsts_disabled [C:2] [TYPE Function]
    @patch.dict(os.environ, {"FORCE_HTTPS": "false"}, clear=False)
    @pytest.mark.asyncio
    async def test_hsts_disabled(self):
        from src.app import HSTSMiddleware
        async def call_next(r):
            resp = MagicMock(); resp.headers = {}; return resp
        mw = HSTSMiddleware(MagicMock())
        assert mw._enabled is False
        resp = await mw.dispatch(_make_mock_request(), call_next)
        assert "Strict-Transport-Security" not in resp.headers
    # #endregion test_hsts_disabled

    # #region test_hsts_empty_env [C:2] [TYPE Function]
    @patch.dict(os.environ, {}, clear=False)
    def test_hsts_empty_env(self):
        os.environ.pop("FORCE_HTTPS", None)
        from src.app import HSTSMiddleware
        mw = HSTSMiddleware(MagicMock())
        assert mw._enabled is False
    # #endregion test_hsts_empty_env


class TestLogRequests:
    """log_requests — HTTP request/response logging."""

    # #region test_log_non_polling [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_log_non_polling(self):
        from src.app import log_requests
        async def call_next(r):
            resp = MagicMock(); resp.status_code = 200; return resp
        resp = await log_requests(_make_mock_request(path="/api/dashboards"), call_next)
        assert resp.status_code == 200
    # #endregion test_log_non_polling

    # #region test_log_polling_skipped [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_log_polling_skipped(self):
        from src.app import log_requests
        async def call_next(r):
            resp = MagicMock(); resp.status_code = 200; return resp
        resp = await log_requests(_make_mock_request(method="GET", path="/api/tasks"), call_next)
        assert resp.status_code == 200
    # #endregion test_log_polling_skipped

    # #region test_log_post_polling_not_skipped [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_log_post_polling_not_skipped(self):
        from src.app import log_requests
        async def call_next(r):
            resp = MagicMock(); resp.status_code = 201; return resp
        resp = await log_requests(_make_mock_request(method="POST", path="/api/tasks"), call_next)
        assert resp.status_code == 201
    # #endregion test_log_post_polling_not_skipped

    # #region test_log_network_error [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_log_network_error(self):
        from src.app import log_requests
        from src.core.utils.network import NetworkError
        async def call_next(r):
            raise NetworkError("timeout")
        with pytest.raises(HTTPException) as e:
            await log_requests(_make_mock_request(path="/api/dashboards"), call_next)
        assert e.value.status_code == 503
    # #endregion test_log_network_error


class TestLogRequestsStderrFallback:
    """log_requests — stderr JSON dump when logger level > INFO."""

    # #region test_stderr_fallback_req [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_stderr_fallback_req(self):
        from src.app import log_requests
        async def call_next(r):
            resp = MagicMock(); resp.status_code = 200; return resp
        with patch("src.app.logger.isEnabledFor", return_value=False):
            resp = await log_requests(_make_mock_request(path="/api/test-stderr"), call_next)
        assert resp.status_code == 200
    # #endregion test_stderr_fallback_req

    # #region test_stderr_fallback_resp [C:2] [TYPE Function]
    @pytest.mark.asyncio
    async def test_stderr_fallback_resp(self):
        from src.app import log_requests
        async def call_next(r):
            resp = MagicMock(); resp.status_code = 500; return resp
        with patch("src.app.logger.isEnabledFor", return_value=False):
            resp = await log_requests(_make_mock_request(path="/api/test-stderr-resp"), call_next)
        assert resp.status_code == 500
    # #endregion test_stderr_fallback_resp


class TestMiddlewareConfig:
    """Middleware registration — Session, CORS, HSTS, TraceContext."""

    def test_session_middleware(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "SessionMiddleware" in names

    def test_cors_middleware(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "CORSMiddleware" in names

    def test_hsts_middleware(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "HSTSMiddleware" in names

    def test_trace_context_middleware(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "TraceContextMiddleware" in names


class TestCorsConfig:
    """CORS and Session configuration edge cases."""

    def test_cors_empty_allowed_origins(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "CORSMiddleware" in names

    def test_session_secret_fallback(self):
        from src.app import app
        names = [m.cls.__name__ for m in app.user_middleware]
        assert "SessionMiddleware" in names


class TestRouterRegistration:
    """Verify all expected route groups are registered."""

    def test_expected_routers_registered(self):
        from src.app import app
        routes = "\n".join(r.path for r in app.routes)
        assert "/api/dashboards" in routes or "/api/plugins" in routes
        assert "/api/tasks" in routes
        assert "/api/settings" in routes
        assert "/api/git" in routes or "/api/mappings" in routes
        assert "/ws/logs/" in routes
        assert "/ws/task-events" in routes
# #endregion Test.AppModule.Middleware