""" Unit tests for DBSavesHandler platform filtering functionality. This module tests the platform filtering fixes for DBSavesHandler to ensure it properly filters by platform_id through the Rom relationship. """ from handler.database import db_save_handler from models.assets import Save from models.platform import Platform from models.rom import Rom from models.user import User class TestDBSavesHandlerPlatformFiltering: """Test suite for platform filtering in DBSavesHandler.""" def test_get_saves_without_platform_filter(self, admin_user: User, save: Save): """Test that get_saves returns all saves when no platform filter is applied.""" saves = db_save_handler.get_saves(user_id=admin_user.id) assert len(saves) >= 1 save_ids = [save.id for save in saves] assert save.id in save_ids def test_get_saves_with_platform_filter( self, admin_user: User, platform: Platform, save: Save ): """Test that get_saves filters correctly by platform_id.""" saves = db_save_handler.get_saves( user_id=admin_user.id, platform_id=platform.id ) assert len(saves) == 1 assert saves[0].id == save.id assert saves[0].file_name == "test_save.sav" def test_get_saves_with_rom_id_and_platform_filter( self, admin_user: User, platform: Platform, rom: Rom, save: Save ): """Test that get_saves works with both rom_id and platform_id filters.""" saves = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, platform_id=platform.id ) assert len(saves) == 1 assert saves[0].id == save.id def test_get_saves_with_nonexistent_platform(self, admin_user: User, save: Save): """Test that get_saves returns empty list for nonexistent platform.""" saves = db_save_handler.get_saves(user_id=admin_user.id, platform_id=999) assert len(saves) == 0 def test_platform_filtering_relationship_integrity( self, admin_user: User, platform: Platform, rom: Rom, save: Save ): """Test that platform filtering correctly uses the Rom relationship.""" assert rom.platform_id == platform.id saves_platform = db_save_handler.get_saves( user_id=admin_user.id, platform_id=platform.id ) assert len(saves_platform) == 1 def test_multiple_saves_same_platform(self, admin_user: User, rom: Rom): """Test filtering with multiple saves on the same platform.""" save1 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="save1.sav", file_name_no_tags="save1", file_name_no_ext="save1", file_extension="sav", emulator="emulator1", file_path=f"{rom.platform_slug}/saves/emulator1", file_size_bytes=100, ) save2 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="save2.sav", file_name_no_tags="save2", file_name_no_ext="save2", file_extension="sav", emulator="emulator2", file_path=f"{rom.platform_slug}/saves/emulator2", file_size_bytes=200, ) db_save_handler.add_save(save1) db_save_handler.add_save(save2) # Filter by platform should return both saves saves = db_save_handler.get_saves( user_id=admin_user.id, platform_id=rom.platform_id ) assert len(saves) == 2 save_names = [save.file_name for save in saves] assert "save1.sav" in save_names assert "save2.sav" in save_names def test_get_save_by_filename_with_platform_filter( self, admin_user: User, rom: Rom, save: Save ): """Test that get_save_by_filename works correctly with platform filtering.""" retrieved_save = db_save_handler.get_save_by_filename( user_id=admin_user.id, rom_id=rom.id, file_name=save.file_name ) assert retrieved_save is not None assert retrieved_save.id == save.id assert retrieved_save.file_name == save.file_name def test_platform_filtering_with_different_emulators( self, admin_user: User, platform: Platform, rom: Rom ): """Test platform filtering with saves from different emulators.""" save_emulator1 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="save_emu1.sav", file_name_no_tags="save_emu1", file_name_no_ext="save_emu1", file_extension="sav", emulator="emulator1", file_path=f"{platform.slug}/saves/emulator1", file_size_bytes=100, ) save_emulator2 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="save_emu2.sav", file_name_no_tags="save_emu2", file_name_no_ext="save_emu2", file_extension="sav", emulator="emulator2", file_path=f"{platform.slug}/saves/emulator2", file_size_bytes=200, ) db_save_handler.add_save(save_emulator1) db_save_handler.add_save(save_emulator2) # Filter by platform should return both saves regardless of emulator saves = db_save_handler.get_saves( user_id=admin_user.id, platform_id=platform.id ) assert len(saves) == 2 emulators = [save.emulator for save in saves] assert "emulator1" in emulators assert "emulator2" in emulators def test_get_save_by_id_with_platform_context( self, admin_user: User, platform: Platform, save: Save ): """Test that get_save works correctly and maintains platform context.""" retrieved_save = db_save_handler.get_save(user_id=admin_user.id, id=save.id) assert retrieved_save is not None assert retrieved_save.id == save.id assert retrieved_save.file_name == "test_save.sav" # Verify the save is associated with the correct platform through ROM assert retrieved_save.rom.platform_id == platform.id class TestDBSavesHandlerSlotFiltering: def test_get_saves_with_slot_filter(self, admin_user: User, rom: Rom): save1 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="slot_test_1.sav", file_name_no_tags="slot_test_1", file_name_no_ext="slot_test_1", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Slot A", ) save2 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="slot_test_2.sav", file_name_no_tags="slot_test_2", file_name_no_ext="slot_test_2", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Slot A", ) save3 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="slot_test_3.sav", file_name_no_tags="slot_test_3", file_name_no_ext="slot_test_3", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Slot B", ) db_save_handler.add_save(save1) db_save_handler.add_save(save2) db_save_handler.add_save(save3) slot_a_saves = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot="Slot A" ) assert len(slot_a_saves) == 2 assert all(s.slot == "Slot A" for s in slot_a_saves) slot_b_saves = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot="Slot B" ) assert len(slot_b_saves) == 1 assert slot_b_saves[0].slot == "Slot B" def test_get_saves_with_null_slot_filter(self, admin_user: User, rom: Rom): save_with_slot = Save( rom_id=rom.id, user_id=admin_user.id, file_name="with_slot.sav", file_name_no_tags="with_slot", file_name_no_ext="with_slot", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Main", ) save_without_slot = Save( rom_id=rom.id, user_id=admin_user.id, file_name="without_slot.sav", file_name_no_tags="without_slot", file_name_no_ext="without_slot", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot=None, ) db_save_handler.add_save(save_with_slot) db_save_handler.add_save(save_without_slot) all_saves = db_save_handler.get_saves(user_id=admin_user.id, rom_id=rom.id) assert len(all_saves) >= 2 def test_get_saves_order_by(self, admin_user: User, rom: Rom): from datetime import datetime, timedelta, timezone base_time = datetime.now(timezone.utc) save1 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="order_test_1.sav", file_name_no_tags="order_test_1", file_name_no_ext="order_test_1", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="order_test", ) save2 = Save( rom_id=rom.id, user_id=admin_user.id, file_name="order_test_2.sav", file_name_no_tags="order_test_2", file_name_no_ext="order_test_2", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="order_test", ) created1 = db_save_handler.add_save(save1) created2 = db_save_handler.add_save(save2) db_save_handler.update_save( created1.id, {"updated_at": base_time - timedelta(hours=2)} ) db_save_handler.update_save( created2.id, {"updated_at": base_time - timedelta(hours=1)} ) ordered_saves_desc = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot="order_test", order_by="updated_at", ) assert len(ordered_saves_desc) == 2 assert ordered_saves_desc[0].id == created2.id assert ordered_saves_desc[1].id == created1.id ordered_saves_asc = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot="order_test", order_by="updated_at", order_dir="asc", ) assert len(ordered_saves_asc) == 2 assert ordered_saves_asc[0].id == created1.id assert ordered_saves_asc[1].id == created2.id class TestDBSavesHandlerGetSaveByContentHash: """Pin the slot filter behavior of get_save_by_content_hash (commit 3d71ef3f6). Legacy callers still pass slot=None and expect cross-slot lookup, while sync negotiation passes a concrete slot value and expects a strict match. """ def test_returns_row_matching_slot_when_multiple_slots_share_hash( self, admin_user: User, rom: Rom ): """Two rows share (rom_id, user_id, content_hash) but differ in slot. A call with a specific slot must return the row with that slot, not the other.""" shared_hash = "abc123abc123abc123abc123abc12345" slot_a = db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="hash_match_a.sav", file_name_no_tags="hash_match_a", file_name_no_ext="hash_match_a", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Slot A", content_hash=shared_hash, ) ) slot_b = db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="hash_match_b.sav", file_name_no_tags="hash_match_b", file_name_no_ext="hash_match_b", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="Slot B", content_hash=shared_hash, ) ) result_a = db_save_handler.get_save_by_content_hash( user_id=admin_user.id, rom_id=rom.id, content_hash=shared_hash, slot="Slot A", ) assert result_a is not None assert result_a.id == slot_a.id result_b = db_save_handler.get_save_by_content_hash( user_id=admin_user.id, rom_id=rom.id, content_hash=shared_hash, slot="Slot B", ) assert result_b is not None assert result_b.id == slot_b.id def test_returns_none_when_slot_does_not_match(self, admin_user: User, rom: Rom): """A single row exists. Querying with a different slot value returns None, even though the (rom_id, user_id, content_hash) tuple matches.""" target_hash = "ffffffffffffffffffffffffffffffff" db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="only_slot.sav", file_name_no_tags="only_slot", file_name_no_ext="only_slot", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="autosave", content_hash=target_hash, ) ) result = db_save_handler.get_save_by_content_hash( user_id=admin_user.id, rom_id=rom.id, content_hash=target_hash, slot="manual", ) assert result is None def test_slot_none_finds_row_with_concrete_slot(self, admin_user: User, rom: Rom): """Legacy / cross-slot lookup: passing slot=None must not filter by slot at all and must still return the row, regardless of the row's own slot value.""" target_hash = "1234567890abcdef1234567890abcdef" created = db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="cross_slot.sav", file_name_no_tags="cross_slot", file_name_no_ext="cross_slot", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="autosave", content_hash=target_hash, ) ) result = db_save_handler.get_save_by_content_hash( user_id=admin_user.id, rom_id=rom.id, content_hash=target_hash, slot=None, ) assert result is not None assert result.id == created.id class TestDBSavesHandlerSlotNotNullFilter: """Pin the slot_not_null kwarg: when true, archival/null-slot saves are excluded; when false (default), they are returned alongside slot-bound rows. Sync callsites use slot_not_null=True so the database does the filter instead of pulling the whole table into Python.""" def test_slot_not_null_false_returns_both(self, admin_user: User, rom: Rom): slot_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="slotted.sav", file_name_no_tags="slotted", file_name_no_ext="slotted", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="autosave", ) archival_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="archival.sav", file_name_no_tags="archival", file_name_no_ext="archival", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot=None, ) db_save_handler.add_save(slot_save) db_save_handler.add_save(archival_save) saves = db_save_handler.get_saves(user_id=admin_user.id, rom_id=rom.id) names = {s.file_name for s in saves} assert "slotted.sav" in names assert "archival.sav" in names def test_slot_not_null_true_excludes_null_slot(self, admin_user: User, rom: Rom): slot_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="slotted_only.sav", file_name_no_tags="slotted_only", file_name_no_ext="slotted_only", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="autosave", ) archival_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="archival_only.sav", file_name_no_tags="archival_only", file_name_no_ext="archival_only", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot=None, ) db_save_handler.add_save(slot_save) db_save_handler.add_save(archival_save) saves = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot_not_null=True ) names = {s.file_name for s in saves} assert "slotted_only.sav" in names assert "archival_only.sav" not in names assert all(s.slot is not None for s in saves) def test_slot_not_null_true_composes_with_slot_value( self, admin_user: User, rom: Rom ): """slot_not_null and slot can both be set; the explicit slot filter wins (and is implicitly not-null), so slot_not_null=True is a no-op in that case. Pin that behavior so callers don't have to reason about the interaction.""" db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="compose_a.sav", file_name_no_tags="compose_a", file_name_no_ext="compose_a", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="A", ) ) db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="compose_b.sav", file_name_no_tags="compose_b", file_name_no_ext="compose_b", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="B", ) ) saves = db_save_handler.get_saves( user_id=admin_user.id, rom_id=rom.id, slot="A", slot_not_null=True ) assert len(saves) == 1 assert saves[0].slot == "A" class TestDBSavesHandlerGetSavesAfterId: """Cover keyset pagination used by the recompute_save_content_hashes maintenance task. Per-page bounded reads avoid materializing the full saves table on instances with very large libraries.""" def test_paginates_in_order_and_terminates(self, admin_user: User, rom: Rom): created_ids = [] for i in range(5): created = db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name=f"page_{i}.sav", file_name_no_tags=f"page_{i}", file_name_no_ext=f"page_{i}", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot=f"slot_{i}", ) ) created_ids.append(created.id) seen: list[int] = [] last_id = 0 while True: batch = db_save_handler.get_saves_after_id(after_id=last_id, limit=2) if not batch: break for s in batch: seen.append(s.id) last_id = s.id for cid in created_ids: assert cid in seen # IDs returned monotonically by primary key assert seen == sorted(seen) def test_after_id_excludes_anchor_row(self, admin_user: User, rom: Rom): created = db_save_handler.add_save( Save( rom_id=rom.id, user_id=admin_user.id, file_name="anchor.sav", file_name_no_tags="anchor", file_name_no_ext="anchor", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="anchor_slot", ) ) batch = db_save_handler.get_saves_after_id(after_id=created.id, limit=10) assert all(s.id > created.id for s in batch) class TestDBSavesHandlerSummary: def test_get_saves_summary_basic(self, admin_user: User, rom: Rom): from datetime import datetime, timedelta, timezone base_time = datetime.now(timezone.utc) configs = [ ("summary_a_1.sav", "Slot A", -3), ("summary_a_2.sav", "Slot A", -1), ("summary_b_1.sav", "Slot B", -2), ("summary_none_1.sav", None, -4), ] for filename, slot, hours_offset in configs: save = Save( rom_id=rom.id, user_id=admin_user.id, file_name=filename, file_name_no_tags=filename.replace(".sav", ""), file_name_no_ext=filename.replace(".sav", ""), file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot=slot, ) created = db_save_handler.add_save(save) db_save_handler.update_save( created.id, {"updated_at": base_time + timedelta(hours=hours_offset)} ) summary = db_save_handler.get_saves_summary( user_id=admin_user.id, rom_id=rom.id ) assert "total_count" in summary assert "slots" in summary assert summary["total_count"] == 4 assert len(summary["slots"]) == 3 def test_get_saves_summary_latest_per_slot(self, admin_user: User, rom: Rom): from datetime import datetime, timedelta, timezone base_time = datetime.now(timezone.utc) old_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="latest_test_old.sav", file_name_no_tags="latest_test_old", file_name_no_ext="latest_test_old", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="latest_test", ) new_save = Save( rom_id=rom.id, user_id=admin_user.id, file_name="latest_test_new.sav", file_name_no_tags="latest_test_new", file_name_no_ext="latest_test_new", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="latest_test", ) old_created = db_save_handler.add_save(old_save) new_created = db_save_handler.add_save(new_save) db_save_handler.update_save( old_created.id, {"updated_at": base_time - timedelta(hours=5)} ) db_save_handler.update_save( new_created.id, {"updated_at": base_time - timedelta(hours=1)} ) summary = db_save_handler.get_saves_summary( user_id=admin_user.id, rom_id=rom.id ) latest_slot = next( (s for s in summary["slots"] if s["slot"] == "latest_test"), None ) assert latest_slot is not None assert latest_slot["count"] == 2 assert latest_slot["latest"].file_name == "latest_test_new.sav" def test_get_saves_summary_empty_rom(self, admin_user: User): summary = db_save_handler.get_saves_summary( user_id=admin_user.id, rom_id=999999 ) assert summary["total_count"] == 0 assert summary["slots"] == [] def test_get_saves_summary_count_accuracy(self, admin_user: User, rom: Rom): for i in range(5): save = Save( rom_id=rom.id, user_id=admin_user.id, file_name=f"count_test_{i}.sav", file_name_no_tags=f"count_test_{i}", file_name_no_ext=f"count_test_{i}", file_extension="sav", emulator="test_emu", file_path=f"{rom.platform_slug}/saves", file_size_bytes=100, slot="count_test", ) db_save_handler.add_save(save) summary = db_save_handler.get_saves_summary( user_id=admin_user.id, rom_id=rom.id ) count_slot = next( (s for s in summary["slots"] if s["slot"] == "count_test"), None ) assert count_slot is not None assert count_slot["count"] == 5