flowdeck/tests/test_v52_infra.py

"""FlowDeck — v5.2.0 Infrastructure & Polish tests.

Covers: API tokens (create/use/revoke), active sessions (list/revoke),
onboarding wizard, automatic backups, projects registry, forge adapters
(mocked HTTP) and multi-user permissions.
"""
import asyncio
import os

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient

# Fixture moved to conftest.py


def _register(client, email="[email protected]", password="secret123", name="Alice"):
    return client.post(
        "/auth/register",
        json={"email": email, "password": password, "name": name},
    )


def _create_collection(name="Projects"):
    from app.db import get_conn
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO collections (name, description, icon, schema_json) VALUES (?, ?, '📋', '[]')",
            (name, ""),
        )
        conn.commit()
        return cur.lastrowid


# ═══════════════ API tokens ═══════════════

def test_api_token_lifecycle(client):
    resp = _register(client)
    assert resp.status_code == 200 and resp.json()["status"] == "ok"
    _create_collection()

    # Create a token via the Settings API.
    create = client.post("/api/settings/tokens", json={"name": "CI script"})
    assert create.status_code == 200, create.text
    data = create.json()
    assert data["token"].startswith("fd_")
    assert data["name"] == "CI script"

    # Use it against the public API.
    ok = client.get("/api/v1/collections", headers={"Authorization": f"Bearer {data['token']}"})
    assert ok.status_code == 200
    assert {"collections"} <= set(ok.json())

    # It is listed in Settings (prefix only, never the secret).
    listing = client.get("/api/settings/tokens").json()
    tokens = listing["tokens"]
    assert len(tokens) == 1
    assert tokens[0]["token_prefix"] == data["token"][:12]
    assert tokens[0]["revoked"] == 0

    # Revoke → the same bearer token is refused.
    revoke = client.delete(f"/api/settings/tokens/{data['id']}")
    assert revoke.json()["status"] == "revoked"
    blocked = client.get("/api/v1/collections", headers={"Authorization": f"Bearer {data['token']}"})
    assert blocked.status_code == 403


def test_api_tokens_require_authentication(client):
    r1 = client.get("/api/settings/tokens")
    assert r1.status_code == 401
    r2 = client.post("/api/settings/tokens", json={"name": "x"})
    assert r2.status_code == 401


# ═══════════════ Active sessions ═══════════════

def test_sessions_listed_and_revocable(client):
    _register(client)
    alice_cookie = client.cookies.get("flowdeck_session")
    assert alice_cookie

    sessions = client.get("/api/settings/sessions").json()["sessions"]
    assert len(sessions) == 1
    assert sessions[0]["is_current"] is True

    # A second login creates another session row.
    _register(client, email="[email protected]", password="secret456")

    from app.db import get_conn
    with get_conn() as conn:
        count = conn.execute("SELECT COUNT(*) FROM user_sessions").fetchone()[0]
    assert count == 2

    # Revoke alice's session using alice's cookie (the test client now has bob's cookie).
    # Get alice's session ID from DB.
    with get_conn() as conn:
        alice_row = conn.execute("SELECT id FROM user_sessions WHERE user_id=(SELECT id FROM users WHERE login='[email protected]')").fetchone()
    alice_sid = alice_row["id"]

    # Create a fresh client with alice's cookie to revoke.
    client_alice = TestClient(client.app)
    client_alice.cookies.set("flowdeck_session", alice_cookie)
    revoke = client_alice.post(f"/api/settings/sessions/{alice_sid}/revoke")
    assert revoke.status_code == 200

    # Verify alice's session is revoked - decode_session should return None.
    from app.auth.session import SessionManager
    assert SessionManager.decode_session(alice_cookie) is None


# ═══════════════ Onboarding ═══════════════

def test_welcome_redirects_unauthenticated(client):
    r = client.get("/welcome", follow_redirects=False)
    assert r.status_code in (302, 200)


def test_onboarding_workspace_and_project(client):
    _register(client)
    r = client.get("/welcome")
    assert r.status_code == 200

    ws = client.post("/api/onboarding/workspace", json={"name": "Équipe Frelon"})
    assert ws.status_code == 200
    data = ws.json()
    assert data["status"] == "ok" and data["id"]

    from app.db import get_conn
    with get_conn() as conn:
        member = conn.execute(
            "SELECT role FROM workspace_members WHERE workspace_id=? AND user_id=(SELECT id FROM users WHERE login='[email protected]')",
            (data["id"],),
        ).fetchone()
    assert member and member["role"] == "owner"

    proj = client.post("/api/onboarding/project", json={"title": "Welcome", "workspace_id": data["id"]})
    assert proj.status_code == 200
    pid = proj.json()["id"]
    with get_conn() as conn:
        page = conn.execute("SELECT title, workspace_id FROM pages WHERE id=?", (pid,)).fetchone()
    assert page["title"] == "Welcome" and page["workspace_id"] == data["id"]

    # Once a workspace exists, /welcome redirects to the app.
    r2 = client.get("/welcome", follow_redirects=False)
    assert r2.status_code == 302 and "workspaces" in r2.headers.get("location", "")


