sales-trainer/backend/tests/test_oauth.py

"""OAuth social login/register tests — provider validation is monkeypatched,
so no live network is ever hit. Config creds are set per-test on the Config
class (read at request time), and the provider validators are swapped out."""
from __future__ import annotations

import pytest

from app.config import Config

# Fully-configured creds for "enabled" tests. Values themselves are irrelevant
# because the validators are monkeypatched; only their non-placeholder presence
# matters to Config.oauth_provider_enabled.
ENABLED = dict(
    OAUTH_GOOGLE_CLIENT_ID="google-client-id.apps.googleusercontent.com",
    OAUTH_GOOGLE_CLIENT_SECRET="GOCSPX-secret",
    OAUTH_FACEBOOK_APP_ID="1234567890",
    OAUTH_FACEBOOK_APP_SECRET="facebookssecret",
    OAUTH_DEFAULT_ORG="org-public",
)


@pytest.fixture()
def oauth_enabled(monkeypatch, user_store):
    """Turn on both providers + default org on Config for the test."""
    for name, value in ENABLED.items():
        monkeypatch.setattr(Config, name, value)
    if user_store.get_org_or_none("org-public") is None:
        user_store.create_org("Public Signups", org_id="org-public")
    yield


@pytest.fixture()
def google_ok(monkeypatch):
    from app.services import oauth as oauth_svc

    def _fake(token):
        return ("new.user@gmail.com", "123456789", "New User")

    monkeypatch.setattr(oauth_svc, "validate_google_token", _fake)
    yield _fake


@pytest.fixture()
def facebook_ok(monkeypatch):
    from app.services import oauth as oauth_svc

    state = {"called": False}

    def _fake(token):
        state["called"] = True
        return ("fb.user@example.com", "987654321", "FB User")

    monkeypatch.setattr(oauth_svc, "validate_facebook_token", _fake)
    yield _fake, state


def _post(client, provider="google", token="valid"):
    return client.post("/api/auth/oauth", json={"provider": provider, "token": token})


def test_new_google_user_created_in_default_org(client, user_store, oauth_enabled, google_ok):
    response = _post(client)
    assert response.status_code == 200, response.get_json()
    body = response.get_json()
    assert body["must_setup"] is True
    user = body["user"]
    assert user["username"] == "g_123456789"
    assert user["id"] == "g_123456789"
    assert user["role"] == "user"
    assert user["email"] == "new.user@gmail.com"
    assert user["org_id"] == "org-public"
    assert user["accepted_terms"] is False
    assert user["accepted_terms_at"] is None
    assert "password_hash" not in user

    created = user_store.get_user("g_123456789")
    assert created["role"] == "user"
    assert created["org_id"] == "org-public"
    assert created["must_setup"] is True
    org = user_store.get_org("org-public")
    assert org["active"] is True
    # Token is usable against an authenticated endpoint.
    me = client.get("/api/auth/me", headers={"Authorization": f"Bearer {body['token']}"})
    assert me.status_code == 200
    protected = client.get("/api/groups", headers={"Authorization": f"Bearer {body['token']}"})
    assert protected.status_code == 403, protected.get_json()


def test_email_match_requires_explicit_account_linking(client, user_store, oauth_enabled, google_ok):
    existing = user_store.create_user(
        org_id="org-public",
        username="existing",
        password="existing-password",
        name="Existing User",
        role="user",
        email="match@example.com",
        must_setup=False,
    )
    # Re-route the fake validator to return the matching email under a new sub.
    from app.services import oauth as oauth_svc

    oauth_svc.validate_google_token = lambda token: ("match@example.com", "999999", "Existing User")

    response = _post(client)
    assert response.status_code == 401, response.get_json()
    assert "token" not in response.get_json()
    assert user_store.get_user_or_none("g_999999") is None  # no duplicate created


def test_email_match_rejects_inactive_organization(client, user_store, oauth_enabled, google_ok):
    user_store.create_user(
        org_id="org-public",
        username="inactive-org-user",
        password="existing-password",
        name="Inactive Org User",
        role="user",
        email="inactive-org@example.com",
        must_setup=False,
    )
    with user_store.orgs.record_lock("org-public"):
        user_store.orgs.update("org-public", active=False)

    from app.services import oauth as oauth_svc

    oauth_svc.validate_google_token = lambda token: (
        "inactive-org@example.com",
        "999998",
        "Inactive Org User",
    )

    response = _post(client)
    assert response.status_code == 401, response.get_json()
    assert "token" not in (response.get_json() or {})


