proofpoint-quarantine-portal/tests/conftest.py

"""Shared test fixtures.

Isolation is the top priority: no test may touch the real config.toml or start a worker
thread. A session-scoped autouse fixture points PPSQ_CONFIG at a temp copy of
config.example.toml BEFORE any app module is imported, and ConfigStore itself refuses to
open the default path under pytest.
"""

from __future__ import annotations

import os
import shutil
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
EXAMPLE = ROOT / "config.example.toml"

# http base_url in the example would fail https validation; allow it for tests.
os.environ.setdefault("PPSQ_ALLOW_INSECURE", "1")


def _write_config(dest: Path) -> Path:
    """Copy the example config and make it bootable for tests: strong secret, dev URL,
    static auth with a known dev account."""
    text = EXAMPLE.read_text()
    text = text.replace(
        'secret_key = "change-me-to-a-random-string"',
        'secret_key = "test-secret-key-that-is-definitely-long-enough-xx"',
    )
    text = text.replace(
        'base_url = "https://pps.example.com:10000"',
        'base_url = "http://pps.example.invalid:10000"',
    )
    text = text.replace('mode = "oidc"', 'mode = "static"')
    # Known dev credentials for the static-login tests.
    text = text.replace('username = "admin"\npassword = "admin"',
                        'username = "dev"\npassword = "devpass"')
    text = text.replace('email = "dev-admin@example.invalid"',
                        'email = "dev@example.invalid"')
    dest.write_text(text)
    return dest


@pytest.fixture(scope="session", autouse=True)
def _isolate_config(tmp_path_factory):
    cfg = tmp_path_factory.mktemp("cfg") / "config.toml"
    _write_config(cfg)
    os.environ["PPSQ_CONFIG"] = str(cfg)
    yield


@pytest.fixture
def config_path(tmp_path):
    """A fresh, writable config file per test."""
    return _write_config(tmp_path / "config.toml")


@pytest.fixture
def store(config_path):
    from config_store import ConfigStore
    return ConfigStore(config_path)


class FakePPS:
    """Records every call; returns canned data for search/get_raw."""

    def __init__(self, records=None):
        self.calls = []
        self._records = records or []

    def act(self, action, folder, localguids, *, targetfolder=None, deletedfolder=None, scan=False):
        self.calls.append(
            {"m": "act", "action": action, "folder": folder,
             "localguids": list(localguids), "targetfolder": targetfolder,
             "deletedfolder": deletedfolder, "scan": scan}
        )
        return {"status": "ok"}

    def search(self, folder, query, limit=200, days_back=7, enddate=None):
        self.calls.append({"m": "search", "folder": folder, "enddate": enddate})
        return list(self._records)

    def get_raw(self, guid):
        self.calls.append({"m": "get_raw", "guid": guid})
        return b"From: x@y\r\nSubject: t\r\n\r\nbody"

    def searches(self):
        return [c for c in self.calls if c["m"] == "search"]

    def acts(self):
        return [c for c in self.calls if c["m"] == "act"]


@pytest.fixture
def fake_pps():
    return FakePPS()


@pytest.fixture
def make_queue(tmp_path):
    """Factory: a JobQueue wired to a FakePPS and a live cfg provider."""
    from worker import JobQueue

    def _make(pps, cfg):
        return JobQueue(
            str(tmp_path / "jobs.db"),
            pps_provider=lambda: pps,
            cfg_provider=lambda: cfg,
            ops_log_path=str(tmp_path / "worker.log"),
        )

    return _make


@pytest.fixture
def app_client(store, fake_pps, tmp_path, monkeypatch):
    """A Flask test client. No worker thread — tests drive queue._process directly."""
    from app import create_app
    from prefs import PrefStore
    from worker import JobQueue

    # Point the store's PPS client at the fake.
    monkeypatch.setattr(store, "pps", lambda: fake_pps)
    queue = JobQueue(
        str(tmp_path / "jobs.db"),
        pps_provider=lambda: fake_pps,
        cfg_provider=lambda: store.snapshot().quarantine,
        ops_log_path=str(tmp_path / "worker.log"),
    )
    prefs = PrefStore(str(tmp_path / "jobs.db"))
    app = create_app(store, queue, prefs)
    app.config["TESTING"] = True
    client = app.test_client()
    client._store = store
    client._queue = queue
    client._pps = fake_pps
    return client


def login_static(client, username="dev", password="devpass"):
    """Log in via static mode and return the CSRF token."""
    client.post("/login", data={"username": username, "password": password})
    return client.get("/api/config").get_json()["csrf_token"]