WealthySmart/backend/tests/test_auth.py

"""JWT lifecycle, API-token validation, constant-time creds, login rate limit."""

from datetime import timedelta

import pytest
from fastapi import HTTPException, Request
from jose import jwt

from app import auth as auth_mod
from app.auth import (
    ALGORITHM,
    LOGIN_RATE_LIMIT,
    _login_attempts,
    _validate_token,
    check_login_rate_limit,
    create_access_token,
    hash_token,
    verify_admin_credentials,
)
from app.config import settings
from app.models.models import APIToken
from app.timeutil import utcnow


def make_request(ip: str) -> Request:
    return Request(
        {
            "type": "http",
            "headers": [(b"x-forwarded-for", ip.encode())],
            "client": ("127.0.0.1", 1234),
        }
    )


class TestJWT:
    def test_roundtrip_and_seven_day_expiry(self):
        token = create_access_token("charlie")
        assert _validate_token(token) == "charlie"

        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[ALGORITHM]
        )
        # exp is a true UTC epoch; a naive datetime's .timestamp() would be
        # interpreted as local time, so compare against an aware now.
        from datetime import datetime, timezone

        lifetime = payload["exp"] - datetime.now(timezone.utc).timestamp()
        assert lifetime == pytest.approx(7 * 24 * 3600, abs=60)

    def test_expired_token_rejected(self, engine, monkeypatch):
        # The API-token fallback path opens a DB session; point it at the
        # test engine so the lookup runs (and finds nothing).
        monkeypatch.setattr("app.db.engine", engine)
        expired = jwt.encode(
            {"sub": "charlie", "exp": utcnow() - timedelta(minutes=1)},
            settings.SECRET_KEY,
            algorithm=ALGORITHM,
        )
        with pytest.raises(HTTPException) as exc:
            _validate_token(expired)
        assert exc.value.status_code == 401

    def test_garbage_token_rejected(self, engine, monkeypatch):
        monkeypatch.setattr("app.db.engine", engine)
        with pytest.raises(HTTPException):
            _validate_token("not-a-jwt")


class TestAPIToken:
    def test_active_api_token_accepted(self, engine, session, monkeypatch):
        monkeypatch.setattr("app.db.engine", engine)
        session.add(
            APIToken(name="n8n", token_hash=hash_token("tok-123"), is_active=True)
        )
        session.commit()
        assert _validate_token("tok-123") == "api:n8n"

    def test_expired_api_token_rejected(self, engine, session, monkeypatch):
        monkeypatch.setattr("app.db.engine", engine)
        session.add(
            APIToken(
                name="old",
                token_hash=hash_token("tok-old"),
                is_active=True,
                expires_at=utcnow() - timedelta(days=1),
            )
        )
        session.commit()
        with pytest.raises(HTTPException) as exc:
            _validate_token("tok-old")
        assert exc.value.status_code == 401

    def test_revoked_api_token_rejected(self, engine, session, monkeypatch):
        monkeypatch.setattr("app.db.engine", engine)
        session.add(
            APIToken(
                name="revoked",
                token_hash=hash_token("tok-rev"),
                is_active=False,
            )
        )
        session.commit()
        with pytest.raises(HTTPException):
            _validate_token("tok-rev")


class TestCredentials:
    def test_correct_credentials_pass(self):
        verify_admin_credentials(
            settings.ADMIN_USERNAME, settings.ADMIN_PASSWORD
        )  # no exception

    @pytest.mark.parametrize(
        "user,pw",
        [
            ("wrong", "test-password"),
            ("testadmin", "wrong"),
            ("", ""),
        ],
    )
    def test_wrong_credentials_rejected(self, user, pw):
        with pytest.raises(HTTPException) as exc:
            verify_admin_credentials(user, pw)
        assert exc.value.status_code == 401


class TestLoginRateLimit:
    def setup_method(self):
        _login_attempts.clear()

    def test_limit_allows_then_blocks(self):
        req = make_request("203.0.113.1")
        for _ in range(LOGIN_RATE_LIMIT):
            check_login_rate_limit(req)
        with pytest.raises(HTTPException) as exc:
            check_login_rate_limit(req)
        assert exc.value.status_code == 429
        assert "Retry-After" in exc.value.headers

    def test_limit_is_per_ip(self):
        for _ in range(LOGIN_RATE_LIMIT):
            check_login_rate_limit(make_request("203.0.113.2"))
        # A different IP is unaffected.
        check_login_rate_limit(make_request("203.0.113.3"))

    def test_window_slides(self):
        req = make_request("203.0.113.4")
        for _ in range(LOGIN_RATE_LIMIT):
            check_login_rate_limit(req)
        # Age every recorded attempt past the window; the next attempt passes.
        attempts = _login_attempts["203.0.113.4"]
        for i in range(len(attempts)):
            attempts[i] -= auth_mod.LOGIN_RATE_WINDOW_SECONDS + 1
        check_login_rate_limit(req)  # no exception