# ═══════════════ Backups ═══════════════

def test_backup_snapshot_and_pruning(client):
    """backup_db() snapshots the SQLite file and prune keeps the newest N."""
    import datetime

    from app.services import backup as backup_svc

    base = datetime.datetime(2026, 1, 1, 0, 0, 0)
    first = backup_svc.backup_db(now=base)
    assert first and first.startswith("flowdeck-") and first.endswith(".db")

    backups = backup_svc.list_backups()
    assert len(backups) == 1
    assert backups[0]["filename"] == first
    assert backups[0]["size"] > 0

    # Distinct timestamps → distinct filenames (the format has 1-second resolution).
    for i in range(1, 4):
        assert backup_svc.backup_db(now=base + datetime.timedelta(seconds=i))
    assert len(backup_svc.list_backups()) == 4

    removed = backup_svc.prune_old_backups(keep=2)
    assert removed == 2
    remaining = backup_svc.list_backups()
    assert len(remaining) == 2
    # Newest first (filename sort == chronological for this format).
    assert remaining[0]["filename"] > remaining[1]["filename"]

    # Age + due logic (file mtime is "now", so a fresh backup is not due).
    assert backup_svc.last_backup_age_hours() is not None
    assert backup_svc.backup_due() is False


def test_backup_disabled_returns_none(client):
    from app.config import settings
    from app.services import backup as backup_svc

    previous = settings.backup_enabled
    settings.backup_enabled = False
    try:
        assert backup_svc.backup_db() is None
    finally:
        settings.backup_enabled = previous


def test_backup_admin_api(client):
    """The backup admin API is admin-only and snapshots on demand."""
    # Unauthenticated → forbidden.
    assert client.post("/api/settings/backups/run").status_code == 403

    # First registered user becomes admin → allowed.
    _register(client)
    run = client.post("/api/settings/backups/run")
    assert run.status_code == 200, run.text
    assert run.json()["status"] == "ok"
    assert run.json()["filename"].startswith("flowdeck-")

    listing = client.get("/api/settings/backups")
    assert listing.status_code == 200
    assert len(listing.json()["backups"]) >= 1


# ═══════════════ Projects registry ═══════════════

def test_projects_registry(client):
    from app.services import projects as projects_svc

    # The test client's database is already initialised by the fixture.
    # Use the SAME connection that the app uses (get_conn()).
    gitea_repo = {
        "id": 17, "name": "flowdeck", "full_name": "bruno/flowdeck",
        "default_branch": "main", "language": "Python", "clone_url": "https://git/x",
    }
    pid = projects_svc.register_repo(gitea_repo, "gitea")
    assert pid is not None
    again = projects_svc.register_repo({**gitea_repo, "language": "Go"}, "gitea")
    assert again == pid  # upsert, not duplicate

    github_repo = {
        "id": 99, "name": "docs", "full_name": "org/docs", "clone_url": "https://github.com/org/docs.git",
    }
    projects_svc.register_repo(github_repo, "github")

    projects = projects_svc.list_projects()
    assert {p["name"] for p in projects} == {"flowdeck", "docs"}
    assert projects_svc.list_projects("gitea")[0]["language"] == "Go"

    builtin = projects_svc.create_builtin_project("Mon Projet", owner="alice")
    assert builtin["id"]
    assert projects_svc.list_projects("builtin")[0]["name"] == "Mon Projet"

    # API listing (uses the same in-memory DB).
    r = client.get("/api/projects")
    assert r.status_code == 200
    assert len(r.json()["projects"]) == 3


# ═══════════════ Forge adapters (mock HTTP) ═══════════════

