padelnomics/web/tests/test_feature_flags.py

"""
Tests for feature flags — migration, is_flag_enabled helper, feature_gate
decorator, admin toggle routes, and per-feature gating across all routes.

Unit tests cover is_flag_enabled and feature_gate in isolation.
Integration tests exercise full request/response flows via Quart test client.
"""

import sqlite3
from datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, patch

import pytest

from padelnomics import core
from padelnomics.migrations.migrate import migrate


# ── Fixtures & helpers ────────────────────────────────────────────


@pytest.fixture(autouse=True)
def mock_csrf_validation():
    """Mock CSRF validation to always pass in all tests in this file."""
    with patch("padelnomics.core.validate_csrf_token", return_value=True):
        yield


@pytest.fixture
async def admin_client(app, db):
    """Test client with an admin-role user session (module-level, follows test_content.py)."""
    now = datetime.utcnow().isoformat()
    async with db.execute(
        "INSERT INTO users (email, name, created_at) VALUES (?, ?, ?)",
        ("flags_admin@test.com", "Flags Admin", now),
    ) as cursor:
        admin_id = cursor.lastrowid
    await db.execute(
        "INSERT INTO user_roles (user_id, role) VALUES (?, 'admin')", (admin_id,)
    )
    await db.commit()
    async with app.test_client() as c:
        async with c.session_transaction() as sess:
            sess["user_id"] = admin_id
        yield c


async def _set_flag(db, name: str, enabled: bool, description: str = ""):
    """Insert or replace a flag in the test DB."""
    await db.execute(
        """INSERT OR REPLACE INTO feature_flags (name, enabled, description)
           VALUES (?, ?, ?)""",
        (name, 1 if enabled else 0, description),
    )
    await db.commit()


async def _flag_value(db, name: str) -> int | None:
    async with db.execute(
        "SELECT enabled FROM feature_flags WHERE name = ?", (name,)
    ) as cursor:
        row = await cursor.fetchone()
        return row[0] if row else None


async def _seed_all_flags(db):
    """Seed the five default flags matching migration 0019 defaults."""
    flags = [
        ("markets", 1),
        ("payments", 0),
        ("planner_export", 0),
        ("supplier_signup", 0),
        ("lead_unlock", 0),
    ]
    for name, enabled in flags:
        await db.execute(
            "INSERT OR REPLACE INTO feature_flags (name, enabled, description) VALUES (?, ?, ?)",
            (name, enabled, ""),
        )
    await db.commit()


def _column_names(conn, table):
    return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()]


def _table_names(conn):
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
        " AND name NOT LIKE 'sqlite_%' ORDER BY name"
    ).fetchall()
    return [r[0] for r in rows]


# ── Migration 0019 ────────────────────────────────────────────────


class TestMigration0019:
    """Migration 0019 creates feature_flags table and seeds initial flags."""

    def test_creates_feature_flags_table(self, tmp_path):
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        assert "feature_flags" in _table_names(conn)
        conn.close()

    def test_table_has_correct_columns(self, tmp_path):
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        cols = _column_names(conn, "feature_flags")
        conn.close()
        assert "name" in cols
        assert "enabled" in cols
        assert "description" in cols
        assert "updated_at" in cols

    def test_seeds_markets_enabled(self, tmp_path):
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        row = conn.execute(
            "SELECT enabled FROM feature_flags WHERE name = 'markets'"
        ).fetchone()
        conn.close()
        assert row is not None and row[0] == 1

    def test_seeds_payments_disabled(self, tmp_path):
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        row = conn.execute(
            "SELECT enabled FROM feature_flags WHERE name = 'payments'"
        ).fetchone()
        conn.close()
        assert row is not None and row[0] == 0

    def test_seeds_all_five_flags(self, tmp_path):
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        names = {r[0] for r in conn.execute("SELECT name FROM feature_flags").fetchall()}
        conn.close()
        assert names == {"markets", "payments", "planner_export", "supplier_signup", "lead_unlock"}

    def test_idempotent_on_rerun(self, tmp_path):
        """Running migrate twice does not duplicate seed rows."""
        db_path = str(tmp_path / "test.db")
        migrate(db_path)
        migrate(db_path)
        conn = sqlite3.connect(db_path)
        count = conn.execute("SELECT COUNT(*) FROM feature_flags").fetchone()[0]
        conn.close()
        assert count == 5


# ── is_flag_enabled ───────────────────────────────────────────────


