fix(plex,audit): E30 review round 2 — only a missing file is a dead headshot, one spelling per portrait

- warm_portraits: only 404/410 mark a headshot dead (measured: Plex answers a dead CDN file 404);
  a refused token or a 5xx is Plex failing, ends the run after three in a row and marks nothing.
  'failed' counts every failure, not only the current streak.
- is_portrait_url: plain path segments only — no dot or empty segment an HTTP client folds onto
  one file; person_thumb_url uses the same predicate (27 852 stored headshots all pass).
- Tests: Plex refusing (401/403/429/5xx), failures between successes (portraits and bios), the kept
  original served from cache, the audit_clear / demo_reset / discovery_purge rows' counts as data,
  and the 0094 backfill on real rows.

backend/tests/test_plex_people_enrich.py

@@ -88,6 +88,14 @@ def test_failures_in_a_row_end_the_batch(db):
     assert tmdb.asked == [1, 2, 3]
 
 
+def test_failures_between_successes_do_not_end_the_batch(db):
+    db.add_all([PlexPerson(name=f"P{i}", tmdb_person_id=i) for i in range(1, 9)])
+    db.commit()
+    tmdb = _Tmdb(fail_on={1, 3, 5, 7})  # fail, ok, fail, ok, …: never three in a row
+    assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 4, "failed": 4}  # type: ignore[arg-type]
+    assert tmdb.asked == list(range(1, 9))
+
+
 def test_the_batch_stores_the_hungarian_biography(db):
     db.add(PlexPerson(name="A", tmdb_person_id=1))
     db.commit()
@@ -108,12 +116,14 @@ def test_the_batch_stores_the_hungarian_biography(db):
 
 
 class _Plex:
-    """A PlexClient stand-in. `down`: Plex does not answer; `dead`: urls Plex answers with an error."""
+    """A PlexClient stand-in. `down`: urls Plex does not answer for (True: none); `dead`: urls Plex
+    answers "not found" for (a dead CDN file); `status`: an error status Plex answers every url with."""
 
-    def __init__(self, down=False, dead=()):
+    def __init__(self, down=False, dead=(), status=None):
         self.asked: list[tuple[str, int | None]] = []
         self.down = down
         self.dead = set(dead)
+        self.status = status
 
     def __enter__(self):
         return self
@@ -123,14 +133,15 @@ class _Plex:
 
     def image_bytes(self, url, width=None, height=None):
         self.asked.append((url, width))
-        if self.down:
+        if self.down is True or (self.down and url in self.down):
             raise plex_people.PlexError("down") from httpx.ConnectError("refused")
-        if url in self.dead:
+        code = 404 if url in self.dead else self.status
+        if code:
             req = httpx.Request("GET", url)
             status = httpx.HTTPStatusError(
-                "404", request=req, response=httpx.Response(404, request=req)
+                str(code), request=req, response=httpx.Response(code, request=req)
             )
-            raise plex_people.PlexError("404") from status
+            raise plex_people.PlexError(str(code)) from status
         return b"small", "image/jpeg"
 
 
@@ -210,6 +221,21 @@ def test_plex_not_answering_ends_the_portrait_run(db, cache_dir, plex_on):
     assert not list(cache_dir.glob("*.miss"))  # Plex was down, the files are not to blame
 
 
+@pytest.mark.parametrize("code", [401, 403, 429, 500, 503])
+def test_plex_refusing_is_plex_failing_not_a_dead_file(db, cache_dir, plex_on, code):
+    _faces(db, 8)
+    plex_on["plex"] = _Plex(status=code)  # a rotated token, a busy or broken transcoder
+    assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 3}
+    assert len(plex_on["plex"].asked) == 3
+    assert not list(cache_dir.glob("*.miss"))  # the files are not to blame: asked again next run
+
+
+def test_plex_failing_between_successes_does_not_end_the_run(db, cache_dir, plex_on):
+    _faces(db, 8)
+    plex_on["plex"] = _Plex(down={CDN + f"{i}.jpg" for i in (0, 2, 4, 6)})  # down, ok, down, ok, …
+    assert plex_people.warm_portraits(db, 100) == {"cached": 4, "failed": 4}
+
+
 def test_dead_headshots_are_left_alone_and_the_rest_still_land(db, cache_dir, plex_on):
     _faces(db, 8)
     dead = {CDN + f"{i}.jpg" for i in range(6)}  # more dead files than any failure limit
@@ -237,9 +263,19 @@ def test_the_person_image_route(db, cache_dir, monkeypatch):
     def get(u):
         return client.get("/api/plex/person-image", params={"u": u})
 
-    # Only a headshot-shaped url on the CDN host: no query variants of one file, no other host.
-    for bad in (url + "?n=1", url + "#x", "https://example.com/people/a.jpg"):
+    # Only a headshot-shaped url on the CDN host: no query variants of one file, no spellings an HTTP
+    # client folds onto it, no other host.
+    for bad in (
+        url + "?n=1",
+        url + "#x",
+        CDN + "x/../face.jpg",
+        CDN + "./face.jpg",
+        plex_people.PERSON_IMG_HOST + "people//face.jpg",
+        CDN + ".face.jpg",
+        "https://example.com/people/a.jpg",
+    ):
         assert get(bad).status_code == 400, bad
+        assert plex_people.person_thumb_url(bad) is None, bad  # never stored either
 
     # No Plex: the original from the CDN, kept under its own key (the job resizes it later).
     def not_configured(d):
@@ -268,6 +304,8 @@ def test_the_person_image_route(db, cache_dir, monkeypatch):
     resp = get(url)
     assert (resp.status_code, resp.content) == (200, b"original")
     assert imgcache.has(original) and not imgcache.has(small)
+    monkeypatch.setattr(plex_people.httpx, "Client", None)  # the CDN is not asked again
+    assert get(url).content == b"original"  # the kept original answers
 
     # With Plex: resized, and it replaces the original.
     monkeypatch.setattr(playback, "PlexClient", lambda d: _Plex())