Vito
2cecfcf0ac
* feat(activity): scan new and unresolved files by default, all files on request
Scan for metadata now skips, before reading any metadata, the files the
Library already links with a resolved link (one generated_images row with
exactly one pair link), found with one streamed query that keeps only the
scanned paths and compares paths with os.path.normcase on both sides.
Skipped files count as Already known ("Skipped · already linked").
Unlinked, fallback-linked, ambiguous and unresolved files are read again.
POST /prompt_manager/jobs/scan takes {scope: "new" | "all"} (default
"new", anything else is a 400). The review dialog shows the concept's
"New and unresolved files" and adds an "All files" switch (D58). The
legacy /prompt_manager/scan route keeps scanning all files with unchanged
SSE payloads.
* feat(activity): download a job report
GET /prompt_manager/jobs/{id}/report returns the summary, counters, every
outcome group's listed rows with truncated flags, the scope and the log as a
JSON attachment (prompt-manager-<kind>-<YYYYMMDD-HHMM>.json), with no absolute
paths. The job view's Download report saves it through the api-client URL, and
the scan log opens at its newest line and follows new lines only while the
reader is at the bottom.
* fix(activity): re-read known files whose size changed, and review follow-ups
The new scope now skips a known, resolved file only while its stored
generated_images.file_size equals its size on disk; a NULL stored size is
re-read. A file replaced at the same path is read again and its conflict
shows under Needs review. The scan stores each file's size when it ingests
it, and both sides of the known-set match go through normalize_image_path.
A failed known-set read logs and falls back to reading every file, and an
empty request body asks for the default scope. The All files switch is
described by its hint, a lost job's Download report is disabled, the report
toast is skipped once the view is gone, and the scan log opens at its
newest line every time. Fixtures use the real scope shape.
* test(activity): wait for the job view's settled detail read, not a fixed delay
The 200-row cap test now starts counting detail reads once one read has
returned the full listing and none is in flight. The job view re-reads after
an in-flight read only when the summary still lists more than that read
returned, so no redundant read is left due after opening.
437 lines
15 KiB
Python
437 lines
15 KiB
Python
"""Unit tests for py/api/jobs.py: the in-memory session job registry.
|
|
|
|
Fake runners only: no files, no database. Ordering is driven by events and
|
|
awaited tasks, never by wall-clock bounds.
|
|
"""
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from py.api.jobs import ( # noqa: E402
|
|
DEFAULT_SCAN_SCOPE,
|
|
LOG_CAP_LINE,
|
|
MAX_FINISHED_JOBS,
|
|
MAX_LOG_LINES,
|
|
MAX_OUTCOME_ROWS,
|
|
SCAN_GROUP_NAMES,
|
|
SCAN_GROUPS,
|
|
SCAN_SCOPES,
|
|
Job,
|
|
JobBusy,
|
|
JobRegistry,
|
|
)
|
|
|
|
|
|
async def _collect(job, stop_after=None):
|
|
seen = []
|
|
async with contextlib.aclosing(job.events()) as events:
|
|
async for payload in events:
|
|
seen.append(payload)
|
|
if stop_after is not None and len(seen) >= stop_after:
|
|
break
|
|
return seen
|
|
|
|
|
|
def _gated_runner(gate, publish=None):
|
|
"""A runner that publishes ``publish`` (if any) and waits for ``gate``."""
|
|
|
|
async def runner(job):
|
|
if publish is not None:
|
|
job.publish(publish)
|
|
await gate.wait()
|
|
|
|
return runner
|
|
|
|
|
|
async def _noop(job):
|
|
return None
|
|
|
|
|
|
class TestJobFanOut(unittest.IsolatedAsyncioTestCase):
|
|
"""The SSE fan-out the scan job always had, now on every Job."""
|
|
|
|
async def test_new_job_is_running_with_no_event(self):
|
|
job = Job("scan", "Scan")
|
|
self.assertTrue(job.running)
|
|
self.assertIsNone(job.last_event)
|
|
self.assertEqual(job.state, "running")
|
|
self.assertEqual(len(job.id), 32)
|
|
|
|
async def test_subscriber_receives_events_until_finish(self):
|
|
job = Job("scan", "Scan")
|
|
consumer = asyncio.ensure_future(_collect(job))
|
|
await asyncio.sleep(0)
|
|
|
|
job.publish({"type": "progress", "progress": 10})
|
|
job.publish({"type": "complete"})
|
|
job.finish("finished")
|
|
|
|
self.assertEqual(
|
|
await asyncio.wait_for(consumer, 2),
|
|
[{"type": "progress", "progress": 10}, {"type": "complete"}],
|
|
)
|
|
self.assertFalse(job.running)
|
|
|
|
async def test_late_subscriber_gets_latest_snapshot_then_live_events(self):
|
|
job = Job("scan", "Scan")
|
|
job.publish({"type": "progress", "progress": 10})
|
|
job.publish({"type": "progress", "progress": 40})
|
|
|
|
consumer = asyncio.ensure_future(_collect(job))
|
|
await asyncio.sleep(0)
|
|
job.publish({"type": "complete"})
|
|
job.finish()
|
|
|
|
self.assertEqual(
|
|
await asyncio.wait_for(consumer, 2),
|
|
[{"type": "progress", "progress": 40}, {"type": "complete"}],
|
|
)
|
|
|
|
async def test_subscriber_after_finish_gets_only_the_final_event(self):
|
|
job = Job("scan", "Scan")
|
|
job.publish({"type": "progress", "progress": 99})
|
|
job.publish({"type": "complete"})
|
|
job.finish()
|
|
|
|
self.assertEqual(await _collect(job), [{"type": "complete"}])
|
|
|
|
async def test_departed_subscriber_is_forgotten(self):
|
|
job = Job("scan", "Scan")
|
|
job.publish({"type": "progress", "progress": 1})
|
|
await _collect(job, stop_after=1)
|
|
|
|
self.assertEqual(job.subscriber_count, 0)
|
|
job.publish({"type": "complete"}) # must not raise or queue anywhere
|
|
job.finish()
|
|
|
|
async def test_second_subscriber_attaches_to_running_job(self):
|
|
registry = JobRegistry()
|
|
gate = asyncio.Event()
|
|
job = registry.start(
|
|
"scan",
|
|
"Scan",
|
|
record_changing=True,
|
|
scope={},
|
|
runner=_gated_runner(gate, {"type": "progress", "progress": 30}),
|
|
)
|
|
await asyncio.sleep(0) # the runner publishes its first event
|
|
first = asyncio.ensure_future(_collect(job))
|
|
second = asyncio.ensure_future(_collect(job))
|
|
await asyncio.sleep(0)
|
|
self.assertEqual(job.subscriber_count, 2)
|
|
self.assertIs(registry.active("scan"), job)
|
|
|
|
job.publish({"type": "progress", "progress": 60})
|
|
gate.set()
|
|
await asyncio.wait_for(job.task, 2)
|
|
|
|
expected = [
|
|
{"type": "progress", "progress": 30},
|
|
{"type": "progress", "progress": 60},
|
|
]
|
|
self.assertEqual(await asyncio.wait_for(first, 2), expected)
|
|
self.assertEqual(await asyncio.wait_for(second, 2), expected)
|
|
self.assertEqual(job.summary()["progress"], 60)
|
|
|
|
async def test_finish_rejects_unknown_state_and_is_final(self):
|
|
job = Job("scan", "Scan")
|
|
with self.assertRaises(ValueError):
|
|
job.finish("running")
|
|
job.finish("cancelled")
|
|
job.finish("finished") # a second finish keeps the first state
|
|
self.assertEqual(job.state, "cancelled")
|
|
self.assertIsNotNone(job.finished_at)
|
|
|
|
|
|
class TestJobRegistry(unittest.IsolatedAsyncioTestCase):
|
|
|
|
async def test_start_lists_and_finishes_a_job(self):
|
|
registry = JobRegistry()
|
|
gate = asyncio.Event()
|
|
job = registry.start(
|
|
"scan",
|
|
"Scan for metadata",
|
|
record_changing=True,
|
|
scope={"files": "new", "output_folders": 2},
|
|
runner=_gated_runner(gate),
|
|
)
|
|
self.assertEqual(job.state, "running")
|
|
self.assertIs(registry.get(job.id), job)
|
|
self.assertIsNone(registry.get("missing"))
|
|
summary = job.summary()
|
|
self.assertEqual(
|
|
set(summary),
|
|
{
|
|
"id",
|
|
"kind",
|
|
"title",
|
|
"state",
|
|
"started_at",
|
|
"finished_at",
|
|
"progress",
|
|
"status_line",
|
|
"processed",
|
|
"total",
|
|
"counters",
|
|
"scope",
|
|
"listed",
|
|
"log_length",
|
|
},
|
|
)
|
|
self.assertEqual(summary["scope"], {"files": "new", "output_folders": 2})
|
|
# What the detail would list, so a view reads it only when that changed.
|
|
self.assertEqual((summary["listed"], summary["log_length"]), ({}, 0))
|
|
self.assertIsNone(summary["finished_at"])
|
|
|
|
gate.set()
|
|
await asyncio.wait_for(job.task, 2)
|
|
self.assertEqual(job.state, "finished")
|
|
self.assertIsNone(registry.active("scan"))
|
|
|
|
# Running jobs list first, then finished ones newest first.
|
|
older = registry.start(
|
|
"thumbs", "Thumbnails", record_changing=False, scope={}, runner=_noop
|
|
)
|
|
await asyncio.wait_for(older.task, 2)
|
|
hold = asyncio.Event()
|
|
running = registry.start(
|
|
"other",
|
|
"Other",
|
|
record_changing=False,
|
|
scope={},
|
|
runner=_gated_runner(hold),
|
|
)
|
|
listed = registry.list()
|
|
self.assertEqual([j.id for j in listed], [running.id, older.id, job.id])
|
|
|
|
# Only the last MAX_FINISHED_JOBS finished jobs are kept.
|
|
for index in range(MAX_FINISHED_JOBS + 5):
|
|
done = registry.start(
|
|
f"k{index}", "Fake", record_changing=False, scope={}, runner=_noop
|
|
)
|
|
await asyncio.wait_for(done.task, 2)
|
|
listed = registry.list()
|
|
finished = [j for j in listed if not j.running]
|
|
self.assertEqual(len(finished), MAX_FINISHED_JOBS)
|
|
self.assertEqual(finished[0].id, done.id)
|
|
self.assertIsNone(registry.get(job.id))
|
|
self.assertIs(listed[0], running)
|
|
hold.set()
|
|
await asyncio.wait_for(running.task, 2)
|
|
|
|
async def test_record_changing_jobs_are_exclusive(self):
|
|
registry = JobRegistry()
|
|
gate = asyncio.Event()
|
|
first = registry.start(
|
|
"scan",
|
|
"Scan for metadata",
|
|
record_changing=True,
|
|
scope={},
|
|
runner=_gated_runner(gate),
|
|
)
|
|
with self.assertRaises(JobBusy) as raised:
|
|
registry.start(
|
|
"maintenance",
|
|
"Maintenance",
|
|
record_changing=True,
|
|
scope={},
|
|
runner=_noop,
|
|
)
|
|
self.assertIs(raised.exception.running, first)
|
|
self.assertIn("Scan for metadata", str(raised.exception))
|
|
|
|
reader = registry.start(
|
|
"report", "Report", record_changing=False, scope={}, runner=_noop
|
|
)
|
|
await asyncio.wait_for(reader.task, 2)
|
|
self.assertEqual(reader.state, "finished")
|
|
|
|
# One job per kind, even a read-only one.
|
|
hold = asyncio.Event()
|
|
registry.start(
|
|
"report",
|
|
"Report",
|
|
record_changing=False,
|
|
scope={},
|
|
runner=_gated_runner(hold),
|
|
)
|
|
with self.assertRaises(JobBusy):
|
|
registry.start(
|
|
"report", "Report", record_changing=False, scope={}, runner=_noop
|
|
)
|
|
gate.set()
|
|
hold.set()
|
|
await asyncio.wait_for(first.task, 2)
|
|
|
|
async def test_cancel_is_idempotent_and_keeps_partial_counts(self):
|
|
registry = JobRegistry()
|
|
reached = asyncio.Event()
|
|
proceed = asyncio.Event()
|
|
|
|
async def runner(job):
|
|
for index in range(3):
|
|
if job.cancelled:
|
|
return
|
|
job.record("added", {"path": f"a{index}.png", "root": "output"}, "+")
|
|
if index == 0:
|
|
reached.set()
|
|
await proceed.wait()
|
|
|
|
job = registry.start(
|
|
"scan",
|
|
"Scan",
|
|
record_changing=True,
|
|
scope={},
|
|
runner=runner,
|
|
groups=SCAN_GROUP_NAMES,
|
|
)
|
|
await asyncio.wait_for(reached.wait(), 2)
|
|
self.assertTrue(registry.cancel(job.id))
|
|
self.assertTrue(registry.cancel(job.id)) # still running: still requested
|
|
proceed.set()
|
|
await asyncio.wait_for(job.task, 2)
|
|
|
|
self.assertEqual(job.state, "cancelled")
|
|
self.assertEqual(job.counters["added"], 1) # the work done before the stop
|
|
self.assertFalse(registry.cancel(job.id)) # finished: a no-op
|
|
self.assertFalse(registry.cancel("missing"))
|
|
self.assertEqual(job.state, "cancelled")
|
|
|
|
async def test_failing_runner_ends_failed_and_frees_the_slot(self):
|
|
registry = JobRegistry()
|
|
|
|
async def runner(job):
|
|
raise RuntimeError("boom")
|
|
|
|
job = registry.start(
|
|
"scan", "Scan", record_changing=True, scope={}, runner=runner
|
|
)
|
|
with self.assertLogs("py.api.jobs", level="ERROR"):
|
|
await asyncio.wait_for(job.task, 2)
|
|
self.assertEqual(job.state, "failed")
|
|
self.assertIsNone(registry.active("scan"))
|
|
|
|
async def test_runner_that_finishes_its_own_job_keeps_its_state(self):
|
|
registry = JobRegistry()
|
|
|
|
async def runner(job):
|
|
job.finish("failed")
|
|
|
|
job = registry.start(
|
|
"scan", "Scan", record_changing=True, scope={}, runner=runner
|
|
)
|
|
await asyncio.wait_for(job.task, 2)
|
|
self.assertEqual(job.state, "failed")
|
|
|
|
|
|
class TestJobOutcomesAndLog(unittest.TestCase):
|
|
|
|
def test_outcome_and_log_caps(self):
|
|
job = Job("scan", "Scan", groups=SCAN_GROUP_NAMES)
|
|
for index in range(MAX_OUTCOME_ROWS + 50):
|
|
job.record("errors", {"path": f"f{index}.png", "root": "output"}, "!")
|
|
self.assertEqual(len(job.outcomes["errors"]), MAX_OUTCOME_ROWS)
|
|
self.assertEqual(job.counters["errors"], MAX_OUTCOME_ROWS + 50)
|
|
self.assertEqual(job.outcomes["errors"][0]["path"], "f0.png")
|
|
|
|
job = Job("scan", "Scan")
|
|
for index in range(MAX_LOG_LINES + 100):
|
|
job.log_line(f"line {index}")
|
|
self.assertEqual(len(job.log), MAX_LOG_LINES + 1)
|
|
self.assertEqual(job.log[-1], LOG_CAP_LINE)
|
|
self.assertTrue(job.log[0].endswith("line 0"))
|
|
self.assertEqual(LOG_CAP_LINE, "[Log capped at 2,000 lines]")
|
|
|
|
def test_paths_are_root_relative(self):
|
|
job = Job("scan", "Scan", groups=SCAN_GROUP_NAMES)
|
|
absolute = os.path.abspath(os.path.join("outputs", "sub", "a.png"))
|
|
job.record(
|
|
"added",
|
|
{"path": absolute, "root": "output", "detail": "x"},
|
|
f"+ {absolute}",
|
|
)
|
|
job.record("known", {"path": "sub/b.png", "root": "output"}, "= sub/b.png")
|
|
job.record("review", {"path": "C:\\out\\c.png", "root": "output"}, "?")
|
|
rows = [row for group in job.outcomes.values() for row in group]
|
|
for row in rows:
|
|
self.assertFalse(os.path.isabs(row["path"]), row)
|
|
self.assertNotIn(":", row["path"])
|
|
self.assertEqual(job.outcomes["added"][0]["path"], "a.png")
|
|
self.assertEqual(job.outcomes["known"][0]["path"], "sub/b.png")
|
|
self.assertEqual(job.outcomes["review"][0]["path"], "c.png")
|
|
self.assertEqual(job.outcomes["known"][0]["detail"], "")
|
|
self.assertNotIn(absolute, "\n".join(job.log))
|
|
|
|
def test_record_rejects_unknown_group(self):
|
|
job = Job("scan", "Scan", groups=SCAN_GROUP_NAMES)
|
|
with self.assertRaises(ValueError):
|
|
job.record("bogus", {"path": "a.png", "root": "output"}, "x")
|
|
|
|
def test_detail_and_progress(self):
|
|
job = Job(
|
|
"scan",
|
|
"Scan",
|
|
scope={"files": "new", "output_folders": 2},
|
|
groups=SCAN_GROUP_NAMES,
|
|
)
|
|
job.update(progress=150, status_line="Processing file 3 of 4", processed=3)
|
|
job.update(total=4)
|
|
job.record("known", {"path": "a.png", "root": "output"}, "= a.png")
|
|
detail = job.detail()
|
|
self.assertEqual(detail["progress"], 100)
|
|
self.assertEqual(detail["status_line"], "Processing file 3 of 4")
|
|
self.assertEqual((detail["processed"], detail["total"]), (3, 4))
|
|
self.assertEqual(detail["counters"]["known"], 1)
|
|
self.assertEqual(detail["outcomes"]["known"][0]["path"], "a.png")
|
|
self.assertEqual(len(detail["log"]), 1)
|
|
detail["outcomes"]["known"].clear() # a copy, not the live list
|
|
self.assertEqual(len(job.outcomes["known"]), 1)
|
|
job.publish({"type": "progress", "progress": -5})
|
|
self.assertEqual(job.summary()["progress"], 0)
|
|
|
|
def test_scan_groups_follow_the_counter_mapping(self):
|
|
self.assertEqual(
|
|
SCAN_GROUPS,
|
|
{
|
|
"added": "added",
|
|
"linked": "added",
|
|
"enriched": "added",
|
|
"unchanged": "known",
|
|
# Activity-2: a file the "new" scope skipped is Already known.
|
|
"skipped": "known",
|
|
"ambiguous": "review",
|
|
"conflict": "review",
|
|
"unresolved": "review",
|
|
"failed": "errors",
|
|
},
|
|
)
|
|
self.assertEqual((SCAN_SCOPES, DEFAULT_SCAN_SCOPE), (("new", "all"), "new"))
|
|
self.assertEqual(
|
|
SCAN_GROUP_NAMES, ("added", "known", "review", "no_metadata", "errors")
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestListedSummary(unittest.TestCase):
|
|
def test_summary_counts_listed_rows_and_log_lines(self):
|
|
job = Job("scan", "Scan", groups=("added", "errors"))
|
|
job.record(
|
|
"added", {"path": "a.png", "root": "output", "root_index": 0}, "+ a.png"
|
|
)
|
|
job.record("errors", {"path": "b.png", "root_index": "x"}, "! b.png")
|
|
summary = job.summary()
|
|
self.assertEqual(summary["listed"], {"added": 1, "errors": 1})
|
|
self.assertEqual(summary["log_length"], 2)
|
|
rows = job.detail()["outcomes"]
|
|
self.assertEqual(rows["added"][0]["root_index"], 0)
|
|
# Only a whole non-negative number is a root index.
|
|
self.assertIsNone(rows["errors"][0]["root_index"])
|