odysseus/tests/test_caldav_test_connection_ssl.py

"""Regression: CalDAV test_connection must trust the operator's CA bundle.

The pre-flight used httpx with trust_env=False, which ignored
SSL_CERT_FILE/REQUESTS_CA_BUNDLE. Self-signed CalDAV servers that the
real sync accepts (via caldav lib -> requests -> honors bundle) were
rejected by the test with CERTIFICATE_VERIFY_FAILED.

These tests exercise the *route handler* directly (via ASGI TestClient)
and capture the verify= kwarg passed to httpx.AsyncClient, ensuring the
route code — not a test-side duplicate — builds the SSL context correctly.
"""
import os
import ssl
import sys
from unittest.mock import MagicMock, patch

import httpx
import pytest

# No module-level sys.modules stubbing here: conftest pre-imports the real
# sqlalchemy/core.database, and stubbing extras (e.g. caldav) at collection
# time leaks MagicMocks into later tests in the same process — it made
# test_caldav_redirect_hardening's real DAVClient a mock that never sent
# the PROPFIND. The route's lazy imports are patched per-request instead.


def _fake_response(status_code=207, headers=None):
    resp = MagicMock()
    resp.status_code = status_code
    resp.headers = headers or {}
    return resp


@pytest.fixture()
def client():
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from routes.calendar_routes import setup_calendar_routes

    with patch("routes.calendar_routes._require_user", return_value="test-owner"):
        router = setup_calendar_routes()
    app = FastAPI()
    app.include_router(router)
    return TestClient(app)


def _make_fake_async_client(captured):
    """Return a fake httpx.AsyncClient class that captures constructor kwargs."""
    class FakeAsyncClient:
        def __init__(self, **kwargs):
            captured.update(kwargs)

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            pass

        async def request(self, *a, **kw):
            return _fake_response(207)

    return FakeAsyncClient


def _post_test(client, captured, env=None):
    """POST /api/calendar/test with credentials in body so no DB lookup needed.

    Patches httpx.AsyncClient at the real module level so the route's
    ``import httpx; httpx.AsyncClient(...)`` picks up the fake class.
    Also stubs validate_caldav_url (lazy-imported from src.caldav_sync).
    """
    fake_cls = _make_fake_async_client(captured)

    # Stub the caldav_sync module so the lazy `from src.caldav_sync import validate_caldav_url`
    # inside the route body resolves to a pass-through.
    caldav_sync_stub = MagicMock()
    caldav_sync_stub.validate_caldav_url = lambda u: u

    ctx_managers = [
        patch.object(httpx, "AsyncClient", fake_cls),
        patch.dict(sys.modules, {"src.caldav_sync": caldav_sync_stub}),
        patch("routes.calendar_routes._require_user", return_value="test-owner"),
    ]
    if env is not None:
        ctx_managers.append(patch.dict(os.environ, env))

    # Enter all context managers
    for cm in ctx_managers:
        cm.__enter__()
    try:
        return client.post(
            "/api/calendar/test",
            json={"url": "https://cal.example.com", "username": "u", "password": "p"},
        )
    finally:
        for cm in reversed(ctx_managers):
            cm.__exit__(None, None, None)


# ---------------------------------------------------------------------------
# Route-level tests
# ---------------------------------------------------------------------------

def test_route_passes_ssl_context_with_correct_flags(client):
    """The route must pass an ssl.SSLContext to httpx.AsyncClient(verify=...)
    with trust_env=False, follow_redirects=False, and VERIFY_X509_STRICT cleared."""
    captured = {}
    resp = _post_test(client, captured)

    assert resp.status_code == 200
    assert isinstance(captured.get("verify"), ssl.SSLContext), (
        f"verify= should be an ssl.SSLContext, got {type(captured.get('verify'))}"
    )
    assert captured.get("trust_env") is False
    assert captured.get("follow_redirects") is False
    ctx = captured["verify"]
    assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT), (
        "VERIFY_X509_STRICT must be cleared for self-signed CA compat"
    )


def test_route_ssl_cert_file_takes_precedence(client, tmp_path):
    """SSL_CERT_FILE is the exact bundle loaded when both variables are set."""
    bundle_a = tmp_path / "ssl-cert-file.pem"
    bundle_b = tmp_path / "requests-ca-bundle.pem"
    bundle_a.write_text("ssl-cert-file", encoding="utf-8")
    bundle_b.write_text("requests-ca-bundle", encoding="utf-8")

    loaded = []

    class FakeSSLContext:
        def __init__(self):
            self.verify_flags = ssl.VERIFY_X509_STRICT

        def load_verify_locations(self, cafile=None, capath=None, cadata=None):
            loaded.append(
                {
                    "cafile": cafile,
                    "capath": capath,
                    "cadata": cadata,
                }
            )

    ssl_context = FakeSSLContext()
    captured = {}
    env = {
        "SSL_CERT_FILE": str(bundle_a),
        "REQUESTS_CA_BUNDLE": str(bundle_b),
    }

    with patch.object(
        ssl,
        "create_default_context",
        return_value=ssl_context,
    ):
        resp = _post_test(client, captured, env=env)

    assert resp.status_code == 200
    assert resp.json() == {"ok": True}
    assert loaded == [
        {
            "cafile": str(bundle_a),
            "capath": None,
            "cadata": None,
        }
    ]
    assert captured.get("verify") is ssl_context
    assert captured.get("trust_env") is False
    assert captured.get("follow_redirects") is False
    assert not (
        ssl_context.verify_flags & ssl.VERIFY_X509_STRICT
    )


def test_route_missing_bundle_does_not_crash(client):
    """A nonexistent CA bundle path must not crash -- fall back to system CAs."""
    captured = {}
    resp = _post_test(client, captured, env={"SSL_CERT_FILE": "/nonexistent/ca-bundle.pem"})

    assert resp.status_code == 200
    ctx = captured["verify"]
    assert isinstance(ctx, ssl.SSLContext)
    assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT)


def test_route_empty_env_vars_use_system_defaults(client):
    """Empty SSL_CERT_FILE and REQUESTS_CA_BUNDLE should not crash."""
    captured = {}
    resp = _post_test(client, captured, env={"SSL_CERT_FILE": "", "REQUESTS_CA_BUNDLE": ""})

    assert resp.status_code == 200
    ctx = captured["verify"]
    assert isinstance(ctx, ssl.SSLContext)
    assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT)