rommapp/romm
1
0
Fork 0
mirror of https://github.com/rommapp/romm.git synced 2026-10-12 02:21:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): check collection ownership in the ROM filters (#5201)

Browse Source 
* fix(security): check collection ownership in the ROM filters

The collection endpoints refuse a private collection to anyone but its
owner, but the ROM listing endpoints accepted the same id as a
collection_id or smart_collection_id filter without checking it. Any
user (or a kiosk visitor) could list, randomly pick from and bulk
download another user's private collections and Favourites, and re-share
one through a public smart collection scoped to it.

The membership filters now match only collections that are public or
owned by the caller. A collection_id inside smart collection criteria is
checked against the smart collection's owner, both when the criteria are
saved and when they are evaluated.

Reported by Takaya Sato (dagadjno).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* fix(security): clear stale smart collection caches across owners

Addresses Greptile's review on #5201. Smart collections cache their
membership on the row, so one scoped to another user's private
collection kept serving the members it cached before the ownership
check. Migration 0152 empties those caches on upgrade, and sharing or
unsharing a collection now refreshes the smart collections scoped to
it. Also covers a second user listing a public smart collection scoped
to its owner's private collection.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* fix(collections): harden smart collection scope refreshes (code-review)

Allow re-saving a smart collection whose stored scope was since lost,
reject non-object criteria in from_stored_criteria, guard the scoped
refresh, and refresh smart collections when a collection's ROMs are
replaced or the collection is deleted.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* refactor(collections): share the refresh wrapper and trim scope tests (simplify)

Move the scoped smart collection refresh next to the existing wrapper,
match the scope on the raw criteria before refreshing, batch the
migration update, and fold the duplicate unshare test into helpers.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* docs(migrations): describe the cleared smart scopes by current behavior (review-polish)

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

* docs(roms): state the collection filter comment plainly (humanizer)

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

