Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c49cee9a20 | |||
| db8b43b0e8 |
@@ -0,0 +1,2 @@
|
||||
-- Estimated seconds remaining for the active download (measured throughput). Nullable.
|
||||
ALTER TABLE "Job" ADD COLUMN "downloadEtaSeconds" INTEGER;
|
||||
@@ -58,6 +58,9 @@ model Job {
|
||||
createdAt DateTime @default(now())
|
||||
updatedAt DateTime @updatedAt
|
||||
downloadProgress Float @default(0)
|
||||
// Estimated seconds until the current download finishes, from measured throughput (slskd byte
|
||||
// transfer rate). Null when unknown (no speed sample yet, or a source that doesn't report bytes).
|
||||
downloadEtaSeconds Int?
|
||||
paused Boolean @default(false)
|
||||
candidates Candidate[]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
"use client";
|
||||
|
||||
import { useEffect, useRef, useState, type ReactNode } from "react";
|
||||
|
||||
/** A vertical "⋮" trigger that opens a small popup menu. Closes on outside click, Escape, or any
|
||||
* click inside the popup (so menu items don't each need to manage it). Children are the items —
|
||||
* use <button className="menu-item">…</button>. */
|
||||
export function KebabMenu({ label = "Actions", children }: { label?: string; children: ReactNode }) {
|
||||
const [open, setOpen] = useState(false);
|
||||
const ref = useRef<HTMLDivElement>(null);
|
||||
|
||||
useEffect(() => {
|
||||
if (!open) return;
|
||||
function onDoc(e: MouseEvent) {
|
||||
if (ref.current && !ref.current.contains(e.target as Node)) setOpen(false);
|
||||
}
|
||||
function onKey(e: KeyboardEvent) {
|
||||
if (e.key === "Escape") setOpen(false);
|
||||
}
|
||||
document.addEventListener("mousedown", onDoc);
|
||||
document.addEventListener("keydown", onKey);
|
||||
return () => {
|
||||
document.removeEventListener("mousedown", onDoc);
|
||||
document.removeEventListener("keydown", onKey);
|
||||
};
|
||||
}, [open]);
|
||||
|
||||
return (
|
||||
<div className="menu" ref={ref}>
|
||||
<button
|
||||
type="button"
|
||||
className="menu-trigger"
|
||||
aria-haspopup="menu"
|
||||
aria-expanded={open}
|
||||
aria-label={label}
|
||||
onClick={() => setOpen((o) => !o)}
|
||||
>
|
||||
⋮
|
||||
</button>
|
||||
{open ? (
|
||||
<div className="menu-pop" role="menu" onClick={() => setOpen(false)}>
|
||||
{children}
|
||||
</div>
|
||||
) : null}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -1,5 +1,20 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { describeJob, timeAgo } from "./status";
|
||||
import { describeJob, etaLabel, timeAgo } from "./status";
|
||||
|
||||
describe("etaLabel", () => {
|
||||
it("formats seconds, minutes, and hours", () => {
|
||||
expect(etaLabel(45)).toBe("~45s left");
|
||||
expect(etaLabel(240)).toBe("~4m left");
|
||||
expect(etaLabel(3900)).toBe("~1h 5m left");
|
||||
});
|
||||
|
||||
it("returns null for missing or non-positive input", () => {
|
||||
expect(etaLabel(null)).toBeNull();
|
||||
expect(etaLabel(undefined)).toBeNull();
|
||||
expect(etaLabel(0)).toBeNull();
|
||||
expect(etaLabel(-5)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe("describeJob", () => {
|
||||
it("maps pipeline states to literal labels + severity kind", () => {
|
||||
|
||||
@@ -27,6 +27,16 @@ export function timeAgo(value: string | Date | null | undefined, now: number): s
|
||||
return `${Math.floor(h / 24)}d`;
|
||||
}
|
||||
|
||||
/** Compact download ETA like "~4m left", "~45s left", "~1h 5m left". Null/≤0 → no label. */
|
||||
export function etaLabel(seconds: number | null | undefined): string | null {
|
||||
if (seconds == null || seconds <= 0) return null;
|
||||
if (seconds < 60) return `~${seconds}s left`;
|
||||
const m = Math.round(seconds / 60);
|
||||
if (m < 60) return `~${m}m left`;
|
||||
const h = Math.floor(m / 60);
|
||||
return `~${h}h ${m % 60}m left`;
|
||||
}
|
||||
|
||||
export function describeJob(state: string, stage: string): { label: string; kind: Kind } {
|
||||
void stage; // stage no longer drives the label; kept for call-site compatibility
|
||||
return (
|
||||
|
||||
@@ -64,6 +64,7 @@ export async function GET() {
|
||||
claimedAt: r.job.claimedAt,
|
||||
updatedAt: r.job.updatedAt,
|
||||
downloadProgress: r.job.downloadProgress,
|
||||
downloadEtaSeconds: r.job.downloadEtaSeconds,
|
||||
paused: r.job.paused,
|
||||
candidateCount: cands.length,
|
||||
chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
|
||||
|
||||
@@ -386,6 +386,32 @@ nav.contents .sep { flex: 1; }
|
||||
.tab:hover { color: var(--ink); }
|
||||
.tab.active { color: var(--ink); border-bottom-color: var(--accent); }
|
||||
.tab .n { color: var(--rule-2); margin-left: 6px; }
|
||||
.tab.has-errors .n { color: var(--alert); }
|
||||
|
||||
/* ── Kebab menu (⋮ popup for press actions) ───────────── */
|
||||
.dept.has-actions { align-items: center; }
|
||||
.dept.has-actions .fill { transform: none; }
|
||||
.menu { position: relative; display: inline-flex; }
|
||||
.menu-trigger {
|
||||
font-size: 1.15rem; line-height: 1; color: var(--graphite); background: transparent;
|
||||
border: 1px solid var(--rule); border-radius: 6px; width: 32px; height: 30px; cursor: pointer;
|
||||
display: inline-flex; align-items: center; justify-content: center;
|
||||
}
|
||||
.menu-trigger:hover { color: var(--ink); border-color: var(--rule-2); }
|
||||
.menu-pop {
|
||||
position: absolute; top: calc(100% + 6px); right: 0; z-index: 30; min-width: 190px;
|
||||
background: var(--paper-2); border: 1px solid var(--rule-2); border-radius: 9px;
|
||||
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.22); padding: 5px;
|
||||
display: flex; flex-direction: column; gap: 1px;
|
||||
}
|
||||
.menu-item {
|
||||
font-family: var(--mono); font-size: 0.68rem; letter-spacing: 0.1em; text-transform: uppercase;
|
||||
text-align: left; background: transparent; border: 0; color: var(--ink);
|
||||
padding: 10px 11px; border-radius: 6px; cursor: pointer; white-space: nowrap;
|
||||
}
|
||||
.menu-item:hover { background: var(--rule); }
|
||||
.menu-item:disabled { color: var(--rule-2); cursor: default; background: transparent; }
|
||||
.menu-item.danger { color: var(--alert); }
|
||||
|
||||
/* ── Discography row (cover thumb + click-to-modal) ───── */
|
||||
.disco-open {
|
||||
|
||||
+121
-72
@@ -1,14 +1,34 @@
|
||||
"use client";
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
import { describeJob, timeAgo } from "./_ui/status";
|
||||
import { describeJob, etaLabel, timeAgo } from "./_ui/status";
|
||||
import { StatTiles } from "./_ui/stat-tiles";
|
||||
import { SectionHeader } from "./_ui/section-header";
|
||||
import { JobRow } from "./_ui/job-row";
|
||||
import { ProgressBar } from "./_ui/progress-bar";
|
||||
import { PressedList } from "./_ui/pressed-list";
|
||||
import { KebabMenu } from "./_ui/kebab-menu";
|
||||
import { toast } from "./_ui/toast";
|
||||
|
||||
type Tab = "active" | "queue" | "errors";
|
||||
|
||||
// Split the on-the-press rows into three honest buckets so active presses aren't buried between
|
||||
// queued items and errors. Active = being worked (searching/downloading/finishing), Queue =
|
||||
// waiting to be claimed, Errors = needs attention.
|
||||
function categoryOf(state: string, stage: string): Tab {
|
||||
const kind = describeJob(state, stage).kind;
|
||||
if (kind === "attention") return "errors";
|
||||
if (state === "requested") return "queue";
|
||||
return "active";
|
||||
}
|
||||
|
||||
const TAB_LABEL: Record<Tab, string> = { active: "Active", queue: "In queue", errors: "Errors" };
|
||||
const TAB_EMPTY: Record<Tab, string> = {
|
||||
active: "Nothing pressing right now.",
|
||||
queue: "The queue is empty.",
|
||||
errors: "No errors — everything's flowing.",
|
||||
};
|
||||
|
||||
type Row = {
|
||||
id: string;
|
||||
artist: string;
|
||||
@@ -26,6 +46,7 @@ type Row = {
|
||||
candidateCount: number;
|
||||
chosen: { source: string; format: string; trackCount: number } | null;
|
||||
downloadProgress: number;
|
||||
downloadEtaSeconds: number | null;
|
||||
paused: boolean;
|
||||
} | null;
|
||||
};
|
||||
@@ -42,6 +63,7 @@ function jobOf(r: Row) {
|
||||
candidateCount: 0,
|
||||
chosen: null,
|
||||
downloadProgress: 0,
|
||||
downloadEtaSeconds: null,
|
||||
paused: false,
|
||||
}
|
||||
);
|
||||
@@ -60,6 +82,9 @@ export function Queue() {
|
||||
const [artist, setArtist] = useState("");
|
||||
const [album, setAlbum] = useState("");
|
||||
const [paused, setPaused] = useState(false);
|
||||
// null = follow the smart default (Active if anything's active, else Queue, else Errors); once
|
||||
// the user picks a tab, their choice sticks.
|
||||
const [tab, setTab] = useState<Tab | null>(null);
|
||||
|
||||
async function refresh() {
|
||||
const [reqRes, wantRes, artRes, floorRes] = await Promise.all([
|
||||
@@ -159,10 +184,18 @@ export function Queue() {
|
||||
}
|
||||
|
||||
const active = rows.filter((r) => jobOf(r).state !== "imported");
|
||||
const downloading = rows.some((r) => {
|
||||
const s = jobOf(r).state;
|
||||
return s === "matching" || s === "matched" || s === "downloading" || s === "tagging";
|
||||
});
|
||||
const buckets: Record<Tab, Row[]> = { active: [], queue: [], errors: [] };
|
||||
for (const r of active) {
|
||||
const j = jobOf(r);
|
||||
buckets[categoryOf(j.state, j.currentStage)].push(r);
|
||||
}
|
||||
const queuedCount = buckets.queue.length;
|
||||
const downloading = buckets.active.length > 0;
|
||||
// Smart default: land on the tab that has something worth looking at.
|
||||
const defaultTab: Tab = buckets.active.length ? "active" : queuedCount ? "queue" : "errors";
|
||||
const activeTab: Tab = tab ?? defaultTab;
|
||||
const shown = buckets[activeTab];
|
||||
|
||||
// Dedupe by artist+album — re-requesting/re-downloading the same album leaves several
|
||||
// completed Requests; show each album once (rows arrive newest-first, so keep the first).
|
||||
const seenPressed = new Set<string>();
|
||||
@@ -174,61 +207,8 @@ export function Queue() {
|
||||
return true;
|
||||
});
|
||||
|
||||
return (
|
||||
<div>
|
||||
<StatTiles
|
||||
items={[
|
||||
{ k: "On the press", v: active.length },
|
||||
{ k: "Pressed", v: pressed.length, accent: true },
|
||||
{ k: "Wanted", v: wanted },
|
||||
{ k: "Watching", v: watching, unit: "artists" },
|
||||
]}
|
||||
/>
|
||||
|
||||
<SectionHeader title="On the Press" note={`${active.length} running`} />
|
||||
|
||||
<div className="press-controls">
|
||||
{paused ? (
|
||||
<button className="btn sm accent" onClick={() => floorAction("resume", "Press resumed")}>
|
||||
Resume the press
|
||||
</button>
|
||||
) : (
|
||||
<button className="btn sm ghost" onClick={() => floorAction("pause", "Press paused")}>
|
||||
Pause the press
|
||||
</button>
|
||||
)}
|
||||
<button className="btn sm ghost" onClick={clearPress}>
|
||||
Clear queue
|
||||
</button>
|
||||
{downloading ? (
|
||||
<button className="btn sm" onClick={stopPress}>
|
||||
Stop now
|
||||
</button>
|
||||
) : null}
|
||||
{paused ? <span className="chip">Press paused</span> : null}
|
||||
</div>
|
||||
|
||||
<form className="request-form" onSubmit={submit}>
|
||||
<label className="field">
|
||||
<span>Artist</span>
|
||||
<input aria-label="artist" placeholder="Artist" value={artist} onChange={(e) => setArtist(e.target.value)} />
|
||||
</label>
|
||||
<label className="field">
|
||||
<span>Album</span>
|
||||
<input aria-label="album" placeholder="Album" value={album} onChange={(e) => setAlbum(e.target.value)} />
|
||||
</label>
|
||||
<button type="submit" className="btn accent">
|
||||
Request
|
||||
</button>
|
||||
</form>
|
||||
|
||||
{active.length === 0 ? (
|
||||
<p className="empty">The press is quiet. Queue a release above, or send one over from Wanted or Discover.</p>
|
||||
) : (
|
||||
<section className="floor">
|
||||
{(() => {
|
||||
const now = Date.now();
|
||||
return active.map((r) => {
|
||||
function renderRow(r: Row) {
|
||||
const j = jobOf(r);
|
||||
const d = describeJob(j.state, j.currentStage);
|
||||
const attention = d.kind === "attention";
|
||||
@@ -285,20 +265,16 @@ export function Queue() {
|
||||
} else if (j.state === "downloading") {
|
||||
const pct = Math.round((j.downloadProgress || 0) * 100);
|
||||
if (pct <= 0) {
|
||||
// Bytes haven't started flowing yet — Qobuz login/metadata/artwork, a
|
||||
// Soulseek peer queue, or a YouTube info-fetch. Show an indeterminate bar
|
||||
// instead of a stuck-looking 0% (it flips to the % bar on the first byte).
|
||||
// Bytes haven't started flowing yet — Qobuz login/metadata/artwork, a Soulseek peer
|
||||
// queue, or a YouTube info-fetch. Show an indeterminate bar instead of a stuck-looking
|
||||
// 0% (it flips to the % bar on the first byte).
|
||||
bar = <ProgressBar kind="working" indeterminate caption="preparing…" />;
|
||||
} else {
|
||||
const total = j.chosen?.trackCount ?? 0;
|
||||
const done = total ? Math.round((pct / 100) * total) : 0;
|
||||
bar = (
|
||||
<ProgressBar
|
||||
kind="working"
|
||||
percent={pct}
|
||||
caption={total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`}
|
||||
/>
|
||||
);
|
||||
const base = total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`;
|
||||
const eta = etaLabel(j.downloadEtaSeconds);
|
||||
bar = <ProgressBar kind="working" percent={pct} caption={eta ? `${base} · ${eta}` : base} />;
|
||||
}
|
||||
} else if (j.state === "tagging") {
|
||||
bar = <ProgressBar kind="verify" indeterminate caption="finishing" />;
|
||||
@@ -318,9 +294,82 @@ export function Queue() {
|
||||
bar={bar}
|
||||
/>
|
||||
);
|
||||
});
|
||||
})()}
|
||||
</section>
|
||||
}
|
||||
|
||||
return (
|
||||
<div>
|
||||
<StatTiles
|
||||
items={[
|
||||
{ k: "On the press", v: active.length },
|
||||
{ k: "Pressed", v: pressed.length, accent: true },
|
||||
{ k: "Wanted", v: wanted },
|
||||
{ k: "Watching", v: watching, unit: "artists" },
|
||||
]}
|
||||
/>
|
||||
|
||||
<div className="dept has-actions">
|
||||
<h2>On the Press</h2>
|
||||
<span className="fill" />
|
||||
{paused ? <span className="count">paused</span> : null}
|
||||
<KebabMenu label="Press actions">
|
||||
{paused ? (
|
||||
<button className="menu-item" onClick={() => floorAction("resume", "Press resumed")}>
|
||||
Resume the press
|
||||
</button>
|
||||
) : (
|
||||
<button className="menu-item" onClick={() => floorAction("pause", "Press paused")}>
|
||||
Pause the press
|
||||
</button>
|
||||
)}
|
||||
<button className="menu-item" onClick={clearPress} disabled={queuedCount === 0}>
|
||||
Clear queue
|
||||
</button>
|
||||
{downloading ? (
|
||||
<button className="menu-item danger" onClick={stopPress}>
|
||||
Stop now
|
||||
</button>
|
||||
) : null}
|
||||
</KebabMenu>
|
||||
</div>
|
||||
|
||||
<form className="request-form" onSubmit={submit}>
|
||||
<label className="field">
|
||||
<span>Artist</span>
|
||||
<input aria-label="artist" placeholder="Artist" value={artist} onChange={(e) => setArtist(e.target.value)} />
|
||||
</label>
|
||||
<label className="field">
|
||||
<span>Album</span>
|
||||
<input aria-label="album" placeholder="Album" value={album} onChange={(e) => setAlbum(e.target.value)} />
|
||||
</label>
|
||||
<button type="submit" className="btn accent">
|
||||
Request
|
||||
</button>
|
||||
</form>
|
||||
|
||||
{active.length === 0 ? (
|
||||
<p className="empty">The press is quiet. Queue a release above, or send one over from Wanted or Discover.</p>
|
||||
) : (
|
||||
<>
|
||||
<div className="tabs" role="tablist">
|
||||
{(["active", "queue", "errors"] as Tab[]).map((t) => (
|
||||
<button
|
||||
key={t}
|
||||
role="tab"
|
||||
aria-selected={activeTab === t}
|
||||
className={`tab${activeTab === t ? " active" : ""}${t === "errors" && buckets.errors.length ? " has-errors" : ""}`}
|
||||
onClick={() => setTab(t)}
|
||||
>
|
||||
{TAB_LABEL[t]}
|
||||
<span className="n">{buckets[t].length}</span>
|
||||
</button>
|
||||
))}
|
||||
</div>
|
||||
{shown.length === 0 ? (
|
||||
<p className="empty">{TAB_EMPTY[activeTab]}</p>
|
||||
) : (
|
||||
<section className="floor">{shown.map(renderRow)}</section>
|
||||
)}
|
||||
</>
|
||||
)}
|
||||
|
||||
{pressed.length > 0 ? (
|
||||
|
||||
@@ -29,6 +29,14 @@ def _dirname(path: str) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def _eta_seconds(total_bytes: int, bytes_now: int, speed_bps: float) -> int | None:
|
||||
"""Seconds until the transfer finishes at the current measured rate, or None when it can't be
|
||||
estimated (no throughput sample yet, unknown total, or already complete). Pure — unit-tested."""
|
||||
if speed_bps <= 0 or total_bytes <= 0 or bytes_now >= total_bytes:
|
||||
return None
|
||||
return int((total_bytes - bytes_now) / speed_bps)
|
||||
|
||||
|
||||
def _norm(s: str) -> str:
|
||||
"""Lowercase, strip diacritics, and reduce to alphanumeric tokens, so 'Beyoncé' matches a
|
||||
peer's 'Beyonce' folder and punctuation ('I Am… Sasha Fierce') doesn't defeat the match."""
|
||||
@@ -206,9 +214,13 @@ class SlskdClient:
|
||||
on_progress(0.0)
|
||||
|
||||
total = len(files)
|
||||
total_bytes = sum(int(f.get("size") or 0) for f in files)
|
||||
completed = False
|
||||
last_bytes = -1
|
||||
stall = 0
|
||||
speed = 0.0 # bytes/sec, EWMA-smoothed so the ETA doesn't jitter poll to poll
|
||||
prev_bytes = 0
|
||||
prev_t = time.monotonic()
|
||||
for _ in range(_MAX_XFER_POLLS):
|
||||
time.sleep(_POLL_SECONDS)
|
||||
data = self._get(f"/api/v0/transfers/downloads/{username}")
|
||||
@@ -224,8 +236,14 @@ class SlskdClient:
|
||||
if states:
|
||||
if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")):
|
||||
raise RuntimeError("soulseek transfer failed")
|
||||
now = time.monotonic()
|
||||
dt = now - prev_t
|
||||
if dt > 0 and bytes_now >= prev_bytes:
|
||||
inst = (bytes_now - prev_bytes) / dt
|
||||
speed = inst if speed == 0.0 else 0.6 * speed + 0.4 * inst
|
||||
prev_bytes, prev_t = bytes_now, now
|
||||
done = sum(1 for s in states if "Completed" in s)
|
||||
on_progress(min(1.0, done / max(total, 1)))
|
||||
on_progress(min(1.0, done / max(total, 1)), _eta_seconds(total_bytes, bytes_now, speed))
|
||||
if done >= total:
|
||||
completed = True
|
||||
break
|
||||
|
||||
@@ -159,29 +159,35 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) ->
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _set_download_progress(conn: psycopg.Connection, job_id: str, frac: float) -> None:
|
||||
def _set_download_progress(
|
||||
conn: psycopg.Connection, job_id: str, frac: float, eta_seconds: int | None = None
|
||||
) -> None:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
|
||||
cur.execute(
|
||||
'UPDATE "Job" SET "downloadProgress" = %s, "downloadEtaSeconds" = %s WHERE id = %s',
|
||||
(frac, eta_seconds, job_id),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _make_on_progress(conn: psycopg.Connection, job_id: str, reports: "threading.Event"):
|
||||
"""Build the on_progress(pct) callback threaded into adapter.download. It marks the
|
||||
adapter as reporting real byte-level progress (so the file-count poller yields) and
|
||||
throttle-writes Job.downloadProgress. Called synchronously from adapter.download on the
|
||||
pipeline's own thread, so it safely reuses `conn`."""
|
||||
"""Build the on_progress(pct, eta_seconds=None) callback threaded into adapter.download. It
|
||||
marks the adapter as reporting real byte-level progress (so the file-count poller yields) and
|
||||
throttle-writes Job.downloadProgress (+ the download ETA when the adapter measures one — slskd
|
||||
does; others pass None). Called synchronously from adapter.download on the pipeline's own
|
||||
thread, so it safely reuses `conn`."""
|
||||
state = {"frac": -1.0, "t": 0.0}
|
||||
|
||||
def on_progress(pct: float) -> None:
|
||||
def on_progress(pct: float, eta_seconds: int | None = None) -> None:
|
||||
reports.set()
|
||||
frac = 0.0 if pct < 0 else 1.0 if pct > 1 else float(pct)
|
||||
now = time.monotonic()
|
||||
# throttle DB writes: on a >=1% move or every 0.5s, not once per received byte
|
||||
if frac - state["frac"] >= 0.01 or (now - state["t"]) >= 0.5:
|
||||
# throttle DB writes: on a >=1% move, a new ETA, or every 0.5s — not once per received byte
|
||||
if frac - state["frac"] >= 0.01 or eta_seconds is not None or (now - state["t"]) >= 0.5:
|
||||
state["frac"] = frac
|
||||
state["t"] = now
|
||||
try:
|
||||
_set_download_progress(conn, job_id, frac)
|
||||
_set_download_progress(conn, job_id, frac, eta_seconds)
|
||||
except Exception as e: # a progress write must never fail the download
|
||||
print(f"pipeline: on_progress write failed: {e}", flush=True)
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@ import json
|
||||
import pytest
|
||||
|
||||
import lyra_worker.adapters._slskd as slskd_mod
|
||||
from lyra_worker.adapters._slskd import SlskdClient, _parse_search_responses
|
||||
from lyra_worker.adapters._slskd import SlskdClient, _eta_seconds, _parse_search_responses
|
||||
|
||||
|
||||
class _Resp:
|
||||
@@ -63,7 +63,7 @@ def test_stalled_peer_is_abandoned_and_cancelled(tmp_path, monkeypatch):
|
||||
fake = slskd_mod.requests
|
||||
|
||||
with pytest.raises(TimeoutError):
|
||||
_client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
|
||||
_client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda *_a: None)
|
||||
assert any(u.endswith("/api/v0/transfers/downloads/peerA/t1") for u in fake.deleted)
|
||||
|
||||
|
||||
@@ -72,7 +72,7 @@ def test_error_state_is_cancelled(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
|
||||
fake = slskd_mod.requests
|
||||
with pytest.raises(RuntimeError):
|
||||
_client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
|
||||
_client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda *_a: None)
|
||||
assert any(u.endswith("/peerA/t1") for u in fake.deleted)
|
||||
|
||||
|
||||
@@ -92,7 +92,7 @@ def test_slow_but_progressing_peer_is_not_abandoned(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
|
||||
monkeypatch.setattr(slskd_mod, "_STALL_POLLS", 2) # would fire early if progress were ignored
|
||||
|
||||
out = _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda _p: None)
|
||||
out = _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda *_a: None)
|
||||
assert out["track_count"] == 1
|
||||
assert slskd_mod.requests.deleted == [] # completed cleanly → nothing cancelled
|
||||
|
||||
@@ -108,3 +108,39 @@ def test_parse_ranks_free_and_fast_peers_first():
|
||||
]
|
||||
peers = [json.loads(c["source_ref"])["username"] for c in _parse_search_responses(responses)]
|
||||
assert peers == ["fast", "slow", "queued"] # free+fast first; no-slot peer last despite speed
|
||||
|
||||
|
||||
def test_eta_seconds_from_measured_rate():
|
||||
assert _eta_seconds(1000, 250, 250.0) == 3 # 750 bytes left / 250 B/s
|
||||
assert _eta_seconds(1000, 0, 100.0) == 10
|
||||
|
||||
|
||||
def test_eta_seconds_unknown_cases():
|
||||
assert _eta_seconds(1000, 250, 0.0) is None # no speed sample yet
|
||||
assert _eta_seconds(0, 0, 100.0) is None # unknown total size
|
||||
assert _eta_seconds(1000, 1000, 100.0) is None # already complete
|
||||
|
||||
|
||||
def test_download_reports_shrinking_eta(tmp_path, monkeypatch):
|
||||
# A 900-byte file arriving 300 bytes per 3s poll → measured speed 100 B/s → ETA counts down
|
||||
# (6s, 3s) then None on completion. on_progress gets (fraction, eta_seconds).
|
||||
(tmp_path / "01.flac").write_bytes(b"x" * 900) # so _retrieve finds the finished file
|
||||
ref = json.dumps({"username": "peerA", "files": [{"filename": r"d\Album\01.flac", "size": 900}]})
|
||||
polls = [
|
||||
{"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 900,
|
||||
"state": "InProgress", "bytesTransferred": b, "id": "t1"}]}]}
|
||||
for b in (300, 600)
|
||||
] + [
|
||||
{"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 900,
|
||||
"state": "Completed, Succeeded", "bytesTransferred": 900, "id": "t1"}]}]}
|
||||
]
|
||||
monkeypatch.setattr(slskd_mod, "requests", _SeqRequests(polls))
|
||||
monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
|
||||
clock = iter(float(i) * 3 for i in range(100)) # 0,3,6,… deterministic monotonic time
|
||||
monkeypatch.setattr(slskd_mod.time, "monotonic", lambda: next(clock))
|
||||
|
||||
calls: list = []
|
||||
_client(tmp_path).download(ref, str(tmp_path / "dest"), lambda *a: calls.append(a))
|
||||
|
||||
etas = [a[1] for a in calls if len(a) > 1]
|
||||
assert etas == [6, 3, None] # counts down at 100 B/s, then None once complete
|
||||
|
||||
Reference in New Issue
Block a user