473 lines
29 KiB
Python
473 lines
29 KiB
Python
import json
|
|
import unittest
|
|
from contextlib import closing
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.app import db
|
|
from backend.app.auth import get_current_user
|
|
from backend.app.clients.jellystat import JellystatClient
|
|
from backend.app.routers import identities
|
|
from backend.app.services import identity_review as review
|
|
from backend.app.services.jellyfin_identity import link_user, linked_user_id
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
|
|
JF = "a" * 32
|
|
OTHER = "b" * 32
|
|
SERVER = "c" * 32
|
|
ADMIN = {"username": "admin", "role": "admin"}
|
|
|
|
|
|
class IdentityReviewTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
db.create_user("Georgia", "jellyfin-user", auth_provider="jellyfin")
|
|
self.user_id = db.get_user_by_username("Georgia")["id"]
|
|
self.runtime = SimpleNamespace(jellyfin_base_url="http://jellyfin", jellyfin_api_key="SECRET-JF",
|
|
jellyseerr_base_url="http://seerr", jellyseerr_api_key="SECRET-SEERR",
|
|
jellystat_base_url="http://jellystat", jellystat_api_key="SECRET-STATS")
|
|
runtime_patch = patch.object(review, "get_runtime_settings", return_value=self.runtime)
|
|
runtime_patch.start()
|
|
self.addCleanup(runtime_patch.stop)
|
|
self.jf = {"state": "available", "server_id": SERVER, "users": [{"id": JF, "name": "Georgia"}]}
|
|
self.seerr = {"state": "available", "users": [{"id": 20, "name": "An unrelated display name", "jellyfin_id": JF}]}
|
|
self.js = {JF: {"state": "matched", "id": JF, "name": "Georgia"}}
|
|
|
|
def build(self):
|
|
local = review.read_snapshot()
|
|
return review.build_report(local, self.jf, self.seerr, self.js, self.runtime), local
|
|
|
|
def row(self, report):
|
|
return next(row for row in report["rows"] if row["user"]["id"] == self.user_id)
|
|
|
|
async def test_manual_selection_resolves_different_username_without_guessing(self):
|
|
self.jf['users'][0]['name'] = 'Different Jellyfin name'
|
|
before = review.read_snapshot()
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)['state'], 'unlinked')
|
|
report = review.build_report(before, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF})
|
|
self.assertTrue(self.row(report)['can_confirm'])
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
review.save_confirmations(report, before, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(linked_user_id('Georgia', self.runtime.jellyfin_base_url), JF)
|
|
self.assertEqual(self.row(self.build()[0])['state'], 'confirmed')
|
|
|
|
async def test_manual_selection_cannot_replace_stored_or_confirmed_identity(self):
|
|
self.jf['users'].append({'id': OTHER, 'name': 'Other'})
|
|
self.seerr['users'].append({'id': 21, 'name': 'Other', 'jellyfin_id': OTHER})
|
|
self.js[OTHER] = {'state': 'matched', 'id': OTHER}
|
|
report, local = self.build()
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
local = review.read_snapshot()
|
|
report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: OTHER})
|
|
self.assertFalse(self.row(report)['can_confirm'])
|
|
with self.assertRaises(HTTPException):
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(review.read_snapshot(), local)
|
|
|
|
async def test_manual_selection_checks_missing_ids_and_duplicate_owners(self):
|
|
self.jf['users'][0]['name'] = 'Different'
|
|
for state in ['missing', 'unavailable', 'not_configured']:
|
|
self.js[JF] = {'state': state}
|
|
report = review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF})
|
|
self.assertFalse(self.row(report)['can_confirm'])
|
|
self.js[JF] = {'state': 'matched', 'id': JF}
|
|
db.create_user('Owner', 'password', auth_provider='local', jellyseerr_user_id=20)
|
|
report = review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF})
|
|
self.assertEqual(self.row(report)['state'], 'conflict')
|
|
report = review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {self.user_id: OTHER})
|
|
self.assertFalse(self.row(report)['can_confirm'])
|
|
with self.assertRaises(HTTPException) as error:
|
|
review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {999: JF})
|
|
self.assertEqual(error.exception.status_code, 404)
|
|
|
|
async def test_resolution_rechecks_live_services_and_rejects_changed_selection(self):
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js):
|
|
before = review.read_snapshot()
|
|
preview = await review.resolve_identity(self.user_id, JF)
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
with self.assertRaises(HTTPException) as error:
|
|
await review.resolve_identity(self.user_id, OTHER, preview['revision'], ADMIN)
|
|
self.assertEqual(error.exception.status_code, 409)
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
result = await review.resolve_identity(self.user_id, JF, preview['revision'], ADMIN)
|
|
self.assertEqual(result['confirmed'], 1)
|
|
|
|
async def test_repair_replaces_wrong_local_link_and_records_before_after(self):
|
|
link_user('Georgia', OTHER, self.runtime.jellyfin_base_url)
|
|
db.set_user_jellyseerr_id('Georgia', 999)
|
|
before = review.read_snapshot()
|
|
report = review.build_report(before, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
|
|
self.assertTrue(self.row(report)['can_confirm'])
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
review.save_confirmations(report, before, self.runtime, [self.user_id], ADMIN, repair=True)
|
|
self.assertEqual(linked_user_id('Georgia', self.runtime.jellyfin_base_url), JF)
|
|
self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 20)
|
|
with closing(db._connect()) as conn:
|
|
audit = conn.execute('SELECT before_json,after_json,repaired_by FROM user_identity_repairs').fetchone()
|
|
self.assertEqual(json.loads(audit[0])['seerr_user_id'], 999)
|
|
self.assertEqual(json.loads(audit[1])['jellyfin_user_id'], JF)
|
|
self.assertEqual(audit[2], 'admin')
|
|
|
|
async def test_repair_preserves_duplicate_ownership_and_server_guards(self):
|
|
db.create_user('Owner', 'password', auth_provider='local', jellyseerr_user_id=20)
|
|
local = review.read_snapshot()
|
|
report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
|
|
self.assertFalse(self.row(report)['can_confirm'])
|
|
with self.assertRaises(HTTPException):
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.execute('DELETE FROM users WHERE username=?', ('Owner',))
|
|
report, local = self.build()
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.jf['server_id'] = OTHER
|
|
report = review.build_report(review.read_snapshot(), self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
|
|
self.assertFalse(self.row(report)['can_confirm'])
|
|
|
|
async def test_repair_does_not_invent_missing_seerr_identity(self):
|
|
self.seerr['users'][0]['jellyfin_id'] = OTHER
|
|
local = review.read_snapshot()
|
|
report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
|
|
self.assertEqual(self.row(report)['state'], 'unlinked')
|
|
with self.assertRaises(HTTPException):
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
|
|
self.assertEqual(local, review.read_snapshot())
|
|
|
|
async def test_repair_audit_failure_rolls_back_links(self):
|
|
link_user('Georgia', OTHER, self.runtime.jellyfin_base_url)
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.execute("CREATE TRIGGER fail_identity_audit BEFORE INSERT ON user_identity_repairs BEGIN SELECT RAISE(ABORT, 'fixture'); END")
|
|
local = review.read_snapshot()
|
|
report = review.build_report(local, self.jf, self.seerr, self.js, self.runtime, {self.user_id: JF}, repair=True)
|
|
with self.assertRaises(HTTPException):
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN, repair=True)
|
|
self.assertEqual(local, review.read_snapshot())
|
|
|
|
async def test_repair_rechecks_revision_and_updates_confirmed_ids(self):
|
|
report, local = self.build()
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.jf['users'][0]['id'] = OTHER
|
|
self.seerr['users'][0]['jellyfin_id'] = OTHER
|
|
self.js = {OTHER: {'state': 'matched', 'id': OTHER}}
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js):
|
|
preview = await review.repair_identity(self.user_id, OTHER)
|
|
with self.assertRaises(HTTPException):
|
|
await review.repair_identity(self.user_id, OTHER, 'f' * 64, ADMIN)
|
|
await review.repair_identity(self.user_id, OTHER, preview['revision'], ADMIN)
|
|
self.assertEqual(review.read_snapshot()['confirmations'][0]['jellyfin_user_id'], OTHER)
|
|
|
|
async def test_single_account_import_is_explicit_and_rechecked_before_local_save(self):
|
|
self.seerr['users'] = []
|
|
async def imported(*args, **kwargs):
|
|
self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': JF}]
|
|
return []
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
|
|
patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=imported) as post:
|
|
before = review.read_snapshot()
|
|
blocked = await review.repair_identity(self.user_id, JF)
|
|
self.assertFalse(blocked['row']['can_confirm'])
|
|
preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
|
|
self.assertEqual(preview['action'], 'import_seerr')
|
|
self.assertTrue(preview['row']['can_confirm'])
|
|
post.assert_not_called()
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
with self.assertRaises(HTTPException):
|
|
await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, False)
|
|
post.assert_not_called()
|
|
await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
|
|
post.assert_awaited_once_with('/api/v1/user/import-from-jellyfin', payload={'jellyfinUserIds': [JF]})
|
|
self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 25)
|
|
|
|
async def test_failed_import_never_writes_local_links_or_retries(self):
|
|
self.seerr['users'] = []
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
|
|
patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=httpx.ReadTimeout('fixture')) as post:
|
|
before = review.read_snapshot()
|
|
preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
|
|
with self.assertRaises(HTTPException) as error:
|
|
await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
|
|
self.assertEqual(error.exception.status_code, 502)
|
|
self.assertEqual(post.await_count, 1)
|
|
self.assertEqual(review.read_snapshot(), before)
|
|
|
|
async def test_import_preserves_upstream_account_when_local_save_is_blocked(self):
|
|
self.seerr['users'] = []
|
|
async def imported(*args, **kwargs):
|
|
self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': JF}]
|
|
db.set_user_jellyseerr_id('Georgia', 999)
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
|
|
patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock, side_effect=imported), \
|
|
patch.object(review.JellyseerrClient, 'delete_user', new_callable=AsyncMock) as delete:
|
|
preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
|
|
with self.assertRaises(HTTPException) as error:
|
|
await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
|
|
self.assertIn('Seerr import completed', error.exception.detail)
|
|
self.assertEqual(db.get_user_by_username('Georgia')['jellyseerr_user_id'], 999)
|
|
self.assertEqual(review.read_snapshot()['confirmations'], [])
|
|
delete.assert_not_called()
|
|
|
|
async def test_import_blocks_existing_name_with_different_jellyfin_id(self):
|
|
self.seerr['users'] = [{'id': 25, 'name': 'Georgia', 'jellyfin_id': OTHER}]
|
|
with patch.object(review, 'jellyfin_directory', new_callable=AsyncMock, return_value=self.jf), \
|
|
patch.object(review, 'seerr_directory', new_callable=AsyncMock, return_value=self.seerr), \
|
|
patch.object(review.JellystatClient, 'check_user_ids', new_callable=AsyncMock, return_value=self.js), \
|
|
patch.object(review.JellyseerrClient, 'post', new_callable=AsyncMock) as post:
|
|
preview = await review.repair_identity(self.user_id, JF, create_seerr=True)
|
|
self.assertFalse(preview['row']['can_confirm'])
|
|
with self.assertRaises(HTTPException):
|
|
await review.repair_identity(self.user_id, JF, preview['revision'], ADMIN, True)
|
|
post.assert_not_called()
|
|
|
|
async def test_georgia_preview_is_read_only_and_uses_seerr_jellyfin_id(self):
|
|
before = review.read_snapshot()
|
|
report, _ = self.build()
|
|
row = self.row(report)
|
|
self.assertEqual(row["basis"], "suggested_username")
|
|
self.assertEqual(row["state"], "ready")
|
|
self.assertEqual(row["seerr"][0]["id"], 20)
|
|
self.assertEqual(before, review.read_snapshot())
|
|
serialized = json.dumps(report)
|
|
for private in ["SECRET", "password_hash", "jellyfin_api_key", "email"]:
|
|
self.assertNotIn(private, serialized)
|
|
|
|
async def test_confirmation_persists_both_links_and_survives_legacy_sync(self):
|
|
report, local = self.build()
|
|
result = review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(result["confirmed"], 1)
|
|
self.assertEqual(linked_user_id("Georgia", self.runtime.jellyfin_base_url), JF)
|
|
self.assertEqual(db.get_user_by_username("Georgia")["jellyseerr_user_id"], 20)
|
|
saved = review.read_snapshot()["confirmations"][0]
|
|
self.assertEqual(saved["jellyfin_server_id"], SERVER)
|
|
self.assertEqual(saved["confirmed_by"], "admin")
|
|
db.set_user_jellyseerr_id("Georgia", 999)
|
|
link_user("Georgia", OTHER, "http://other-server")
|
|
self.assertEqual(db.get_user_by_username("Georgia")["jellyseerr_user_id"], 20)
|
|
self.assertIsNone(linked_user_id("Georgia", "http://other-server"))
|
|
refreshed, _ = self.build()
|
|
self.assertEqual(self.row(refreshed)["state"], "confirmed")
|
|
self.assertFalse(self.row(refreshed)["can_confirm"])
|
|
|
|
async def test_hidden_duplicate_seerr_and_jellyfin_rows_block_confirmation(self):
|
|
db.set_user_jellyseerr_id("Georgia", 20)
|
|
db.create_user("georgia@example.com", "jellyseerr-user", auth_provider="jellyseerr", jellyseerr_user_id=20)
|
|
report, local = self.build()
|
|
self.assertEqual(len(db.get_all_users()), 1)
|
|
self.assertEqual(len(report["rows"]), 2)
|
|
self.assertTrue(all(row["state"] == "conflict" for row in report["rows"]))
|
|
with self.assertRaises(HTTPException):
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(review.read_snapshot()["confirmations"], [])
|
|
|
|
async def test_whitespace_accounts_and_wrong_seerr_mapping_are_conflicts(self):
|
|
self.jf["users"].append({"id": OTHER, "name": "Georgia "})
|
|
self.seerr["users"].append({"id": 21, "name": "Georgia ", "jellyfin_id": OTHER})
|
|
db.set_user_jellyseerr_id("Georgia", 21)
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "conflict")
|
|
self.assertFalse(self.row(report)["can_confirm"])
|
|
|
|
async def test_case_duplicates_in_magent_remain_visible_and_blocked(self):
|
|
# Legacy duplicate predates the normalized-name creation guard.
|
|
with db._connect() as conn:
|
|
conn.execute("INSERT INTO users(username,password_hash,role,auth_provider,created_at) VALUES('georgia','unused','user','jellyfin','2026-01-01')")
|
|
report, _ = self.build()
|
|
self.assertEqual(report["counts"]["conflict"], 2)
|
|
self.assertEqual(report["counts"]["ready"], 0)
|
|
|
|
async def test_email_prefix_and_local_username_do_not_claim_an_identity(self):
|
|
db.create_user("Georgia@example.com", "jellyseerr-user", auth_provider="jellyseerr")
|
|
self.jf["users"].append({"id": OTHER, "name": "local"})
|
|
db.create_user("local", "password", auth_provider="local")
|
|
report, _ = self.build()
|
|
for row in report["rows"]:
|
|
if row["user"]["id"] != self.user_id:
|
|
self.assertIsNone(row["candidate_jellyfin_id"])
|
|
self.assertFalse(row["can_confirm"])
|
|
|
|
async def test_missing_or_unavailable_services_never_confirm(self):
|
|
for state in ["missing", "unavailable", "not_configured"]:
|
|
self.js[JF] = {"state": state}
|
|
report, _ = self.build()
|
|
self.assertFalse(self.row(report)["can_confirm"])
|
|
self.js = {JF: {"state": "matched", "id": JF}}
|
|
self.seerr = {"state": "unavailable", "users": []}
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "unavailable")
|
|
|
|
async def test_duplicate_upstream_id_and_orphaned_reservations_are_blocked(self):
|
|
self.seerr["users"].append({"id": 21, "name": "Other", "jellyfin_id": JF})
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "conflict")
|
|
self.seerr["users"].pop()
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.execute("INSERT INTO jellyfin_user_links VALUES (?,?,?)", (review.source_key(self.runtime.jellyfin_base_url), 9999, JF))
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "conflict")
|
|
|
|
async def test_wrong_stored_id_is_not_silently_replaced(self):
|
|
link_user("Georgia", OTHER, self.runtime.jellyfin_base_url)
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "conflict")
|
|
self.assertEqual(self.row(report)["candidate_jellyfin_id"], OTHER)
|
|
|
|
async def test_account_changes_reject_whole_save(self):
|
|
report, local = self.build()
|
|
db.set_user_jellyseerr_id("Georgia", 99)
|
|
with self.assertRaises(HTTPException) as raised:
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
self.assertEqual(review.read_snapshot()["confirmations"], [])
|
|
self.assertIsNone(linked_user_id("Georgia", self.runtime.jellyfin_base_url))
|
|
|
|
async def test_settings_changes_reject_save(self):
|
|
report, local = self.build()
|
|
db.set_setting("jellyfin_base_url", "http://changed")
|
|
with self.assertRaises(HTTPException) as raised:
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
|
|
async def test_save_reads_real_runtime_settings_inside_transaction(self):
|
|
from backend.app.runtime import get_runtime_settings
|
|
for key in review.CONFIG_KEYS:
|
|
db.set_setting(key, getattr(self.runtime, key))
|
|
report, local = self.build()
|
|
with patch.object(review, "get_runtime_settings", get_runtime_settings):
|
|
result = review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.assertEqual(result["confirmed"], 1)
|
|
|
|
async def test_batch_rolls_back_all_links_if_a_later_write_fails(self):
|
|
db.create_user("Second", "jellyfin-user", auth_provider="jellyfin")
|
|
second_id = db.get_user_by_username("Second")["id"]
|
|
self.jf["users"].append({"id": OTHER, "name": "Second"})
|
|
self.seerr["users"].append({"id": 21, "name": "Second", "jellyfin_id": OTHER})
|
|
self.js[OTHER] = {"state": "matched", "id": OTHER}
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.execute(f"""CREATE TRIGGER fail_second_confirmation BEFORE INSERT ON user_identity_confirmations
|
|
WHEN NEW.local_user_id={second_id} BEGIN SELECT RAISE(ABORT, 'fixture conflict'); END""")
|
|
report, local = self.build()
|
|
with self.assertRaises(HTTPException) as raised:
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id, second_id], ADMIN)
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
after = review.read_snapshot()
|
|
self.assertEqual(after["confirmations"], [])
|
|
self.assertEqual(after["links"], [])
|
|
self.assertTrue(all(row["jellyseerr_user_id"] is None for row in after["users"]))
|
|
|
|
async def test_confirmation_rechecks_live_report_and_rejects_stale_revision(self):
|
|
report, local = self.build()
|
|
changed = {**report, "revision": "f" * 64}
|
|
with patch.object(review, "review_identities", new_callable=AsyncMock, return_value=(changed, local, self.runtime)), \
|
|
patch.object(review, "save_confirmations") as save:
|
|
with self.assertRaises(HTTPException) as raised:
|
|
await review.confirm_identities(report["revision"], [self.user_id], ADMIN)
|
|
self.assertEqual(raised.exception.status_code, 409)
|
|
save.assert_not_called()
|
|
|
|
async def test_different_server_cannot_reuse_confirmed_id(self):
|
|
report, local = self.build()
|
|
review.save_confirmations(report, local, self.runtime, [self.user_id], ADMIN)
|
|
self.jf["server_id"] = OTHER
|
|
report, _ = self.build()
|
|
self.assertEqual(self.row(report)["state"], "conflict")
|
|
|
|
async def test_report_revision_ignores_time_but_detects_mapping_changes(self):
|
|
a, _ = self.build()
|
|
b, _ = self.build()
|
|
self.assertEqual(a["revision"], b["revision"])
|
|
self.seerr["users"][0]["id"] = 99
|
|
c, _ = self.build()
|
|
self.assertNotEqual(a["revision"], c["revision"])
|
|
|
|
async def test_seerr_directory_requires_complete_unique_pages(self):
|
|
users = [{"id": i, "jellyfinUserId": f"{i:032x}", "displayName": f"User {i}"} for i in range(1, 102)]
|
|
pages = [{"pageInfo": {"results": 101}, "results": users[:100]}, {"pageInfo": {"results": 101}, "results": users[100:]}]
|
|
with patch.object(review.JellyseerrClient, "get_users", new_callable=AsyncMock, side_effect=pages) as get:
|
|
result = await review.seerr_directory(self.runtime)
|
|
self.assertEqual(result["state"], "available")
|
|
self.assertEqual(len(result["users"]), 101)
|
|
self.assertEqual(get.await_args.kwargs["skip"], 100)
|
|
for broken in [[], users[:2], users[:1] * 100]:
|
|
with patch.object(review.JellyseerrClient, "get_users", new_callable=AsyncMock, return_value={"pageInfo": {"results": 101}, "results": broken}):
|
|
self.assertEqual((await review.seerr_directory(self.runtime))["state"], "unavailable")
|
|
|
|
async def test_jellyfin_server_and_directory_must_agree(self):
|
|
with patch.object(review.JellyfinClient, "get_system_info", new_callable=AsyncMock, return_value={"Id": SERVER}), \
|
|
patch.object(review.JellyfinClient, "get_users", new_callable=AsyncMock, return_value=[{"Id": JF, "Name": "Georgia", "ServerId": OTHER}]):
|
|
self.assertEqual((await review.jellyfin_directory(self.runtime))["state"], "unavailable")
|
|
|
|
|
|
class JellystatIdentityTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_unconfigured_client_never_calls_upstream(self):
|
|
with patch("backend.app.clients.jellystat.httpx.AsyncClient") as http:
|
|
result = await JellystatClient(None, None).check_user_ids([JF])
|
|
http.assert_not_called()
|
|
self.assertEqual(result[JF]["state"], "not_configured")
|
|
|
|
async def test_missing_wrong_and_failed_ids_are_distinguished_without_details(self):
|
|
ids = [f"{i:032x}" for i in range(1, 6)]
|
|
def handler(request):
|
|
self.assertEqual(request.headers["x-api-token"], "PRIVATE")
|
|
user_id = json.loads(request.content)["userid"]
|
|
if user_id == ids[0]: return httpx.Response(200, json={"Id": user_id, "Name": "Georgia", "PRIVATE": "hidden"})
|
|
if user_id == ids[1]: return httpx.Response(200, content=b"")
|
|
if user_id == ids[2]: return httpx.Response(200, json={"Id": OTHER})
|
|
if user_id == ids[3]: return httpx.Response(401, text="PRIVATE error")
|
|
return httpx.Response(503, text="PRIVATE error")
|
|
real = httpx.AsyncClient
|
|
with patch("backend.app.clients.jellystat.httpx.AsyncClient", side_effect=lambda **kwargs: real(transport=httpx.MockTransport(handler), **kwargs)):
|
|
data = await JellystatClient("http://jellystat", "PRIVATE").check_user_ids(ids)
|
|
self.assertEqual([data[key]["state"] for key in ids], ["matched", "missing", "unavailable", "unavailable", "unavailable"])
|
|
self.assertNotIn("PRIVATE", json.dumps(data))
|
|
|
|
|
|
class IdentityRouteTests(unittest.TestCase):
|
|
def client(self, role=None):
|
|
app = FastAPI()
|
|
app.include_router(identities.router)
|
|
if role:
|
|
app.dependency_overrides[get_current_user] = lambda: {"username": "viewer", "role": role}
|
|
return TestClient(app)
|
|
|
|
def test_admin_only_read_and_write(self):
|
|
for role, status in [(None, 401), ("user", 403)]:
|
|
client = self.client(role)
|
|
self.assertEqual(client.get("/admin/identities").status_code, status)
|
|
self.assertEqual(client.post("/admin/identities/confirm", json={"revision": "a" * 64, "user_ids": [1]}).status_code, status)
|
|
|
|
def test_resolution_requires_admin_and_strict_ids(self):
|
|
for endpoint in ['check', 'confirm']:
|
|
body = {'user_id': 1, 'jellyfin_user_id': JF}
|
|
if endpoint == 'confirm': body['revision'] = 'a' * 64
|
|
for role, status in [(None, 401), ('user', 403)]:
|
|
self.assertEqual(self.client(role).post('/admin/identities/resolve/' + endpoint, json=body).status_code, status)
|
|
self.assertEqual(self.client(role).post('/admin/identities/repair/' + endpoint, json=body).status_code, status)
|
|
for invalid in [{'user_id': True}, {'jellyfin_user_id': 'invalid'}, {'seerr_user_id': 22}]:
|
|
self.assertEqual(self.client('admin').post('/admin/identities/resolve/' + endpoint, json={**body, **invalid}).status_code, 422)
|
|
self.assertEqual(self.client('admin').post('/admin/identities/repair/' + endpoint, json={**body, **invalid}).status_code, 422)
|
|
with patch.object(identities, 'resolve_identity', new_callable=AsyncMock, return_value={'row': {}}):
|
|
result = self.client('admin').post('/admin/identities/resolve/check', json={'user_id': 1, 'jellyfin_user_id': JF})
|
|
self.assertEqual(result.headers['cache-control'], 'no-store')
|
|
|
|
def test_no_store_and_no_browser_supplied_identity(self):
|
|
with patch.object(identities, "review_identities", new_callable=AsyncMock, return_value=({"rows": []}, {}, None)):
|
|
response = self.client("admin").get("/admin/identities")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.headers["cache-control"], "no-store")
|
|
for body in [{"revision": "a" * 64, "user_ids": [1, 1]}, {"revision": "a" * 64, "user_ids": []},
|
|
{"revision": "a" * 64, "user_ids": [1], "jellyfin_id": OTHER}]:
|
|
self.assertEqual(self.client("admin").post("/admin/identities/confirm", json=body).status_code, 422)
|