def test_github_adapter_mocked_http(client):
    import httpx

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/user":
            return httpx.Response(200, json={"login": "alice"})
        if request.url.path == "/user/repos":
            return httpx.Response(200, json=[
                {"id": 1, "name": "repos_a", "full_name": "alice/repos_a",
                 "default_branch": "main", "clone_url": "https://x", "language": "Python"},
            ])
        if request.url.path.startswith("/repos/alice/repos_a"):
            if request.url.path.endswith("/languages"):
                return httpx.Response(200, json={"Python": 100, "HTML": 20})
            return httpx.Response(200, json={
                "id": 1, "name": "repos_a", "full_name": "alice/repos_a",
                "default_branch": "main", "clone_url": "https://x", "language": "Python",
                "owner": {"login": "alice"},
            })
        return httpx.Response(404, json={"message": "not found"})

    from app.services.github_adapter import GitHubAdapter
    adapter = GitHubAdapter("gh-token", transport=httpx.MockTransport(handler))

    async def run():
        assert await adapter.validate_token() is True
        repos = await adapter.list_repos(page=1)
        assert repos[0]["full_name"] == "alice/repos_a"
        info = await adapter.get_repo_info("alice", "repos_a")
        assert info["default_branch"] == "main"
        assert info["language"] == "Python"
        langs = await adapter.get_languages("alice", "repos_a")
        assert langs["Python"] == 100

    asyncio.run(run())


def test_forge_repo_normalization(client):
    from app.services.forge_adapter import GiteaAdapter, normalize_repo
    repo = {"full_name": "org/repo", "name": "repo", "default_branch": "master",
            "clone_url": "https://git/org/repo.git", "language": "Rust", "id": 5}
    normalized = normalize_repo(repo, "gitea")
    assert normalized["proj_type"] == "gitea"
    assert normalized["owner"] == "org"
    assert normalized["name"] == "repo"
    assert GiteaAdapter.kind == "gitea"


def test_projects_sync_with_mock_client(client):
    from app.db import get_conn
    from app.services.projects import sync_all_projects

    # Give the sync a gitea token → records a repo through a smoke path.
    with get_conn() as conn:
        uid = conn.execute("SELECT id FROM users WHERE login='[email protected]'").fetchone()
        if uid:
            conn.execute(
                "INSERT INTO user_oauth_tokens (user_id, provider, access_token) VALUES (?, 'gitea', 'tok')",
                (uid["id"],),
            )
            conn.commit()

    stats = asyncio.run(sync_all_projects())
    # Gitea call will fail (no real network) → counted as an error, not a crash.
    assert "error" in stats


os.environ.setdefault("PROJECT_SYNC_ENABLED", "false")

# ═══════════════ OAuth integration (mocked providers) ═══════════════


class _FakeProvider:
    """Stand-in OAuth provider — no network calls."""

    name = "gitea"
    icon = "🔗"

    def __init__(self, login="octocat", full_name="Octo Cat", email="[email protected]"):
        self.login = login
        self.full_name = full_name
        self.email = email

    def is_enabled(self) -> bool:
        return True

    def get_authorize_url(self, state, redirect_uri=None, force_login=False):
        return f"https://gitea.test/login/oauth/authorize?client_id=cid&state={state}"

    async def exchange_code(self, code, redirect_uri=None):
        return {"access_token": "tok-123", "refresh_token": "ref-123"}

    async def get_user(self, access_token):
        return {
            "login": self.login,
            "full_name": self.full_name,
            "email": self.email,
            "avatar_url": "https://gitea.test/avatar.png",
            "provider_id": "42",
        }

    async def list_repositories(self, access_token):
        return []


def _patch_provider(monkeypatch, provider=None):
    from app.auth import providers
    fake = provider or _FakeProvider()
    monkeypatch.setattr(providers, "get_provider", lambda name: fake if name in ("gitea", "github") else None)
    return fake


def _extract_state(location: str) -> str:
    from urllib.parse import parse_qs, urlparse
    return parse_qs(urlparse(location).query)["state"][0]


def test_oauth_login_callback_flow(client, monkeypatch):
    """Full login flow: authorize redirect → callback → user + token + session."""
    _patch_provider(monkeypatch)

    login = client.get("/auth/login?provider=gitea", follow_redirects=False)
    assert login.status_code == 302
    assert "gitea.test/login/oauth/authorize" in login.headers["location"]
    state = _extract_state(login.headers["location"])

    cb = client.get(f"/auth/callback?code=abc&state={state}", follow_redirects=False)
    assert cb.status_code == 302
    assert "/workspaces" in cb.headers["location"]
    assert client.cookies.get("flowdeck_session")

    from app.db import get_conn
    with get_conn() as conn:
        user = conn.execute("SELECT id, auth_method FROM users WHERE login='gitea_octocat'").fetchone()
        assert user and user["auth_method"] == "gitea"
        token = conn.execute(
            "SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='gitea'",
            (user["id"],),
        ).fetchone()
        assert token and token["access_token"] == "tok-123"

    me = client.get("/auth/user").json()
    assert me["authenticated"] is True
    assert me["user"]["login"] == "gitea_octocat"


