|
11 | 11 | from handler.database import db_device_handler, db_play_session_handler, db_rom_handler |
12 | 12 | from models.device import Device |
13 | 13 | from models.platform import Platform |
14 | | -from models.rom import Rom |
| 14 | +from models.rom import Rom, RomUserStatus |
15 | 15 | from models.user import User |
16 | 16 | from utils.datetime import to_utc |
17 | 17 |
|
@@ -68,6 +68,14 @@ def _ingest(device_id=None, sessions=None): |
68 | 68 | return {"device_id": device_id, "sessions": sessions or []} |
69 | 69 |
|
70 | 70 |
|
| 71 | +def _prime_rom_user(rom_id, user_id, **fields): |
| 72 | + """Fetch (or create) the rom_user and seed it with the given fields.""" |
| 73 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom_id, user_id=user_id) |
| 74 | + if rom_user is None: |
| 75 | + rom_user = db_rom_handler.add_rom_user(rom_id=rom_id, user_id=user_id) |
| 76 | + return db_rom_handler.update_rom_user(rom_user.id, fields) |
| 77 | + |
| 78 | + |
71 | 79 | class TestPlaySessionIngest: |
72 | 80 | def test_single_session(self, client, access_token: str, device: Device, rom: Rom): |
73 | 81 | payload = _ingest(device_id=device.id, sessions=[_session(rom_id=rom.id)]) |
@@ -472,6 +480,96 @@ def test_last_played_is_max_end_time( |
472 | 480 | last_played_utc = to_utc(rom_user.last_played) |
473 | 481 | assert abs((last_played_utc - expected_latest).total_seconds()) < 2 |
474 | 482 |
|
| 483 | + def test_now_playing_set_on_ingest( |
| 484 | + self, client, access_token: str, admin_user: User, rom: Rom |
| 485 | + ): |
| 486 | + payload = _ingest(sessions=[_session(rom_id=rom.id)]) |
| 487 | + client.post( |
| 488 | + "/api/play-sessions", |
| 489 | + json=payload, |
| 490 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 491 | + ) |
| 492 | + |
| 493 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 494 | + assert rom_user is not None |
| 495 | + assert rom_user.now_playing is True |
| 496 | + |
| 497 | + def test_empty_status_rewound_to_incomplete( |
| 498 | + self, client, access_token: str, admin_user: User, rom: Rom |
| 499 | + ): |
| 500 | + client.post( |
| 501 | + "/api/play-sessions", |
| 502 | + json=_ingest(sessions=[_session(rom_id=rom.id)]), |
| 503 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 504 | + ) |
| 505 | + |
| 506 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 507 | + assert rom_user.status == RomUserStatus.INCOMPLETE |
| 508 | + |
| 509 | + def test_finished_status_rewound_to_incomplete( |
| 510 | + self, client, access_token: str, admin_user: User, rom: Rom |
| 511 | + ): |
| 512 | + _prime_rom_user(rom.id, admin_user.id, status=RomUserStatus.FINISHED) |
| 513 | + |
| 514 | + client.post( |
| 515 | + "/api/play-sessions", |
| 516 | + json=_ingest(sessions=[_session(rom_id=rom.id)]), |
| 517 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 518 | + ) |
| 519 | + |
| 520 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 521 | + assert rom_user.status == RomUserStatus.INCOMPLETE |
| 522 | + |
| 523 | + @pytest.mark.parametrize( |
| 524 | + "protected_status", |
| 525 | + [ |
| 526 | + RomUserStatus.COMPLETED_100, |
| 527 | + RomUserStatus.RETIRED, |
| 528 | + RomUserStatus.NEVER_PLAYING, |
| 529 | + ], |
| 530 | + ) |
| 531 | + def test_deliberate_status_preserved( |
| 532 | + self, client, access_token, admin_user, rom, protected_status |
| 533 | + ): |
| 534 | + _prime_rom_user(rom.id, admin_user.id, status=protected_status) |
| 535 | + |
| 536 | + client.post( |
| 537 | + "/api/play-sessions", |
| 538 | + json=_ingest(sessions=[_session(rom_id=rom.id)]), |
| 539 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 540 | + ) |
| 541 | + |
| 542 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 543 | + # now_playing is orthogonal, so it still flips on... |
| 544 | + assert rom_user.now_playing is True |
| 545 | + # ...but the deliberate enum status is left untouched. |
| 546 | + assert rom_user.status == protected_status |
| 547 | + |
| 548 | + def test_older_session_does_not_resurrect_now_playing( |
| 549 | + self, client, access_token: str, admin_user: User, rom: Rom |
| 550 | + ): |
| 551 | + # A recent play sets last_played and now_playing. |
| 552 | + client.post( |
| 553 | + "/api/play-sessions", |
| 554 | + json=_ingest(sessions=[_session(rom_id=rom.id, start_offset_hours=-1)]), |
| 555 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 556 | + ) |
| 557 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 558 | + db_rom_handler.update_rom_user( |
| 559 | + rom_user.id, {"now_playing": False, "status": RomUserStatus.FINISHED} |
| 560 | + ) |
| 561 | + |
| 562 | + # Backfilling an older session must not advance state. |
| 563 | + client.post( |
| 564 | + "/api/play-sessions", |
| 565 | + json=_ingest(sessions=[_session(rom_id=rom.id, start_offset_hours=-10)]), |
| 566 | + headers={"Authorization": f"Bearer {access_token}"}, |
| 567 | + ) |
| 568 | + |
| 569 | + rom_user = db_rom_handler.get_rom_user(rom_id=rom.id, user_id=admin_user.id) |
| 570 | + assert rom_user.now_playing is False |
| 571 | + assert rom_user.status == RomUserStatus.FINISHED |
| 572 | + |
475 | 573 | def test_rom_user_created_if_not_exists( |
476 | 574 | self, client, access_token: str, admin_user: User, platform: Platform |
477 | 575 | ): |
|
0 commit comments