340 lines
13 KiB
Python
340 lines
13 KiB
Python
# #region Test.Auth.Jwt [C:3] [TYPE Module] [SEMANTICS test,auth,jwt,token]
|
|
# @BRIEF Tests for core/auth/jwt.py — create_access_token, decode_token, blacklist, is_blacklisted.
|
|
# @RELATION BINDS_TO -> [Auth.Jwt]
|
|
# @TEST_EDGE: invalid_token -> decode_token raises JWTError
|
|
# @TEST_EDGE: expired_token -> decode_token raises JWTError
|
|
# @TEST_EDGE: blacklist_missing_jti -> blacklist_token returns early
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
import os
|
|
|
|
# Must set auth env vars before any src.core.auth imports
|
|
os.environ.setdefault("AUTH_SECRET_KEY", "test-secret-key-for-jwt-testing")
|
|
os.environ.setdefault("JWT_AUDIENCE", "test-audience")
|
|
os.environ.setdefault("JWT_ISSUER", "test-issuer")
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from jose import jwt as jose_jwt
|
|
|
|
|
|
class TestCreateAccessToken:
|
|
"""Verify create_access_token @POST guarantees."""
|
|
|
|
def test_creates_token_with_minimal_data(self):
|
|
from src.core.auth.jwt import create_access_token, decode_token
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
assert isinstance(token, str)
|
|
payload = decode_token(token)
|
|
assert payload["sub"] == "user1"
|
|
assert "exp" in payload
|
|
assert "iat" in payload
|
|
assert "jti" in payload
|
|
assert "aud" in payload
|
|
assert "iss" in payload
|
|
|
|
def test_creates_token_with_scopes(self):
|
|
from src.core.auth.jwt import create_access_token, decode_token
|
|
|
|
token = create_access_token({"sub": "user1", "scopes": ["read", "write"]})
|
|
payload = decode_token(token)
|
|
assert payload["scopes"] == ["read", "write"]
|
|
|
|
def test_creates_token_with_custom_expiry(self):
|
|
from src.core.auth.jwt import create_access_token, decode_token
|
|
|
|
token = create_access_token({"sub": "user1"}, expires_delta=timedelta(hours=2))
|
|
payload = decode_token(token)
|
|
assert payload["sub"] == "user1"
|
|
|
|
def test_token_contains_expected_claims(self):
|
|
from src.core.auth.jwt import create_access_token, decode_token, auth_config
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
payload = decode_token(token)
|
|
assert payload["aud"] == auth_config.JWT_AUDIENCE
|
|
assert payload["iss"] == auth_config.JWT_ISSUER
|
|
assert payload["sub"] == "user1"
|
|
|
|
|
|
class TestDecodeToken:
|
|
"""Verify decode_token @POST guarantees."""
|
|
|
|
def test_decode_valid_token(self):
|
|
from src.core.auth.jwt import create_access_token, decode_token
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
payload = decode_token(token)
|
|
assert payload["sub"] == "user1"
|
|
|
|
def test_decode_expired_token_raises(self):
|
|
from src.core.auth.jwt import decode_token, auth_config
|
|
from jose import jwt as jose_jwt
|
|
|
|
expired = jose_jwt.encode(
|
|
{
|
|
"sub": "user1",
|
|
"exp": 0,
|
|
"iat": 0,
|
|
"jti": "test-jti",
|
|
"aud": auth_config.JWT_AUDIENCE,
|
|
"iss": auth_config.JWT_ISSUER,
|
|
},
|
|
auth_config.SECRET_KEY,
|
|
algorithm=auth_config.ALGORITHM,
|
|
)
|
|
with pytest.raises(Exception):
|
|
decode_token(expired)
|
|
|
|
def test_decode_invalid_signature_raises(self):
|
|
from src.core.auth.jwt import decode_token
|
|
|
|
# Token with different key
|
|
bad_token = jose_jwt.encode(
|
|
{"sub": "user1", "exp": 9999999999, "iat": 0, "jti": "test", "aud": "test-audience", "iss": "test-issuer"},
|
|
"wrong-secret-key",
|
|
algorithm="HS256",
|
|
)
|
|
with pytest.raises(Exception):
|
|
decode_token(bad_token)
|
|
|
|
def test_decode_malformed_token_raises(self):
|
|
from src.core.auth.jwt import decode_token
|
|
|
|
with pytest.raises(Exception):
|
|
decode_token("not-a-jwt-token")
|
|
|
|
def test_decode_missing_required_claims_raises(self):
|
|
from src.core.auth.jwt import decode_token, auth_config
|
|
|
|
# Token missing 'sub' — jose may or may not enforce require for sub
|
|
# Test with a genuinely invalid token instead
|
|
with pytest.raises(Exception):
|
|
decode_token("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.missing-parts")
|
|
|
|
def test_decode_empty_token_raises(self):
|
|
from src.core.auth.jwt import decode_token
|
|
|
|
with pytest.raises(Exception):
|
|
decode_token("")
|
|
|
|
|
|
class TestHashToken:
|
|
"""Verify _hash_token utility."""
|
|
|
|
def test_hash_token_returns_sha256_hex(self):
|
|
from src.core.auth.jwt import _hash_token
|
|
|
|
result = _hash_token("test-token")
|
|
assert isinstance(result, str)
|
|
assert len(result) == 64 # SHA-256 hex digest length
|
|
|
|
def test_hash_token_is_deterministic(self):
|
|
from src.core.auth.jwt import _hash_token
|
|
|
|
assert _hash_token("same-token") == _hash_token("same-token")
|
|
|
|
def test_hash_token_different_for_different_inputs(self):
|
|
from src.core.auth.jwt import _hash_token
|
|
|
|
assert _hash_token("token-a") != _hash_token("token-b")
|
|
|
|
|
|
class TestPruneBlacklist:
|
|
"""Verify _prune_blacklist deletes expired entries."""
|
|
|
|
def test_deletes_expired_entries(self):
|
|
from src.core.auth.jwt import _prune_blacklist
|
|
from src.models.mapping import Base
|
|
from src.models.auth import TokenBlacklist
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(bind=engine)
|
|
db = Session(bind=engine)
|
|
|
|
past = datetime.now(UTC) - timedelta(hours=2)
|
|
db.add(TokenBlacklist(jti="expired1", token_hash="a" * 64, expires_at=past))
|
|
db.add(TokenBlacklist(jti="expired2", token_hash="b" * 64, expires_at=past))
|
|
db.commit()
|
|
|
|
_prune_blacklist(db)
|
|
|
|
remaining = db.query(TokenBlacklist).all()
|
|
assert len(remaining) == 0
|
|
|
|
def test_keeps_future_entries(self):
|
|
from src.core.auth.jwt import _prune_blacklist
|
|
from src.models.mapping import Base
|
|
from src.models.auth import TokenBlacklist
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(bind=engine)
|
|
db = Session(bind=engine)
|
|
|
|
future = datetime.now(UTC) + timedelta(hours=2)
|
|
db.add(TokenBlacklist(jti="valid1", token_hash="c" * 64, expires_at=future))
|
|
db.commit()
|
|
|
|
_prune_blacklist(db)
|
|
|
|
remaining = db.query(TokenBlacklist).all()
|
|
assert len(remaining) == 1
|
|
assert remaining[0].jti == "valid1"
|
|
|
|
def test_empty_db_does_not_raise(self):
|
|
from src.core.auth.jwt import _prune_blacklist
|
|
from src.models.mapping import Base
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(bind=engine)
|
|
db = Session(bind=engine)
|
|
|
|
# Should not raise
|
|
_prune_blacklist(db)
|
|
|
|
|
|
class TestBlacklistToken:
|
|
"""Verify blacklist_token @POST guarantees."""
|
|
|
|
@pytest.fixture
|
|
def db_session(self):
|
|
from src.models.mapping import Base
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session(bind=engine)
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_blacklist_valid_token(self, db_session):
|
|
from src.core.auth.jwt import create_access_token, blacklist_token, is_token_blacklisted
|
|
from src.models.auth import TokenBlacklist
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
blacklist_token(token, db_session)
|
|
|
|
assert is_token_blacklisted(token, db_session) is True
|
|
entry = db_session.query(TokenBlacklist).first()
|
|
assert entry is not None
|
|
assert entry.jti is not None
|
|
|
|
def test_blacklist_invalid_token_skips(self, db_session):
|
|
from src.core.auth.jwt import blacklist_token
|
|
from src.models.auth import TokenBlacklist
|
|
|
|
# Should return early without error
|
|
blacklist_token("invalid-token", db_session)
|
|
count = db_session.query(TokenBlacklist).count()
|
|
assert count == 0
|
|
|
|
def test_blacklist_existing_entry_skips(self, db_session):
|
|
from src.core.auth.jwt import create_access_token, blacklist_token
|
|
from src.models.auth import TokenBlacklist
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
blacklist_token(token, db_session)
|
|
# Blacklist again — should skip without error
|
|
blacklist_token(token, db_session)
|
|
count = db_session.query(TokenBlacklist).count()
|
|
assert count == 1
|
|
|
|
def test_blacklist_token_missing_jti_returns(self, db_session):
|
|
"""Covers line 133: decode succeeds but payload has no jti."""
|
|
from src.core.auth.jwt import blacklist_token
|
|
from src.models.auth import TokenBlacklist
|
|
from unittest.mock import patch
|
|
|
|
with patch("src.core.auth.jwt.decode_token", return_value={"sub": "user1"}):
|
|
blacklist_token("some-token", db_session)
|
|
count = db_session.query(TokenBlacklist).count()
|
|
assert count == 0
|
|
|
|
def test_blacklist_token_with_datetime_exp_works(self, db_session):
|
|
"""Covers line 136: exp_raw is a datetime instance."""
|
|
from src.core.auth.jwt import blacklist_token, is_token_blacklisted
|
|
from src.models.auth import TokenBlacklist
|
|
from unittest.mock import patch
|
|
from datetime import datetime, timedelta
|
|
|
|
# Use naive datetime since SQLite stores datetimes without timezone
|
|
exp_dt = datetime.now() + timedelta(hours=1)
|
|
with patch("src.core.auth.jwt.decode_token", return_value={
|
|
"sub": "user1", "jti": "test-jti-dt", "exp": exp_dt,
|
|
}):
|
|
blacklist_token("some-token", db_session)
|
|
entry = db_session.query(TokenBlacklist).filter(TokenBlacklist.jti == "test-jti-dt").first()
|
|
assert entry is not None
|
|
# SQLite stores datetime precision differently, compare timestamps
|
|
assert abs((entry.expires_at - exp_dt).total_seconds()) < 1
|
|
|
|
|
|
class TestIsTokenBlacklisted:
|
|
"""Verify is_token_blacklisted @POST guarantees."""
|
|
|
|
@pytest.fixture
|
|
def db_session(self):
|
|
from src.models.mapping import Base
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
|
event.listen(engine, "connect", lambda c, _: c.execute("PRAGMA foreign_keys=ON"))
|
|
Base.metadata.create_all(bind=engine)
|
|
session = Session(bind=engine)
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_not_blacklisted_returns_false(self, db_session):
|
|
from src.core.auth.jwt import create_access_token, is_token_blacklisted
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
assert is_token_blacklisted(token, db_session) is False
|
|
|
|
def test_blacklisted_returns_true(self, db_session):
|
|
from src.core.auth.jwt import create_access_token, blacklist_token, is_token_blacklisted
|
|
|
|
token = create_access_token({"sub": "user1"})
|
|
blacklist_token(token, db_session)
|
|
assert is_token_blacklisted(token, db_session) is True
|
|
|
|
def test_invalid_token_returns_true(self, db_session):
|
|
from src.core.auth.jwt import is_token_blacklisted
|
|
|
|
assert is_token_blacklisted("invalid-token", db_session) is True
|
|
|
|
def test_token_missing_jti_returns_false(self, db_session):
|
|
"""Covers line 171: decode succeeds but payload has no jti."""
|
|
from src.core.auth.jwt import is_token_blacklisted
|
|
from unittest.mock import patch
|
|
|
|
with patch("src.core.auth.jwt.decode_token", return_value={"sub": "user1"}):
|
|
result = is_token_blacklisted("some-token", db_session)
|
|
assert result is False
|
|
|
|
def test_deleted_token_returns_true(self, db_session):
|
|
from src.core.auth.jwt import is_token_blacklisted
|
|
|
|
# Token missing jti claim
|
|
assert is_token_blacklisted("malformed.token.here", db_session) is True
|
|
# #endregion Test.Auth.Jwt
|