Warnings fixed: - datetime.utcnow() → datetime.now(UTC) across 48+ files (src/ + tests/) - datetime.utcnow (callback ref) → lambda: datetime.now(UTC) in model fields (18 files) - Pydantic class Config → model_config = ConfigDict(...) (16 files) - Pydantic .dict() → .model_dump() (8 files) - ConfigDict(allow_population_by_field_name=True) → validate_by_name=True - SQLAlchemy declarative_base() import path updated - FastAPI on_event → lifespan context manager (app.py) - Import sorting (ruff I001) auto-fixed across all files - Fixed broken re-export chains that ruff F401 cleanup broke: _validate_bcp47: service.py now imports from dictionary_validation directly job_to_response: _job_routes.py and test imports from service_utils directly fetch_datasource_metadata: restored re-export in service.py - Added missing TranslateJobService import in _job_routes.py (was deleted by F401) - Added ConfigDict(protected_namespaces=()) for DashboardDatasetItem schema field - pytest.ini: replaced deprecated importmode with asyncio_mode All 440 tests pass with zero deprecation warnings.
124 lines
4.0 KiB
Python
124 lines
4.0 KiB
Python
# #region TestAPIKeyModel [C:2] [TYPE Module] [SEMANTICS test, api_key, model]
|
|
# @BRIEF Contract tests for the APIKey SQLAlchemy model — creation, hash storage, and field constraints.
|
|
# @RELATION BINDS_TO -> [APIKeyModel]
|
|
# @TEST_CONTRACT: APIKey model stores SHA-256 hash and prefix; raw key never persisted.
|
|
# @TEST_EDGE: key_hash is unique and indexed
|
|
# @TEST_EDGE: prefix is exactly 11 chars ("ssk_" + 7)
|
|
# @TEST_EDGE: active defaults to True
|
|
import pytest
|
|
|
|
|
|
# #region clean_db [C:1] [TYPE Fixture]
|
|
# @BRIEF In-memory SQLite fixture for APIKey model tests.
|
|
@pytest.fixture
|
|
def clean_db():
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from src.models.mapping import Base
|
|
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
yield session
|
|
session.close()
|
|
# #endregion clean_db
|
|
|
|
|
|
class TestAPIKeyModel:
|
|
"""Verify APIKey model fields and constraints."""
|
|
|
|
# #region test_create_api_key [C:2] [TYPE Function]
|
|
# @BRIEF Create APIKey with required fields and verify defaults.
|
|
def test_create_api_key(self, clean_db):
|
|
from src.models.api_key import APIKey
|
|
|
|
api_key = APIKey(
|
|
key_hash="a" * 64,
|
|
prefix="ssk_abc1234",
|
|
name="Test Key",
|
|
permissions=["maintenance:start"],
|
|
)
|
|
clean_db.add(api_key)
|
|
clean_db.commit()
|
|
clean_db.refresh(api_key)
|
|
|
|
assert api_key.id is not None
|
|
assert api_key.key_hash == "a" * 64
|
|
assert api_key.prefix == "ssk_abc1234"
|
|
assert api_key.name == "Test Key"
|
|
assert api_key.permissions == ["maintenance:start"]
|
|
assert api_key.active is True
|
|
assert api_key.created_at is not None
|
|
assert api_key.environment_id is None
|
|
assert api_key.expires_at is None
|
|
assert api_key.last_used_at is None
|
|
# #endregion test_create_api_key
|
|
|
|
# #region test_key_hash_unique [C:2] [TYPE Function]
|
|
# @BRIEF key_hash is unique — duplicate raises IntegrityError.
|
|
def test_key_hash_unique(self, clean_db):
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from src.models.api_key import APIKey
|
|
|
|
clean_db.add(APIKey(
|
|
key_hash="b" * 64,
|
|
prefix="ssk_xyz7890",
|
|
name="Key 1",
|
|
permissions=["maintenance:start"],
|
|
))
|
|
clean_db.commit()
|
|
|
|
with pytest.raises(IntegrityError):
|
|
clean_db.add(APIKey(
|
|
key_hash="b" * 64, # Same hash
|
|
prefix="ssk_def5678",
|
|
name="Key 2",
|
|
permissions=["maintenance:end"],
|
|
))
|
|
clean_db.commit()
|
|
# #endregion test_key_hash_unique
|
|
|
|
# #region test_prefix_length [C:2] [TYPE Function]
|
|
# @BRIEF prefix is exactly 11 characters.
|
|
def test_prefix_length(self, clean_db):
|
|
from src.models.api_key import APIKey
|
|
|
|
api_key = APIKey(
|
|
key_hash="c" * 64,
|
|
prefix="ssk_short1", # 10 chars — should be ok in SQL
|
|
name="Short Prefix",
|
|
permissions=["maintenance:start"],
|
|
)
|
|
clean_db.add(api_key)
|
|
clean_db.commit()
|
|
# The DB allows any prefix length; the app layer enforces 11 chars
|
|
# via generate_api_key. Model just stores what it's given.
|
|
assert len(api_key.prefix) <= 11
|
|
# #endregion test_prefix_length
|
|
|
|
# #region test_active_default_true [C:2] [TYPE Function]
|
|
# @BRIEF active column defaults to True.
|
|
def test_active_default_true(self, clean_db):
|
|
from src.models.api_key import APIKey
|
|
|
|
api_key = APIKey(
|
|
key_hash="d" * 64,
|
|
prefix="ssk_def1234",
|
|
name="Default Active",
|
|
permissions=[],
|
|
)
|
|
clean_db.add(api_key)
|
|
clean_db.commit()
|
|
|
|
assert api_key.active is True
|
|
# #endregion test_active_default_true
|
|
# #endregion TestAPIKeyModel
|