"""实体合并的墓碑（migration 0040）：旧 id 解析到并进去的那一条。 账本之外按实体 id 记下的东西（头像文件名、复核 CSV、页面上的 `/entity-image?id=`）在合并后 还拿着旧 id。墓碑让它们跳到现在的实体；目标没了就和目标一样查不到，不悬空。 """ from __future__ import annotations import importlib.util import io import json import shutil import sqlite3 import tempfile import unittest from pathlib import Path from unittest import mock from urllib.parse import parse_qs, urlsplit from PIL import Image, ImageDraw from peach.entities import (memoized_resolution, merge_entity, normalize_entity_name, prefer_entity_name, resolve_entity, resolve_entity_id, upsert_asset_entity) from peach.web_review import _use_canonical_entity_names from support.ledger import fresh_ledger HAS_DEPS = all(importlib.util.find_spec(name) for name in ("fastapi", "httpx")) def add_person(connection, entity_id: int, name: str) -> None: connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,'performer',?,?,'t','t')", (entity_id, name, name.casefold())) def merge(connection, source_id: int, target_id: int, name: str) -> dict: moved = merge_entity(connection, target_id=target_id, source_id=source_id, source_name=name, alias_source="test:merge") connection.commit() return moved def picture(colour: str) -> bytes: buffer = io.BytesIO() image = Image.new("RGB", (40, 60), colour) ImageDraw.Draw(image).rectangle((0, 0, 40, 30), fill="black") image.save(buffer, format="JPEG") return buffer.getvalue() class EntityRedirectTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.db = fresh_ledger(self.tmp.name) # 和服务、脚本的连接一样不开外键：ON DELETE 声明不会替我们收拾墓碑。 self.con = sqlite3.connect(self.db) self.addCleanup(self.con.close) for entity_id, name in ((10, "橋本ありな"), (11, "新ありな"), (12, "桥本有菜")): add_person(self.con, entity_id, name) self.con.commit() def redirects(self) -> list[tuple]: return self.con.execute( "SELECT old_id,target_id FROM entity_redirect ORDER BY old_id").fetchall() def test_a_read_request_resolves_each_name_once(self): """只读请求里同一个名字只查一次库；块外照旧每次现查，写路径看得到刚写的实体。""" queries = [] self.con.set_trace_callback(queries.append) with memoized_resolution(): first = [resolve_entity(self.con, "performer", "橋本ありな")[0] for _ in range(3)] within = len(queries) resolve_entity(self.con, "performer", "橋本ありな") resolve_entity(self.con, "performer", "橋本ありな") self.con.set_trace_callback(None) self.assertEqual(first, [10, 10, 10]) self.assertEqual(within, 1) self.assertEqual(len(queries) - within, 2) def test_resolution_matches_exactly_and_never_scans_the_name_tables(self): """解析按原文逐字比对：只差大小写的写法认不出，别名同理；两条查询都走索引，不扫整张表。""" add_person(self.con, 20, "Mia") self.con.execute("INSERT INTO entity_alias(entity_id,alias,normalized_alias,source,confidence)" " VALUES(20,'MIA Nanasawa','mia nanasawa','test',1.0)") queries = [] self.con.set_trace_callback(queries.append) found = [resolve_entity(self.con, "performer", name) for name in ("Mia", "mia", "MIA Nanasawa", "mia nanasawa")] self.con.set_trace_callback(None) self.assertEqual([row[0] if row else None for row in found], [20, None, 20, None]) plans = [detail for sql in queries for *_, detail in self.con.execute("EXPLAIN QUERY PLAN " + sql)] self.assertFalse([detail for detail in plans if detail.startswith("SCAN")], plans) # 规范名那条只用上 `kind` 前缀时也不算扫表，却仍要把同类实体逐个比对；两个归一列都得用上。 for key in ("normalized_name=?", "normalized_alias=?"): self.assertTrue([detail for detail in plans if key in detail], plans) def test_entity_writes_keep_the_normalized_name_the_resolver_looks_up(self): """建实体、合并留别名、改用别名作规范名之后，归一列都等于对原文现算的结果，名字都解析得到。""" self.con.row_factory = sqlite3.Row entity_id = upsert_asset_entity(self.con, kind="performer", name=" Yua\u200b Mikami ", asset_id=1, role="performer", source="test") first = self.con.execute("SELECT canonical_name FROM entity WHERE id=?", (entity_id,)).fetchone()[0] merge(self.con, 11, entity_id, "新ありな") prefer_entity_name(self.con, entity_id, "新ありな") rows = [*self.con.execute("SELECT canonical_name,normalized_name FROM entity"), *self.con.execute("SELECT alias,normalized_alias FROM entity_alias")] self.assertEqual([(text, key) for text, key in rows if normalize_entity_name(text) != key], []) self.assertEqual(resolve_entity(self.con, "performer", "新ありな")[0], entity_id) self.assertEqual(resolve_entity(self.con, "performer", first)[0], entity_id) def test_the_merged_id_resolves_to_the_target(self): merge(self.con, 11, 10, "新ありな") self.assertEqual(resolve_entity_id(self.con, 11), 10) self.assertEqual(resolve_entity_id(self.con, 10), 10) self.assertIsNone(resolve_entity_id(self.con, 999)) self.assertEqual(self.con.execute( "SELECT source FROM entity_redirect WHERE old_id=11").fetchone(), ("test:merge",)) def test_a_chain_of_merges_points_every_old_id_at_the_last_target(self): merge(self.con, 11, 10, "新ありな") moved = merge(self.con, 10, 12, "橋本ありな") self.assertEqual(self.redirects(), [(10, 12), (11, 12)]) self.assertEqual(moved["redirects"], 1) self.assertEqual(resolve_entity_id(self.con, 11), 12) self.assertEqual(self.con.execute("PRAGMA foreign_key_check").fetchall(), []) def test_deleting_the_target_removes_its_redirects(self): """目标被删（标签改名、清理脚本都直接删 entity）时旧 id 和目标一样查不到。 悬空的墓碑会让下一次合并后的 `foreign_key_check` 不为 0，ADR-0064 的后继按它回滚。 """ merge(self.con, 11, 10, "新ありな") self.con.execute("DELETE FROM entity WHERE id=10") self.con.commit() self.assertEqual(self.redirects(), []) self.assertIsNone(resolve_entity_id(self.con, 11)) self.assertEqual(self.con.execute("PRAGMA foreign_key_check(entity_redirect)").fetchall(), []) def test_a_dangling_redirect_resolves_to_nothing(self): """绕开触发器留下的悬空行（比如重建 entity 表时丢了触发器）也不解析到死实体。""" merge(self.con, 11, 10, "新ありな") self.con.execute("DROP TRIGGER entity_redirect_target_delete") self.con.execute("DELETE FROM entity WHERE id=10") self.assertIsNone(resolve_entity_id(self.con, 11)) def test_a_new_entity_that_reuses_the_id_wins_over_the_redirect(self): """`entity.id` 没有 AUTOINCREMENT，删掉的最大 id 会再发给下一条新实体。""" merge(self.con, 12, 10, "桥本有菜") add_person(self.con, 12, "別の人") self.con.commit() self.assertEqual(resolve_entity_id(self.con, 12), 12) self.assertEqual(self.redirects(), []) def test_favorites_follow_merges_when_sqlite_reuses_the_highest_id(self): from peach.web_contract import WebContract, dispatch_api_get, dispatch_api_post contract = WebContract(self.db) for entity_id in (12, 10): dispatch_api_post(contract, "/api/library-shortcuts", { "action": "favorite", "entity_id": entity_id, "on": True}) saved = dispatch_api_post(contract, "/api/library-shortcuts", { "action": "save_filter", "name": "常用", "href": "/?loc="})['filters'] self.con.execute("UPDATE profile SET settings_json=json_set(settings_json,'$.searchHistoryLimit',25)" " WHERE id='local-default'") self.con.execute("INSERT INTO profile(id,user_id,name,is_default,settings_json,created_at,updated_at)" " SELECT 'second',user_id,'另一个',0,settings_json,created_at,updated_at" " FROM profile WHERE id='local-default'") self.con.execute("INSERT INTO profile(id,user_id,name,is_default,settings_json,created_at,updated_at)" " SELECT 'broken',user_id,'损坏',0,'{',created_at,updated_at" " FROM profile WHERE id='local-default'") self.con.commit() merge(self.con, 12, 10, "桥本有菜") inserted = self.con.execute("INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES('performer','別の人','別の人','t','t')").lastrowid self.con.commit() self.assertEqual(inserted, 12) self.assertEqual(self.redirects(), []) result = dispatch_api_get(contract, "/api/library-shortcuts", {}) self.assertEqual([row['entity_id'] for row in result['favorites']], [10]) self.assertEqual(result['filters'], saved) other = json.loads(self.con.execute("SELECT settings_json FROM profile WHERE id='second'").fetchone()[0]) self.assertEqual([row['entity_id'] for row in other['libraryShortcuts']['favorites']], [10]) self.assertEqual(other['libraryShortcuts']['filters'], saved) self.assertEqual(other['searchHistoryLimit'], 25) self.assertEqual(self.con.execute("SELECT settings_json FROM profile WHERE id='broken'").fetchone()[0], '{') self.assertEqual(dispatch_api_post(contract, "/api/library-shortcuts", { "action": "favorite", "entity_id": 10, "on": False})['favorites'], []) def test_favorite_transfer_rolls_back_with_entity_merge(self): original = json.dumps({'sidebarOrder': ['tags'], 'libraryShortcuts': { 'favorites': [{'entity_id': 12}], 'filters': []}}) self.con.execute("UPDATE profile SET settings_json=? WHERE id='local-default'", (original,)) self.con.commit() merge_entity(self.con, target_id=10, source_id=12, source_name="桥本有菜", alias_source="test:merge") self.con.rollback() self.assertEqual(self.con.execute("SELECT settings_json FROM profile WHERE id='local-default'").fetchone()[0], original) self.assertIsNotNone(self.con.execute("SELECT id FROM entity WHERE id=12").fetchone()) def test_merging_into_a_missing_or_the_same_entity_is_refused(self): with self.assertRaises(ValueError): merge_entity(self.con, target_id=999, source_id=11, source_name="新ありな", alias_source="test:merge") with self.assertRaises(ValueError): merge_entity(self.con, target_id=11, source_id=11, source_name="新ありな", alias_source="test:merge") self.assertIsNotNone(self.con.execute("SELECT 1 FROM entity WHERE id=11").fetchone()) def test_follows_and_feed_subscriptions_follow_the_merge(self): """关注与订阅绑在人身上，合并后跟着走，不悬空在已删的实体上。""" self.con.executescript(""" INSERT INTO follow_source(id,entity_id,provider,ref,label,url,created_at,updated_at) VALUES(1,11,'x','arina','arina','https://x.test/arina','t','t'); INSERT INTO feed_source(id,kind,url,entity_id,created_at) VALUES(1,'javdb_actor','https://j.test/1',11,'t'); INSERT INTO feed_discovery(id,code,discovered_at) VALUES(1,'ABC-001','t'),(2,'ABC-002','t'); INSERT INTO feed_discovery_entity(discovery_id,entity_id) VALUES(1,10),(1,11),(2,11); """) moved = merge(self.con, 11, 10, "新ありな") self.assertEqual((moved["follows"], moved["feeds"], moved["discoveries"]), (1, 1, 1)) self.assertEqual(self.con.execute("SELECT entity_id FROM follow_source").fetchall(), [(10,)]) self.assertEqual(self.con.execute("SELECT entity_id FROM feed_source").fetchall(), [(10,)]) self.assertEqual(self.con.execute( "SELECT discovery_id,entity_id FROM feed_discovery_entity ORDER BY 1").fetchall(), [(1, 10), (2, 10)]) self.assertEqual(self.con.execute("PRAGMA foreign_key_check").fetchall(), []) def test_a_review_row_carrying_the_old_id_shows_the_target(self): """候选 CSV 的 `entity_id` 停在出候选那一刻；实体并入别人之后，这一行认现在那一条。""" merge(self.con, 11, 10, "新ありな") self.con.row_factory = sqlite3.Row rows = [{"entity_id": "11", "current_name": "Arina Arata"}, {"entity_id": "12", "current_name": "桥本有菜"}, {"entity_id": "999", "current_name": "nobody"}] _use_canonical_entity_names(self.con, rows) self.assertEqual([(row["entity_id"], row["current_name"]) for row in rows], [("10", "橋本ありな"), ("12", "桥本有菜"), ("999", "nobody")]) self.assertEqual(rows[0]["source_name"], "Arina Arata") @unittest.skipUnless(HAS_DEPS, "FastAPI/httpx 尚未安装") class EntityRedirectRouteTests(unittest.TestCase): """页面与接口拿着旧 id 或旧名来，落到并进去的那一条。""" def setUp(self): from fastapi.testclient import TestClient from peach.api import create_app from peach.config import PeachSettings self.folder = Path(tempfile.mkdtemp()).resolve() self.addCleanup(shutil.rmtree, self.folder, True) self.avatars = self.folder / "avatars" self.avatars.mkdir() database = fresh_ledger(self.folder) connection = sqlite3.connect(database) add_person(connection, 10, "橋本ありな") add_person(connection, 11, "新ありな") connection.execute( "INSERT INTO asset(id,location,path,name,medium) VALUES(1,'local',?,'1.mp4','video')", (r"R:\media\1.mp4",)) connection.execute( "INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence)" " VALUES(1,11,'performer','test',1.0)") connection.commit() merge(connection, 11, 10, "新ありな") connection.close() self.face = mock.patch("peach.avatar_provider.FaceProbe").start() self.face.return_value.return_value = None self.addCleanup(mock.patch.stopall) self.app = create_app(PeachSettings( db_path=database, configured=True, token="secret", avatar_root=self.avatars, candidate_root=self.folder / "generated", cover_root=self.folder / "covers")) self.client = TestClient(self.app) self.addCleanup(self.client.close) def test_the_old_name_opens_the_target_page(self): response = self.client.get("/api/entity", params={ "kind": "performer", "name": "新ありな", "t": "secret"}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["id"], 10) self.assertEqual(response.json()["canonical_name"], "橋本ありな") def test_entity_and_photo_queries_accept_current_and_merged_ids(self): for path in ('/api/entity', '/api/photos'): named = self.client.get(path, params={ 'kind': 'performer', 'name': '橋本ありな', 't': 'secret'}) self.assertEqual(named.status_code, 200, named.text) for entity_id in (10, 11): response = self.client.get(path, params={ 'kind': 'performer', 'id': entity_id, 'name': 'ignored', 't': 'secret'}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json(), named.json()) def test_entity_id_queries_refuse_invalid_missing_or_other_kind_ids(self): for path in ('/api/entity', '/api/photos'): for entity_id in ('', '0', '-1', '1.0', 'ten', '１０', str(2**63), '9' * 50): response = self.client.get(path, params={ 'kind': 'performer', 'id': entity_id, 't': 'secret'}) self.assertEqual(response.json(), {'error': 'invalid entity'}, (path, entity_id)) for kind, entity_id in (('studio', '10'), ('performer', '999')): response = self.client.get(path, params={'kind': kind, 'id': entity_id, 't': 'secret'}) self.assertEqual(response.json(), {'error': 'not found'}, (path, kind, entity_id)) def test_verified_cross_kind_merge_ids_use_the_final_identity(self): with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(20,'creator','Old Account','old account','t','t')") merge_entity(connection, target_id=10, source_id=20, source_name='Old Account', alias_source='user:identity-merge:creator') add_person(connection, 12, 'Final Person') merge_entity(connection, target_id=12, source_id=10, source_name='橋本ありな', alias_source='test:merge') expected = {'redirect': {'id': 12, 'kind': 'performer', 'name': 'Final Person'}} for identity in ({'id': 20}, {'name': 'Old Account'}): response = self.client.get('/api/entity', params={'kind': 'creator', 't': 'secret', **identity}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json(), expected) query = '?tag=a%2Fb&tag=second&media=photos&t=secret' for method in ('GET', 'HEAD'): response = self.client.request(method, '/creators/20' + query, follow_redirects=False) self.assertEqual(response.status_code, 307) self.assertEqual(response.headers['location'], '/performers/12' + query) self.assertEqual(response.headers['cache-control'], 'no-store') self.assertEqual(self.client.get('/api/photos', params={ 'kind': 'creator', 'id': 20, 't': 'secret'}).json(), {'error': 'not found'}) def test_cross_kind_id_redirect_requires_source_kind_evidence(self): with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.executemany( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,'creator',?,?,'t','t')", [(20, 'Explicit', 'explicit'), (21, 'Generic', 'generic')]) merge_entity(connection, target_id=10, source_id=20, source_name='Explicit', alias_source='user:identity-merge:creator') merge_entity(connection, target_id=10, source_id=21, source_name='Generic', alias_source='test:merge') for kind, entity_id in (('creator', 10), ('studio', 20), ('creator', 21)): response = self.client.get('/api/entity', params={'kind': kind, 'id': entity_id, 't': 'secret'}) self.assertEqual(response.json(), {'error': 'not found'}, (kind, entity_id)) prefix = 'studios' if kind == 'studio' else 'creators' for method in ('GET', 'HEAD'): page = self.client.request(method, f'/{prefix}/{entity_id}?t=secret', follow_redirects=False) self.assertEqual(page.status_code, 404, (kind, entity_id, method)) def test_a_reused_cross_kind_merge_id_keeps_its_current_identity(self): with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.execute( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(20,'creator','Old Account','old account','t','t')") merge_entity(connection, target_id=10, source_id=20, source_name='Old Account', alias_source='user:identity-merge:creator') add_person(connection, 20, 'New Person') self.assertEqual(self.client.get('/api/entity', params={ 'kind': 'creator', 'id': 20, 't': 'secret'}).json(), {'error': 'not found'}) self.assertEqual(self.client.get('/creators/20?t=secret', follow_redirects=False).status_code, 404) current = self.client.get('/api/entity', params={'kind': 'performer', 'id': 20, 't': 'secret'}) self.assertEqual(current.json()['id'], 20) self.assertEqual(current.json()['canonical_name'], 'New Person') def test_tag_profiles_and_photos_use_the_same_identity_by_name_and_id(self): with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.execute("INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(20,'tag','Tag A','tag a','t','t')") connection.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(1,20,'tag','test',1)") connection.execute("INSERT INTO asset(id,location,path,name,medium) VALUES(2,'local',?,'2.jpg','image')", (r'R:\media\2.jpg',)) connection.execute("INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(2,20,'tag','test',1)") for path in ('/api/entity', '/api/photos'): named = self.client.get(path, params={'kind': 'tag', 'name': 'Tag A', 't': 'secret'}) by_id = self.client.get(path, params={'kind': 'tag', 'id': '20', 't': 'secret'}) self.assertEqual(named.status_code, 200, named.text) self.assertEqual(by_id.json(), named.json()) if path == '/api/entity': self.assertEqual(by_id.json()['asset_count'], 1) else: self.assertEqual(by_id.json()['total'], 1) self.assertEqual(by_id.json()['items'][0]['id'], 2) def test_tag_hidden_assets_are_excluded_from_profiles_photos_and_samples(self): with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.executemany( "INSERT INTO entity(id,kind,canonical_name,normalized_name,created_at,updated_at)" " VALUES(?,'tag',?,?,'t','t')", [(20, 'Tag A', 'tag a'), (21, 'Hidden only', 'hidden only'), (22, 'Visible only', 'visible only')]) add_person(connection, 12, 'Hidden cast') connection.execute("UPDATE asset SET code='ABC-001',size=10,snapshot_path='visible.jpg' WHERE id=1") connection.executemany( "INSERT INTO asset(id,location,path,name,medium,size,code,snapshot_path,play_count)" " VALUES(?,'local',?,?,?,?,?,?,?)", [(2, r'R:\media\visible\2.jpg', '2.jpg', 'image', 10, None, None, 0), (3, r'R:\media\3.mp4', '3.mp4', 'video', 1000, 'ABC-002', 'hidden.jpg', 100), (4, r'R:\media\hidden\4.jpg', '4.jpg', 'image', 10, None, None, 0), (5, r'R:\media\5.mp4', '5.mp4', 'video', 500, 'ABC-003', 'hidden-cast.jpg', 0)]) connection.executemany( "INSERT INTO asset_entity(asset_id,entity_id,role,source,confidence) VALUES(?,?,?,'test',1)", [(asset_id, 20, 'tag') for asset_id in (1, 2, 3, 4, 5)] + [(1, 22, 'tag'), (3, 21, 'tag'), (3, 10, 'performer'), (5, 12, 'performer')]) connection.executemany( "INSERT INTO asset_tag_preference(profile_id,asset_id,normalized_tag,hidden,updated_at)" " VALUES('local-default',?,'tag a',1,0)", [(3,), (4,), (5,)]) connection.executemany( "INSERT INTO code_sample_image(code,position,url,site,source,fetched_at)" " VALUES(?,1,'https://example.test/sample.jpg','dmm','test','t')", [('ABC-001',), ('ABC-002',), ('ABC-003',)]) def read(path, kind='tag', entity_id=20): response = self.client.get(path, params={'kind': kind, 'id': entity_id, 't': 'secret'}) self.assertEqual(response.status_code, 200, response.text) return response.json() profile = read('/api/entity') self.assertEqual(profile['asset_count'], 1) self.assertEqual(profile['representative_asset_id'], 1) self.assertEqual({tag['id']: tag['n'] for tag in profile['tags']}, {20: 1, 22: 1}) self.assertEqual([(person['id'], person['n'], person['rep']) for person in profile['related_performers']], [(10, 1, 1)]) photos = read('/api/photos') self.assertEqual(photos['total'], 1) self.assertEqual([item['id'] for item in photos['items']], [2]) self.assertEqual({item['id']: item['n'] for item in photos['sets']}, {'code:ABC-001': 1, 2: 1}) self.assertEqual(photos['sample_total'], 1) self.assertEqual(read('/api/entity', 'performer', 10)['asset_count'], 2) from peach.web_catalog import _entity_asset_filter, q_facets from peach.web_state import WebContract contract = WebContract(self.app.state.settings.db_path) with contract.read_connection() as connection: clause, params = _entity_asset_filter(connection, 'tag', 'Tag A') scoped_assets = connection.execute(f'SELECT a.id FROM asset a WHERE {clause} ORDER BY a.id', params).fetchall() self.assertEqual([row[0] for row in scoped_assets], [1, 2]) facets = q_facets(contract, filters={'tag': 'Tag A'}) self.assertEqual(facets['stats']['total'], profile['asset_count']) self.assertEqual({item['k']: item['n'] for item in facets['tags']}, {'Tag A': 1, 'Visible only': 1}) scoped_facets = q_facets(contract, scope_kind='tag', scope_name='Tag A') self.assertEqual(scoped_facets['stats']['total'], profile['asset_count']) self.assertEqual(scoped_facets['tags'], facets['tags']) with sqlite3.connect(self.app.state.settings.db_path) as connection: connection.execute("UPDATE asset_tag_preference SET hidden=0 WHERE normalized_tag='tag a'") self.assertEqual(read('/api/entity')['asset_count'], 3) restored = read('/api/photos') self.assertEqual(restored['total'], 2) self.assertEqual(restored['sample_total'], 3) self.assertEqual(q_facets(contract, filters={'tag': 'Tag A'})['stats']['total'], 3) def test_an_old_image_address_redirects_to_the_target(self): body = picture("navy") (self.avatars / "performer-10.img").write_bytes(body) response = self.client.get( "/entity-image", params={"kind": "performer", "id": 11, "thumb": 1, "t": "secret"}, follow_redirects=False) self.assertEqual(response.status_code, 307) location = urlsplit(response.headers["location"]) self.assertEqual((location.scheme, location.netloc, location.path), ("", "", "/entity-image")) self.assertEqual(parse_qs(location.query), {"kind": ["performer"], "id": ["10"], "thumb": ["1"], "t": ["secret"]}) followed = self.client.get( "/entity-image", params={"kind": "performer", "id": 11, "t": "secret"}) self.assertEqual(followed.status_code, 200) self.assertEqual(followed.content, body) def test_an_unknown_id_still_404s(self): response = self.client.get( "/entity-image", params={"kind": "performer", "id": 999, "t": "secret"}, follow_redirects=False) self.assertEqual(response.status_code, 404) def test_a_picture_picked_under_the_old_id_lands_on_the_target(self): body = picture("olive") response = self.client.post( "/api/avatar-pick?id=11&kind=performer&name=me.jpg&t=secret", content=body, headers={"content-type": "image/jpeg"}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["id"], 10) self.assertEqual((self.avatars / "performer-10.img").read_bytes(), body) self.assertFalse((self.avatars / "performer-11.img").exists()) if __name__ == "__main__": unittest.main()