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 unittest.mock import AsyncMock, patch

import pytest
from padelnomics.core import utcnow_iso
from padelnomics.migrations.migrate import migrate

from padelnomics import core

# ── 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 = utcnow_iso()
    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_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 = 'markets'"
        ).fetchone()
        conn.close()
        assert row is not None and row[0] == 0

    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)."""
        import inspect

        from padelnomics.suppliers.routes import unlock_lead
        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