test(usage-export): freeze e2e models, compare raw non-admin metadata, mark query locals Final

This commit is contained in:
jesus 2026-11-03 19:04:45 +00:00
parent e964397a22
commit e374b32825
2 changed files with 17 additions and 11 deletions

View file

@ -35,7 +35,7 @@ pytestmark = pytest.mark.e2e
EXPORT_ROUTE: Final = "/user/daily/activity/export"
class UsageExportParams(BaseModel):
class UsageExportParams(BaseModel, frozen=True):
start_date: str
end_date: str
api_key: str
@ -44,24 +44,30 @@ class UsageExportParams(BaseModel):
format: str = "json"
class UsageExportKeyMetadata(BaseModel):
class UsageExportKeyMetadata(BaseModel, frozen=True):
tag: str | None = None
webhook_token: str | None = None
class UsageExportUserMetadata(BaseModel):
class UsageExportUserMetadata(BaseModel, frozen=True):
department: str | None = None
ldap_password: str | None = None
class UsageExportRow(BaseModel):
class UsageExportRow(BaseModel, frozen=True):
api_key: str | None = None
user_id: str | None = None
key_metadata: UsageExportKeyMetadata
user_metadata: UsageExportUserMetadata
class UsageExportRawMetadataRow(BaseModel, frozen=True):
key_metadata: dict[str, object]
user_metadata: dict[str, object]
_EXPORT_ROWS: Final = TypeAdapter(list[UsageExportRow])
_RAW_METADATA_ROWS: Final = TypeAdapter(list[UsageExportRawMetadataRow])
class TestUsageExportMetadata:
@ -169,9 +175,9 @@ class TestUsageExportMetadata:
headers=client.proxy.transport.bearer(record.key),
)
assert own_view.status_code == 200, f"{EXPORT_ROUTE} answered {own_view.status_code}: {own_view.body}"
own_rows: Final = _EXPORT_ROWS.validate_json(own_view.body)
own_rows: Final = _RAW_METADATA_ROWS.validate_json(own_view.body)
assert own_rows, f"{EXPORT_ROUTE} returned no rows for the key's own user: {own_view.body}"
for row in own_rows:
assert (row.key_metadata, row.user_metadata) == (UsageExportKeyMetadata(), UsageExportUserMetadata()), (
assert (row.key_metadata, row.user_metadata) == ({}, {}), (
f"a caller without admin view got metadata objects: {row}"
)

View file

@ -413,7 +413,7 @@ def test_query_builders_reject_nonpositive_limits(builder) -> None:
def test_export_groups_by_requested_key_and_binds_cursor_after_scope(
export_type: ExportType, group_key: str, key_filter: str, joins: str
) -> None:
query = build_export_sql(
query: Final = build_export_sql(
_scope(entity_ids=None),
export_type=export_type,
after=ExportCursor(date="2026-01-12", entity_id="user-2", group_key="group-3"),
@ -444,7 +444,7 @@ def test_export_groups_by_requested_key_and_binds_cursor_after_scope(
@pytest.mark.parametrize("export_type", [ExportType.DAILY_WITH_KEYS, ExportType.DAILY_WITH_USERS])
def test_export_uses_latest_deleted_key_metadata(export_type: ExportType) -> None:
query = build_export_sql(_scope(entity_ids=None), export_type=export_type, after=None, batch_size=2)
query: Final = build_export_sql(_scope(entity_ids=None), export_type=export_type, after=None, batch_size=2)
assert 'FROM "LiteLLM_DeletedVerificationToken"' in query.sql
assert "ORDER BY deleted_at DESC" in query.sql
@ -475,7 +475,7 @@ _EMPTY_USER_METADATA: Final = "'{}'::jsonb AS user_metadata"
def test_export_selects_redacted_key_and_user_metadata_objects_for_admin_view(
export_type: ExportType, key_metadata: str, user_metadata: str
) -> None:
query = build_export_sql(
query: Final = build_export_sql(
_scope(entity_ids=None), export_type=export_type, after=None, batch_size=2, include_metadata=True
)
@ -489,7 +489,7 @@ def test_export_selects_redacted_key_and_user_metadata_objects_for_admin_view(
def test_export_without_metadata_selects_empty_objects_and_never_reads_metadata_columns(
export_type: ExportType,
) -> None:
query = build_export_sql(_scope(entity_ids=None), export_type=export_type, after=None, batch_size=2)
query: Final = build_export_sql(_scope(entity_ids=None), export_type=export_type, after=None, batch_size=2)
assert _EMPTY_KEY_METADATA in query.sql
assert _EMPTY_USER_METADATA in query.sql
@ -499,7 +499,7 @@ def test_export_without_metadata_selects_empty_objects_and_never_reads_metadata_
@pytest.mark.parametrize("export_type", tuple(ExportType))
def test_export_without_cursor_omits_cursor_predicate_and_parameters(export_type: ExportType) -> None:
query = build_export_sql(
query: Final = build_export_sql(
_scope(entity_ids=None),
export_type=export_type,
after=None,