def test_invalid_or_unverified_email_token_rejected(client, user_store, oauth_enabled, monkeypatch):
    from app.services import oauth as oauth_svc

    def _reject(token):
        raise oauth_svc.OAuthError("provider validation failed (email not verified)")

    monkeypatch.setattr(oauth_svc, "validate_google_token", _reject)
    response = _post(client)
    assert response.status_code == 401, response.get_json()
    # No user created out of an unverified token.
    assert len(user_store.users.all()) == 1  # only bootstrap admin


def test_disabled_provider_returns_404(client, user_store):
    # No creds configured → OAuth disabled.
    response = _post(client)
    assert response.status_code == 404
    assert user_store.get_user_or_none("g_123456789") is None


def test_username_collision_gets_unique_suffix(client, user_store, oauth_enabled, google_ok):
    user_store.create_user(
        org_id="org-public",
        username="g_1",
        password="collide-password",
        name="Collision",
        role="user",
        must_setup=False,
    )
    # Fake validator returns sub "1" → base username "g_1" collides.
    from app.services import oauth as oauth_svc

    oauth_svc.validate_google_token = lambda token: ("collide@gmail.com", "1", "Collide User")

    response = _post(client)
    assert response.status_code == 200, response.get_json()
    assert response.get_json()["user"]["username"] == "g_1_1"


def test_rate_limit_enforced(client, user_store, oauth_enabled, google_ok):
    # Per-email limit is 8/300s; each call increments the email bucket.
    for _ in range(8):
        assert _post(client).status_code == 200
    response = _post(client)
    assert response.status_code == 429, response.get_json()
    # No extra user was created (the same email always matches/linked during the
    # window would actually create once; here all 8 successful are new-link-less
    # creations of the same email? They'd each be rejected as dupes. Instead just
    # assert the 429 and that the endpoint is rate-limited on repeated calls.)
    assert response.get_json()["error"]


def test_facebook_path_works_and_uses_verification(client, user_store, oauth_enabled, facebook_ok):
    _fake, state = facebook_ok
    response = _post(client, provider="facebook", token="fb-token")
    assert response.status_code == 200, response.get_json()
    body = response.get_json()
    assert body["user"]["username"] == "fb_987654321"
    assert body["user"]["email"] == "fb.user@example.com"
    assert state["called"] is True  # the validator ran (app-token verification flow)


def test_oauth_config_endpoint_no_secrets(client, oauth_enabled):
    response = client.get("/api/auth/oauth/config")
    assert response.status_code == 200
    body = response.get_json()
    assert body["google"] is True
    assert body["facebook"] is True
    assert body["google_client_id"] == "google-client-id.apps.googleusercontent.com"
    assert body["facebook_app_id"] == "1234567890"
    # Secrets must never be exposed.
    assert "google_client_secret" not in body
    assert "facebook_app_secret" not in body


def test_oauth_config_disabled_when_creds_absent(client):
    response = client.get("/api/auth/oauth/config")
    assert response.status_code == 200
    body = response.get_json()
    assert body["google"] is False
    assert body["facebook"] is False


def test_oauth_requires_provider_and_token(client, oauth_enabled):
    for payload in ({}, {"provider": "google"}, {"token": "abc"}, {"provider": "twitter", "token": "x"}):
        resp = client.post("/api/auth/oauth", json=payload)
        assert resp.status_code == 400, (payload, resp.get_json())


def test_social_signup_is_seat_checked(client, user_store, oauth_enabled, google_ok):
    # Default org with a single seat already consumed → new signup must be
    # rejected (fail closed) once seats are exhausted.
    # Consume the only seat.
    user_store.create_user(
        org_id="org-public",
        username="g_1",
        password="seat-password",
        name="Seat User",
        role="user",
        must_setup=False,
    )
    # Shrink seats so no room remains (seat check happens in create_user).
    with user_store.orgs.record_lock("org-public"):
        user_store.orgs.update("org-public", seats=1)

    response = _post(client)
    assert response.status_code == 401, response.get_json()  # AuthError → generic 401
    assert user_store.get_user_or_none("g_123456789") is None