GarminHealthLab/backend/tests/test_auth.py

"""Unit tests for password hashing, JWT handling, and the auth endpoints."""
import datetime
import hashlib
import os

import jwt
import pytest

import auth
from config import JWT_SECRET


# --- password hashing -------------------------------------------------------
class TestPasswordHashing:
    def test_hash_is_verifiable(self):
        stored = auth.hash_password("correct horse")
        assert auth.verify_password("correct horse", stored) is True

    def test_wrong_password_rejected(self):
        stored = auth.hash_password("correct horse")
        assert auth.verify_password("wrong horse", stored) is False

    def test_salt_makes_hashes_unique(self):
        a = auth.hash_password("same")
        b = auth.hash_password("same")
        assert a != b, "identical passwords must not produce identical hashes"

    def test_hash_format_is_self_describing(self):
        stored = auth.hash_password("pw")
        prefix, iterations, salt, digest = stored.split("$")
        assert prefix == auth.PBKDF2_PREFIX
        assert int(iterations) == auth.PBKDF2_ITERATIONS
        assert len(bytes.fromhex(salt)) == 16
        assert len(bytes.fromhex(digest)) == 64

    def test_password_never_appears_in_hash(self):
        stored = auth.hash_password("supersecret")
        assert "supersecret" not in stored

    @pytest.mark.parametrize(
        "stored",
        ["", None, "garbage", "pbkdf2_sha256$notanint$aa$bb", "nothex:nothex"],
    )
    def test_malformed_hashes_rejected_not_raised(self, stored):
        assert auth.verify_password("anything", stored) is False

    def test_hashing_does_not_require_scrypt(self, monkeypatch):
        """Regression: macOS system Python (LibreSSL) has no hashlib.scrypt.

        Registration used to raise AttributeError -> HTTP 500 on those builds.
        """
        monkeypatch.delattr(hashlib, "scrypt", raising=False)
        stored = auth.hash_password("pw")
        assert auth.verify_password("pw", stored) is True

    def test_unicode_password(self):
        stored = auth.hash_password("密码🔒")
        assert auth.verify_password("密码🔒", stored) is True
        assert auth.verify_password("密码", stored) is False

    @pytest.mark.skipif(
        not hasattr(hashlib, "scrypt"), reason="interpreter built without scrypt"
    )
    def test_legacy_scrypt_hash_still_verifies(self):
        salt = os.urandom(16)
        digest = hashlib.scrypt(
            b"legacy", salt=salt, n=16384, r=8, p=1, dklen=64
        ).hex()
        assert auth.verify_password("legacy", f"{salt.hex()}:{digest}") is True
        assert auth.verify_password("nope", f"{salt.hex()}:{digest}") is False


# --- JWT --------------------------------------------------------------------
class TestTokens:
    def test_sign_and_verify_roundtrip(self):
        token = auth.sign_token("user-123")
        assert auth.verify_token(token) == {"user_id": "user-123"}

    def test_expired_token_rejected(self):
        past = datetime.datetime.utcnow() - datetime.timedelta(days=1)
        token = jwt.encode(
            {"sub": "u", "iat": past, "exp": past}, JWT_SECRET, algorithm="HS256"
        )
        with pytest.raises(jwt.ExpiredSignatureError):
            auth.verify_token(token)

    def test_token_signed_with_other_secret_rejected(self):
        token = jwt.encode({"sub": "u"}, "a-different-secret", algorithm="HS256")
        with pytest.raises(jwt.InvalidTokenError):
            auth.verify_token(token)

    def test_tampered_token_rejected(self):
        token = auth.sign_token("user-123")
        head, payload, sig = token.split(".")
        with pytest.raises(jwt.InvalidTokenError):
            auth.verify_token(f"{head}.{payload}.{sig[:-2]}xx")