---------

Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Georges-Antoine Assi
2026-10-09 13:20:26 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 09d8360337
commit f89f6ef185
11 changed files with 567 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/alembic/versions/0152_cross_user_smart_scopes.py
+77
View File
@@ -0,0 +1,77 @@
"""clear smart collections scoped to another user's private collection
Revision ID: 0152_cross_user_smart_scopes
Revises: 0151_state_content_hash
Create Date: 2026-10-09 00:00:00.000000
"""
import sqlalchemy as sa
from alembic import op
from utils.database import CustomJSON
# revision identifiers, used by Alembic.
revision = "0152_cross_user_smart_scopes"
down_revision = "0151_state_content_hash"
branch_labels = None
depends_on = None
collections = sa.table(
"collections",
sa.column("id", sa.Integer),
sa.column("user_id", sa.Integer),
sa.column("is_public", sa.Boolean),
)
smart_collections = sa.table(
"smart_collections",
sa.column("id", sa.Integer),
sa.column("user_id", sa.Integer),
sa.column("filter_criteria", CustomJSON()),
sa.column("rom_ids", CustomJSON()),
sa.column("path_covers_small", CustomJSON()),
sa.column("path_covers_large", CustomJSON()),
)
def upgrade() -> None:
# A scope the owner can't see matches nothing, so its cached membership is
# emptied now rather than left readable until the next refresh.
conn = op.get_bind()
visible = {
row.id: (row.user_id, bool(row.is_public))
for row in conn.execute(sa.select(collections))
}
rows = conn.execute(
sa.select(
smart_collections.c.id,
smart_collections.c.user_id,
smart_collections.c.filter_criteria,
)
).all()
stale_ids = []
for smart_id, owner_id, criteria in rows:
if not isinstance(criteria, dict):
continue
try:
collection_id = int(criteria.get("collection_id") or 0)
except (TypeError, ValueError):
continue
if not collection_id:
continue
user_id, is_public = visible.get(collection_id, (None, False))
if not is_public and user_id != owner_id:
stale_ids.append(smart_id)
if stale_ids:
conn.execute(
sa.update(smart_collections)
.where(smart_collections.c.id.in_(stale_ids))
.values(rom_ids=[], path_covers_small=[], path_covers_large=[])
)
def downgrade() -> None:
pass
backend/endpoints/collections.py
+43 -3
View File
@@ -14,7 +14,10 @@ from endpoints.responses.collection import (
SmartCollectionSchema,
VirtualCollectionSchema,
)
from endpoints.roms import refresh_affected_smart_collections
from endpoints.roms import (
refresh_affected_smart_collections,
refresh_scoped_smart_collections,
)
from exceptions.endpoint_exceptions import (
CollectionAlreadyExistsException,
CollectionNotFoundInDatabaseException,
@@ -24,6 +27,7 @@ from handler.audit_handler import AuditTarget, changed_fields, record
from handler.auth.constants import Scope
from handler.auth.dependencies import get_rom_visibility_filter
from handler.database import db_collection_handler, db_rom_handler
from handler.database.rom_filters import RomFilterParams
from handler.filesystem import fs_resource_handler
from handler.filesystem.assets_handler import validate_image_upload
from handler.filesystem.base_handler import CoverSize
@@ -90,6 +94,23 @@ def _record_collection(
record(action, request, AuditTarget.of_collection(collection), data)
def _criteria_collection_id(criteria: object) -> int | None:
filters = RomFilterParams.from_stored_criteria(criteria)
return filters.collection_id if filters else None
def _check_criteria_collection(
criteria: object, user_id: int, kept_id: int | None = None
) -> None:
"""Refuse criteria newly scoped to a collection the owner can't see."""
collection_id = _criteria_collection_id(criteria)
if not collection_id or collection_id == kept_id:
return
collection = db_collection_handler.get_collection(collection_id)
if not collection or (collection.user_id != user_id and not collection.is_public):
raise CollectionNotFoundInDatabaseException(collection_id)
@protected_route(router.post, "", [Scope.COLLECTIONS_WRITE])
async def add_collection(
request: Request,
@@ -191,6 +212,7 @@ async def add_smart_collection(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid JSON for filter_criteria field",
) from e
_check_criteria_collection(parsed_filter_criteria, request.user.id)
cleaned_data = {
"name": name,
@@ -529,6 +551,8 @@ async def update_collection(
changed = changed_fields(
collection, cleaned_data, ("name", "description", "is_public")
)
if "is_public" in changed:
refresh_scoped_smart_collections(id)
new_artwork = artwork is not None and artwork.filename is not None
if (
remove_cover
@@ -537,6 +561,10 @@ async def update_collection(
):
changed.append("cover")
before_ids, after_ids = set(collection.rom_ids), set(updated_collection.rom_ids)
if before_ids != after_ids:
refresh_affected_smart_collections(
sorted(before_ids ^ after_ids), membership_only=True
)
if changed or before_ids != after_ids:
_record_collection(
request,
@@ -571,9 +599,13 @@ def update_collection_visibility(
if collection.user_id != request.user.id:
raise CollectionPermissionError(id)
return CollectionSchema.model_validate(
db_collection_handler.update_collection(id, {"is_public": is_public})
updated_collection = db_collection_handler.update_collection(
id, {"is_public": is_public}
)
if collection.is_public != is_public:
refresh_scoped_smart_collections(id)
return CollectionSchema.model_validate(updated_collection)
class CollectionRomsPayload(PydanticBaseModel):
@@ -701,6 +733,13 @@ async def update_smart_collection(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid JSON for filter_criteria field",
) from e
# The form resends the stored scope on every edit; the ROM filters
# already match nothing for one the owner has since lost.
_check_criteria_collection(
parsed_filter_criteria,
request.user.id,
kept_id=_criteria_collection_id(smart_collection.filter_criteria),
)
cleaned_data = {
"name": name if name is not None else smart_collection.name,
@@ -768,6 +807,7 @@ async def delete_collection(
log.info(f"Deleting {hl(collection.name, color=BLUE)} from database")
db_collection_handler.delete_collection(id)
refresh_scoped_smart_collections(id)
_record_collection(request, AuditAction.COLLECTION_DELETE, collection)
try:
backend/endpoints/roms/__init__.py
+10 -9
View File
@@ -231,6 +231,15 @@ def refresh_affected_smart_collections(
log.error(f"Couldn't refresh smart collections for {rom_ids}: {e}")
def refresh_scoped_smart_collections(collection_id: int) -> None:
"""Like `refresh_affected_smart_collections`, for the smart collections
scoped to a collection whose visibility or existence changed."""
try:
db_collection_handler.refresh_smart_collections_scoped_to(collection_id)
except Exception as e:
log.error(f"Couldn't refresh smart collections scoped to {collection_id}: {e}")
def build_unscoped_sidecar_cache_key(
user_id: int,
order_by: str,
@@ -916,7 +925,6 @@ def get_random_rom(
def _bulk_download_target(
user_id: int,
platform_id: int | None,
collection_id: int | None,
smart_collection_id: int | None,
@@ -938,13 +946,7 @@ def _bulk_download_target(
collection = db_collection_handler.get_smart_collection(smart_collection_id)
else:
return None
if collection is None:
return None
target = AuditTarget.of_collection(collection)
if collection.is_public or collection.user_id == user_id:
return target
# Someone else's private collection is kept by id; its name stays theirs.
return AuditTarget(target.type, target.id, None)
return AuditTarget.of_collection(collection) if collection else None
@protected_route(
@@ -1053,7 +1055,6 @@ async def download_roms(
record_download(
request,
lambda: _bulk_download_target(
request.user.id,
platform_id,
collection_id,
smart_collection_id,
backend/handler/database/collections_handler.py
+20 -1
View File
@@ -1,5 +1,5 @@
import functools
from collections.abc import Callable, Sequence
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime, timezone
from typing import Any, cast
@@ -546,6 +546,7 @@ class DBCollectionsHandler(DBBaseHandler):
query=query,
filters=filters,
user_id=user_id,
collection_user_id=smart_collection.user_id,
include_related=False,
session=session,
)
@@ -649,6 +650,24 @@ class DBCollectionsHandler(DBBaseHandler):
return len(ids)
@begin_session
def refresh_smart_collections_scoped_to(
self,
collection_id: int,
session: Session = INJECTED_SESSION,
) -> None:
"""Refresh the smart collections whose criteria name `collection_id`, since
sharing, unsharing or deleting it changes what their criteria match."""
rows = session.execute(
select(SmartCollection.id, SmartCollection.filter_criteria)
).all()
for smart_collection_id, criteria in rows:
# A raw match, so rows with unrelated invalid criteria stay unparsed.
if isinstance(criteria, Mapping) and str(
criteria.get("collection_id")
) == str(collection_id):
self.refresh_smart_collection(smart_collection_id, session=session)
@begin_session
def refresh_smart_collections_for_roms(
self,
backend/handler/database/rom_filters.py
+5 -3
View File
@@ -372,15 +372,17 @@ class RomFilterParams(BaseModel):
] = None
@classmethod
def from_stored_criteria(
cls, criteria: Mapping[str, Any]
) -> "RomFilterParams | None":
def from_stored_criteria(cls, criteria: object) -> "RomFilterParams | None":
"""Build from a smart collection's stored `filter_criteria`, or None
when a stored value cannot be honoured.
`smart_collection_id` is dropped: the create dialog records the route it
was opened from, so following it would nest (and could cycle).
"""
if not isinstance(criteria, Mapping):
log.warning("Smart collection criteria rejected: not an object")
return None
values: dict[str, Any] = dict(criteria)
values.pop("smart_collection_id", None)
backend/handler/database/roms_handler.py
+19 -7
View File
@@ -1120,14 +1120,17 @@ class DBRomsHandler(DBBaseHandler):
return query.filter(Rom.platform_id.in_(platform_ids))
def _filter_by_collection_id[S: Select[*tuple[Any, ...]]](
self, query: S, collection_id: int
self, query: S, collection_id: int, user_id: int | None
) -> S:
# `collections_roms` is keyed on (collection_id, rom_id), so membership
# is an indexed subquery rather than a list of ids fetched into Python.
# Membership stays an indexed subquery, so no ids are fetched into Python.
# Another user's private collection matches nothing, like an unknown id.
return query.filter(
Rom.id.in_(
select(CollectionRom.rom_id).where(
CollectionRom.collection_id == collection_id
select(CollectionRom.rom_id)
.join(Collection, Collection.id == CollectionRom.collection_id)
.where(
CollectionRom.collection_id == collection_id,
or_(Collection.is_public, Collection.user_id == user_id),
)
)
)
@@ -1157,7 +1160,9 @@ class DBRomsHandler(DBBaseHandler):
smart_collection = db_collection_handler.get_smart_collection(
smart_collection_id, session=session
)
if not smart_collection:
if not smart_collection or not (
smart_collection.is_public or smart_collection.user_id == user_id
):
return query.filter(false())
# MariaDB's semi-join plan for this IN (subquery) rescans the whole
@@ -1541,6 +1546,9 @@ class DBRomsHandler(DBBaseHandler):
order_by: str = "",
order_dir: str = "asc",
user_id: int | None = None,
# Whose access a `collection_id` filter is checked against, when not
# `user_id`: a smart collection's criteria answer for its owner.
collection_user_id: int | None = None,
updated_after: datetime | None = None,
released_days: Sequence[tuple[int, int]] | None = None,
released_before_year: int | None = None,
@@ -1614,7 +1622,11 @@ class DBRomsHandler(DBBaseHandler):
query = self._filter_by_platform_ids(query, filters.platform_ids)
if filters.collection_id:
query = self._filter_by_collection_id(query, filters.collection_id)
query = self._filter_by_collection_id(
query,
filters.collection_id,
collection_user_id if collection_user_id is not None else user_id,
)
if filters.virtual_collection_id:
query = self._filter_by_virtual_collection_id(
backend/tests/endpoints/roms/test_download_audit.py
+4 -4
View File
@@ -170,7 +170,7 @@ def test_a_bulk_download_is_one_event_on_its_platform(
assert event.data["count"] == 1
def test_someone_elses_private_collection_is_kept_by_id_only(
def test_someone_elses_private_collection_is_not_downloaded(
client: TestClient,
access_token: str,
editor_user: User,
@@ -188,11 +188,11 @@ def test_someone_elses_private_collection_is_kept_by_id_only(
)
db_collection_handler.add_roms_to_collection(private.id, [rom.id])
client.get(
response = client.get(
"/api/roms/download",
params={"collection_id": private.id},
headers=_auth(access_token),
)
[event] = recorded_events()
assert (event.target_id, event.target_name) == (str(private.id), None)
assert response.status_code == 404
assert recorded_events() == []
backend/tests/endpoints/test_collection.py
+319
View File
@@ -911,3 +911,322 @@ class TestCollectionVisibility:
assert refreshed is not None
assert refreshed.is_public is True
assert refreshed.filter_criteria == criteria
def _scope_to(
collection: Collection, owner: User, is_public: bool = False
) -> SmartCollection:
return db_collection_handler.add_smart_collection(
SmartCollection(
name="Scoped",
description="",
user_id=owner.id,
is_public=is_public,
filter_criteria={"collection_id": collection.id},
)
)
def _cached_rom_ids(smart_collection_id: int) -> list[int]:
smart_collection = db_collection_handler.get_smart_collection(smart_collection_id)
assert smart_collection is not None
return smart_collection.rom_ids
class TestRomFiltersRespectCollectionPrivacy:
"""The ROM endpoints take a collection id as a filter, so they must refuse
the same private collections the collection endpoints do."""
@pytest.fixture
def private_collection(self, admin_user: User, rom: Rom) -> Collection:
collection = db_collection_handler.add_collection(
Collection(name="Secret", description="", user_id=admin_user.id)
)
db_collection_handler.add_roms_to_collection(collection.id, [rom.id])
return collection
def test_listing_another_users_private_collection_is_empty(
self, client, other_user_token: str, private_collection: Collection
):
response = client.get(
f"/api/roms?collection_id={private_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["items"] == []
def test_random_pick_from_another_users_private_collection_is_null(
self, client, other_user_token: str, private_collection: Collection
):
response = client.get(
f"/api/roms/random?collection_id={private_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.json() is None
def test_downloading_another_users_private_collection_is_refused(
self, client, other_user_token: str, private_collection: Collection
):
response = client.get(
f"/api/roms/download?collection_id={private_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_another_users_public_collection_is_listed(
self,
client,
other_user_token: str,
private_collection: Collection,
rom: Rom,
):
db_collection_handler.update_collection(
private_collection.id, {"is_public": True}
)
response = client.get(
f"/api/roms?collection_id={private_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert [item["id"] for item in response.json()["items"]] == [rom.id]
def test_another_users_private_smart_collection_is_empty(
self, client, other_user_token: str, admin_user: User, rom: Rom
):
smart_collection = db_collection_handler.add_smart_collection(
SmartCollection(
name="Mine",
description="",
user_id=admin_user.id,
filter_criteria={"platform_ids": [rom.platform_id]},
)
)
response = client.get(
f"/api/roms?smart_collection_id={smart_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.json()["items"] == []
def test_smart_collection_cannot_be_scoped_to_another_users_private_collection(
self, client, other_user_token: str, private_collection: Collection
):
response = client.post(
"/api/collections/smart?is_public=true",
data={
"name": "Pivot",
"filter_criteria": json.dumps({"collection_id": private_collection.id}),
},
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_stored_scope_to_another_users_private_collection_matches_nothing(
self,
client,
access_token: str,
editor_user: User,
private_collection: Collection,
):
smart_collection = _scope_to(private_collection, editor_user, is_public=True)
refreshed = db_collection_handler.refresh_smart_collection(smart_collection.id)
response = client.get(
f"/api/roms?smart_collection_id={smart_collection.id}",
headers={"Authorization": f"Bearer {access_token}"},
)
assert refreshed is not None
assert refreshed.rom_ids == []
assert response.json()["items"] == []
def test_owner_can_scope_a_smart_collection_to_their_private_collection(
self,
client,
access_token: str,
private_collection: Collection,
rom: Rom,
):
response = client.post(
"/api/collections/smart",
data={
"name": "Within secret",
"filter_criteria": json.dumps({"collection_id": private_collection.id}),
},
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert response.json()["rom_ids"] == [rom.id]
def test_owners_public_smart_scope_to_their_private_collection_is_shared(
self,
client,
other_user_token: str,
admin_user: User,
private_collection: Collection,
rom: Rom,
):
smart_collection = _scope_to(private_collection, admin_user, is_public=True)
response = client.get(
f"/api/roms?smart_collection_id={smart_collection.id}",
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert [item["id"] for item in response.json()["items"]] == [rom.id]
def test_unsharing_a_collection_empties_other_owners_smart_scopes(
self,
client,
access_token: str,
editor_user: User,
private_collection: Collection,
):
db_collection_handler.update_collection(
private_collection.id, {"is_public": True}
)
# Unparseable criteria elsewhere must not block this refresh.
db_collection_handler.add_smart_collection(
SmartCollection(
name="Broken",
description="",
user_id=editor_user.id,
filter_criteria=[1],
)
)
smart_collection = _scope_to(private_collection, editor_user, is_public=True)
shared = db_collection_handler.refresh_smart_collection(smart_collection.id)
response = client.put(
f"/api/collections/{private_collection.id}/visibility",
json={"is_public": False},
headers={"Authorization": f"Bearer {access_token}"},
)
assert shared is not None and shared.rom_ids != []
assert response.status_code == status.HTTP_200_OK
assert _cached_rom_ids(smart_collection.id) == []
def test_sharing_through_the_edit_form_fills_other_owners_smart_scopes(
self,
client,
access_token: str,
editor_user: User,
private_collection: Collection,
rom: Rom,
):
smart_collection = _scope_to(private_collection, editor_user)
response = client.put(
f"/api/collections/{private_collection.id}",
params={"is_public": True},
data={"rom_ids": json.dumps([rom.id])},
headers={"Authorization": f"Bearer {access_token}"},
)
assert response.status_code == status.HTTP_200_OK
assert _cached_rom_ids(smart_collection.id) == [rom.id]
def test_replacing_a_collections_roms_refreshes_its_smart_scopes(
self,
client,
access_token: str,
admin_user: User,
private_collection: Collection,
rom: Rom,
):
smart_collection = _scope_to(private_collection, admin_user)
before = db_collection_handler.refresh_smart_collection(smart_collection.id)
response = client.put(
f"/api/collections/{private_collection.id}",
data={"rom_ids": "[]"},
headers={"Authorization": f"Bearer {access_token}"},
)
assert before is not None and before.rom_ids == [rom.id]
assert response.status_code == status.HTTP_200_OK
assert _cached_rom_ids(smart_collection.id) == []
def test_deleting_a_collection_empties_its_smart_scopes(
self,
client,
access_token: str,
admin_user: User,
private_collection: Collection,
):
smart_collection = _scope_to(private_collection, admin_user)
before = db_collection_handler.refresh_smart_collection(smart_collection.id)
response = client.delete(
f"/api/collections/{private_collection.id}",
headers={"Authorization": f"Bearer {access_token}"},
)
assert before is not None and before.rom_ids != []
assert response.status_code == status.HTTP_200_OK
assert _cached_rom_ids(smart_collection.id) == []
def test_a_smart_collection_that_lost_its_scope_can_still_be_renamed(
self,
client,
access_token: str,
other_user_token: str,
editor_user: User,
private_collection: Collection,
):
db_collection_handler.update_collection(
private_collection.id, {"is_public": True}
)
criteria = {"collection_id": private_collection.id}
smart_collection = _scope_to(private_collection, editor_user)
unshared = client.put(
f"/api/collections/{private_collection.id}/visibility",
json={"is_public": False},
headers={"Authorization": f"Bearer {access_token}"},
)
response = client.put(
f"/api/collections/smart/{smart_collection.id}",
data={"name": "Renamed", "filter_criteria": json.dumps(criteria)},
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert unshared.status_code == status.HTTP_200_OK
assert response.status_code == status.HTTP_200_OK
assert response.json()["name"] == "Renamed"
assert response.json()["rom_ids"] == []
def test_smart_collection_cannot_be_rescoped_to_another_users_private_collection(
self,
client,
other_user_token: str,
editor_user: User,
private_collection: Collection,
):
smart_collection = db_collection_handler.add_smart_collection(
SmartCollection(
name="Mine",
description="",
user_id=editor_user.id,
filter_criteria={"matched": True},
)
)
response = client.put(
f"/api/collections/smart/{smart_collection.id}",
data={
"filter_criteria": json.dumps({"collection_id": private_collection.id})
},
headers={"Authorization": f"Bearer {other_user_token}"},
)
assert response.status_code == status.HTTP_404_NOT_FOUND
backend/tests/handler/database/test_collections_handler.py
+21 -1
View File
@@ -157,12 +157,32 @@ def test_filter_roms_by_collection(platform: Platform, admin_user: User):
collection.id, [rom.id for rom in in_collection]
)
roms = db_rom_handler.get_roms_scalar(collection_id=collection.id)
roms = db_rom_handler.get_roms_scalar(
collection_id=collection.id, user_id=admin_user.id
)
assert {rom.id for rom in roms} == {rom.id for rom in in_collection}
assert outside.id not in {rom.id for rom in roms}
def test_filter_roms_by_another_users_private_collection_returns_nothing(
platform: Platform, admin_user: User, editor_user: User
):
rom = _add_rom(platform, 0, {})
collection = db_collection_handler.add_collection(
Collection(name="Shelf", description="", user_id=admin_user.id)
)
db_collection_handler.add_roms_to_collection(collection.id, [rom.id])
assert db_rom_handler.get_roms_scalar(collection_id=collection.id) == []
assert (
db_rom_handler.get_roms_scalar(
collection_id=collection.id, user_id=editor_user.id
)
== []
)
def test_filter_roms_by_unknown_collection_returns_nothing(
platform: Platform, admin_user: User
):
backend/tests/handler/database/test_rom_filter_params.py
+6
View File
@@ -149,6 +149,12 @@ class TestFromStoredCriteria:
is None
)
@pytest.mark.parametrize("not_an_object", [[1], "genres", 5, None])
def test_criteria_that_are_not_an_object_are_rejected_rather_than_raised(
self, not_an_object: object
):
assert RomFilterParams.from_stored_criteria(not_an_object) is None
class TestRegistryColumns:
def test_every_filter_matches_against_the_facets_mirror(self):
backend/tests/test_migrations.py
+43 -1
View File
@@ -24,7 +24,7 @@ import models
from handler.database import db_collection_handler, db_rom_handler
from handler.database.base_handler import sync_engine
from models.base import BaseModel
from models.collection import SmartCollection
from models.collection import Collection, SmartCollection
from models.platform import Platform
from models.rom import FULL_PATH_HASH_LENGTH, Rom, compute_full_path_hash
from models.user import User
@@ -341,6 +341,48 @@ def test_the_state_content_hash_revision_reverses_and_replays():
assert _schema_of(connection, "states") == before
def test_the_cross_user_smart_scopes_revision_empties_only_those_caches(
admin_user: User, editor_user: User
):
migration = _load_migration("0152_cross_user_smart_scopes.py")
private, public = (
db_collection_handler.add_collection(
Collection(
name=name, description="", user_id=admin_user.id, is_public=is_public
)
)
for name, is_public in (("Secret", False), ("Shared", True))
)
borrowed, own, borrowed_public = (
db_collection_handler.add_smart_collection(
SmartCollection(
name="Scoped",
user_id=owner.id,
rom_ids=[1, 2],
path_covers_small=["cover"],
filter_criteria={"collection_id": scope.id},
)
)
for owner, scope in (
(editor_user, private),
(admin_user, private),
(editor_user, public),
)
)
with sync_engine.begin() as connection:
with Operations.context(MigrationContext.configure(connection)):
migration.upgrade()
cleared = db_collection_handler.get_smart_collection(borrowed.id)
assert cleared is not None
assert (cleared.rom_ids, cleared.path_covers_small) == ([], [])
for kept_id in (own.id, borrowed_public.id):
kept = db_collection_handler.get_smart_collection(kept_id)
assert kept is not None
assert (kept.rom_ids, kept.path_covers_small) == ([1, 2], ["cover"])
def test_the_user_oidc_sub_revision_reverses_and_replays():
migration = _load_migration("0144_user_oidc_sub.py")