feat(autotag): model downloads run as Activity jobs
This commit is contained in:
+135
-46
@@ -1,6 +1,8 @@
|
||||
"""AutoTag API routes for PromptManager."""
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import functools
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -9,7 +11,8 @@ from pathlib import Path
|
||||
from typing import Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
from .jobs import AUTOTAG_BATCH_JOB_KIND
|
||||
from .job_routes import busy_response
|
||||
from .jobs import AUTOTAG_BATCH_JOB_KIND, MODEL_DOWNLOAD_JOB_KIND, JobBusy
|
||||
from .prompts import thumbnail_url_for
|
||||
from . import thumbnail_settings
|
||||
from .. import model_gate
|
||||
@@ -36,6 +39,9 @@ OWNER_LABELS = {
|
||||
"image-annotations": "Generate suggestions",
|
||||
}
|
||||
ANOTHER_OPERATION = "Another tagging operation"
|
||||
_DOWNLOAD_FAILED = "The download failed. Check server logs for details."
|
||||
# POST /jobs/model-download carries only {model_type}.
|
||||
MODEL_DOWNLOAD_BODY_MAX = 1024
|
||||
# The running download's gate owner and job title. Only one download runs at a
|
||||
# time and each needs the gate, so a new start replaces the entry.
|
||||
_download_titles = {}
|
||||
@@ -135,6 +141,11 @@ def path_is_within(candidate, root) -> bool:
|
||||
return child != parent and child.is_relative_to(parent)
|
||||
|
||||
|
||||
async def _single(event):
|
||||
"""A stream of one event."""
|
||||
yield event
|
||||
|
||||
|
||||
async def _refusal(title):
|
||||
"""The single error event of an auto-tag batch refused by another job."""
|
||||
message = f"{title} already running"
|
||||
@@ -232,55 +243,133 @@ class AutotagRoutesMixin:
|
||||
return response
|
||||
|
||||
async def download_autotag_model(self, request):
|
||||
"""Download an AutoTag model with streaming progress (POST only)."""
|
||||
"""Download a tagging model, streaming the legacy events (POST only).
|
||||
|
||||
The download is the model_download Activity job: this request
|
||||
attaches to a running download of the same model or starts one, and
|
||||
streams the legacy frames until the job ends. A client disconnect
|
||||
ends only the stream; the download goes on. A running download of
|
||||
another model answers 409 model-busy (it holds the gate).
|
||||
"""
|
||||
from ..autotag import MODELS, get_autotag_service
|
||||
|
||||
model_type = request.match_info.get("model_type")
|
||||
owner = model_gate.new_owner("legacy-download")
|
||||
|
||||
async def events():
|
||||
try:
|
||||
from ..autotag import get_autotag_service
|
||||
|
||||
service = get_autotag_service()
|
||||
if model_type not in service.models_config:
|
||||
yield {
|
||||
"type": "error",
|
||||
"message": f"Invalid model type: {model_type}",
|
||||
}
|
||||
return
|
||||
|
||||
yield {
|
||||
"type": "progress",
|
||||
"progress": 0,
|
||||
"status": "Starting download...",
|
||||
}
|
||||
|
||||
def progress_callback(status: str, progress: float):
|
||||
pass
|
||||
|
||||
success = await model_gate.run(
|
||||
owner, service.download_model, model_type, progress_callback
|
||||
)
|
||||
if success:
|
||||
yield {
|
||||
"type": "complete",
|
||||
"progress": 100,
|
||||
"status": "Download complete",
|
||||
}
|
||||
else:
|
||||
yield {"type": "error", "message": "Download failed"}
|
||||
except Exception:
|
||||
self.logger.exception("Download model error")
|
||||
yield {
|
||||
"type": "error",
|
||||
"message": _INTERNAL_ERROR,
|
||||
}
|
||||
|
||||
if not model_gate.try_acquire(owner):
|
||||
if model_type not in MODELS:
|
||||
invalid = {"type": "error", "message": f"Invalid model type: {model_type}"}
|
||||
return await self._stream_sse(request, _single(invalid))
|
||||
job = self._jobs.active(MODEL_DOWNLOAD_JOB_KIND)
|
||||
if job is not None and job.scope.get("model_type") != model_type:
|
||||
return model_busy_response()
|
||||
if job is None:
|
||||
job, refusal = self._start_model_download(model_type, get_autotag_service())
|
||||
if refusal is not None:
|
||||
return refusal
|
||||
return await self._stream_sse(request, self._download_frames(job))
|
||||
|
||||
@staticmethod
|
||||
async def _download_frames(job):
|
||||
"""The legacy download frames for job: start, then its end."""
|
||||
yield {"type": "progress", "progress": 0, "status": "Starting download..."}
|
||||
async with contextlib.aclosing(job.events()) as events:
|
||||
async for _payload in events:
|
||||
pass
|
||||
if job.state == "finished":
|
||||
yield {"type": "complete", "progress": 100, "status": "Download complete"}
|
||||
else:
|
||||
yield {"type": "error", "message": "Download failed"}
|
||||
|
||||
# ── model download job (#328) ───────────────────────────────────────
|
||||
|
||||
async def start_model_download_job(self, request):
|
||||
"""``POST /jobs/model-download {model_type}``: 202 with the job.
|
||||
|
||||
400 for a bad body, 409 ``busy`` while a download (or an exclusive
|
||||
job) runs, 409 ``model-busy`` while another operation holds the gate.
|
||||
"""
|
||||
data, error = await self._read_json_body(
|
||||
request, max_bytes=MODEL_DOWNLOAD_BODY_MAX
|
||||
)
|
||||
if error is not None:
|
||||
return error
|
||||
if set(data) != {"model_type"}:
|
||||
return self._bad_request("The body must be exactly {model_type}")
|
||||
from ..autotag import MODELS, get_autotag_service
|
||||
|
||||
model_type = data["model_type"]
|
||||
if not isinstance(model_type, str) or model_type not in MODELS:
|
||||
return self._bad_request(
|
||||
"model_type must be one of " + ", ".join(sorted(MODELS))
|
||||
)
|
||||
job, refusal = self._start_model_download(model_type, get_autotag_service())
|
||||
if refusal is not None:
|
||||
return refusal
|
||||
return web.json_response({"success": True, "job": job.summary()}, status=202)
|
||||
|
||||
def _start_model_download(self, model_type, service):
|
||||
"""Register the download job: ``(job, None)`` or ``(None, refusal)``.
|
||||
|
||||
Synchronous (no await between the checks and the start). The gate is
|
||||
taken before the job registers and released when the runner ends,
|
||||
also when the task is cancelled before the runner ever starts.
|
||||
"""
|
||||
from ..autotag import MODELS
|
||||
|
||||
busy = self._jobs.blocker(MODEL_DOWNLOAD_JOB_KIND, admission="read")
|
||||
if busy is not None:
|
||||
return None, busy_response(JobBusy(busy))
|
||||
owner = model_gate.new_owner(MODEL_DOWNLOAD_OWNER_KIND)
|
||||
if not model_gate.try_acquire(owner):
|
||||
return None, model_busy_response()
|
||||
config = MODELS[model_type]
|
||||
label = config["label"]
|
||||
title = f"Tagging model download · {label}"
|
||||
_download_titles.clear()
|
||||
_download_titles[owner] = title
|
||||
try:
|
||||
return await self._stream_sse(request, events())
|
||||
finally:
|
||||
job = self._jobs.start(
|
||||
MODEL_DOWNLOAD_JOB_KIND,
|
||||
title,
|
||||
admission="read",
|
||||
cancellable=False,
|
||||
scope={
|
||||
"model_type": model_type,
|
||||
"label": label,
|
||||
"size_label": size_label(config["size"]),
|
||||
},
|
||||
runner=functools.partial(
|
||||
self._run_model_download, service, model_type, owner
|
||||
),
|
||||
)
|
||||
except JobBusy as refused:
|
||||
model_gate.release(owner)
|
||||
return None, busy_response(refused)
|
||||
job.task.add_done_callback(lambda _task: model_gate.release(owner))
|
||||
return job, None
|
||||
|
||||
async def _run_model_download(self, service, model_type, owner, job):
|
||||
label = job.scope["label"]
|
||||
job.update(status_line=f"Downloading {label} ({job.scope['size_label']})")
|
||||
job.log_line(f"Download started: {label}")
|
||||
job.publish({"type": "status", "status": job.status_line})
|
||||
try:
|
||||
ok = await model_gate.run(owner, service.download_model, model_type, None)
|
||||
except Exception:
|
||||
self.logger.exception(f"Tagging model download failed: {model_type}")
|
||||
ok = False
|
||||
finally:
|
||||
# Before the job ends, so a reader that sees the end finds the
|
||||
# gate free (a running download thread defers the release).
|
||||
model_gate.release(owner)
|
||||
if ok:
|
||||
job.update(status_line=f"Downloaded {label}")
|
||||
job.log_line(f"Download finished: {label}")
|
||||
job.publish({"type": "status", "status": job.status_line})
|
||||
job.finish("finished")
|
||||
return
|
||||
job.update(status_line=_DOWNLOAD_FAILED)
|
||||
job.log_line(f"Download failed: {label}")
|
||||
job.publish({"type": "status", "status": job.status_line})
|
||||
job.finish("failed")
|
||||
|
||||
async def _read_start_params(self, request) -> Optional[dict]:
|
||||
"""Merge query-string and optional JSON-body parameters for autotag/start.
|
||||
|
||||
@@ -130,6 +130,10 @@ class JobRoutesMixin:
|
||||
async def start_image_tag_job_route(request):
|
||||
return await self.start_image_tag_job(request)
|
||||
|
||||
@routes.post("/prompt_manager/jobs/model-download")
|
||||
async def start_model_download_job_route(request):
|
||||
return await self.start_model_download_job(request)
|
||||
|
||||
@routes.post("/prompt_manager/jobs/thumbnails")
|
||||
async def start_thumbnail_job_route(request):
|
||||
return await self.start_thumbnail_job(request)
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
// #328: the legacy admin and gallery auto-tag tools keep working beside System › Auto-tagging. The model download is
|
||||
// the `model_download` Activity job; the pages' stream attaches to it. No model is ever downloaded: the fixture's
|
||||
// fake download waits until the test releases it (`/__test__/tagging/download`).
|
||||
import { startCoverage, saveCoverage } from '../../tools/vendor/browser-coverage.mjs';
|
||||
import { test, expect } from './fixtures.mjs';
|
||||
|
||||
const TITLE = 'Tagging model download · WD14 ViT';
|
||||
|
||||
async function post(request, url, data) {
|
||||
await expect(await request.post(url, data === undefined ? {} : { data })).toBeOK();
|
||||
}
|
||||
|
||||
async function downloads(request) {
|
||||
const response = await request.get('/prompt_manager/jobs');
|
||||
await expect(response).toBeOK();
|
||||
return (await response.json()).jobs.filter(job => job.kind === 'model_download');
|
||||
}
|
||||
|
||||
test.beforeEach(async ({ page, request, browserName }) => {
|
||||
await startCoverage(page, browserName);
|
||||
// The 3.2.0 update notice would cover the admin page's toolbar.
|
||||
await page.addInitScript(() => { localStorage.setItem('pm_update_notice_dismissed', '3.2.0'); });
|
||||
await post(request, '/__test__/jobs/restart');
|
||||
await post(request, '/__test__/tagging/reset');
|
||||
await post(request, '/__test__/tagging/models', { unavailable: ['wd14-vit'] });
|
||||
await post(request, '/__test__/tagging/download', { mode: 'hold' });
|
||||
});
|
||||
|
||||
test.afterEach(async ({ page, request, browserName }) => {
|
||||
await saveCoverage(page, browserName);
|
||||
// Releases a download a failed test left held, and puts the raising default back.
|
||||
await post(request, '/__test__/tagging/reset');
|
||||
await post(request, '/__test__/jobs/restart');
|
||||
});
|
||||
|
||||
async function startDownload(page, request) {
|
||||
await page.locator('#autoTagBtn').click();
|
||||
const status = page.locator('#wd14VitModelStatus');
|
||||
await status.getByRole('button', { name: 'Download', exact: true }).click();
|
||||
const modal = page.locator('#autoTagDownloadModal');
|
||||
await expect(modal).toBeVisible();
|
||||
await expect(page.locator('#downloadStatus')).toHaveText('Downloading…');
|
||||
await expect(page.locator('#downloadProgressBar')).toHaveClass(/animate-pulse/);
|
||||
await expect(modal).toContainText('The download continues in Activity.');
|
||||
await expect(page.locator('#cancelDownloadBtn')).toHaveText('Close');
|
||||
const running = (await downloads(request)).filter(job => job.state === 'running');
|
||||
expect(running.map(job => [job.title, job.cancellable])).toEqual([[TITLE, false]]);
|
||||
return { modal, status };
|
||||
}
|
||||
|
||||
test('admin download modal closes on complete and Activity lists the job without Cancel', async ({ page, request }) => {
|
||||
await page.goto('/prompt_manager/admin');
|
||||
const { modal, status } = await startDownload(page, request);
|
||||
|
||||
await page.goto('/prompt_manager/workspace#/activity');
|
||||
const item = page.locator('#workspace-main section.surface > .activity-item')
|
||||
.filter({ has: page.getByRole('link', { name: new RegExp(`${TITLE}$`) }) });
|
||||
await expect(item.locator('.chip')).toHaveText('Running');
|
||||
await expect(item.locator('progress')).not.toHaveAttribute('value');
|
||||
await expect(item.getByRole('button', { name: 'Cancel job', exact: true })).toHaveCount(0);
|
||||
|
||||
// Back on the admin page a second download of the same model attaches to the running job.
|
||||
await page.goto('/prompt_manager/admin');
|
||||
await startDownload(page, request);
|
||||
await post(request, '/__test__/tagging/download', { mode: 'succeed' });
|
||||
await expect(modal).toBeHidden();
|
||||
await expect(status).toContainText('Downloaded');
|
||||
expect((await downloads(request)).map(job => [job.title, job.state])).toEqual([[TITLE, 'finished']]);
|
||||
});
|
||||
|
||||
test('gallery download modal closes on complete', async ({ page, request }) => {
|
||||
await page.goto('/prompt_manager/gallery');
|
||||
const { modal, status } = await startDownload(page, request);
|
||||
await post(request, '/__test__/tagging/download', { mode: 'succeed' });
|
||||
await expect(modal).toBeHidden();
|
||||
await expect(page.locator('#autoTagModal')).toBeVisible();
|
||||
await expect(status).toContainText('Downloaded');
|
||||
expect((await downloads(request)).map(job => job.state)).toEqual(['finished']);
|
||||
});
|
||||
|
||||
test('a failed download closes the modal and says so', async ({ page, request }) => {
|
||||
await page.goto('/prompt_manager/gallery');
|
||||
const { modal, status } = await startDownload(page, request);
|
||||
await post(request, '/__test__/tagging/download', { mode: 'fail' });
|
||||
await expect(modal).toBeHidden();
|
||||
await expect(page.getByText('Download failed: Download failed')).toBeVisible();
|
||||
await expect(status.getByRole('button', { name: 'Download', exact: true })).toBeVisible();
|
||||
expect((await downloads(request)).map(job => job.state)).toEqual(['failed']);
|
||||
});
|
||||
@@ -1088,6 +1088,36 @@ test('a held legacy job and the health check show no tiles, no Results and no ma
|
||||
await expect(toast(page)).toHaveText('Health check finished.');
|
||||
});
|
||||
|
||||
// #328: a tagging model download is a job of its own kind with no byte progress and no Cancel; Activity shows it
|
||||
// through the generic held view (its title, status line and an indeterminate bar), then its end toast.
|
||||
test('a held model download shows its title, an indeterminate bar and no Cancel; its end toast follows', async ({ page, request }) => {
|
||||
await page.setViewportSize({ width: 1600, height: 1100 });
|
||||
const title = 'Tagging model download · WD14 ViT';
|
||||
await expect(await request.post('/__test__/tagging/reset')).toBeOK();
|
||||
await expect(await request.post('/__test__/tagging/download', { data: { mode: 'hold' } })).toBeOK();
|
||||
try {
|
||||
const response = await request.post('/prompt_manager/jobs/model-download', { data: { model_type: 'wd14-vit' } });
|
||||
expect(response.status()).toBe(202);
|
||||
const job = (await response.json()).job;
|
||||
expect([job.kind, job.title, job.cancellable]).toEqual(['model_download', title, false]);
|
||||
await page.goto(`${workspace}#/activity`);
|
||||
const item = items(page).filter({ has: page.getByRole('link', { name: new RegExp(`${title}$`) }) });
|
||||
await expect(item.locator('progress')).not.toHaveAttribute('value');
|
||||
await expect(item.getByRole('button', { name: 'Cancel job', exact: true })).toHaveCount(0);
|
||||
await item.getByRole('link').click();
|
||||
await expect(heading(page).locator('h1')).toHaveText(title);
|
||||
await expect(progressSection(page).locator('.panel-heading > span.small')).toHaveText('Downloading WD14 ViT (~400 MB)');
|
||||
await expect(progressSection(page).locator('progress')).not.toHaveAttribute('value');
|
||||
await expect(progressSection(page).locator('dl.stat-row')).toHaveCount(0);
|
||||
await expect(heading(page).getByRole('button', { name: /^Cancel/ })).toHaveCount(0);
|
||||
await expect(await request.post('/__test__/tagging/download', { data: { mode: 'succeed' } })).toBeOK();
|
||||
await expect(toast(page)).toHaveText(`${title} finished.`);
|
||||
await expect(heading(page).locator('p.muted')).toHaveText(/^Finished \d\d:\d\d$/);
|
||||
} finally {
|
||||
await request.post('/__test__/tagging/reset');
|
||||
}
|
||||
});
|
||||
|
||||
// S5 review I3: a thumbnail job with nothing to generate ends on the server's final line ("No previews to generate · N
|
||||
// already up to date"), never on a stale "Processing file 0 of 0" under the Completed chip.
|
||||
test('a finished thumbnail job with nothing to do says so', async ({ page, request }) => {
|
||||
|
||||
@@ -12,6 +12,7 @@ import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import threading
|
||||
import time
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
@@ -181,6 +182,20 @@ def register_tagging_commands(
|
||||
routes: web.RouteTableDef, api: PromptManagerAPI, root: Path, model
|
||||
) -> None:
|
||||
"""Register fixed seed, snapshot and fake-model commands; no paths or SQL input."""
|
||||
# The fake download (``/__test__/tagging/download``): waits on ``event``,
|
||||
# then answers ``result``. Every reset sets the event and puts back the
|
||||
# model's raising default, so no test can leave a worker hanging.
|
||||
download = {"event": threading.Event(), "result": True}
|
||||
|
||||
def restore_download() -> None:
|
||||
download["event"].set()
|
||||
model.__dict__.pop("download_model", None)
|
||||
|
||||
def fake_download(model_type: str, progress_callback=None) -> bool:
|
||||
download["event"].wait(DOWNLOAD_HOLD_LIMIT_S)
|
||||
if download["result"]:
|
||||
model.unavailable = model.unavailable - {model_type}
|
||||
return download["result"]
|
||||
|
||||
@routes.post("/__test__/tagging/reset")
|
||||
async def reset(request: web.Request) -> web.Response:
|
||||
@@ -195,6 +210,7 @@ def register_tagging_commands(
|
||||
model.unavailable = set()
|
||||
model.suggestions = None
|
||||
model.loaded = None
|
||||
restore_download()
|
||||
_set_defaults(model, DEFAULTS)
|
||||
# A test that ends with the dialog open leaves its capture behind; drop
|
||||
# every capture so the next test starts below the store's limits.
|
||||
@@ -223,6 +239,33 @@ def register_tagging_commands(
|
||||
model.unavailable = set(data["unavailable"])
|
||||
return web.json_response({"success": True})
|
||||
|
||||
@routes.post("/__test__/tagging/download")
|
||||
async def download_mode(request: web.Request) -> web.Response:
|
||||
"""Switch the model's download to a fake (never a real download).
|
||||
|
||||
``{"mode": "hold"}``: the next download waits until a later
|
||||
``succeed`` or ``fail`` (or a reset) releases it. ``succeed`` and
|
||||
``fail`` release a held download with that result and answer every
|
||||
later one at once. A successful fake marks the model downloaded.
|
||||
"""
|
||||
data = await request.json() if request.can_read_body else None
|
||||
if (
|
||||
request.query
|
||||
or not isinstance(data, dict)
|
||||
or set(data) != {"mode"}
|
||||
or data["mode"] not in ("hold", "succeed", "fail")
|
||||
):
|
||||
raise web.HTTPBadRequest()
|
||||
if data["mode"] == "hold":
|
||||
download["event"].set()
|
||||
download["event"] = threading.Event()
|
||||
download["result"] = True
|
||||
else:
|
||||
download["result"] = data["mode"] == "succeed"
|
||||
download["event"].set()
|
||||
model.download_model = fake_download
|
||||
return web.json_response({"success": True})
|
||||
|
||||
@routes.post("/__test__/tagging/suggestions")
|
||||
async def suggestions(request: web.Request) -> web.Response:
|
||||
"""Answer Generate with fixed scored names (the concept's own sample).
|
||||
@@ -273,6 +316,9 @@ def register_tagging_commands(
|
||||
return web.json_response({"captures": count})
|
||||
|
||||
|
||||
# A held fake download gives up after this long, whatever the test does.
|
||||
DOWNLOAD_HOLD_LIMIT_S = 60
|
||||
|
||||
# The standard tagging defaults: WD14 general and character thresholds, and
|
||||
# the suggestion count (``AutotagConfig``).
|
||||
DEFAULTS = (0.35, 0.85, 10)
|
||||
|
||||
+256
-1
@@ -34,6 +34,7 @@ from aiohttp.test_utils import AioHTTPTestCase
|
||||
|
||||
from database.operations import PromptDatabase
|
||||
from py.api import PromptManagerAPI
|
||||
from py.api.jobs import JobBusy
|
||||
from utils.hashing import generate_prompt_hash
|
||||
|
||||
try: # guarded so a missing module fails assertions, not test loading
|
||||
@@ -1068,7 +1069,9 @@ class TestLegacyRoutesShareTheModelGate(AutotagAPITestCase):
|
||||
self.assertIsNone(model_gate.holder())
|
||||
|
||||
self.assertEqual(len(seen), 3)
|
||||
self.assertTrue(all(owner and owner.startswith("legacy-") for owner in seen))
|
||||
# The download runs as the model_download job, which owns the gate.
|
||||
kinds = [owner.split(":", 1)[0] for owner in seen]
|
||||
self.assertEqual(kinds, ["legacy-single", "legacy-single", "model-download"])
|
||||
|
||||
async def test_validation_errors_win_over_the_gate(self):
|
||||
self.assertTrue(model_gate.try_acquire("test-holder"))
|
||||
@@ -1440,3 +1443,255 @@ class TestModelBusyNamesTheOperation(AutotagAPITestCase):
|
||||
self.assertEqual(data["operation"], "Generate suggestions")
|
||||
self.assertEqual(self.service.unload_calls, 0)
|
||||
self.assertEqual(self.service.get_loaded_model_type(), "wd14-vit")
|
||||
|
||||
|
||||
class DownloadJobTestCase(AutotagAPITestCase):
|
||||
"""Helpers for the ``model_download`` job: a download that waits for a
|
||||
signal, so a test sees the job while it runs."""
|
||||
|
||||
TITLE = "Tagging model download · WD14 ViT"
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.started = threading.Event()
|
||||
self.proceed = threading.Event()
|
||||
|
||||
def _hold_download(self, result=True, error=None):
|
||||
def download(model_type, progress_callback=None):
|
||||
self.service.download_calls.append(model_type)
|
||||
self.started.set()
|
||||
self.proceed.wait(30)
|
||||
if error is not None:
|
||||
raise error
|
||||
return result
|
||||
|
||||
self.service.download_model = download
|
||||
|
||||
async def _started(self):
|
||||
loop = asyncio.get_running_loop()
|
||||
self.assertTrue(await loop.run_in_executor(None, self.started.wait, 30))
|
||||
|
||||
async def _start(self, model_type="wd14-vit"):
|
||||
return await self._post_json(
|
||||
"/prompt_manager/jobs/model-download", {"model_type": model_type}
|
||||
)
|
||||
|
||||
async def _ended(self, job_id):
|
||||
job = self.api._jobs.get(job_id)
|
||||
self.proceed.set()
|
||||
await asyncio.wait_for(asyncio.shield(job.task), 30)
|
||||
return job
|
||||
|
||||
|
||||
class TestModelDownloadJob(DownloadJobTestCase):
|
||||
|
||||
async def test_download_job_is_indeterminate_and_not_cancellable(self):
|
||||
self._hold_download()
|
||||
resp = await self._start()
|
||||
self.assertEqual(resp.status, 202)
|
||||
summary = (await resp.json())["job"]
|
||||
try:
|
||||
await self._started()
|
||||
self.assertEqual(summary["kind"], "model_download")
|
||||
self.assertEqual(summary["title"], self.TITLE)
|
||||
self.assertFalse(summary["cancellable"])
|
||||
self.assertEqual(summary["admission"], "read")
|
||||
self.assertEqual(
|
||||
summary["scope"],
|
||||
{
|
||||
"model_type": "wd14-vit",
|
||||
"label": "WD14 ViT",
|
||||
"size_label": "~400 MB",
|
||||
},
|
||||
)
|
||||
detail = (
|
||||
await (
|
||||
await self.client.get(f"/prompt_manager/jobs/{summary['id']}")
|
||||
).json()
|
||||
)["job"]
|
||||
self.assertEqual((detail["progress"], detail["total"]), (0, 0))
|
||||
self.assertEqual(detail["status_line"], "Downloading WD14 ViT (~400 MB)")
|
||||
cancel = await self.client.post(
|
||||
f"/prompt_manager/jobs/{summary['id']}/cancel"
|
||||
)
|
||||
self.assertFalse((await cancel.json())["cancellation_requested"])
|
||||
finally:
|
||||
job = await self._ended(summary["id"])
|
||||
self.assertEqual(job.state, "finished")
|
||||
self.assertEqual(job.status_line, "Downloaded WD14 ViT")
|
||||
self.assertEqual(self.service.download_calls, ["wd14-vit"])
|
||||
self.assertEqual(len(job.log), 2)
|
||||
|
||||
async def test_start_body_is_checked_before_anything_runs(self):
|
||||
for body in (
|
||||
{"model_type": "nope"},
|
||||
{"model_type": "wd14-vit", "extra": 1},
|
||||
{},
|
||||
["wd14-vit"],
|
||||
{"model_type": 3},
|
||||
):
|
||||
resp = await self._post_json("/prompt_manager/jobs/model-download", body)
|
||||
self.assertEqual(resp.status, 400, body)
|
||||
resp = await self._post_json(
|
||||
"/prompt_manager/jobs/model-download",
|
||||
{"model_type": "wd14-vit", "pad": "x" * 2048},
|
||||
)
|
||||
self.assertIn(resp.status, (400, 413))
|
||||
self.assertEqual(self.service.download_calls, [])
|
||||
self.assertIsNone(model_gate.holder())
|
||||
self.assertEqual(self.api._jobs.list(), [])
|
||||
|
||||
async def test_second_download_is_409_busy(self):
|
||||
self._hold_download()
|
||||
first = (await (await self._start()).json())["job"]
|
||||
try:
|
||||
await self._started()
|
||||
resp = await self._start("gguf")
|
||||
data = await resp.json()
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(data["code"], "busy")
|
||||
self.assertEqual(data["running"]["title"], self.TITLE)
|
||||
finally:
|
||||
await self._ended(first["id"])
|
||||
self.assertEqual(self.service.download_calls, ["wd14-vit"])
|
||||
|
||||
async def test_download_holds_gate_until_runner_exits(self):
|
||||
self._hold_download()
|
||||
self.service._loaded_type = "wd14-swinv2"
|
||||
first = (await (await self._start()).json())["job"]
|
||||
try:
|
||||
await self._started()
|
||||
self.assertTrue(model_gate.holder().startswith("model-download:"))
|
||||
resp = await self.client.post("/prompt_manager/autotag/unload")
|
||||
data = await resp.json()
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(data["operation"], self.TITLE)
|
||||
finally:
|
||||
await self._ended(first["id"])
|
||||
self.assertIsNone(model_gate.holder())
|
||||
self.assertEqual(self.service.unload_calls, 0)
|
||||
|
||||
async def test_download_releases_gate_when_start_is_refused(self):
|
||||
class Running:
|
||||
id = "other"
|
||||
title = "Database restore"
|
||||
|
||||
with patch.object(self.api._jobs, "start", side_effect=JobBusy(Running())):
|
||||
resp = await self._start()
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual((await resp.json())["code"], "busy")
|
||||
self.assertIsNone(model_gate.holder())
|
||||
self.assertEqual(self.service.download_calls, [])
|
||||
|
||||
async def test_gate_held_elsewhere_is_409_model_busy(self):
|
||||
owner = model_gate.new_owner("image-annotations")
|
||||
self.assertTrue(model_gate.try_acquire(owner))
|
||||
try:
|
||||
resp = await self._start()
|
||||
data = await resp.json()
|
||||
finally:
|
||||
model_gate.release(owner)
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(data["operation"], "Generate suggestions")
|
||||
self.assertEqual(self.api._jobs.list(), [])
|
||||
|
||||
async def test_failed_download_fails_the_job_without_paths(self):
|
||||
for outcome in ({"result": False}, {"error": OSError("/srv/models is full")}):
|
||||
self.started.clear()
|
||||
self.proceed.clear()
|
||||
self._hold_download(**outcome)
|
||||
summary = (await (await self._start()).json())["job"]
|
||||
await self._started()
|
||||
job = await self._ended(summary["id"])
|
||||
self.assertEqual(job.state, "failed", outcome)
|
||||
self.assertEqual(
|
||||
job.status_line, "The download failed. Check server logs for details."
|
||||
)
|
||||
report = await self.client.get(
|
||||
f"/prompt_manager/jobs/{summary['id']}/report"
|
||||
)
|
||||
text = await report.text()
|
||||
self.assertNotIn("/srv/models", text)
|
||||
leaked = [s for s in _strings(json.loads(text)) if _absolute(s)]
|
||||
self.assertEqual(leaked, [])
|
||||
self.assertIsNone(model_gate.holder())
|
||||
|
||||
|
||||
class TestLegacyDownloadAttaches(DownloadJobTestCase):
|
||||
"""POST /autotag/download/{type}: the legacy stream rides on the job."""
|
||||
|
||||
GOLDEN_OK = [
|
||||
{"type": "progress", "progress": 0, "status": "Starting download..."},
|
||||
{"type": "complete", "progress": 100, "status": "Download complete"},
|
||||
]
|
||||
GOLDEN_FAILED = [
|
||||
{"type": "progress", "progress": 0, "status": "Starting download..."},
|
||||
{"type": "error", "message": "Download failed"},
|
||||
]
|
||||
|
||||
async def test_legacy_download_events_unchanged(self):
|
||||
resp = await self.client.post("/prompt_manager/autotag/download/wd14-vit")
|
||||
self.assertEqual(resp.status, 200)
|
||||
self.assertEqual(self._sse_events(await resp.text()), self.GOLDEN_OK)
|
||||
jobs = self.api._jobs.list()
|
||||
self.assertEqual(
|
||||
[(j.kind, j.title, j.state) for j in jobs],
|
||||
[("model_download", self.TITLE, "finished")],
|
||||
)
|
||||
|
||||
self.service.download_result = False
|
||||
resp = await self.client.post("/prompt_manager/autotag/download/wd14-vit")
|
||||
self.assertEqual(self._sse_events(await resp.text()), self.GOLDEN_FAILED)
|
||||
|
||||
async def test_legacy_download_attaches_to_running_job(self):
|
||||
self._hold_download()
|
||||
summary = (await (await self._start()).json())["job"]
|
||||
await self._started()
|
||||
legacy = asyncio.ensure_future(
|
||||
self.client.post("/prompt_manager/autotag/download/wd14-vit")
|
||||
)
|
||||
try:
|
||||
# Wait until the legacy stream is subscribed to the running job.
|
||||
job = self.api._jobs.get(summary["id"])
|
||||
for _ in range(300):
|
||||
if job.subscriber_count:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
self.assertEqual(job.subscriber_count, 1)
|
||||
finally:
|
||||
await self._ended(summary["id"])
|
||||
resp = await legacy
|
||||
self.assertEqual(self._sse_events(await resp.text()), self.GOLDEN_OK)
|
||||
self.assertEqual(self.service.download_calls, ["wd14-vit"])
|
||||
self.assertEqual(len(self.api._jobs.list()), 1)
|
||||
|
||||
async def test_legacy_download_of_other_type_is_409_model_busy(self):
|
||||
self._hold_download()
|
||||
summary = (await (await self._start()).json())["job"]
|
||||
try:
|
||||
await self._started()
|
||||
resp = await self.client.post("/prompt_manager/autotag/download/gguf")
|
||||
data = await resp.json()
|
||||
finally:
|
||||
await self._ended(summary["id"])
|
||||
self.assertEqual(resp.status, 409)
|
||||
self.assertEqual(data["code"], "model-busy")
|
||||
self.assertEqual(data["operation"], self.TITLE)
|
||||
self.assertEqual(self.service.download_calls, ["wd14-vit"])
|
||||
|
||||
async def test_legacy_disconnect_keeps_the_job_running(self):
|
||||
self._hold_download()
|
||||
resp = await self.client.post("/prompt_manager/autotag/download/wd14-vit")
|
||||
first = await resp.content.readline()
|
||||
self.assertTrue(first.startswith(b"data: "))
|
||||
await self._started()
|
||||
resp.close() # the page closed the modal
|
||||
(job,) = self.api._jobs.list()
|
||||
for _ in range(300):
|
||||
if not job.subscriber_count:
|
||||
break
|
||||
await asyncio.sleep(0.01)
|
||||
self.assertTrue(job.running)
|
||||
job = await self._ended(job.id)
|
||||
self.assertEqual(job.state, "finished")
|
||||
self.assertIsNone(model_gate.holder())
|
||||
|
||||
+2
-1
@@ -1428,12 +1428,13 @@ No sentences, no commentary, no captions. Only tags. Keep length short but descr
|
||||
<div class="w-full bg-pm-input rounded-full h-1.5">
|
||||
<div id="downloadProgressBar" class="bg-pm-accent h-1.5 rounded-full transition-all duration-300" style="width: 0%"></div>
|
||||
</div>
|
||||
<p class="text-pm-secondary text-xs mt-2">The download continues in Activity.</p>
|
||||
</div>
|
||||
|
||||
<div class="flex justify-center">
|
||||
<button id="cancelDownloadBtn"
|
||||
class="px-3 py-1.5 text-pm-error bg-transparent hover:bg-pm-error-tint border border-pm text-xs font-medium rounded-pm-sm transition-colors">
|
||||
Cancel
|
||||
Close
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
+2
-1
@@ -730,13 +730,14 @@ No sentences, no commentary, no captions. Only tags. Keep length short but descr
|
||||
<div class="w-full bg-pm-input rounded-full h-2">
|
||||
<div id="downloadProgressBar" class="bg-pm-accent h-2 rounded-full transition-all duration-300" style="width: 0%"></div>
|
||||
</div>
|
||||
<p class="text-pm-secondary text-xs mt-2">The download continues in Activity.</p>
|
||||
</div>
|
||||
|
||||
<!-- Cancel Button -->
|
||||
<div class="flex justify-center">
|
||||
<button id="cancelDownloadBtn"
|
||||
class="px-3 py-1 bg-pm-error hover:bg-pm-error text-pm text-xs font-medium rounded-pm-sm transition-colors">
|
||||
Cancel
|
||||
Close
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
+11
-7
@@ -3689,6 +3689,7 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
document.getElementById('downloadStatus').textContent = 'Preparing...';
|
||||
document.getElementById('downloadProgressPercent').textContent = '0%';
|
||||
document.getElementById('downloadProgressBar').style.width = '0%';
|
||||
document.getElementById('downloadProgressBar').classList.remove('animate-pulse');
|
||||
|
||||
this.hideModal("autoTagModal");
|
||||
this.showModal("autoTagDownloadModal");
|
||||
@@ -3699,18 +3700,21 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
this.autoTagState.downloadEventSource.onmessage = (event) => {
|
||||
const data = JSON.parse(event.data);
|
||||
|
||||
if (data.status === 'downloading') {
|
||||
const percent = Math.round(data.progress * 100);
|
||||
document.getElementById('downloadStatus').textContent = data.message || 'Downloading...';
|
||||
document.getElementById('downloadProgressPercent').textContent = `${percent}%`;
|
||||
document.getElementById('downloadProgressBar').style.width = `${percent}%`;
|
||||
} else if (data.status === 'complete') {
|
||||
// The server sends `type` (progress / complete / error); the download is an
|
||||
// Activity job with no byte progress, so the bar is indeterminate (#328).
|
||||
if (data.type === 'progress') {
|
||||
document.getElementById('downloadStatus').textContent = 'Downloading\u2026';
|
||||
document.getElementById('downloadProgressPercent').textContent = '';
|
||||
const bar = document.getElementById('downloadProgressBar');
|
||||
bar.style.width = '100%';
|
||||
bar.classList.add('animate-pulse');
|
||||
} else if (data.type === 'complete') {
|
||||
this.autoTagState.downloadEventSource.close();
|
||||
this.hideModal("autoTagDownloadModal");
|
||||
this.showModal("autoTagModal");
|
||||
this.checkAutoTagModels();
|
||||
this.showNotification('Model downloaded successfully!', 'success');
|
||||
} else if (data.status === 'error') {
|
||||
} else if (data.type === 'error') {
|
||||
this.autoTagState.downloadEventSource.close();
|
||||
this.hideModal("autoTagDownloadModal");
|
||||
this.showModal("autoTagModal");
|
||||
|
||||
+11
-7
@@ -3421,6 +3421,7 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
document.getElementById('downloadStatus').textContent = 'Preparing...';
|
||||
document.getElementById('downloadProgressPercent').textContent = '0%';
|
||||
document.getElementById('downloadProgressBar').style.width = '0%';
|
||||
document.getElementById('downloadProgressBar').classList.remove('animate-pulse');
|
||||
|
||||
this.hideModal("autoTagModal");
|
||||
this.showModal("autoTagDownloadModal");
|
||||
@@ -3431,18 +3432,21 @@ Seed: ${this.currentMetadata.seed || 'Unknown'}`;
|
||||
this.autoTagState.downloadEventSource.onmessage = (event) => {
|
||||
const data = JSON.parse(event.data);
|
||||
|
||||
if (data.status === 'downloading') {
|
||||
const percent = Math.round(data.progress * 100);
|
||||
document.getElementById('downloadStatus').textContent = data.message || 'Downloading...';
|
||||
document.getElementById('downloadProgressPercent').textContent = `${percent}%`;
|
||||
document.getElementById('downloadProgressBar').style.width = `${percent}%`;
|
||||
} else if (data.status === 'complete') {
|
||||
// The server sends `type` (progress / complete / error); the download is an
|
||||
// Activity job with no byte progress, so the bar is indeterminate (#328).
|
||||
if (data.type === 'progress') {
|
||||
document.getElementById('downloadStatus').textContent = 'Downloading\u2026';
|
||||
document.getElementById('downloadProgressPercent').textContent = '';
|
||||
const bar = document.getElementById('downloadProgressBar');
|
||||
bar.style.width = '100%';
|
||||
bar.classList.add('animate-pulse');
|
||||
} else if (data.type === 'complete') {
|
||||
this.autoTagState.downloadEventSource.close();
|
||||
this.hideModal("autoTagDownloadModal");
|
||||
this.showModal("autoTagModal");
|
||||
this.checkAutoTagModels();
|
||||
this.showNotification('Model downloaded successfully!', 'success');
|
||||
} else if (data.status === 'error') {
|
||||
} else if (data.type === 'error') {
|
||||
this.autoTagState.downloadEventSource.close();
|
||||
this.hideModal("autoTagDownloadModal");
|
||||
this.showModal("autoTagModal");
|
||||
|
||||
Reference in New Issue
Block a user