class TestIsFlagEnabled:
    """Unit tests for is_flag_enabled() helper."""

    @pytest.mark.asyncio
    async def test_returns_true_for_enabled_flag(self, db):
        await _set_flag(db, "markets", True)
        assert await core.is_flag_enabled("markets") is True

    @pytest.mark.asyncio
    async def test_returns_false_for_disabled_flag(self, db):
        await _set_flag(db, "payments", False)
        assert await core.is_flag_enabled("payments") is False

    @pytest.mark.asyncio
    async def test_returns_default_false_for_unknown_flag(self, db):
        assert await core.is_flag_enabled("nonexistent_flag") is False

    @pytest.mark.asyncio
    async def test_returns_custom_default_for_unknown_flag(self, db):
        assert await core.is_flag_enabled("nonexistent_flag", default=True) is True

    @pytest.mark.asyncio
    async def test_reflects_live_db_change(self, db):
        """Change takes effect without restart — reads live DB every call."""
        await _set_flag(db, "payments", False)
        assert await core.is_flag_enabled("payments") is False
        await _set_flag(db, "payments", True)
        assert await core.is_flag_enabled("payments") is True


# ── feature_gate decorator ────────────────────────────────────────


class TestFeatureGateDecorator:
    """feature_gate blocks GET when flag is disabled, passes through when enabled."""

    @pytest.mark.asyncio
    async def test_shows_waitlist_on_get_when_flag_disabled(self, client, db):
        """GET /auth/signup shows waitlist page when payments flag is off."""
        await _set_flag(db, "payments", False)
        response = await client.get("/auth/signup")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        # payments waitlist.html shows "join" or "launching soon" messaging
        assert "launching soon" in html.lower() or "join" in html.lower() or "waitlist" in html.lower()

    @pytest.mark.asyncio
    async def test_shows_normal_form_on_get_when_flag_enabled(self, client, db):
        """GET /auth/signup shows normal signup form when payments flag is on."""
        await _set_flag(db, "payments", True)
        response = await client.get("/auth/signup")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        # Normal signup shows create account form
        assert "create" in html.lower() or "sign up" in html.lower() or "email" in html.lower()

    @pytest.mark.asyncio
    async def test_post_passes_through_when_flag_disabled(self, client, db):
        """POST is never blocked by feature_gate (gates GET only)."""
        await _set_flag(db, "payments", False)
        with patch("padelnomics.worker.enqueue", new_callable=AsyncMock):
            response = await client.post("/auth/signup", form={
                "csrf_token": "test_token",
                "email": "test@example.com",
            })
        assert response.status_code in (200, 302)

    @pytest.mark.asyncio
    async def test_markets_route_gated_by_markets_flag(self, client, db):
        """markets flag controls /en/markets access."""
        # Disabled → shows gated page
        await _set_flag(db, "markets", False)
        response = await client.get("/en/markets")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "coming soon" in html.lower() or "intelligence" in html.lower()

        # Enabled → passes through
        await _set_flag(db, "markets", True)
        response = await client.get("/en/markets")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        # Normal markets page doesn't show the "coming soon" waitlist title
        assert "coming soon" not in html.lower()

    @pytest.mark.asyncio
    async def test_supplier_signup_gated_by_flag(self, client, db):
        """supplier_signup flag controls /en/suppliers/signup."""
        await _set_flag(db, "supplier_signup", False)
        response = await client.get("/en/suppliers/signup")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "waitlist" in html.lower() or "supplier" in html.lower()

    @pytest.mark.asyncio
    async def test_planner_export_gated_for_authenticated_user(self, auth_client, db):
        """planner_export flag controls /en/planner/export."""
        await _set_flag(db, "planner_export", False)
        response = await auth_client.get("/en/planner/export")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "coming soon" in html.lower() or "waitlist" in html.lower()

    @pytest.mark.asyncio
    async def test_planner_export_accessible_when_enabled(self, auth_client, db):
        """Normal planner export page shown when planner_export flag is on."""
        await _set_flag(db, "planner_export", True)
        response = await auth_client.get("/en/planner/export")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "coming soon" not in html.lower()


# ── lead_unlock gate ──────────────────────────────────────────────