def test_oauth_github_callback_creates_github_user(client, monkeypatch):
    _patch_provider(monkeypatch, _FakeProvider(login="hubber", full_name="Hub Ber"))

    login = client.get("/auth/login?provider=github", follow_redirects=False)
    assert login.status_code == 302
    state = _extract_state(login.headers["location"])

    cb = client.get(f"/auth/callback?code=xyz&state={state}", follow_redirects=False)
    assert cb.status_code == 302

    from app.db import get_conn
    with get_conn() as conn:
        user = conn.execute("SELECT id, auth_method FROM users WHERE login='github_hubber'").fetchone()
        assert user and user["auth_method"] == "github"
        token = conn.execute(
            "SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='github'",
            (user["id"],),
        ).fetchone()
        assert token and token["access_token"] == "tok-123"


def test_oauth_link_mode_attaches_to_current_user(client, monkeypatch):
    """mode=link must attach the forge token to the already-logged-in user."""
    _register(client)
    _patch_provider(monkeypatch)

    login = client.get("/auth/login?provider=gitea&mode=link", follow_redirects=False)
    state = _extract_state(login.headers["location"])
    assert state.endswith(":link")

    cb = client.get(f"/auth/callback?code=abc&state={state}", follow_redirects=False)
    assert cb.status_code == 302
    assert "settings" in cb.headers["location"]

    from app.db import get_conn
    with get_conn() as conn:
        user = conn.execute("SELECT id FROM users WHERE login='[email protected]'").fetchone()
        token = conn.execute(
            "SELECT access_token FROM user_oauth_tokens WHERE user_id=? AND provider='gitea'",
            (user["id"],),
        ).fetchone()
        assert token and token["access_token"] == "tok-123"
        # No new OAuth user was created.
        count = conn.execute("SELECT COUNT(*) FROM users WHERE login LIKE 'gitea_%'").fetchone()[0]
        assert count == 0


def test_oauth_callback_rejects_invalid_state(client, monkeypatch):
    _patch_provider(monkeypatch)
    client.get("/auth/login?provider=gitea", follow_redirects=False)
    bad = client.get("/auth/callback?code=abc&state=tampered", follow_redirects=False)
    assert bad.status_code == 400


def test_oauth_provider_authorize_urls():
    from app.auth.providers import GiteaProvider, GitHubProvider

    gitea = GiteaProvider("https://git.example.com", "cid", "sec", "https://app/auth/callback")
    assert gitea.is_enabled()
    gitea_url = gitea.get_authorize_url("st", redirect_uri="https://app/auth/callback")
    assert gitea_url.startswith("https://git.example.com/login/oauth/authorize?")
    assert "client_id=cid" in gitea_url and "state=st" in gitea_url

    github = GitHubProvider("cid", "sec", "https://app/auth/callback")
    assert github.is_enabled()
    assert "github.com/login/oauth/authorize" in github.get_authorize_url("st")
    assert not GitHubProvider("", "", "https://app/auth/callback").is_enabled()


# ═══════════════ Multi-user permissions ═══════════════

def test_permission_manager_roles(client):
    from app.db import get_conn
    from app.password_utils import hash_password
    from app.services.permission_manager import PermissionManager

    with get_conn() as conn:
        ua = conn.execute(
            "INSERT INTO users (login, full_name, email, password_hash) VALUES ('pma', 'A', 'a@x', ?)",
            (hash_password("secret123"),),
        ).lastrowid
        ub = conn.execute(
            "INSERT INTO users (login, full_name, email, password_hash) VALUES ('pmb', 'B', 'b@x', ?)",
            (hash_password("secret123"),),
        ).lastrowid
        uc = conn.execute(
            "INSERT INTO users (login, full_name, email, password_hash) VALUES ('pmc', 'C', 'c@x', ?)",
            (hash_password("secret123"),),
        ).lastrowid
        ws = conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('Team', ?)", (ua,)).lastrowid
        conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?, 'editor')", (ws, ub))
        conn.commit()

    owner_pm = PermissionManager(ua)
    editor_pm = PermissionManager(ub)
    outsider_pm = PermissionManager(uc)  # uc exists but is not a member → "viewer" role

    assert owner_pm.can_read(ws) and owner_pm.can_write(ws) and owner_pm.can_destructive(ws)
    assert editor_pm.can_read(ws) and editor_pm.can_write(ws)
    assert not editor_pm.can_destructive(ws)
    # Non-members get "viewer" role → can read but not write
    assert outsider_pm.can_read(ws)
    assert not outsider_pm.can_write(ws)
    assert not outsider_pm.can_destructive(ws)

    # Tool gating.
    editor_pm.assert_can("create_page", {}, ws)  # editor OK
    with pytest.raises(HTTPException):
        editor_pm.assert_can("delete_page", {}, ws, approval_mode="auto")
    with pytest.raises(HTTPException):
        outsider_pm.assert_can("update_page", {}, ws)  # viewer cannot write