Vito
9ac471eb72
* perf(images): share and warm one listing per output root (#348) The Images page, its sidebar counts, the subfolder list and the legacy gallery each rebuilt the output listing when the 30 s cache expired, all at once, and the subfolder list computed relative paths on the event loop. Thumbnail writes also threw the listing away. Keep one RootListing per output root (py/api/gallery_listing.py): - one walk per root at a time; concurrent readers wait for it (shielded) - a listing older than 30 s is served at once and refreshed in the background - the walk, the comparison with the previous listing and both folder sets run in the executor; an unchanged walk keeps the same list object - os.scandir probe: a missing root fails the walk and keeps the previous listing (one warning per failure streak) - Rescan waits for a walk started after the click and is a 500 when it fails, even with a previous listing - deletes keep the blocking invalidation; a generation counter that survives drops discards walks that raced it - thumbnail jobs, conversion and the legacy generate/clear routes no longer drop the listing - warm-up on the host's on_startup, cancelled on on_cleanup Adds tools/vendor/perf/images_load_probe.mjs (maintainer tool, not CI) and the TESTING.md section "Images listing cache (#348)". * fix(images): drop listings on the loop when invalidated from a worker thread (#348) A settings save (gallery roots change) and the browser-test seeds call invalidate_gallery_cache() from the executor. Listing records are only touched on the event loop, so an off-loop call now hands the drop to the loop and waits for it, keeping the call synchronous for its callers. The shared-Rescan test waits for the later Rescans to take their walk number instead of sleeping. * test(images): bound the shared-listing test waits (#348) A macOS py3.12 run stopped inside the shared-listing tests and sat until the 6 h job limit with no traceback. It does not reproduce locally (about 300 runs on 3.12 and 3.14, pinned to one core). Requests and background settles in TestSharedListing now give up after 30 s, so a recurrence fails with the stuck await instead of hanging the run. * test(images): let the settle loops yield so a finished walk leaves the set (#348) On CI (ubuntu and macOS py3.12) the runner's watchdog dumped test_a_root_that_recovers_warns_again_on_its_next_failure spinning in _settle: asyncio.gather over tasks that are already done returns without yielding, so the set's discard callback, queued when the walk finished, never ran and the loop never ended (wait_for could not fire either). Delaying discard by one loop turn reproduces it locally; both settle loops now yield one turn after each gather. * test(browser): poll the import review's rows instead of reading them once gridOf reads the dialog's rows once; right after the click the review may not have rendered them yet (firefox CI: received undefined). * perf(images): walk the output with scandir entries and build folders from strings (#348) The live test showed a read during a background refresh taking 2.2 s instead of 0.07 s: the walk held the GIL for about 2 s in the server. py-spy split that walk into two halves: Path.relative_to per file in folder_sets, and Path.stat + Path.is_symlink + Path joins per file in the scan. - _walk_media visits folders in os.walk's order with os.scandir entries: the file type comes from the listing (no lstat per file) and the mtime from entry.stat (one call, none on Windows). _iter_media_files is now a thin wrapper, so its other caller is unchanged. - folder_sets slices the path strings (in the path flavour's separator) instead of calling relative_to, falling back to it for a path that does not start with the root string. On the maintainer's 19,935-file output folder: scan 0.386 s -> 0.112 s, folder sets 0.452 s -> 0.007 s, both results identical to the previous code. A reference copy of the os.walk version checks order, filters, the depth cap, symlinks and the file cap. * perf(images): build the workspace caches behind the warm-up and changed walks (#348) The live test after the faster walk still took 2.9-5.2 s for the first Images open after a restart. py-spy showed the walk was now small; the time went to the workspace's own caches, cold after a restart and built by three parallel first requests at once: the favorites listing (Path.relative_to per file), the per-file values and newest order, and the prompt-link read. - Once the host's startup warm-up has run, a stored walk that changed a listing schedules one executor job that builds what a first page reads, through the request path itself (_select_workspace_page with the default filters, then _workspace_favorite_paths). A walk stored during a build asks for one more build; an unchanged walk (same list kept) builds nothing; a failed build only warns; on_cleanup stops it. Servers that never ran the warm-up (the test servers) start no build. - _favorite_listing slices the path strings as _listed does, falling back to relative_to for a path outside the root string. * perf(images): start a stale listing's refresh walk a second after the read (#348) After the faster walk, a read that found the listing stale still took 0.5-1.0 s against 0.07 s: it was served from the old listing but ran beside the refresh walk it had just started, sharing the GIL. The refresh now starts REFRESH_DELAY (1 s) after that read through a per-root rewalk_handle, so the page's own requests finish first. One pending refresh per root and none while a walk runs; any walk start (a cold read, Rescan, the warm-up) cancels a pending one; on_cleanup cancels them before its first await. * test(browser): accept KB or MB for the picked safety backup's size The restore preview showed '2.5 MB' for the test server's database copy in a CI coverage shard: its size depends on what earlier tests on that server wrote, and the formatter switches to MB at 1 MiB. * fix(images): keep the 3.10-3.13 suffix rule in the scandir walk and harden the warm-up (#348) Review of the walk speedup: - os.path.splitext drops names like ..png that Path.suffix listed on Python 3.10-3.13 (3.14 changed Path.suffix, which hid it from the parity test on a 3.14 venv). _name_suffix spells out the 3.10-3.13 rule; the test now asserts ..png and ...webp are listed and g. is not. - _warm_workspace reads the roots inside its try, so a failing _get_all_output_dirs warns instead of leaving an unretrieved task error. - The folder_sets fallback test now takes the fallback (a Windows case mismatch); the coalescing test pins the walk count.
445 lines
19 KiB
Python
445 lines
19 KiB
Python
"""Convert outdated thumbnails (#317): an explicit, cancellable Activity job.
|
|
|
|
``POST /prompt_manager/jobs/thumbnails {mode: "convert"}`` runs under the
|
|
thumbnail job's kind, admission (``guarded``) and worker threads. For every
|
|
media file whose preview is in another format than the one written now, it
|
|
writes the new preview from the SOURCE (verified, atomic), then removes that
|
|
file's old-format previews under ``thumbnails/``. A failure keeps the old
|
|
preview; previews whose source is gone (orphans) are never touched. The final
|
|
line reports converted / failed and bytes before → after.
|
|
|
|
Ordering is driven by events, never by timers.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
try:
|
|
from tests.test_api_images import ThumbnailTestCase, make_png
|
|
except ImportError: # discovered with tests/ as the top-level directory
|
|
from test_api_images import ThumbnailTestCase, make_png
|
|
|
|
from PIL import Image # noqa: E402
|
|
|
|
from py.api import thumbnail_encode # noqa: E402
|
|
from py.config import PromptManagerConfig # noqa: E402
|
|
|
|
JOB = "/prompt_manager/jobs/thumbnails"
|
|
PLAN = "/prompt_manager/images/thumbnails/plan"
|
|
STATUS = "/prompt_manager/images/thumbnails/status"
|
|
WAIT = 10
|
|
|
|
|
|
def big_png(path, size=(512, 512)):
|
|
"""A noisy PNG: its PNG preview is large, a WebP one much smaller."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.effect_noise(size, 40).convert("RGB").save(path, "PNG")
|
|
return path
|
|
|
|
|
|
class ConvertCase(ThumbnailTestCase):
|
|
async def asyncSetUp(self):
|
|
await super().asyncSetUp()
|
|
saved = PromptManagerConfig.get_config()["thumbnails"]
|
|
self.addCleanup(PromptManagerConfig.update_config, {"thumbnails": saved})
|
|
PromptManagerConfig.update_config(
|
|
{"thumbnails": {"format": "webp", "quality": 80, "edge": 512}}
|
|
)
|
|
thumbnail_encode.reset_encoders()
|
|
self.addCleanup(thumbnail_encode.reset_encoders)
|
|
self.addCleanup(self.api.close_auto_thumbnails)
|
|
|
|
def _legacy(self, name, suffix=".png", source=True):
|
|
"""A media file (unless ``source`` is False) and its old-format preview."""
|
|
if source:
|
|
big_png(self.output_dir / f"{name}.png")
|
|
preview = big_png(self._thumb_path(name, suffix), size=(256, 256))
|
|
if source:
|
|
later = (self.output_dir / f"{name}.png").stat().st_mtime + 5
|
|
os.utime(preview, (later, later))
|
|
return preview
|
|
|
|
async def _convert(self):
|
|
resp = await self.client.post(JOB, json={"mode": "convert"})
|
|
body = await resp.json()
|
|
self.assertEqual(resp.status, 202, body)
|
|
job = self.api._jobs.get(body["job"]["id"])
|
|
await asyncio.wait_for(job.task, WAIT)
|
|
return body["job"], job
|
|
|
|
|
|
class TestConvertOutdated(ConvertCase):
|
|
async def test_plan_counts_the_outdated_previews(self):
|
|
self._legacy("a")
|
|
self._legacy("b")
|
|
make_png(self.output_dir / "new.png") # missing: not this job's work
|
|
plan = await (await self.client.get(PLAN, params={"mode": "convert"})).json()
|
|
self.assertEqual(
|
|
(plan["mode"], plan["targets"], plan["outdated"]), ("convert", 2, 2)
|
|
)
|
|
self.assertEqual(self.api._jobs.list(), [])
|
|
|
|
async def test_conversion_keeps_the_listing(self):
|
|
self._legacy("a")
|
|
await self.client.get("/prompt_manager/images/output?mode=workspace")
|
|
files = self.api._root_listings[str(self.output_dir)].files
|
|
_summary, job = await self._convert()
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertIs(self.api._root_listings[str(self.output_dir)].files, files)
|
|
|
|
async def test_converts_from_the_source_and_removes_the_old_preview(self):
|
|
old = self._legacy("a")
|
|
original = (self.output_dir / "a.png").read_bytes()
|
|
orphan = self._legacy("gone", source=False)
|
|
make_png(self.output_dir / "new.png")
|
|
before = old.stat().st_size
|
|
|
|
summary, job = await self._convert()
|
|
|
|
self.assertEqual(
|
|
(summary["kind"], summary["admission"], summary["cancellable"]),
|
|
("thumbnails", "guarded", True),
|
|
)
|
|
self.assertEqual(summary["title"], "Convert outdated thumbnails")
|
|
self.assertEqual(summary["scope"]["mode"], "convert")
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertEqual(job.counters["generated"], 1)
|
|
self.assertEqual(job.counters["errors"], 0)
|
|
new = self._thumb_path("a")
|
|
with Image.open(new) as img:
|
|
self.assertEqual((img.format, max(img.size)), ("WEBP", 512))
|
|
self.assertFalse(old.exists(), "the old PNG preview is removed")
|
|
self.assertEqual((self.output_dir / "a.png").read_bytes(), original)
|
|
self.assertTrue(orphan.exists(), "an orphan preview is never touched")
|
|
self.assertFalse(self._thumb_path("new").exists(), "missing is not converted")
|
|
after = new.stat().st_size
|
|
self.assertRegex(job.status_line, r"^1 converted · 0 failed · .+ → .+$")
|
|
self.assertTrue(any("Conversion finished" in line for line in job.log))
|
|
self.assertLess(after, before)
|
|
status = await (await self.client.get(STATUS)).json()
|
|
self.assertEqual((status["outdated"], status["pending"]), (0, 1))
|
|
|
|
async def test_a_source_that_fails_keeps_its_old_preview(self):
|
|
old = self._legacy("bad")
|
|
(self.output_dir / "bad.png").write_bytes(b"\x89PNG broken")
|
|
later = old.stat().st_mtime + 10
|
|
os.utime(old, (later, later))
|
|
keep = old.read_bytes()
|
|
|
|
_, job = await self._convert()
|
|
|
|
self.assertEqual(job.counters["errors"], 1)
|
|
self.assertEqual(job.counters["generated"], 0)
|
|
self.assertEqual(old.read_bytes(), keep)
|
|
self.assertFalse(self._thumb_path("bad").exists())
|
|
self.assertRegex(job.status_line, r"^0 converted · 1 failed")
|
|
|
|
async def test_a_video_without_a_frame_keeps_its_old_preview(self):
|
|
(self.output_dir / "clip.mp4").write_bytes(b"no decoder")
|
|
old = big_png(self._thumb_path("clip", ".jpg"), size=(64, 64))
|
|
keep = old.read_bytes()
|
|
with (
|
|
patch.object(self.api, "_opencv_frame", return_value=None),
|
|
patch.object(self.api, "_ffmpeg_frame", return_value=None),
|
|
):
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.counters["errors"], 1)
|
|
self.assertEqual(old.read_bytes(), keep)
|
|
self.assertFalse(self._thumb_path("clip").exists(), "never the placeholder")
|
|
|
|
async def test_an_old_preview_that_is_a_link_is_left(self):
|
|
big_png(self.output_dir / "l.png")
|
|
target = big_png(self.comfy_dir / "elsewhere.png", size=(32, 32))
|
|
link = self._thumb_path("l", ".png")
|
|
link.parent.mkdir(parents=True, exist_ok=True)
|
|
self._make_symlink(target, link)
|
|
_, job = await self._convert()
|
|
self.assertTrue(link.is_symlink())
|
|
self.assertTrue(target.exists())
|
|
self.assertEqual(job.counters["errors"], 0)
|
|
|
|
async def test_a_preview_already_current_only_loses_its_leftovers(self):
|
|
old = self._legacy("c")
|
|
current = big_png(self._thumb_path("c"), size=(64, 64))
|
|
later = old.stat().st_mtime + 5
|
|
os.utime(current, (later, later))
|
|
written = current.read_bytes()
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.counters["generated"], 1)
|
|
self.assertEqual(current.read_bytes(), written, "not written again")
|
|
self.assertFalse(old.exists())
|
|
|
|
async def test_cancel_stops_at_a_file_boundary_keeping_what_converted(self):
|
|
original = PromptManagerConfig.WORKER_THREADS
|
|
PromptManagerConfig.WORKER_THREADS = 1
|
|
self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
|
|
olds = [self._legacy(f"p{i}") for i in range(6)]
|
|
done = []
|
|
real = self.api._convert_one
|
|
job_ids = []
|
|
|
|
def counting(*args, **kwargs):
|
|
result = real(*args, **kwargs)
|
|
done.append(result)
|
|
if len(done) == 2:
|
|
self.assertTrue(self.api._jobs.cancel(job_ids[0]))
|
|
return result
|
|
|
|
self.api._convert_one = counting
|
|
resp = await self.client.post(JOB, json={"mode": "convert"})
|
|
job_ids.append((await resp.json())["job"]["id"])
|
|
job = self.api._jobs.get(job_ids[0])
|
|
await asyncio.wait_for(job.task, WAIT)
|
|
|
|
self.assertEqual(job.state, "cancelled")
|
|
self.assertEqual(job.counters["generated"], 2)
|
|
converted = [p for p in olds if not p.exists()]
|
|
self.assertEqual(len(converted), 2)
|
|
for old in olds:
|
|
webp = old.with_suffix(".webp")
|
|
self.assertEqual(webp.exists(), not old.exists(), old.name)
|
|
self.assertTrue(any("Conversion cancelled" in line for line in job.log))
|
|
|
|
async def test_refused_beside_a_running_thumbnail_job_or_scan(self):
|
|
self._legacy("a")
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.addCleanup(release.set)
|
|
real = self.api._convert_one
|
|
|
|
def gated(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(WAIT)
|
|
return real(*args, **kwargs)
|
|
|
|
self.api._convert_one = gated
|
|
resp = await self.client.post(JOB, json={"mode": "convert"})
|
|
first = (await resp.json())["job"]["id"]
|
|
loop = asyncio.get_running_loop()
|
|
self.assertTrue(await loop.run_in_executor(None, entered.wait, WAIT))
|
|
again = await self.client.post(JOB, json={"mode": "missing"})
|
|
self.assertEqual(again.status, 409)
|
|
scan = await self.client.post("/prompt_manager/jobs/scan")
|
|
self.assertEqual(scan.status, 409)
|
|
release.set()
|
|
await asyncio.wait_for(self.api._jobs.get(first).task, WAIT)
|
|
|
|
async def test_the_legacy_gallery_stream_never_attaches_to_a_conversion(self):
|
|
try:
|
|
from tests.test_api_images import parse_sse
|
|
except ImportError:
|
|
from test_api_images import parse_sse
|
|
|
|
self._legacy("a")
|
|
entered, release = threading.Event(), threading.Event()
|
|
self.addCleanup(release.set)
|
|
real = self.api._convert_one
|
|
|
|
def gated(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(WAIT)
|
|
return real(*args, **kwargs)
|
|
|
|
self.api._convert_one = gated
|
|
resp = await self.client.post(JOB, json={"mode": "convert"})
|
|
first = (await resp.json())["job"]["id"]
|
|
loop = asyncio.get_running_loop()
|
|
self.assertTrue(await loop.run_in_executor(None, entered.wait, WAIT))
|
|
stream = await self.client.post(self.PROGRESS_URL)
|
|
events = parse_sse(await stream.text())
|
|
self.assertEqual(events[-1][0], "error")
|
|
self.assertIn("Convert outdated thumbnails", events[-1][1]["error"])
|
|
release.set()
|
|
await asyncio.wait_for(self.api._jobs.get(first).task, WAIT)
|
|
|
|
async def test_nothing_outdated_says_so(self):
|
|
big_png(self.output_dir / "a.png")
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertEqual(job.status_line, "No outdated previews to convert")
|
|
|
|
async def test_the_source_vanishing_keeps_the_old_preview(self):
|
|
old = self._legacy("v")
|
|
real = self.api._generate_one
|
|
|
|
def vanish(src, *args, **kwargs):
|
|
os.remove(src)
|
|
return real(src, *args, **kwargs)
|
|
|
|
self.api._generate_one = vanish
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.counters["errors"], 1)
|
|
self.assertTrue(old.exists(), "a source gone mid-run makes an orphan")
|
|
|
|
async def test_same_stem_sources_convert_once_and_count_bytes_once(self):
|
|
# Review I2 / R2: x.png and x.jpg share x_thumb.webp. The pair is
|
|
# converted once, from the image first in the source order (.png),
|
|
# and every removed file's bytes count once.
|
|
original = PromptManagerConfig.WORKER_THREADS
|
|
PromptManagerConfig.WORKER_THREADS = 2
|
|
self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
|
|
from py.api.thumbnail_convert import readable_bytes
|
|
|
|
png = self.output_dir / "x.png"
|
|
Image.new("RGB", (300, 300), (220, 20, 20)).save(png)
|
|
jpg = self.output_dir / "x.jpg"
|
|
Image.new("RGB", (300, 300), (20, 20, 220)).save(jpg)
|
|
old_png = big_png(self._thumb_path("x", ".png"), size=(200, 200))
|
|
old_jpg = self._thumb_path("x", ".jpg")
|
|
Image.effect_noise((200, 200), 40).convert("RGB").save(old_jpg, "JPEG")
|
|
later = max(png.stat().st_mtime, jpg.stat().st_mtime) + 5
|
|
for path in (old_png, old_jpg):
|
|
os.utime(path, (later, later))
|
|
before = old_png.stat().st_size + old_jpg.stat().st_size
|
|
|
|
_, job = await self._convert()
|
|
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertEqual(
|
|
(job.counters["generated"], job.counters["errors"]), (2, 0), job.log
|
|
)
|
|
new = self._thumb_path("x")
|
|
self.assertFalse(old_png.exists() or old_jpg.exists())
|
|
with Image.open(new) as img:
|
|
red, _green, blue = img.convert("RGB").getpixel((5, 5))
|
|
self.assertGreater(red, blue, "the image first in the source order wins")
|
|
self.assertEqual(
|
|
job.status_line,
|
|
f"2 converted · 0 failed · {readable_bytes(before)} → "
|
|
f"{readable_bytes(new.stat().st_size)}",
|
|
)
|
|
|
|
async def test_an_old_preview_already_gone_counts_as_removed(self):
|
|
old = self._legacy("g")
|
|
real = self.api._generate_one
|
|
|
|
def remove_first(src, *args, **kwargs):
|
|
os.remove(old)
|
|
return real(src, *args, **kwargs)
|
|
|
|
self.api._generate_one = remove_first
|
|
_, job = await self._convert()
|
|
self.assertEqual((job.counters["generated"], job.counters["errors"]), (1, 0))
|
|
self.assertRegex(job.status_line, r"^1 converted · 0 failed · 0 B → ")
|
|
|
|
async def test_a_current_preview_that_does_not_decode_is_written_again(self):
|
|
# Review M4: a kept current preview is decoded before old ones go.
|
|
old = self._legacy("m")
|
|
current = self._thumb_path("m")
|
|
current.write_bytes(b"RIFF not a webp")
|
|
later = old.stat().st_mtime + 5
|
|
os.utime(current, (later, later))
|
|
_, job = await self._convert()
|
|
self.assertEqual((job.counters["generated"], job.counters["errors"]), (1, 0))
|
|
with Image.open(current) as img:
|
|
self.assertEqual(img.format, "WEBP")
|
|
self.assertFalse(old.exists())
|
|
|
|
async def test_a_format_change_underneath_stops_the_job(self):
|
|
# Review I4 (defense in depth): the job never removes previews in a
|
|
# format chosen after it started.
|
|
original = PromptManagerConfig.WORKER_THREADS
|
|
PromptManagerConfig.WORKER_THREADS = 1
|
|
self.addCleanup(setattr, PromptManagerConfig, "WORKER_THREADS", original)
|
|
olds = [self._legacy(f"f{i}") for i in range(4)]
|
|
real = self.api._convert_one
|
|
calls = []
|
|
|
|
def switching(*args, **kwargs):
|
|
result = real(*args, **kwargs)
|
|
calls.append(result)
|
|
if len(calls) == 1:
|
|
PromptManagerConfig.update_config(
|
|
{"thumbnails": {"format": "jpeg", "quality": 80, "edge": 512}}
|
|
)
|
|
jpeg = self._thumb_path("f3", ".jpg")
|
|
Image.new("RGB", (16, 16)).save(jpeg, "JPEG")
|
|
return result
|
|
|
|
self.api._convert_one = switching
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.state, "cancelled")
|
|
self.assertIn("format changed", job.status_line)
|
|
self.assertEqual(job.counters["generated"], 1)
|
|
self.assertTrue(self._thumb_path("f3", ".jpg").exists(), "the new format stays")
|
|
self.assertEqual(sum(1 for p in olds if p.exists()), 3)
|
|
|
|
async def test_the_new_preview_is_never_removed_as_an_old_one(self):
|
|
# Re-review N1: on a case-insensitive disk x_thumb.JPG (an old
|
|
# spelling) can be the x_thumb.jpg just written. A hard link stands
|
|
# in for that second name of one file.
|
|
PromptManagerConfig.update_config(
|
|
{"thumbnails": {"format": "jpeg", "quality": 80, "edge": 512}}
|
|
)
|
|
src = self.output_dir / "x.jpg"
|
|
Image.new("RGB", (64, 64), (200, 10, 10)).save(src, "JPEG")
|
|
alias = self._thumb_path("x", ".JPG")
|
|
alias.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (16, 16)).save(alias, "JPEG")
|
|
later = src.stat().st_mtime + 5
|
|
os.utime(alias, (later, later))
|
|
real = self.api._generate_one
|
|
|
|
def same_file(source, dst, *args, **kwargs):
|
|
made = real(source, dst, *args, **kwargs)
|
|
os.remove(alias)
|
|
os.link(made["path"], alias)
|
|
return made
|
|
|
|
self.api._generate_one = same_file
|
|
_, job = await self._convert()
|
|
self.assertTrue(self._thumb_path("x", ".jpg").exists())
|
|
self.assertTrue(alias.exists(), "the new preview's other name is kept")
|
|
self.assertEqual(job.counters["errors"], 0)
|
|
|
|
def test_same_file_falls_back_to_the_path_without_inode_numbers(self):
|
|
from py.api import thumbnail_convert as tc
|
|
|
|
a = self.output_dir / "a_thumb.jpg"
|
|
b = self.output_dir / "b_thumb.jpg"
|
|
for path in (a, b):
|
|
path.write_bytes(b"x")
|
|
real_stat = os.stat
|
|
|
|
def no_inode(path, *args, **kwargs):
|
|
result = real_stat(path, *args, **kwargs)
|
|
return os.stat_result(
|
|
(result.st_mode, 0, result.st_dev) + tuple(result)[3:]
|
|
)
|
|
|
|
with patch.object(tc.os, "stat", side_effect=no_inode):
|
|
self.assertFalse(tc._same_file(a, b))
|
|
self.assertTrue(tc._same_file(a, a))
|
|
self.assertTrue(tc._same_file(a, self.output_dir / "." / "a_thumb.jpg"))
|
|
self.assertFalse(tc._same_file(a, self.output_dir / "gone.jpg"))
|
|
|
|
async def test_convert_goes_on_past_the_per_batch_cap(self):
|
|
import py.api.images as images_module
|
|
|
|
olds = [self._legacy(f"c{i}") for i in range(8)]
|
|
with patch.object(images_module, "MAX_THUMBNAIL_FILES", 3):
|
|
_, job = await self._convert()
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertEqual(job.counters["generated"], 8)
|
|
self.assertFalse(any(old.exists() for old in olds))
|
|
self.assertEqual(job.summary()["total"], 8)
|
|
self.assertRegex(job.status_line, r"^8 converted · 0 failed · ")
|
|
self.assertNotIn("next run", job.status_line)
|
|
|
|
def test_byte_sizes_are_readable(self):
|
|
from py.api.thumbnail_convert import readable_bytes
|
|
|
|
self.assertEqual(readable_bytes(0), "0 B")
|
|
self.assertEqual(readable_bytes(1536), "1.5 KB")
|
|
self.assertEqual(readable_bytes(5 * 1024 * 1024), "5.0 MB")
|
|
self.assertEqual(readable_bytes(3 * 1024**3), "3.0 GB")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|