+
+
+
+ {results.download && }
+ }
+
+ {phase === 'idle' &&
+ Runs {PING_SAMPLES} pings, then a {PHASE_SECONDS} second download and a {PHASE_SECONDS} second
+ upload between this device and the WROLPi. Nothing is saved.
+ }
+
+}
+
+function PingVerdict({median}) {
+ const {kind, text} = describePing(median);
+ const color = {success: 'green', warning: 'yellow', danger: 'red'}[kind] || 'grey';
+ return ;
+}
+
+export function ConnectionContext({info, hotspotConnected, others}) {
+ if (!info) {
+ return null;
+ }
+ const viaHotspot = hotspotConnected && isHotspotAddress(info.client_ip);
+ return
+ Tested from {info.client_ip}{' '}
+ {viaHotspot
+ ?
+ : }
+ {others > 0 && {others === 1 ? '1 other speed test was' : `${others} other speed tests were`} running
+ at the same time, so the link was shared.}
+ ;
+}
+
+export function StreamCapacity({mbps}) {
+ const rows = streamCapacity(mbps);
+ return
+ Simultaneous video streams
+
+
+
+ Quality
+ About
+ Streams at once
+
+
+
+ {rows.map(row =>
+ {row.label}
+ {row.mbps} Mbps
+ {row.streams}
+ )}
+
+
+
+ Measures the network between this device and the WROLPi. Videos are served from the drive, which
+ can also limit how many play at once. Bitrates are typical for each resolution; individual videos
+ vary.
+
+
;
+}
diff --git a/app/src/components/calculators/SpeedTestCalculator.test.js b/app/src/components/calculators/SpeedTestCalculator.test.js
new file mode 100644
index 00000000..a7f1a075
--- /dev/null
+++ b/app/src/components/calculators/SpeedTestCalculator.test.js
@@ -0,0 +1,158 @@
+import React from "react";
+import {act, fireEvent, screen, waitFor} from "@testing-library/react";
+import {renderWithProviders} from "../../test-utils";
+import {ConnectionContext, SpeedTestCalculator, StreamCapacity} from "./SpeedTestCalculator";
+import * as speedtest from "./speedtest";
+
+jest.mock('./speedtest', () => {
+ const actual = jest.requireActual('./speedtest');
+ return {
+ ...actual,
+ fetchInfo: jest.fn(),
+ runPing: jest.fn(),
+ runDownload: jest.fn(),
+ runUpload: jest.fn(),
+ };
+});
+
+const deferred = () => {
+ let resolve;
+ const promise = new Promise(r => {
+ resolve = r;
+ });
+ return {promise, resolve};
+};
+
+beforeEach(() => {
+ jest.clearAllMocks();
+ speedtest.fetchInfo.mockResolvedValue({client_ip: '10.0.0.50', active_tests: 0});
+ speedtest.runPing.mockResolvedValue({min: 4, median: 6, jitter: 1.5, lost: 0});
+ speedtest.runDownload.mockResolvedValue({mbps: 87.5, peak: 95, bytes: 1e8, samples: []});
+ speedtest.runUpload.mockResolvedValue({mbps: 21.2, peak: 25, bytes: 3e7, samples: []});
+});
+
+describe('SpeedTestCalculator', () => {
+ test('idle state shows Start and no results', () => {
+ renderWithProviders();
+ expect(screen.getByRole('button', {name: /start test/i})).toBeInTheDocument();
+ expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Ready');
+ expect(screen.queryByTestId('speedtest-results')).not.toBeInTheDocument();
+ });
+
+ test('runs ping, download, upload in order and shows results and stream capacity', async () => {
+ renderWithProviders();
+ fireEvent.click(screen.getByRole('button', {name: /start test/i}));
+
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Complete'));
+
+ expect(speedtest.runPing).toHaveBeenCalledTimes(1);
+ expect(speedtest.runDownload).toHaveBeenCalledTimes(1);
+ expect(speedtest.runUpload).toHaveBeenCalledTimes(1);
+ // Order: ping before download before upload.
+ expect(speedtest.runPing.mock.invocationCallOrder[0])
+ .toBeLessThan(speedtest.runDownload.mock.invocationCallOrder[0]);
+ expect(speedtest.runDownload.mock.invocationCallOrder[0])
+ .toBeLessThan(speedtest.runUpload.mock.invocationCallOrder[0]);
+
+ const results = screen.getByTestId('speedtest-results');
+ expect(results).toHaveTextContent('87.5 Mbps');
+ expect(results).toHaveTextContent('21.2 Mbps');
+ expect(results).toHaveTextContent('6.0 ms');
+
+ // 87.5 Mbps: 35 SD, 17 HD, 10 Full HD, 5 2K, 3 4K streams.
+ const capacity = screen.getByTestId('speedtest-capacity');
+ const cells = Array.from(capacity.querySelectorAll('tbody td:last-child')).map(td => td.textContent);
+ expect(cells).toEqual(['35', '17', '10', '5', '3']);
+
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('10.0.0.50');
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('via LAN');
+ expect(screen.getByRole('button', {name: /run again/i})).toBeInTheDocument();
+ });
+
+ test('Start becomes Cancel before the first await resolves, so a second run cannot be started', async () => {
+ // Hold the first await so Start is clicked again before the ping phase begins.
+ const pending = deferred();
+ speedtest.fetchInfo.mockImplementation(() => pending.promise);
+ renderWithProviders();
+ fireEvent.click(screen.getByRole('button', {name: /start test/i}));
+ // The button has already become Cancel, and a stray click on it is the abort path,
+ // so hunt for any Start/Run again button and click it if one exists.
+ expect(screen.queryByRole('button', {name: /start test|run again/i})).toBeNull();
+ expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Connecting');
+
+ pending.resolve({client_ip: '10.0.0.50', active_tests: 0});
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Complete'));
+ expect(speedtest.fetchInfo).toHaveBeenCalledTimes(1);
+ expect(speedtest.runPing).toHaveBeenCalledTimes(1);
+ });
+
+ test('Cancel aborts the run and shows no results for unfinished phases', async () => {
+ const pending = deferred();
+ speedtest.runDownload.mockImplementation(({signal}) => new Promise((resolve, reject) => {
+ signal.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')));
+ pending.promise.then(resolve);
+ }));
+ renderWithProviders();
+ fireEvent.click(screen.getByRole('button', {name: /start test/i}));
+
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Downloading'));
+ const {signal} = speedtest.runDownload.mock.calls[0][0];
+ fireEvent.click(screen.getByRole('button', {name: /cancel/i}));
+
+ expect(signal.aborted).toBe(true);
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Cancelled'));
+ expect(speedtest.runUpload).not.toHaveBeenCalled();
+ // Ping finished before the cancel, so it is shown; download is not.
+ expect(screen.getByTestId('speedtest-results')).toHaveTextContent('Ping');
+ expect(screen.getByTestId('speedtest-results')).not.toHaveTextContent('Download');
+ });
+
+ test('an unreachable API is reported', async () => {
+ speedtest.fetchInfo.mockRejectedValue(new Error('HTTP 502'));
+ renderWithProviders();
+ fireEvent.click(screen.getByRole('button', {name: /start test/i}));
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Failed'));
+ expect(screen.getByText(/could not be reached/i)).toBeInTheDocument();
+ expect(speedtest.runPing).not.toHaveBeenCalled();
+ });
+
+ test('a failing download phase is reported', async () => {
+ speedtest.runDownload.mockRejectedValue(new Error('Download test failed: HTTP 500'));
+ renderWithProviders();
+ fireEvent.click(screen.getByRole('button', {name: /start test/i}));
+ await waitFor(() => expect(screen.getByTestId('speedtest-phase')).toHaveTextContent('Failed'));
+ expect(screen.getByText(/HTTP 500/)).toBeInTheDocument();
+ });
+});
+
+describe('ConnectionContext', () => {
+ test('hotspot badge needs both a hotspot address and the hotspot to be up', () => {
+ const {rerender} = renderWithProviders(
+ );
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('via WROLPi hotspot');
+
+ rerender();
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('via LAN');
+
+ rerender();
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('via LAN');
+ });
+
+ test('mentions concurrent tests', () => {
+ renderWithProviders();
+ expect(screen.getByTestId('speedtest-context')).toHaveTextContent('2 other speed tests were running');
+ });
+
+ test('renders nothing without info', () => {
+ const {container} = renderWithProviders();
+ expect(container.querySelector('[data-testid="speedtest-context"]')).toBeNull();
+ });
+});
+
+describe('StreamCapacity', () => {
+ test('shows every tier with the disk caveat', () => {
+ renderWithProviders();
+ expect(screen.getByText('1080p (Full HD)')).toBeInTheDocument();
+ expect(screen.getByText(/served from the drive/i)).toBeInTheDocument();
+ });
+});
diff --git a/app/src/components/calculators/speedtest.js b/app/src/components/calculators/speedtest.js
new file mode 100644
index 00000000..a6d2f92a
--- /dev/null
+++ b/app/src/components/calculators/speedtest.js
@@ -0,0 +1,339 @@
+/**
+ * Speed test logic with no React in it, so Jest can drive it with fake fetch/streams.
+ *
+ * Units: Mbps means decimal megabits per second (1 Mbps = 1,000,000 bits/s), the unit video
+ * bitrates are quoted in. MB/s is decimal megabytes.
+ */
+import {API_URI} from "../Vars";
+
+export const SPEEDTEST_API = `${API_URI}/speedtest`;
+
+// The browser runs each throughput phase this long. The server clamps download at 15 s.
+export const PHASE_SECONDS = 10;
+// Throughput is tallied into buckets this wide; the live number and sparkline are per bucket.
+export const BUCKET_MS = 250;
+// TCP ramps up over the first second; the steady-state figure ignores it.
+export const RAMP_MS = 1000;
+export const PING_SAMPLES = 20;
+export const PING_GAP_MS = 100;
+export const PING_TIMEOUT_MS = 5000;
+// One body reused for every upload POST. 1 MiB so a slow hotspot (a few Mbps) still completes
+// several requests inside the phase; the server's per-request cap is 8 MiB.
+export const UPLOAD_CHUNK_BYTES = 1024 * 1024;
+
+// Typical bitrates for a well-encoded video at each resolution. Deliberately round; the table
+// says "about".
+export const STREAM_TIERS = [
+ {key: '480p', label: '480p (SD)', mbps: 2.5},
+ {key: '720p', label: '720p (HD)', mbps: 5},
+ {key: '1080p', label: '1080p (Full HD)', mbps: 8},
+ {key: '1440p', label: '1440p (2K)', mbps: 16},
+ {key: '2160p', label: '2160p (4K)', mbps: 25},
+];
+
+export const bytesToMbps = (bytes, ms) => ms > 0 ? (bytes * 8) / (ms / 1000) / 1e6 : 0;
+
+export const mbpsToMBps = mbps => mbps / 8;
+
+export function formatMbps(mbps) {
+ if (mbps === null || mbps === undefined || !Number.isFinite(mbps)) {
+ return '—';
+ }
+ if (mbps >= 100) {
+ return mbps.toFixed(0);
+ }
+ return mbps >= 10 ? mbps.toFixed(1) : mbps.toFixed(2);
+}
+
+export function formatMs(ms) {
+ if (ms === null || ms === undefined || !Number.isFinite(ms)) {
+ return '—';
+ }
+ return ms >= 100 ? ms.toFixed(0) : ms.toFixed(1);
+}
+
+/**
+ * How many simultaneous streams of each tier fit in the measured download throughput.
+ */
+export function streamCapacity(downloadMbps) {
+ const mbps = Number.isFinite(downloadMbps) && downloadMbps > 0 ? downloadMbps : 0;
+ return STREAM_TIERS.map(tier => ({...tier, streams: Math.floor(mbps / tier.mbps)}));
+}
+
+/**
+ * Average the per-bucket throughput after the ramp-up window. `samples` are
+ * `{ms, bytes}` buckets in order, `ms` being the bucket's end offset from the phase start.
+ * Falls back to the whole run when it was too short to have a steady state.
+ */
+export function steadyStateMbps(samples) {
+ if (!samples || samples.length === 0) {
+ return 0;
+ }
+ let steady = samples.filter(s => s.ms > RAMP_MS);
+ if (steady.length === 0) {
+ steady = samples;
+ }
+ const bytes = steady.reduce((acc, s) => acc + s.bytes, 0);
+ const first = steady[0];
+ const start = first.ms - first.width;
+ const ms = steady[steady.length - 1].ms - start;
+ return bytesToMbps(bytes, ms);
+}
+
+export const peakMbps = samples => Math.max(0, ...(samples || []).map(s => bytesToMbps(s.bytes, s.width)));
+
+export function pingStats(rtts) {
+ const good = (rtts || []).filter(Number.isFinite);
+ if (good.length === 0) {
+ return {min: null, median: null, jitter: null, lost: (rtts || []).length};
+ }
+ const sorted = [...good].sort((a, b) => a - b);
+ const mid = Math.floor(sorted.length / 2);
+ const median = sorted.length % 2 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2;
+ let jitter = 0;
+ if (good.length > 1) {
+ let diffs = 0;
+ for (let i = 1; i < good.length; i++) {
+ diffs += Math.abs(good[i] - good[i - 1]);
+ }
+ jitter = diffs / (good.length - 1);
+ }
+ return {min: sorted[0], median, jitter, lost: (rtts || []).length - good.length};
+}
+
+export function describePing(medianMs) {
+ if (medianMs === null || medianMs === undefined || !Number.isFinite(medianMs)) {
+ return {kind: 'warning', text: 'No response'};
+ }
+ if (medianMs < 20) {
+ return {kind: 'success', text: 'Excellent'};
+ }
+ if (medianMs < 60) {
+ return {kind: 'success', text: 'Good'};
+ }
+ if (medianMs < 150) {
+ return {kind: 'warning', text: 'Fair'};
+ }
+ return {kind: 'danger', text: 'Poor'};
+}
+
+/**
+ * NetworkManager's shared-connection mode (which `nmcli device wifi hotspot` uses) hands out
+ * 10.42.x.x addresses.
+ */
+export function isHotspotAddress(ip) {
+ if (!ip || typeof ip !== 'string') {
+ return false;
+ }
+ const v4 = ip.startsWith('::ffff:') ? ip.slice(7) : ip;
+ return /^10\.42\.\d{1,3}\.\d{1,3}$/.test(v4);
+}
+
+/**
+ * Accumulates byte counts into fixed-width buckets and reports them for the live display.
+ */
+export class Buckets {
+ constructor(width = BUCKET_MS, now = () => performance.now()) {
+ this.width = width;
+ this.now = now;
+ this.start = now();
+ this.samples = [];
+ this.pending = 0;
+ this.total = 0;
+ }
+
+ add(bytes) {
+ this.pending += bytes;
+ this.total += bytes;
+ this.flush(false);
+ }
+
+ /** Close every bucket that has fully elapsed (or all of them when `force`). */
+ flush(force) {
+ const elapsed = this.now() - this.start;
+ const closed = this.samples.length * this.width;
+ if (elapsed - closed >= this.width) {
+ // Every complete bucket since the last flush gets the pending bytes. Almost always
+ // this is one bucket; a stall spreads its bytes over the buckets it spanned.
+ const count = Math.floor((elapsed - closed) / this.width);
+ const per = this.pending / count;
+ for (let i = 1; i <= count; i++) {
+ this.samples.push({ms: closed + i * this.width, width: this.width, bytes: per});
+ }
+ this.pending = 0;
+ } else if (force && this.pending > 0) {
+ const width = Math.max(1, elapsed - closed);
+ this.samples.push({ms: elapsed, width, bytes: this.pending});
+ this.pending = 0;
+ }
+ }
+
+ get elapsedMs() {
+ return this.now() - this.start;
+ }
+
+ /** Mbps of the most recently closed bucket, for the big live number. */
+ get currentMbps() {
+ const last = this.samples[this.samples.length - 1];
+ return last ? bytesToMbps(last.bytes, last.width) : 0;
+ }
+}
+
+const defaultDeps = () => ({
+ fetch: (...args) => window.fetch(...args),
+ now: () => performance.now(),
+ sleep: ms => new Promise(resolve => setTimeout(resolve, ms)),
+});
+
+/**
+ * Time PING_SAMPLES sequential round trips. Calls `onSample(rtts)` after each.
+ * A sample that times out is recorded as `null`.
+ */
+export async function runPing({signal, onSample, samples = PING_SAMPLES, gapMs = PING_GAP_MS, deps} = {}) {
+ const {fetch, now, sleep} = {...defaultDeps(), ...deps};
+ const rtts = [];
+ for (let i = 0; i < samples; i++) {
+ if (signal?.aborted) {
+ break;
+ }
+ const timeout = new AbortController();
+ const timer = setTimeout(() => timeout.abort(), PING_TIMEOUT_MS);
+ const onAbort = () => timeout.abort();
+ signal?.addEventListener('abort', onAbort);
+ const start = now();
+ try {
+ const response = await fetch(`${SPEEDTEST_API}/ping`, {cache: 'no-store', signal: timeout.signal});
+ await response.arrayBuffer();
+ rtts.push(response.ok ? now() - start : null);
+ } catch (e) {
+ if (signal?.aborted) {
+ break;
+ }
+ rtts.push(null);
+ } finally {
+ clearTimeout(timer);
+ signal?.removeEventListener('abort', onAbort);
+ }
+ if (onSample) {
+ onSample([...rtts]);
+ }
+ if (i < samples - 1) {
+ await sleep(gapMs);
+ }
+ }
+ return pingStats(rtts);
+}
+
+/**
+ * Read a download stream for `seconds`, tallying bytes into buckets. Calls `onProgress(buckets)`
+ * as data arrives. Resolves to `{mbps, peak, bytes, samples}`.
+ */
+export async function runDownload({signal, onProgress, seconds = PHASE_SECONDS, deps} = {}) {
+ const {fetch, now} = {...defaultDeps(), ...deps};
+ const buckets = new Buckets(BUCKET_MS, now);
+ // The server also stops at `seconds`; the client deadline covers a slow last chunk.
+ const deadline = buckets.start + seconds * 1000;
+ const response = await fetch(`${SPEEDTEST_API}/download?seconds=${seconds}`, {cache: 'no-store', signal});
+ if (!response.ok) {
+ throw new Error(`Download test failed: HTTP ${response.status}`);
+ }
+ const reader = response.body.getReader();
+ try {
+ while (true) {
+ const {done, value} = await reader.read();
+ if (done) {
+ break;
+ }
+ buckets.add(value.byteLength);
+ if (onProgress) {
+ onProgress(buckets);
+ }
+ if (now() >= deadline) {
+ break;
+ }
+ }
+ } finally {
+ try {
+ await reader.cancel();
+ } catch (e) {
+ // Already closed.
+ }
+ }
+ buckets.flush(true);
+ return {mbps: steadyStateMbps(buckets.samples), peak: peakMbps(buckets.samples), bytes: buckets.total, samples: buckets.samples};
+}
+
+export function randomBlob(size = UPLOAD_CHUNK_BYTES) {
+ const bytes = new Uint8Array(size);
+ // getRandomValues refuses more than 64 KiB per call.
+ const step = 65536;
+ for (let offset = 0; offset < size; offset += step) {
+ window.crypto.getRandomValues(bytes.subarray(offset, Math.min(offset + step, size)));
+ }
+ return new Blob([bytes], {type: 'application/octet-stream'});
+}
+
+/**
+ * POST the same random body over and over for `seconds`, tallying the bytes each request
+ * delivered. Sequential requests rather than a streamed body because Safari cannot stream a
+ * request body. A request still in flight when the phase deadline passes is aborted, so the
+ * phase ends on time on a slow link rather than after the current body finishes; only completed
+ * requests count. Resolves to `{mbps, peak, bytes, samples}`.
+ */
+export async function runUpload({signal, onProgress, seconds = PHASE_SECONDS, body, deps} = {}) {
+ const {fetch, now} = {...defaultDeps(), ...deps};
+ const blob = body || randomBlob();
+ const buckets = new Buckets(BUCKET_MS, now);
+ const deadline = buckets.start + seconds * 1000;
+ // Aborts the in-flight request at the deadline or when the caller cancels.
+ const phase = new AbortController();
+ const onAbort = () => phase.abort();
+ signal?.addEventListener('abort', onAbort);
+ const timer = setTimeout(() => phase.abort(), seconds * 1000);
+ try {
+ while (now() < deadline && !phase.signal.aborted) {
+ let response;
+ try {
+ response = await fetch(`${SPEEDTEST_API}/upload`, {
+ method: 'POST', body: blob, cache: 'no-store', signal: phase.signal,
+ });
+ } catch (e) {
+ if (phase.signal.aborted && !signal?.aborted) {
+ // The deadline cut this request short; the phase is simply over.
+ break;
+ }
+ throw e;
+ }
+ if (!response.ok) {
+ throw new Error(`Upload test failed: HTTP ${response.status}`);
+ }
+ const json = await response.json();
+ buckets.add(json.bytes);
+ if (onProgress) {
+ onProgress(buckets);
+ }
+ }
+ } finally {
+ clearTimeout(timer);
+ signal?.removeEventListener('abort', onAbort);
+ }
+ buckets.flush(true);
+ return {mbps: steadyStateMbps(buckets.samples), peak: peakMbps(buckets.samples), bytes: buckets.total, samples: buckets.samples};
+}
+
+export async function fetchInfo({deps} = {}) {
+ const {fetch} = {...defaultDeps(), ...deps};
+ const response = await fetch(`${SPEEDTEST_API}/info`, {cache: 'no-store'});
+ if (!response.ok) {
+ throw new Error(`HTTP ${response.status}`);
+ }
+ return await response.json();
+}
+
+/** "2 other tests" given the counter sampled before and during our own run. */
+export function otherActiveTests(before, during) {
+ const b = Number.isFinite(before) ? before : 0;
+ // While our own download or upload is in flight the counter includes us.
+ const d = Number.isFinite(during) ? Math.max(0, during - 1) : 0;
+ return Math.max(b, d);
+}
diff --git a/app/src/components/calculators/speedtest.test.js b/app/src/components/calculators/speedtest.test.js
new file mode 100644
index 00000000..df058792
--- /dev/null
+++ b/app/src/components/calculators/speedtest.test.js
@@ -0,0 +1,318 @@
+import {
+ Buckets,
+ bytesToMbps,
+ describePing,
+ formatMbps,
+ isHotspotAddress,
+ otherActiveTests,
+ peakMbps,
+ pingStats,
+ runDownload,
+ runPing,
+ runUpload,
+ steadyStateMbps,
+ streamCapacity,
+} from "./speedtest";
+
+// A controllable clock so bucket boundaries are deterministic.
+const makeClock = () => {
+ let t = 0;
+ const now = () => t;
+ now.advance = ms => {
+ t += ms;
+ };
+ return now;
+};
+
+describe('bytesToMbps', () => {
+ test('converts bytes over milliseconds to decimal megabits per second', () => {
+ // 1,250,000 bytes in one second is 10,000,000 bits/s = 10 Mbps.
+ expect(bytesToMbps(1_250_000, 1000)).toBeCloseTo(10, 6);
+ expect(bytesToMbps(1_250_000, 500)).toBeCloseTo(20, 6);
+ });
+
+ test('zero or negative time is zero, not infinity', () => {
+ expect(bytesToMbps(100, 0)).toBe(0);
+ });
+});
+
+describe('formatMbps', () => {
+ test('shows more digits for slower links', () => {
+ expect(formatMbps(543.21)).toBe('543');
+ expect(formatMbps(54.321)).toBe('54.3');
+ expect(formatMbps(5.4321)).toBe('5.43');
+ });
+
+ test('missing values are a dash', () => {
+ expect(formatMbps(null)).toBe('—');
+ expect(formatMbps(NaN)).toBe('—');
+ });
+});
+
+describe('streamCapacity', () => {
+ test('floors the measured throughput against each tier bitrate', () => {
+ const rows = Object.fromEntries(streamCapacity(8).map(r => [r.key, r.streams]));
+ expect(rows).toEqual({'480p': 3, '720p': 1, '1080p': 1, '1440p': 0, '2160p': 0});
+ });
+
+ test('a fast link streams many at once', () => {
+ const rows = Object.fromEntries(streamCapacity(250).map(r => [r.key, r.streams]));
+ expect(rows['2160p']).toBe(10);
+ expect(rows['480p']).toBe(100);
+ });
+
+ test('nothing measured means zero streams everywhere', () => {
+ expect(streamCapacity(0).every(r => r.streams === 0)).toBe(true);
+ expect(streamCapacity(NaN).every(r => r.streams === 0)).toBe(true);
+ });
+});
+
+describe('steadyStateMbps', () => {
+ // 250 ms buckets; 312,500 bytes per bucket is 10 Mbps.
+ const bucket = (ms, bytes) => ({ms, width: 250, bytes});
+
+ test('ignores the ramp-up second and averages the rest', () => {
+ const samples = [
+ bucket(250, 0), bucket(500, 0), bucket(750, 0), bucket(1000, 0), // slow start
+ bucket(1250, 312_500), bucket(1500, 312_500), bucket(1750, 312_500), bucket(2000, 312_500),
+ ];
+ expect(steadyStateMbps(samples)).toBeCloseTo(10, 6);
+ });
+
+ test('falls back to the whole run when it never left the ramp window', () => {
+ expect(steadyStateMbps([bucket(250, 312_500), bucket(500, 312_500)])).toBeCloseTo(10, 6);
+ });
+
+ test('empty run is zero', () => {
+ expect(steadyStateMbps([])).toBe(0);
+ expect(steadyStateMbps(undefined)).toBe(0);
+ });
+
+ test('peak is the fastest single bucket', () => {
+ expect(peakMbps([bucket(250, 312_500), bucket(500, 625_000)])).toBeCloseTo(20, 6);
+ });
+});
+
+describe('pingStats', () => {
+ test('reports min, median and mean successive jitter', () => {
+ const stats = pingStats([10, 12, 30, 14]);
+ expect(stats.min).toBe(10);
+ expect(stats.median).toBe(13);
+ // |12-10| + |30-12| + |14-30| = 2 + 18 + 16 = 36; / 3 = 12.
+ expect(stats.jitter).toBeCloseTo(12, 6);
+ expect(stats.lost).toBe(0);
+ });
+
+ test('timeouts are counted as lost and excluded from the numbers', () => {
+ const stats = pingStats([10, null, 20]);
+ expect(stats.lost).toBe(1);
+ expect(stats.median).toBe(15);
+ });
+
+ test('all lost gives nulls', () => {
+ expect(pingStats([null, null])).toEqual({min: null, median: null, jitter: null, lost: 2});
+ });
+});
+
+describe('describePing', () => {
+ test('thresholds', () => {
+ expect(describePing(5).text).toBe('Excellent');
+ expect(describePing(40).text).toBe('Good');
+ expect(describePing(100).text).toBe('Fair');
+ expect(describePing(400).text).toBe('Poor');
+ expect(describePing(null).text).toBe('No response');
+ });
+});
+
+describe('isHotspotAddress', () => {
+ test("matches NetworkManager's shared range", () => {
+ expect(isHotspotAddress('10.42.0.7')).toBe(true);
+ expect(isHotspotAddress('10.42.1.200')).toBe(true);
+ expect(isHotspotAddress('::ffff:10.42.0.7')).toBe(true);
+ });
+
+ test('rejects LAN, loopback and garbage', () => {
+ expect(isHotspotAddress('10.0.0.5')).toBe(false);
+ expect(isHotspotAddress('192.168.1.10')).toBe(false);
+ expect(isHotspotAddress('::1')).toBe(false);
+ expect(isHotspotAddress('10.420.0.1')).toBe(false);
+ expect(isHotspotAddress(undefined)).toBe(false);
+ });
+});
+
+describe('otherActiveTests', () => {
+ test('subtracts our own in-flight request from the during sample', () => {
+ expect(otherActiveTests(0, 1)).toBe(0);
+ expect(otherActiveTests(0, 3)).toBe(2);
+ expect(otherActiveTests(2, 1)).toBe(2);
+ expect(otherActiveTests(undefined, undefined)).toBe(0);
+ });
+});
+
+describe('Buckets', () => {
+ test('closes a bucket once its width has elapsed', () => {
+ const now = makeClock();
+ const b = new Buckets(250, now);
+ b.add(100);
+ expect(b.samples).toHaveLength(0);
+ now.advance(250);
+ b.add(200);
+ expect(b.samples).toHaveLength(1);
+ expect(b.samples[0]).toEqual({ms: 250, width: 250, bytes: 300});
+ expect(b.total).toBe(300);
+ });
+
+ test('a stall spreads its bytes over the buckets it spanned', () => {
+ const now = makeClock();
+ const b = new Buckets(250, now);
+ now.advance(1000);
+ b.add(400);
+ expect(b.samples.map(s => s.bytes)).toEqual([100, 100, 100, 100]);
+ });
+
+ test('force flush closes a partial bucket', () => {
+ const now = makeClock();
+ const b = new Buckets(250, now);
+ now.advance(100);
+ b.add(50);
+ b.flush(true);
+ expect(b.samples).toHaveLength(1);
+ expect(b.samples[0].width).toBe(100);
+ expect(b.samples[0].bytes).toBe(50);
+ });
+});
+
+// A fake fetch that serves a ReadableStream of `chunks`, advancing the clock per chunk.
+const streamingFetch = (chunks, now, perChunkMs) => jest.fn(async () => {
+ let i = 0;
+ return {
+ ok: true,
+ status: 200,
+ body: {
+ getReader: () => ({
+ read: async () => {
+ if (i >= chunks.length) {
+ return {done: true};
+ }
+ now.advance(perChunkMs);
+ return {done: false, value: new Uint8Array(chunks[i++])};
+ },
+ cancel: async () => {
+ },
+ }),
+ },
+ };
+});
+
+describe('runDownload', () => {
+ test('totals the bytes and reports steady-state throughput', async () => {
+ const now = makeClock();
+ // 12 chunks of 312,500 bytes, one every 250 ms: 10 Mbps for 3 s.
+ const fetch = streamingFetch(Array(12).fill(312_500), now, 250);
+ const progress = jest.fn();
+ const result = await runDownload({seconds: 10, onProgress: progress, deps: {fetch, now}});
+ expect(fetch).toHaveBeenCalledWith(expect.stringContaining('/speedtest/download?seconds=10'),
+ expect.objectContaining({cache: 'no-store'}));
+ expect(result.bytes).toBe(12 * 312_500);
+ expect(result.mbps).toBeCloseTo(10, 3);
+ expect(progress).toHaveBeenCalled();
+ });
+
+ test('stops reading at the client deadline even if the server keeps sending', async () => {
+ const now = makeClock();
+ const fetch = streamingFetch(Array(100).fill(1000), now, 250);
+ const result = await runDownload({seconds: 1, deps: {fetch, now}});
+ // 1 s deadline / 250 ms per chunk = 4 chunks read.
+ expect(result.bytes).toBe(4000);
+ });
+
+ test('a non-2xx response is an error', async () => {
+ const fetch = jest.fn(async () => ({ok: false, status: 502}));
+ await expect(runDownload({deps: {fetch, now: makeClock()}})).rejects.toThrow('HTTP 502');
+ });
+});
+
+describe('runUpload', () => {
+ test('posts until the deadline and sums the server-reported bytes', async () => {
+ const now = makeClock();
+ const fetch = jest.fn(async (url, init) => {
+ now.advance(500);
+ return {ok: true, status: 200, json: async () => ({bytes: init.body.size, seconds: 0.5})};
+ });
+ const body = {size: 625_000}; // 625,000 bytes per 500 ms = 10 Mbps
+ const result = await runUpload({seconds: 2, body, deps: {fetch, now}});
+ expect(fetch).toHaveBeenCalledTimes(4);
+ expect(fetch.mock.calls[0][1]).toEqual(expect.objectContaining({method: 'POST', body}));
+ expect(result.bytes).toBe(4 * 625_000);
+ expect(result.mbps).toBeCloseTo(10, 3);
+ });
+
+ test('a request still in flight at the deadline is aborted and the phase ends on time', async () => {
+ jest.useFakeTimers();
+ try {
+ const now = makeClock();
+ let calls = 0;
+ const fetch = jest.fn((url, init) => new Promise((resolve, reject) => {
+ calls++;
+ if (calls === 1) {
+ // Fast first request.
+ now.advance(200);
+ resolve({ok: true, status: 200, json: async () => ({bytes: 1000, seconds: 0.2})});
+ return;
+ }
+ // Second request never completes on its own: it must be aborted at the deadline.
+ init.signal.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')));
+ }));
+ const promise = runUpload({seconds: 1, body: {size: 1000}, deps: {fetch, now}});
+ // Let the first request settle and the second start.
+ await Promise.resolve();
+ await Promise.resolve();
+ await Promise.resolve();
+ now.advance(2000);
+ jest.advanceTimersByTime(1000);
+ const result = await promise;
+ expect(fetch).toHaveBeenCalledTimes(2);
+ expect(fetch.mock.calls[1][1].signal.aborted).toBe(true);
+ // Only the completed request counts.
+ expect(result.bytes).toBe(1000);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
+ test('an aborted signal ends the loop', async () => {
+ const now = makeClock();
+ const controller = new AbortController();
+ const fetch = jest.fn(async () => {
+ now.advance(100);
+ controller.abort();
+ return {ok: true, status: 200, json: async () => ({bytes: 10, seconds: 0.1})};
+ });
+ const result = await runUpload({seconds: 10, body: {size: 10}, signal: controller.signal, deps: {fetch, now}});
+ expect(fetch).toHaveBeenCalledTimes(1);
+ expect(result.bytes).toBe(10);
+ });
+});
+
+describe('runPing', () => {
+ test('times each round trip and records timeouts as lost', async () => {
+ const now = makeClock();
+ let call = 0;
+ const fetch = jest.fn(async () => {
+ call++;
+ now.advance(call === 2 ? 30 : 10);
+ if (call === 3) {
+ throw new Error('network');
+ }
+ return {ok: true, arrayBuffer: async () => new ArrayBuffer(0)};
+ });
+ const sleep = jest.fn(async () => {
+ });
+ const stats = await runPing({samples: 4, deps: {fetch, now, sleep}});
+ expect(fetch).toHaveBeenCalledTimes(4);
+ expect(stats.lost).toBe(1);
+ expect(stats.min).toBe(10);
+ expect(stats.median).toBe(10);
+ expect(sleep).toHaveBeenCalledTimes(3);
+ });
+});
diff --git a/app/src/components/ui/Sparkline.tsx b/app/src/components/ui/Sparkline.tsx
new file mode 100644
index 00000000..9e6f4bb0
--- /dev/null
+++ b/app/src/components/ui/Sparkline.tsx
@@ -0,0 +1,75 @@
+import React from 'react';
+
+export interface SparklineProps {
+ /** The series, oldest first. Non-finite values are drawn as gaps. */
+ values: Array;
+ /** Fixed vertical ceiling; defaults to the series maximum so the line fills the box. */
+ max?: number;
+ /** Token color name (`blue`, `green`, `warning`…). Resolves through `var(--name)`. */
+ color?: string;
+ /** Pixel height. Width always fills the container. */
+ height?: number;
+ /** Draw a faint fill under the line. */
+ area?: boolean;
+ /** Accessible summary, e.g. "Download throughput over the last 10 seconds". */
+ label?: string;
+ className?: string;
+}
+
+/** A tiny inline-SVG line chart with no axes; the figure beside it carries the scale. */
+export function Sparkline({values, max, color = 'blue', height = 48, area = true, label, className}: SparklineProps) {
+ const width = 100; // viewBox units; the SVG stretches to its container.
+ const finite = values.map(v => (typeof v === 'number' && Number.isFinite(v) ? v : null));
+ const ceiling = max && max > 0 ? max : Math.max(1e-9, ...finite.map(v => v ?? 0));
+ const n = finite.length;
+ const step = n > 1 ? width / (n - 1) : width;
+ const pad = 2;
+ const usable = height - pad * 2;
+
+ // Break the polyline at gaps so a lost sample shows as missing rather than a plunge.
+ const segments: string[] = [];
+ let current: string[] = [];
+ finite.forEach((v, i) => {
+ if (v === null) {
+ if (current.length) {
+ segments.push(current.join(' '));
+ current = [];
+ }
+ return;
+ }
+ const x = n > 1 ? i * step : width;
+ const y = pad + usable - (Math.min(v, ceiling) / ceiling) * usable;
+ current.push(`${x.toFixed(2)},${y.toFixed(2)}`);
+ });
+ if (current.length) {
+ segments.push(current.join(' '));
+ }
+
+ const stroke = `var(--${color})`;
+
+ return ;
+}
diff --git a/app/src/components/ui/index.ts b/app/src/components/ui/index.ts
index bac5c82e..da22a30a 100644
--- a/app/src/components/ui/index.ts
+++ b/app/src/components/ui/index.ts
@@ -15,6 +15,7 @@ export * from './scale';
export * from './Icon';
export * from './Button';
export * from './Feedback';
+export * from './Sparkline';
export * from './Surfaces';
export * from './DataTable';
export * from './Navigation';
diff --git a/docker/web/Caddyfile b/docker/web/Caddyfile
index a7b095c8..1d745032 100644
--- a/docker/web/Caddyfile
+++ b/docker/web/Caddyfile
@@ -44,6 +44,13 @@
redir * /controller/ 301
}
+ # Speed test streams generated bytes; never buffer it.
+ handle /api/speedtest/* {
+ reverse_proxy api:8081 {
+ flush_interval -1
+ }
+ }
+
# Main API.
handle /api/* {
reverse_proxy api:8081
diff --git a/etc/raspberrypios/Caddyfile b/etc/raspberrypios/Caddyfile
index 95989f58..5a6f2484 100644
--- a/etc/raspberrypios/Caddyfile
+++ b/etc/raspberrypios/Caddyfile
@@ -44,6 +44,13 @@
redir * /controller/ 301
}
+ # Speed test streams generated bytes; never buffer it.
+ handle /api/speedtest/* {
+ reverse_proxy 0.0.0.0:8081 {
+ flush_interval -1
+ }
+ }
+
# Main API.
handle /api/* {
reverse_proxy 0.0.0.0:8081
diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py
index ac80ad10..e1c0b73b 100644
--- a/wrolpi/contexts.py
+++ b/wrolpi/contexts.py
@@ -74,6 +74,9 @@ def attach_shared_contexts(app: Sanic):
# held until the API restarts; that is accepted, transcoding is not critical work.
app.shared_ctx.transcode_lock = multiprocessing.Lock()
+ # Speed tests in flight across all workers (wrolpi.speedtest). Informational only, no lock.
+ app.shared_ctx.speedtest_active = multiprocessing.Value(ctypes.c_int, 0)
+
# Switches
app.shared_ctx.switches = manager.dict()
app.shared_ctx.switches_lock = multiprocessing.Lock()
@@ -125,6 +128,7 @@ def reset_shared_contexts(app: Sanic):
app.shared_ctx.refresh.clear()
app.shared_ctx.uploaded_files.clear()
app.shared_ctx.status.clear()
+ app.shared_ctx.speedtest_active.value = 0
app.shared_ctx.status.update(dict(
cpu_stats=dict(),
load_stats=dict(),
diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py
index 0e76f5e8..95d36512 100644
--- a/wrolpi/root_api.py
+++ b/wrolpi/root_api.py
@@ -36,6 +36,7 @@ from wrolpi.events import get_events, Events
from wrolpi.files import files_bp
from wrolpi.files.lib import get_file_statistics, search_file_suggestion_count
from wrolpi.log_levels import int_to_name
+from wrolpi.speedtest import speedtest_bp
from wrolpi.tags import Tag
from wrolpi.vars import DOCKERIZED, IS_RPI, IS_RPI4, IS_RPI5, API_HOST, API_PORT, API_WORKERS, API_DEBUG, \
API_ACCESS_LOG, truthy_arg, API_AUTO_RELOAD
@@ -44,6 +45,9 @@ from wrolpi.version import __version__
logger = logger.getChild(__name__)
api_app.config.FALLBACK_ERROR_FORMAT = 'json'
+# Caddy is the single proxy in front of the API; trust its X-Forwarded-For so request.remote_addr
+# is the browser's address (used by the speed test).
+api_app.config.PROXIES_COUNT = 1
api_bp = Blueprint('RootAPI', url_prefix='/api')
@@ -57,6 +61,7 @@ api_app.blueprint(files_bp)
api_app.blueprint(flasher_bp)
api_app.blueprint(inventory_bp)
api_app.blueprint(map_bp)
+api_app.blueprint(speedtest_bp)
api_app.blueprint(videos_bp)
api_app.blueprint(zim_bp)
diff --git a/wrolpi/speedtest.py b/wrolpi/speedtest.py
new file mode 100644
index 00000000..86678602
--- /dev/null
+++ b/wrolpi/speedtest.py
@@ -0,0 +1,173 @@
+"""Speed test endpoints: ping, a bounded download stream, a bounded upload sink, and context.
+
+Nothing is stored. Several clients may test at once; the server only counts tests in flight.
+"""
+import os
+import time
+from http import HTTPStatus
+from typing import Iterator, Optional
+
+from sanic import Blueprint
+from sanic.request import Request
+from sanic_ext.extensions.openapi import openapi
+
+from wrolpi.api_utils import json_response, api_app
+from wrolpi.common import logger
+from wrolpi.errors import ValidationError, APIError
+
+logger = logger.getChild(__name__)
+
+speedtest_bp = Blueprint('SpeedTest', url_prefix='/api/speedtest')
+
+# Bounds on how long a download stream may hold a worker.
+MIN_DOWNLOAD_SECONDS = 0.1
+DEFAULT_DOWNLOAD_SECONDS = 10.0
+MAX_DOWNLOAD_SECONDS = 15.0
+# Bytes sent per `await response.send()`. Large enough that Python overhead does not cap a fast
+# LAN, small enough that the loop yields often.
+CHUNK_SIZE = 256 * 1024
+# The browser uploads in sequential POSTs of a few MiB. Both caps exist so a request cannot hold a
+# worker: size alone does not, a slow or paused body could trickle in for minutes.
+MAX_UPLOAD_CHUNK = 8 * 1024 * 1024
+MAX_UPLOAD_SECONDS = 15.0
+
+# One random buffer built at import; chunks are slices of it so no bytes are generated per
+# request. Random so no proxy or compression can flatter the result. A whole multiple of
+# CHUNK_SIZE so every slice is a full chunk.
+_RANDOM = os.urandom(CHUNK_SIZE * 16)
+
+NO_STORE_HEADERS = {
+ 'Cache-Control': 'no-store',
+ # Ask proxies not to buffer the stream.
+ 'X-Accel-Buffering': 'no',
+}
+
+
+class UploadTooLarge(APIError):
+ code = 'UPLOAD_TOO_LARGE'
+ summary = f'Speed test upload requests are limited to {MAX_UPLOAD_CHUNK} bytes'
+ status_code = HTTPStatus.REQUEST_ENTITY_TOO_LARGE
+
+
+class UploadTooSlow(APIError):
+ code = 'UPLOAD_TOO_SLOW'
+ summary = f'Speed test upload requests must finish within {MAX_UPLOAD_SECONDS} seconds'
+ status_code = HTTPStatus.REQUEST_TIMEOUT
+
+
+def clamp_seconds(value: Optional[str]) -> float:
+ """Parse the requested download duration and keep it within the server's bounds."""
+ if value is None or value == '':
+ return DEFAULT_DOWNLOAD_SECONDS
+ try:
+ seconds = float(value)
+ except (TypeError, ValueError):
+ raise ValidationError(f'seconds must be a number, got {value!r}')
+ if seconds != seconds: # NaN
+ raise ValidationError('seconds must be a number, got NaN')
+ if seconds < 0:
+ raise ValidationError('seconds must not be negative')
+ return max(MIN_DOWNLOAD_SECONDS, min(seconds, MAX_DOWNLOAD_SECONDS))
+
+
+def iter_chunks() -> Iterator[bytes]:
+ """Yield full CHUNK_SIZE slices of the random buffer forever, wrapping around."""
+ offset = 0
+ while True:
+ yield _RANDOM[offset:offset + CHUNK_SIZE]
+ offset = (offset + CHUNK_SIZE) % len(_RANDOM)
+
+
+def _counter():
+ """The cross-worker count of speed tests in flight (created pre-fork in wrolpi.contexts)."""
+ return api_app.shared_ctx.speedtest_active
+
+
+def _adjust_active(delta: int):
+ counter = _counter()
+ with counter.get_lock():
+ counter.value = max(0, counter.value + delta)
+
+
+def get_client_ip(request: Request) -> str:
+ """The address Caddy forwarded, falling back to the socket peer for direct requests."""
+ return request.remote_addr or request.ip
+
+
+@speedtest_bp.get('/ping')
+@openapi.definition(
+ summary='Tiny response the browser times to measure round trip latency.',
+)
+async def ping(_: Request):
+ return json_response(dict(now=time.monotonic() * 1000), headers=NO_STORE_HEADERS)
+
+
+@speedtest_bp.get('/info')
+@openapi.definition(
+ summary='Context for a speed test: the client address the server saw and how many tests are running.',
+)
+async def info(request: Request):
+ return json_response(dict(
+ client_ip=get_client_ip(request),
+ active_tests=_counter().value,
+ max_download_seconds=MAX_DOWNLOAD_SECONDS,
+ max_upload_chunk=MAX_UPLOAD_CHUNK,
+ max_upload_seconds=MAX_UPLOAD_SECONDS,
+ ), headers=NO_STORE_HEADERS)
+
+
+@speedtest_bp.get('/download')
+@openapi.definition(
+ summary='Stream random bytes for `seconds` (clamped) so the browser can measure download throughput.',
+)
+async def download(request: Request):
+ seconds = clamp_seconds(request.args.get('seconds'))
+ response = await request.respond(content_type='application/octet-stream', headers=NO_STORE_HEADERS)
+ _adjust_active(1)
+ try:
+ deadline = time.monotonic() + seconds
+ for chunk in iter_chunks():
+ if time.monotonic() >= deadline:
+ break
+ await response.send(chunk)
+ await response.eof()
+ except Exception as e:
+ # The browser aborts the fetch when its own timer ends; that is normal.
+ logger.debug(f'speed test download ended early: {e!r}')
+ finally:
+ _adjust_active(-1)
+
+
+async def drain_upload(request: Request) -> int:
+ """Read and discard the request body, enforcing both the size and the time cap."""
+ deadline = time.monotonic() + MAX_UPLOAD_SECONDS
+ total = 0
+ while True:
+ body = await request.stream.read()
+ if body is None:
+ return total
+ total += len(body)
+ if total > MAX_UPLOAD_CHUNK:
+ raise UploadTooLarge()
+ if time.monotonic() > deadline:
+ raise UploadTooSlow()
+
+
+@speedtest_bp.post('/upload', stream=True)
+@openapi.definition(
+ summary='Read and discard the request body, reporting how many bytes arrived and how long it took.',
+)
+async def upload(request: Request):
+ declared = request.headers.get('content-length')
+ if declared and declared.isdigit() and int(declared) > MAX_UPLOAD_CHUNK:
+ raise UploadTooLarge()
+
+ _adjust_active(1)
+ try:
+ start = time.monotonic()
+ total = await drain_upload(request)
+ seconds = time.monotonic() - start
+ finally:
+ _adjust_active(-1)
+
+ return json_response(dict(bytes=total, seconds=seconds), headers=NO_STORE_HEADERS)
diff --git a/wrolpi/test/test_speedtest.py b/wrolpi/test/test_speedtest.py
new file mode 100644
index 00000000..0e3d1db1
--- /dev/null
+++ b/wrolpi/test/test_speedtest.py
@@ -0,0 +1,190 @@
+import time
+from http import HTTPStatus
+
+import pytest
+
+from wrolpi import speedtest
+from wrolpi.api_utils import api_app
+
+
+def _active() -> int:
+ return api_app.shared_ctx.speedtest_active.value
+
+
+def test_clamp_seconds():
+ assert speedtest.clamp_seconds(None) == speedtest.DEFAULT_DOWNLOAD_SECONDS
+ assert speedtest.clamp_seconds('') == speedtest.DEFAULT_DOWNLOAD_SECONDS
+ assert speedtest.clamp_seconds('2.5') == 2.5
+ # Too long is clamped to the ceiling, too short to the floor.
+ assert speedtest.clamp_seconds('999') == speedtest.MAX_DOWNLOAD_SECONDS
+ assert speedtest.clamp_seconds('0') == speedtest.MIN_DOWNLOAD_SECONDS
+ with pytest.raises(speedtest.ValidationError):
+ speedtest.clamp_seconds('abc')
+ with pytest.raises(speedtest.ValidationError, match='negative'):
+ speedtest.clamp_seconds('-1')
+ with pytest.raises(speedtest.ValidationError, match='NaN'):
+ speedtest.clamp_seconds('nan')
+
+
+def test_chunks_never_empty():
+ """The chunk generator wraps around the random buffer without ever yielding a short chunk."""
+ chunks = speedtest.iter_chunks()
+ seen = [next(chunks) for _ in range(len(speedtest._RANDOM) // speedtest.CHUNK_SIZE + 3)]
+ assert all(len(i) == speedtest.CHUNK_SIZE for i in seen)
+ # Wrapped around to the start.
+ assert seen[0] == seen[len(speedtest._RANDOM) // speedtest.CHUNK_SIZE]
+
+
+@pytest.mark.asyncio
+async def test_ping(async_client):
+ request, response = await async_client.get('/api/speedtest/ping')
+ assert response.status_code == HTTPStatus.OK
+ assert isinstance(response.json['now'], (int, float))
+ assert response.headers['Cache-Control'] == 'no-store'
+
+
+@pytest.mark.asyncio
+async def test_download_streams_for_requested_seconds(async_client):
+ assert _active() == 0
+ start = time.monotonic()
+ request, response = await async_client.get('/api/speedtest/download?seconds=0.3')
+ elapsed = time.monotonic() - start
+ assert response.status_code == HTTPStatus.OK
+ assert response.headers['Content-Type'] == 'application/octet-stream'
+ assert response.headers['Cache-Control'] == 'no-store'
+ assert len(response.body) >= speedtest.CHUNK_SIZE
+ assert len(response.body) % speedtest.CHUNK_SIZE == 0
+ # Ran for about the requested duration. Loose upper bound for slow CI.
+ assert 0.25 <= elapsed < 5
+ # Counter is released when the stream finishes.
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_download_clamps_seconds(async_client):
+ """A client cannot hold a worker longer than MAX_DOWNLOAD_SECONDS."""
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'MAX_DOWNLOAD_SECONDS', 0.2)
+ start = time.monotonic()
+ request, response = await async_client.get('/api/speedtest/download?seconds=999')
+ elapsed = time.monotonic() - start
+ assert response.status_code == HTTPStatus.OK
+ assert elapsed < 3
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_download_invalid_seconds(async_client):
+ request, response = await async_client.get('/api/speedtest/download?seconds=abc')
+ assert response.status_code == HTTPStatus.BAD_REQUEST
+ request, response = await async_client.get('/api/speedtest/download?seconds=-1')
+ assert response.status_code == HTTPStatus.BAD_REQUEST
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_download_client_disconnect_releases_counter(async_client):
+ """The active-test counter is released even when the stream dies part way through."""
+
+ def broken_chunks():
+ yield speedtest._RANDOM[:speedtest.CHUNK_SIZE]
+ raise ConnectionResetError('client went away')
+
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'iter_chunks', broken_chunks)
+ try:
+ await async_client.get('/api/speedtest/download?seconds=0.5')
+ except Exception:
+ # The test client may surface the broken stream; the counter is what matters.
+ pass
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_upload_counts_and_discards(async_client):
+ body = b'x' * (1024 * 1024)
+ request, response = await async_client.post('/api/speedtest/upload', content=body)
+ assert response.status_code == HTTPStatus.OK
+ assert response.json['bytes'] == len(body)
+ assert response.json['seconds'] >= 0
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_upload_too_large(async_client):
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'MAX_UPLOAD_CHUNK', 1024)
+ body = b'x' * 2048
+ request, response = await async_client.post('/api/speedtest/upload', content=body)
+ assert response.status_code == HTTPStatus.REQUEST_ENTITY_TOO_LARGE
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_upload_too_large_mid_stream_releases_counter(async_client):
+ """A body that only reveals its size while streaming is refused, and the counter is released."""
+
+ async def body():
+ yield b'x' * 1024
+ yield b'x' * 1024
+
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'MAX_UPLOAD_CHUNK', 1024)
+ request, response = await async_client.post('/api/speedtest/upload', content=body())
+ assert response.status_code == HTTPStatus.REQUEST_ENTITY_TOO_LARGE
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_upload_too_slow(async_client):
+ """A body that trickles in past the time cap is refused, and the counter is released."""
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'MAX_UPLOAD_SECONDS', 0)
+ request, response = await async_client.post('/api/speedtest/upload', content=b'x' * 2048)
+ assert response.status_code == HTTPStatus.REQUEST_TIMEOUT
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_upload_broken_stream_releases_counter(async_client):
+ """The counter is released when reading the body blows up part way through."""
+
+ async def broken(_request):
+ raise ConnectionResetError('client went away')
+
+ with pytest.MonkeyPatch.context() as mp:
+ mp.setattr(speedtest, 'drain_upload', broken)
+ try:
+ await async_client.post('/api/speedtest/upload', content=b'x' * 1024)
+ except Exception:
+ pass
+ assert _active() == 0
+
+
+@pytest.mark.asyncio
+async def test_info(async_client):
+ request, response = await async_client.get('/api/speedtest/info')
+ assert response.status_code == HTTPStatus.OK
+ assert response.json['client_ip']
+ assert response.json['active_tests'] == 0
+ assert response.json['max_download_seconds'] == speedtest.MAX_DOWNLOAD_SECONDS
+ assert response.json['max_upload_chunk'] == speedtest.MAX_UPLOAD_CHUNK
+ assert response.json['max_upload_seconds'] == speedtest.MAX_UPLOAD_SECONDS
+
+ # Caddy forwards the real client address; Sanic is configured to trust one proxy hop.
+ request, response = await async_client.get('/api/speedtest/info',
+ headers={'X-Forwarded-For': '10.42.0.7'})
+ assert response.json['client_ip'] == '10.42.0.7'
+
+
+def test_speedtest_counter_in_shared_context():
+ """The counter must be created pre-fork so all Sanic workers share it."""
+ from sanic import Sanic
+ from wrolpi.contexts import attach_shared_contexts, reset_shared_contexts
+
+ test_app = Sanic(f'test_context_{id(test_speedtest_counter_in_shared_context)}')
+ attach_shared_contexts(test_app)
+ assert test_app.shared_ctx.speedtest_active.value == 0
+ test_app.shared_ctx.speedtest_active.value = 3
+ reset_shared_contexts(test_app)
+ assert test_app.shared_ctx.speedtest_active.value == 0