balanceboard/tests/conftest.py

"""Shared pytest fixtures for BalanceBoard.

The app fixture runs against an in-memory SQLite database (no Postgres
required) by monkeypatching ``database.init_db``. Filter-engine and
polling-service singletons are stubbed so request handlers do not start
background threads or hit the real filter pipeline during API contract
tests.
"""

import pytest
from sqlalchemy.pool import StaticPool


def _sqlite_init_db(app):
    """Replace the Postgres init_db with an in-memory SQLite setup."""
    from database import db

    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
    # A single in-memory DB shared across the test's connections.
    app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {"poolclass": StaticPool}
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    db.init_app(app)
    with app.app_context():
        db.create_all()


@pytest.fixture
def app(monkeypatch):
    import database
    from app import create_app

    monkeypatch.setattr(database, "init_db", _sqlite_init_db)
    application = create_app()
    application.config.update(TESTING=True)
    return application


@pytest.fixture
def client(app):
    return app.test_client()


class _StubPolling:
    """No-op polling service so ``before_request`` does not start threads."""

    def init_app(self, app):
        pass

    def start(self):
        pass


class _StubFilterEngine:
    """Passthrough filter engine for API contract tests."""

    class _Config:
        def get_filterset(self, name):
            return None

    config = _Config()

    def apply_filterset(self, posts, filterset_name="no_filter", use_cache=True):
        for p in posts:
            p.setdefault("_filter_score", 0.5)
            p.setdefault("_filter_categories", [])
            p.setdefault("_filter_tags", [])
        return posts

    def filter_comments(self, comments, filterset_name="no_filter"):
        return comments

    def get_available_filtersets(self):
        return ["no_filter"]


@pytest.fixture
def stub_services(app, monkeypatch):
    """Patch the app's lazy service accessors so requests stay hermetic."""
    import app as app_module

    monkeypatch.setattr(app_module, "get_filter_engine", lambda: _StubFilterEngine())
    monkeypatch.setattr(app_module, "get_polling_service", lambda: _StubPolling())
    return app


class StubPostService:
    """In-memory post/comment store for /api/v1 contract tests."""

    def __init__(self, posts=None, comments=None):
        self._posts = posts or {}
        self._comments = comments or {}

    def load(self):
        return self._posts, self._comments

    @staticmethod
    def build_comment_tree(comments):
        comment_dict = {c["uuid"]: {**c, "replies": []} for c in comments}
        roots = []
        for c in comments:
            parent = c.get("parent_comment_uuid")
            if parent and parent in comment_dict:
                comment_dict[parent]["replies"].append(comment_dict[c["uuid"]])
            else:
                roots.append(comment_dict[c["uuid"]])
        return roots

    def source_counts(self):
        return {}

    def latest_content_mtime(self):
        return 0