class TestLeadUnlockGate:
    """lead_unlock flag controls whether the unlock endpoint is reachable."""

    @pytest.mark.asyncio
    async def test_is_flag_disabled_by_default(self, db):
        """lead_unlock flag seeded as disabled — is_flag_enabled returns False."""
        await _set_flag(db, "lead_unlock", False)
        assert await core.is_flag_enabled("lead_unlock") is False

    @pytest.mark.asyncio
    async def test_is_flag_enabled_after_toggle(self, db):
        """is_flag_enabled returns True after flag is enabled."""
        await _set_flag(db, "lead_unlock", True)
        assert await core.is_flag_enabled("lead_unlock") is True

    @pytest.mark.asyncio
    async def test_route_imports_is_flag_enabled(self):
        """suppliers/routes.py imports is_flag_enabled (gate is wired up)."""
        from padelnomics.suppliers.routes import unlock_lead
        import inspect
        src = inspect.getsource(unlock_lead)
        assert "is_flag_enabled" in src
        assert "lead_unlock" in src
        assert "403" in src


# ── Admin flag toggle routes ──────────────────────────────────────


class TestAdminFlagRoutes:
    """Admin /admin/flags endpoints require admin role and toggle flags correctly."""

    @pytest.mark.asyncio
    async def test_flags_page_requires_admin(self, client, db):
        """Anonymous request to /admin/flags redirects to login."""
        response = await client.get("/admin/flags", follow_redirects=False)
        assert response.status_code == 302

    @pytest.mark.asyncio
    async def test_flags_page_accessible_to_admin(self, admin_client, db):
        await _seed_all_flags(db)
        response = await admin_client.get("/admin/flags")
        assert response.status_code == 200

    @pytest.mark.asyncio
    async def test_flags_page_lists_all_seeded_flags(self, admin_client, db):
        await _seed_all_flags(db)
        response = await admin_client.get("/admin/flags")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        for flag in ("markets", "payments", "planner_export", "supplier_signup", "lead_unlock"):
            assert flag in html

    @pytest.mark.asyncio
    async def test_toggle_enables_disabled_flag(self, admin_client, db):
        await _set_flag(db, "payments", False)
        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "payments",
        })
        assert await _flag_value(db, "payments") == 1

    @pytest.mark.asyncio
    async def test_toggle_disables_enabled_flag(self, admin_client, db):
        await _set_flag(db, "markets", True)
        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "markets",
        })
        assert await _flag_value(db, "markets") == 0

    @pytest.mark.asyncio
    async def test_toggle_unknown_flag_redirects_with_flash(self, admin_client, db):
        response = await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "nonexistent_flag",
        }, follow_redirects=True)
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "not found" in html.lower()

    @pytest.mark.asyncio
    async def test_toggle_requires_admin(self, client, db):
        """Anonymous POST to toggle is rejected."""
        response = await client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "markets",
        }, follow_redirects=False)
        assert response.status_code == 302


# ── Full toggle flow (e2e) ────────────────────────────────────────


class TestFlagToggleFlow:
    """End-to-end: admin toggles flag → route behaviour changes immediately."""

    @pytest.mark.asyncio
    async def test_disable_markets_shows_gated_page(self, admin_client, client, db):
        """Disable markets → /en/markets shows coming soon. Enable → shows content."""
        # Seed markets as enabled
        await _set_flag(db, "markets", True)
        response = await client.get("/en/markets")
        html = await response.get_data(as_text=True)
        assert "coming soon" not in html.lower()

        # Admin disables via toggle
        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "markets",
        })

        # Now shows gated page
        response = await client.get("/en/markets")
        html = await response.get_data(as_text=True)
        assert "coming soon" in html.lower()

    @pytest.mark.asyncio
    async def test_enable_supplier_signup_passes_through(self, admin_client, client, db):
        """Enable supplier_signup → normal wizard shown (no waitlist page)."""
        # Start with flag disabled
        await _set_flag(db, "supplier_signup", False)
        response = await client.get("/en/suppliers/signup")
        html = await response.get_data(as_text=True)
        # Gated: shows waitlist/supplier promo content
        assert "waitlist" in html.lower() or "supplier" in html.lower()

        # Admin enables
        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token",
            "name": "supplier_signup",
        })

        # Now passes through (wizard is shown)
        response = await client.get("/en/suppliers/signup")
        assert response.status_code == 200
        html = await response.get_data(as_text=True)
        assert "waitlist" not in html.lower()

    @pytest.mark.asyncio
    async def test_double_toggle_restores_original_state(self, admin_client, db):
        """Toggling a flag twice returns it to its original value."""
        await _set_flag(db, "payments", False)
        assert await _flag_value(db, "payments") == 0

        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token", "name": "payments",
        })
        assert await _flag_value(db, "payments") == 1

        await admin_client.post("/admin/flags/toggle", form={
            "csrf_token": "test_token", "name": "payments",
        })
        assert await _flag_value(db, "payments") == 0