# --- register ---------------------------------------------------------------
class TestRegister:
    def test_returns_201_and_token(self, client):
        r = client.post(
            "/api/auth/register",
            json={
                "email": "a@example.com",
                "garminEmail": "g@example.com",
                "garminPassword": "pw123456",
            },
        )
        assert r.status_code == 201
        body = r.get_json()
        assert body["email"] == "a@example.com"
        assert auth.verify_token(body["token"])["user_id"] == body["id"]

    def test_duplicate_email_conflicts(self, client, user):
        r = client.post(
            "/api/auth/register",
            json={
                "email": user["email"],
                "garminEmail": "other@example.com",
                "garminPassword": "pw123456",
            },
        )
        assert r.status_code == 409

    @pytest.mark.parametrize(
        "payload",
        [
            {},
            {"email": "a@example.com"},
            {"email": "a@example.com", "garminEmail": "g@example.com"},
            {"email": "", "garminEmail": "g@example.com", "garminPassword": "x"},
        ],
    )
    def test_missing_fields_rejected(self, client, payload):
        assert client.post("/api/auth/register", json=payload).status_code == 400

    def test_password_stored_only_as_hash(self, client, db):
        client.post(
            "/api/auth/register",
            json={
                "email": "h@example.com",
                "garminEmail": "g@example.com",
                "garminPassword": "plaintext-secret",
            },
        )
        row = db.query_one(
            "SELECT garmin_password_hash FROM users WHERE email = ?", ["h@example.com"]
        )
        assert "plaintext-secret" not in row["garmin_password_hash"]
        assert auth.verify_password("plaintext-secret", row["garmin_password_hash"])


# --- login ------------------------------------------------------------------
class TestLogin:
    def test_valid_credentials(self, client, user):
        r = client.post(
            "/api/auth/login",
            json={"email": user["email"], "password": user["password"]},
        )
        assert r.status_code == 200
        assert r.get_json()["id"] == user["id"]

    def test_wrong_password(self, client, user):
        r = client.post(
            "/api/auth/login", json={"email": user["email"], "password": "nope"}
        )
        assert r.status_code == 401

    def test_unknown_email(self, client):
        r = client.post(
            "/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
        )
        assert r.status_code == 401

    def test_error_does_not_reveal_which_field_was_wrong(self, client, user):
        unknown = client.post(
            "/api/auth/login", json={"email": "ghost@example.com", "password": "pw"}
        ).get_json()
        bad_pw = client.post(
            "/api/auth/login", json={"email": user["email"], "password": "nope"}
        ).get_json()
        assert unknown == bad_pw, "responses must not distinguish the two cases"


# --- require_auth -----------------------------------------------------------
class TestRequireAuth:
    def test_missing_header(self, client):
        assert client.get("/api/health/summary").status_code == 401

    @pytest.mark.parametrize(
        "header",
        ["", "Bearer", "Token abc", "bearer abc", "Bearer not.a.jwt"],
    )
    def test_malformed_headers(self, client, header):
        r = client.get("/api/health/summary", headers={"Authorization": header})
        assert r.status_code == 401

    def test_expired_token_gets_401_not_500(self, client):
        past = datetime.datetime.utcnow() - datetime.timedelta(days=1)
        token = jwt.encode(
            {"sub": "u", "iat": past, "exp": past}, JWT_SECRET, algorithm="HS256"
        )
        r = client.get(
            "/api/health/summary", headers={"Authorization": f"Bearer {token}"}
        )
        assert r.status_code == 401
        assert "expired" in r.get_json()["error"]

    def test_valid_token_passes(self, client, auth):
        assert client.get("/api/health/summary", headers=auth).status_code == 200


class TestLogoutAndRefresh:
    def test_logout_clears_stored_token(self, client, auth, user, db):
        assert client.post("/api/auth/logout", headers=auth).status_code == 200
        row = db.query_one("SELECT jwt_token FROM users WHERE id = ?", [user["id"]])
        assert row["jwt_token"] is None

    def test_refresh_returns_usable_token(self, client, auth):
        r = client.post("/api/auth/refresh", headers=auth)
        assert r.status_code == 200
        new_token = r.get_json()["token"]
        assert (
            client.get(
                "/api/health/summary",
                headers={"Authorization": f"Bearer {new_token}"},
            ).status_code
            == 200
        )

    def test_refresh_requires_auth(self, client):
        assert client.post("/api/auth/refresh").status_code == 401