diff --git a/src/audit/crawler/js_fetcher.py b/src/audit/crawler/js_fetcher.py index 7252e59..0735b80 100644 --- a/src/audit/crawler/js_fetcher.py +++ b/src/audit/crawler/js_fetcher.py @@ -15,6 +15,7 @@ from __future__ import annotations +import asyncio import contextlib import io from types import TracebackType @@ -125,6 +126,7 @@ def __init__( visual_probe: VisualProbe | None = None, capture_screenshots: bool = False, shared_context: BrowserContext | None = None, + shared_pages: tuple[Page, ...] = (), private_context: bool = False, max_rendered_html_chars: int | None = None, headless: bool = True, @@ -160,6 +162,14 @@ def __init__( # context per page. Public scans keep the original isolated-context # behavior. self._shared_context = shared_context + if shared_pages and shared_context is None: + raise ValueError("shared_pages requires shared_context") + self._shared_pages = shared_pages + self._shared_page_pool: asyncio.Queue[Page] | None = None + if shared_pages: + self._shared_page_pool = asyncio.Queue(maxsize=len(shared_pages)) + for shared_page in shared_pages: + self._shared_page_pool.put_nowait(shared_page) # A protected companion must never put target paths, redirect URLs, or # browser exception text (which can contain query/session material) # into the application log. Public crawling keeps the useful existing @@ -217,8 +227,13 @@ async def fetch(self, url: str) -> FetchResult: ) owns_context = True page: Page | None = None + pooled_page = False try: - page = await ctx.new_page() + if self._shared_page_pool is not None: + page = await self._shared_page_pool.get() + pooled_page = True + else: + page = await ctx.new_page() try: resp = await page.goto(url, timeout=self._nav_timeout_ms, wait_until="load") except Exception as exc: @@ -357,7 +372,9 @@ async def fetch(self, url: str) -> FetchResult: # wipes it only when the scan reaches a terminal state. Close the # per-page tab, never the session itself. Public scans retain the # old "new context per fetch" isolation guarantee. - if owns_context: + if pooled_page and page is not None and self._shared_page_pool is not None: + self._shared_page_pool.put_nowait(page) + elif owns_context: await ctx.close() elif page is not None: with contextlib.suppress(Exception): diff --git a/src/audit/crawler/orchestrator.py b/src/audit/crawler/orchestrator.py index e6b9c05..a676111 100644 --- a/src/audit/crawler/orchestrator.py +++ b/src/audit/crawler/orchestrator.py @@ -495,6 +495,7 @@ def config_json_for_scan(config: CrawlConfig) -> str: "max_depth": config.max_depth, "allow_subdomains": config.allow_subdomains, "rps": config.rps, + "workers": config.workers, "ignore_robots": config.ignore_robots, "concurrency_per_host": config.concurrency_per_host, "user_agent": config.user_agent, @@ -784,9 +785,12 @@ async def _worker(ctx: _WorkerContext) -> None: while True: if _page_limit_reached(ctx): return - job = queue.lease(ctx.conn, JOB_KIND, lease_secs=120) + job = queue.lease(ctx.conn, JOB_KIND, lease_secs=120, scan_id=ctx.scan_id) if job is None: - if ctx.in_flight == 0 and queue.pending_count(ctx.conn, JOB_KIND) == 0: + if ( + ctx.in_flight == 0 + and queue.pending_count(ctx.conn, JOB_KIND, scan_id=ctx.scan_id) == 0 + ): return await asyncio.sleep(0.05) continue diff --git a/src/audit/db/queue.py b/src/audit/db/queue.py index 2526c83..dafa1bb 100644 --- a/src/audit/db/queue.py +++ b/src/audit/db/queue.py @@ -64,6 +64,7 @@ def lease( kind: str, lease_secs: float, *, + scan_id: int | None = None, now: datetime | None = None, ) -> Job | None: """Atomically claim one pending job of ``kind``. Returns ``None`` when queue is empty. @@ -73,23 +74,43 @@ def lease( """ current = _now(now) lease_until = current + timedelta(seconds=lease_secs) - cur = conn.execute( - """ - UPDATE jobs - SET state = 'leased', - lease_until = ?, - attempts = attempts + 1, - updated_at = ? - WHERE id = ( - SELECT id FROM jobs - WHERE state = 'pending' AND kind = ? - ORDER BY id - LIMIT 1 - ) - RETURNING id, kind, payload_json, attempts - """, - (_iso(lease_until), _iso(current), kind), - ) + if scan_id is None: + cur = conn.execute( + """ + UPDATE jobs + SET state = 'leased', + lease_until = ?, + attempts = attempts + 1, + updated_at = ? + WHERE id = ( + SELECT id FROM jobs + WHERE state = 'pending' AND kind = ? + ORDER BY id + LIMIT 1 + ) + RETURNING id, kind, payload_json, attempts + """, + (_iso(lease_until), _iso(current), kind), + ) + else: + cur = conn.execute( + """ + UPDATE jobs + SET state = 'leased', + lease_until = ?, + attempts = attempts + 1, + updated_at = ? + WHERE id = ( + SELECT id FROM jobs + WHERE state = 'pending' AND kind = ? + AND json_extract(payload_json, '$.scan_id') = ? + ORDER BY id + LIMIT 1 + ) + RETURNING id, kind, payload_json, attempts + """, + (_iso(lease_until), _iso(current), kind, scan_id), + ) row = cur.fetchone() if row is None: return None @@ -153,14 +174,31 @@ def reclaim_expired(conn: sqlite3.Connection, *, now: datetime | None = None) -> return int(cur.rowcount) -def pending_count(conn: sqlite3.Connection, kind: str | None = None) -> int: +def pending_count( + conn: sqlite3.Connection, + kind: str | None = None, + *, + scan_id: int | None = None, +) -> int: """Count jobs currently in ``pending`` state.""" - if kind is None: + if kind is None and scan_id is None: cur = conn.execute("SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending'") - else: + elif kind is not None and scan_id is None: cur = conn.execute( "SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' AND kind = ?", (kind,), ) + elif kind is None: + cur = conn.execute( + "SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' " + "AND json_extract(payload_json, '$.scan_id') = ?", + (scan_id,), + ) + else: + cur = conn.execute( + "SELECT COUNT(*) AS n FROM jobs WHERE state = 'pending' AND kind = ? " + "AND json_extract(payload_json, '$.scan_id') = ?", + (kind, scan_id), + ) row = cur.fetchone() return int(row["n"]) if row is not None else 0 diff --git a/src/audit/protected/session.py b/src/audit/protected/session.py index 09621fe..4ad2fac 100644 --- a/src/audit/protected/session.py +++ b/src/audit/protected/session.py @@ -429,7 +429,34 @@ def verify_authenticated_target(self, url: str | None = None) -> ValidatedUrl: self._state = ManualAuthState.AUTHENTICATED return verified - async def minimize_for_background_scan(self) -> bool: + async def prepare_background_scan_pages(self, count: int) -> tuple[Page, ...]: + """Create the fixed authenticated tab pool before hiding Chromium. + + macOS restores a minimized Chromium window whenever Playwright creates + a new top-level page. Preparing every worker tab first lets the crawl + reuse those pages without repeatedly bringing the browser to the + foreground. The pages and their shared authenticated context remain + memory-only and are destroyed together at session close. + """ + + if self._state is not ManualAuthState.AUTHENTICATED: + raise ManualAuthenticationError( + "Complete and verify manual sign-in before preparing scan tabs." + ) + if count <= 0: + raise ValueError("Background scan page count must be positive.") + pages: list[Page] = [] + try: + for _ in range(count): + pages.append(await self.context.new_page()) + except Exception: + for page in pages: + with contextlib.suppress(Exception): + await page.close(run_before_unload=False) + raise + return tuple(pages) + + async def minimize_for_background_scan(self, page: Page | None = None) -> bool: """Minimize the headed sign-in window before reusing its context. Chromium cannot switch a live authenticated context from headed to @@ -444,12 +471,12 @@ async def minimize_for_background_scan(self) -> bool: raise ManualAuthenticationError( "Complete and verify manual sign-in before backgrounding the browser." ) - page = self._page - if page is None: + target_page = page or self._page + if target_page is None: return False cdp_session: Any | None = None try: - cdp_session = await self.context.new_cdp_session(page) + cdp_session = await self.context.new_cdp_session(target_page) window = await cdp_session.send("Browser.getWindowForTarget") window_id = window.get("windowId") if not isinstance(window_id, int): @@ -552,6 +579,7 @@ def create_shared_js_fetcher( visual_probe: VisualProbe | None = None, capture_screenshots: bool = False, max_rendered_html_chars: int | None = None, + shared_pages: tuple[Page, ...] = (), ) -> JsFetcher: """Return a fetcher that reuses this in-memory authenticated context. @@ -573,6 +601,7 @@ def create_shared_js_fetcher( visual_probe=visual_probe, capture_screenshots=capture_screenshots, shared_context=self.context, + shared_pages=shared_pages, private_context=True, max_rendered_html_chars=max_rendered_html_chars, ) diff --git a/src/audit/web/frontend/src/api/types.ts b/src/audit/web/frontend/src/api/types.ts index 85df87f..ee4c7bd 100644 --- a/src/audit/web/frontend/src/api/types.ts +++ b/src/audit/web/frontend/src/api/types.ts @@ -312,6 +312,8 @@ export interface LocalLoginScanPayload { max_pages: number; max_depth: number; rps: number; + /** Concurrent authenticated tabs; the local login API caps this at four. */ + workers: number; whole_host: boolean; /** DOM rule engines run against the signed-in application scope. */ scan_engine: ProtectedScanEngine; diff --git a/src/audit/web/frontend/src/components/LocalLoginScan.tsx b/src/audit/web/frontend/src/components/LocalLoginScan.tsx index b846f66..76d113c 100644 --- a/src/audit/web/frontend/src/components/LocalLoginScan.tsx +++ b/src/audit/web/frontend/src/components/LocalLoginScan.tsx @@ -54,6 +54,7 @@ function LocalLoginForm({ showSteps }: { showSteps: boolean }) { max_pages: 2500, max_depth: 10, rps: 1, + workers: 2, whole_host: false, scan_engine: "axe", axe_level: "AA", @@ -278,6 +279,7 @@ function LocalLoginForm({ showSteps }: { showSteps: boolean }) { ? "Local OCR image-text analysis" : "Local OCR and loopback VLM analysis", `Up to ${form.max_pages.toLocaleString()} pages`, + `${form.workers} concurrent authenticated ${form.workers === 1 ? "tab" : "tabs"}`, ].map((label) => (
  • undefined} + max={4} + onChange={(value) => update("workers", value)} /> +

    + Workers open concurrent tabs inside the same temporary, + authenticated browser session. Two is recommended; four is + the safety maximum for login and 2FA scans. +

    diff --git a/src/audit/web/server.py b/src/audit/web/server.py index 293d184..6abd0a0 100644 --- a/src/audit/web/server.py +++ b/src/audit/web/server.py @@ -122,6 +122,7 @@ class LocalLoginScanRequest(BaseModel): max_pages: int = Field(default=2500, ge=1, le=2500) max_depth: int = Field(default=10, ge=1, le=20) rps: float = Field(default=1.0, ge=0.1, le=5.0) + workers: int = Field(default=2, ge=1, le=4) whole_host: bool = False scan_engine: Literal["axe", "alfa", "both"] = "axe" axe_level: Literal["A", "AA", "AAA"] = "AA" @@ -938,8 +939,12 @@ async def api_create_local_login_scan( max_depth=body.max_depth, rps=body.rps, whole_host=body.whole_host, - concurrency_per_host=1, - workers=1, + # Login scans reuse one authenticated BrowserContext, but each + # crawler worker opens and closes its own tab. Keep the cap small + # so concurrent pages cannot overwhelm the shared session or the + # target application while still using modern laptop capacity. + concurrency_per_host=body.workers, + workers=body.workers, user_agent=settings.user_agent, request_timeout_s=settings.request_timeout_s, # Optional protected-image analysis uses the same authenticated @@ -969,7 +974,10 @@ async def api_create_local_login_scan( capture_screenshots=False, ignore_robots=True, ) - scan_id = _prepare_scan_row(resolved_db, config) + # A manual-login session is deliberately memory-only and cannot be + # resumed after interruption. Always allocate a fresh report and job + # frontier even when the auditor scans the same URL again. + scan_id = _prepare_scan_row(resolved_db, config, resume_interrupted=False) session = ManualAuthenticationSession( seed_url=body.seed_url, approved_target_origins=(body.target_origin,), @@ -1153,6 +1161,7 @@ async def _runner() -> None: async def api_cancel_scan(scan_id: int) -> JSONResponse: task = crawl_state.get("task") active_scan_id = crawl_state.get("scan_id") + is_local_login = scan_id in local_login_runs if isinstance(task, asyncio.Task) and not task.done() and active_scan_id == scan_id: task.cancel() with get_conn() as conn: @@ -1171,11 +1180,20 @@ async def api_cancel_scan(scan_id: int) -> JSONResponse: "finished_at = CURRENT_TIMESTAMP WHERE id = ?", (scan_id,), ) - conn.execute( - "DELETE FROM jobs WHERE json_extract(payload_json, '$.scan_id') = ? " - "AND state = 'pending'", - (scan_id,), - ) + if is_local_login: + conn.execute( + "DELETE FROM jobs " + "WHERE json_extract(payload_json, '$.scan_id') = ? " + "AND state IN ('pending', 'leased')", + (scan_id,), + ) + else: + conn.execute( + "DELETE FROM jobs " + "WHERE json_extract(payload_json, '$.scan_id') = ? " + "AND state = 'pending'", + (scan_id,), + ) return JSONResponse({"ok": True}) @app.delete("/api/scans/{scan_id:int}") @@ -2431,7 +2449,12 @@ def _build_crawl_config(form: dict[str, Any], settings: Settings) -> CrawlConfig ) -def _prepare_scan_row(db_path: Path, config: CrawlConfig) -> int: +def _prepare_scan_row( + db_path: Path, + config: CrawlConfig, + *, + resume_interrupted: bool = True, +) -> int: """Create the scan row up front so the UI can redirect immediately. ``run_crawl`` canonicalizes directory-like seeds before discovering its @@ -2445,7 +2468,9 @@ def _prepare_scan_row(db_path: Path, config: CrawlConfig) -> int: seed_url = _canonical_scan_seed(config.seed_url) conn = connect(db_path) try: - if _protected_scan_table_exists(conn): + if not resume_interrupted: + existing = None + elif _protected_scan_table_exists(conn): existing = conn.execute( "SELECT id FROM scans " "WHERE seed_url = ? AND status IN ('running', 'interrupted') " @@ -2531,8 +2556,12 @@ async def _run_local_login_background( await run.confirmation.wait() run.status = "verifying_authentication" run.session.verify_authenticated_target() - run.browser_backgrounded = await run.session.minimize_for_background_scan() + # Chromium on macOS restores a minimized window whenever a new page is + # created. Allocate one reusable tab per worker before minimizing so + # the authenticated crawl stays out of the auditor's way throughout. + scan_pages = await run.session.prepare_background_scan_pages(config.workers) await run.session.discard_manual_auth_page() + run.browser_backgrounded = await run.session.minimize_for_background_scan(scan_pages[0]) # The orchestrator normally constructs these around a fresh browser. # For an authenticated scan they must be attached before we inject the @@ -2553,6 +2582,7 @@ async def _run_local_login_background( ), focus_probe=FocusProbe(suppress_diagnostics=True), capture_screenshots=False, + shared_pages=scan_pages, ) run.status = "scanning" authenticated_alfa: _AuthenticatedAlfaRunner | None = None diff --git a/tests/ui/test_accessibility_axe.py b/tests/ui/test_accessibility_axe.py index a85fd76..e6ce954 100644 --- a/tests/ui/test_accessibility_axe.py +++ b/tests/ui/test_accessibility_axe.py @@ -630,6 +630,13 @@ async def test_login_scan_is_visible_and_explains_login_before_crawl( page.get_by_role("textbox", name="Page to scan after login") ).to_be_visible() await page.get_by_text("Advanced settings", exact=True).click() + workers = page.get_by_role("spinbutton", name="Workers") + await playwright_async.expect(workers).to_be_enabled() + await playwright_async.expect(workers).to_have_value("2") + await workers.fill("4") + await playwright_async.expect( + page.get_by_text("4 concurrent authenticated tabs", exact=True) + ).to_be_visible() both_engines = page.get_by_role( "radio", name=re.compile(r"^axe-core \+ Siteimprove Alfa") ) diff --git a/tests/ui/test_routes.py b/tests/ui/test_routes.py index 76a5aa0..9915ca9 100644 --- a/tests/ui/test_routes.py +++ b/tests/ui/test_routes.py @@ -351,6 +351,13 @@ def test_local_login_scan_starts_from_same_loopback_origin( captured: dict[str, object] = {} + with connect(db_path) as conn: + interrupted = conn.execute( + "INSERT INTO scans (seed_url, status, config_json) VALUES (?, 'interrupted', '{}')", + ("https://app.example.test/secure/",), + ) + interrupted_scan_id = int(interrupted.lastrowid or 0) + async def _no_browser_run( _db_path: object, _blob_dir: object, config: object, _run: object ) -> None: @@ -384,6 +391,7 @@ def __init__(self, **_kwargs: object) -> None: "max_pages": 10, "max_depth": 4, "rps": 0.5, + "workers": 4, "whole_host": True, "scan_engine": "both", "axe_level": "AAA", @@ -399,6 +407,7 @@ def __init__(self, **_kwargs: object) -> None: assert response.status_code == 201 assert response.json()["status"] == "opening_browser" assert isinstance(response.json()["scan_id"], int) + assert response.json()["scan_id"] != interrupted_scan_id config = captured["config"] assert isinstance(config, server.CrawlConfig) assert config.max_pages == 10 @@ -409,12 +418,27 @@ def __init__(self, **_kwargs: object) -> None: assert config.axe_enabled is True assert config.keyboard_probe_enabled is False assert config.responsive_checks_enabled is False - assert config.workers == 1 + assert config.workers == 4 + assert config.concurrency_per_host == 4 assert config.alfa_enabled is True assert config.browser_only is True assert config.image_extraction_enabled is False +def test_local_login_scan_rejects_more_than_four_workers(client: TestClient) -> None: + response = client.post( + "/api/local-login-scans", + json={ + "seed_url": "https://app.example.test/secure/", + "authorization_acknowledged": True, + "workers": 5, + }, + ) + + # Request validation runs before the loopback-only convenience-flow guard. + assert response.status_code == 422 + + def test_local_login_scan_rejects_selected_unavailable_alfa( seeded_db: tuple[Path, Path, int], monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/test_js_fetcher_pool.py b/tests/unit/test_js_fetcher_pool.py new file mode 100644 index 0000000..dd224ca --- /dev/null +++ b/tests/unit/test_js_fetcher_pool.py @@ -0,0 +1,76 @@ +"""Authenticated JsFetcher tab-pool behavior.""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field + +import pytest + +from audit.crawler.js_fetcher import JsFetcher + + +@dataclass +class _Response: + url: str + status: int = 200 + headers: dict[str, str] = field(default_factory=lambda: {"content-type": "text/html"}) + + +@dataclass +class _Page: + goto_calls: list[str] = field(default_factory=list) + closed: bool = False + + async def goto(self, url: str, *, timeout: int, wait_until: str) -> _Response: + assert timeout == 30_000 + assert wait_until == "load" + self.goto_calls.append(url) + await asyncio.sleep(0) + return _Response(url) + + async def wait_for_load_state(self, state: str, *, timeout: int) -> None: + assert state == "networkidle" + assert timeout == 10_000 + + async def content(self) -> str: + return "pooled" + + async def close(self) -> None: + self.closed = True + + +@dataclass +class _Context: + new_page_calls: int = 0 + + async def new_page(self) -> _Page: + self.new_page_calls += 1 + return _Page() + + +@pytest.mark.asyncio +async def test_shared_page_pool_reuses_prepared_tabs_without_opening_new_windows() -> None: + context = _Context() + pages = (_Page(), _Page()) + fetcher = JsFetcher( + user_agent="test", + shared_context=context, # type: ignore[arg-type] + shared_pages=pages, # type: ignore[arg-type] + private_context=True, + ) + + first, second = await asyncio.gather( + fetcher.fetch("https://app.example.test/one"), + fetcher.fetch("https://app.example.test/two"), + ) + third = await fetcher.fetch("https://app.example.test/three") + + assert {first.url, second.url, third.url} == { + "https://app.example.test/one", + "https://app.example.test/two", + "https://app.example.test/three", + } + assert context.new_page_calls == 0 + assert sum(len(page.goto_calls) for page in pages) == 3 + assert all(not page.closed for page in pages) diff --git a/tests/unit/test_protected_session.py b/tests/unit/test_protected_session.py index f5aa4c9..645afc3 100644 --- a/tests/unit/test_protected_session.py +++ b/tests/unit/test_protected_session.py @@ -239,6 +239,8 @@ class _FakeContext: web_socket_route_calls: list[tuple[str, object]] = field(default_factory=list) event_handlers: dict[str, object] = field(default_factory=dict) init_scripts: list[str] = field(default_factory=list) + additional_pages: list[_FakePage] = field(default_factory=list) + new_page_calls: int = 0 closed: bool = False async def route(self, pattern: str, handler: object) -> None: @@ -254,10 +256,15 @@ def on(self, event: str, handler: object) -> None: self.event_handlers[event] = handler async def new_page(self) -> _FakePage: - return self.page + self.new_page_calls += 1 + if self.new_page_calls == 1: + return self.page + page = _FakePage() + self.additional_pages.append(page) + return page async def new_cdp_session(self, page: _FakePage) -> _FakeCdpSession: - assert page is self.page + assert page is self.page or page in self.additional_pages return self.cdp_session async def close(self) -> None: @@ -376,7 +383,18 @@ async def test_manual_session_is_headed_ephemeral_and_tightens_after_target_veri assert verified.url == "https://app.example.edu/dashboard" assert session.state is ManualAuthState.AUTHENTICATED - assert await session.minimize_for_background_scan() + scan_pages = await session.prepare_background_scan_pages(2) + assert scan_pages == tuple(context.additional_pages) + assert len(set(map(id, scan_pages))) == 2 + + # Prepare the fixed worker tabs before minimizing. Creating a new tab + # after this point restores a minimized Chromium window on macOS. + await session.discard_manual_auth_page() + assert page.closed + with pytest.raises(ManualAuthenticationError): + _ = session.page + + assert await session.minimize_for_background_scan(scan_pages[0]) assert context.cdp_session.calls == [ ("Browser.getWindowForTarget", None), ( @@ -401,13 +419,6 @@ async def test_manual_session_is_headed_ephemeral_and_tightens_after_target_veri ] assert context.cdp_session.detached - # The authenticated context remains available for new crawl tabs, but - # the original headed sign-in tab is discarded as soon as it is safe. - await session.discard_manual_auth_page() - assert page.closed - with pytest.raises(ManualAuthenticationError): - _ = session.page - after_auth_idp = _FakeRoute(_FakeRequest("GET", "https://login.example.edu/authorize")) unsafe_scan_post = _FakeRoute(_FakeRequest("POST", "https://app.example.edu/form", "fetch")) unsafe_worker = _FakeRoute(_FakeRequest("GET", "https://app.example.edu/worker.js", "worker")) @@ -431,8 +442,9 @@ async def test_manual_session_is_headed_ephemeral_and_tightens_after_target_veri assert chunked_image.actions == [("abort", "blockedbyclient")] assert session.consume_scan_egress_block() == "egress_policy_blocked" - fetcher = session.create_shared_js_fetcher() + fetcher = session.create_shared_js_fetcher(shared_pages=scan_pages) assert fetcher._shared_context is context # Shared context, never exported state. + assert fetcher._shared_pages == scan_pages await session.close() assert session.state is ManualAuthState.CLOSED diff --git a/tests/unit/test_queue.py b/tests/unit/test_queue.py index f1cceae..765f4b9 100644 --- a/tests/unit/test_queue.py +++ b/tests/unit/test_queue.py @@ -57,6 +57,21 @@ def test_lease_filters_by_kind(tmp_db: sqlite3.Connection) -> None: assert queue.pending_count(tmp_db, "fetch") == 1 +def test_lease_and_pending_count_can_be_scoped_to_scan(tmp_db: sqlite3.Connection) -> None: + queue.enqueue(tmp_db, "fetch", {"url": "old", "scan_id": 12}) + queue.enqueue(tmp_db, "fetch", {"url": "current", "scan_id": 13}) + + assert queue.pending_count(tmp_db, "fetch", scan_id=12) == 1 + assert queue.pending_count(tmp_db, "fetch", scan_id=13) == 1 + + job = queue.lease(tmp_db, "fetch", lease_secs=30, scan_id=13) + + assert job is not None + assert job.payload == {"url": "current", "scan_id": 13} + assert queue.pending_count(tmp_db, "fetch", scan_id=13) == 0 + assert queue.pending_count(tmp_db, "fetch", scan_id=12) == 1 + + def test_complete_marks_done(tmp_db: sqlite3.Connection) -> None: queue.enqueue(tmp_db, "fetch", {"u": "x"}) job = queue.lease(tmp_db, "fetch", lease_secs=30)