From 59597ae0f191494dcb79ead5d0ca3165f1394b26 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:22:15 +0200 Subject: [PATCH 1/5] feat(fleet): accept a paired client's own earnings history A client that has been reading a provider account on its own -- CashPilot Desktop before it was paired -- had no way to hand that history to the server, so the fleet view began on the day of pairing and every earlier day was simply lost from the total. POST /api/workers/earnings-import takes those readings and stores them under the importing client's own source, which the source-aware schema added in the previous change makes possible. Separate series matter here: earnings are clamped deltas between consecutive readings of the same balance, so interleaving two samplers of one account makes every apparent drop clamp to zero and understates the total. Each series is differenced on its own and the results are summed. Two properties carry the security of it: * The source comes from the AUTHENTICATED worker, never the request body, so no client can write into another's history or into the server's own. * Only a fully enrolled worker may import. A caller still presenting the shared enrollment key gets 403 with instructions to heartbeat first -- every worker holds that key, and this writes durable money data. Re-sending a day updates it rather than appending, so a retried or repeated import is safe by construction. Refs: CashPilot-Desktop-xjr --- app/main.py | 81 ++++++++++++++ docs/fleet.md | 34 ++++++ tests/test_beads_batch_63.py | 202 +++++++++++++++++++++++++++++++++++ 3 files changed, 317 insertions(+) create mode 100644 tests/test_beads_batch_63.py diff --git a/app/main.py b/app/main.py index 046a75f9..28b2f9f5 100644 --- a/app/main.py +++ b/app/main.py @@ -3713,6 +3713,87 @@ async def _earnings_for_worker(body: WorkerHeartbeat, days: int = 30) -> dict[st } +class EarningsReading(BaseModel): + """One historical balance reading from a paired client.""" + + slug: str + balance: float + date: str + currency: str = "USD" + fx_rate_usd: float | None = None + + +class EarningsImport(BaseModel): + """A client pushing the history it collected before it was paired. + + Deliberately carries NO source field. The source is taken from the + AUTHENTICATED worker, never from the body -- otherwise any enrolled client + could write into the 'server' series, or into another machine's, and + overwrite readings it never took. + """ + + client_id: str + readings: list[EarningsReading] = [] + + +@app.post("/api/workers/earnings-import") +async def api_worker_earnings_import(request: Request, body: EarningsImport) -> dict[str, Any]: + """Accept a paired client's historical earnings under its own source. + + This exists because the server and a Desktop can both have been reading the + SAME provider account. Their readings must not be merged into one series: + earnings are clamped deltas between consecutive readings, so interleaving two + samplers makes every drop clamp to zero and the total comes out + systematically understated. Each client's rows therefore land under its own + ``source`` and are differenced separately. + + Idempotent by construction: the (platform, source, date) unique index means a + re-pair or a retried import UPDATES a day rather than adding a second reading + for it, which would difference against itself and read as zero. + """ + cid = (body.client_id or "").strip() + if not cid: + raise HTTPException(status_code=400, detail="client_id required") + + state = await _authenticate_worker_heartbeat(request, cid) + # Only a CONFIRMED worker may import. "enroll" and "reissue" both mean the + # caller presented the SHARED key, which every worker holds -- accepting it + # here would let anyone with that token write a history for any client_id + # they cared to name. A heartbeat is idempotent status; this is durable + # money data, so it gets the stricter bar. + if state != "ok": + raise HTTPException( + status_code=403, + detail=( + "Importing earnings requires this worker's own key. Send a heartbeat first " + "to complete enrollment, then retry." + ), + ) + + known = {s["slug"] for s in catalog.get_services()} + written = 0 + skipped: list[str] = [] + for reading in body.readings: + slug = (reading.slug or "").strip() + # An unknown slug is dropped rather than stored: it would create a + # platform the catalog cannot render, name or ever collect for again. + if not slug or slug not in known: + skipped.append(slug or "(blank)") + continue + await database.upsert_earnings( + platform=slug, + balance=float(reading.balance), + currency=(reading.currency or "USD").upper(), + date=reading.date, + fx_rate_usd=reading.fx_rate_usd, + source=cid, + ) + written += 1 + + logger.info("Imported %d earnings reading(s) from worker '%s' (%d skipped)", written, cid, len(skipped)) + return {"status": "ok", "imported": written, "skipped": skipped, "source": cid} + + @app.post("/api/workers/heartbeat") async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[str, Any]: """Receive a heartbeat from a worker. Registers or updates the worker.""" diff --git a/docs/fleet.md b/docs/fleet.md index 01c21e27..edb2974e 100644 --- a/docs/fleet.md +++ b/docs/fleet.md @@ -34,6 +34,40 @@ Workers use **REST HTTP** to communicate with the UI: Workers must be reachable from the UI for commands. The UI must be reachable from workers for heartbeats. +### Importing earnings a client collected on its own + +`POST /api/workers/earnings-import` lets a client that has been reading a +provider account by itself — CashPilot Desktop, typically, before it was paired — +hand that history to the UI, so the fleet view shows the complete picture rather +than starting from the day of pairing. + +```json +{ + "client_id": "desktop-macbook", + "readings": [ + {"slug": "honeygain", "balance": 12.4, "date": "2026-07-01", "currency": "USD"}, + {"slug": "mysterium", "balance": 88.0, "date": "2026-07-01", "currency": "MYST", "fx_rate_usd": 0.41} + ] +} +``` + +Three things about it are deliberate: + +- **Each client's readings are stored under their own source, not merged with the + server's.** Earnings are clamped deltas between consecutive readings of the same + balance, so interleaving two samplers of one account makes every apparent drop + clamp to zero and understates the total. Separate series are differenced + separately and then summed. +- **The source is taken from the authenticated worker, never from the request + body.** Otherwise any enrolled client could write into another's history, or + into the server's own. +- **Only a fully enrolled worker may import.** A caller still presenting the + shared enrollment key gets `403` with instructions to heartbeat first: every + worker holds that key, and this writes durable money data. + +Re-sending the same day updates it rather than appending, so a retried or +repeated import is safe. + ## Setting Up the Fleet ### Main server (UI + local worker) diff --git a/tests/test_beads_batch_63.py b/tests/test_beads_batch_63.py new file mode 100644 index 00000000..1bbca4d2 --- /dev/null +++ b/tests/test_beads_batch_63.py @@ -0,0 +1,202 @@ +"""CashPilot-Desktop-xjr: a paired client pushes the history it collected alone. + +The write path for the source-aware schema. A Desktop that ran standalone has +months of readings the server never saw; pairing should surrender them so the +fleet shows the complete picture, and unlinking should leave that machine able to +show what it earned by itself. + +Two properties carry the whole design: + +* **The source comes from AUTHENTICATION, never the body.** Otherwise any + enrolled client could write into the ``'server'`` series, or into another + machine's, and overwrite readings it never took. +* **Only a CONFIRMED worker may import.** ``"enroll"`` and ``"reissue"`` both + mean the caller presented the SHARED key, which every worker holds. A + heartbeat is idempotent status; this is durable money data, so it gets the + stricter bar. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi.testclient import TestClient + +from app.main import app + + +@pytest.fixture +def client(): + # Not a context manager: entering one runs the lifespan, which starts + # APScheduler and fails outside the main thread. These tests exercise the + # route, not startup. + return TestClient(app, raise_server_exceptions=False) + + +def _import(client, *, cid="desktop-a", readings=None, token="worker-own-key"): + return client.post( + "/api/workers/earnings-import", + json={"client_id": cid, "readings": readings if readings is not None else []}, + headers={"Authorization": f"Bearer {token}"}, + ) + + +class TestOnlyAConfirmedWorkerMayImport: + """The shared enrolment key must not be enough to write money data.""" + + @pytest.mark.parametrize("state", ["enroll", "reissue"]) + def test_a_shared_key_holder_is_refused(self, client, state): + # Both states mean "presented the shared key". Every worker holds it, so + # accepting it here would let any of them write a history for any + # client_id they cared to name -- including 'server'. + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value=state), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "2026-01-01"}]) + assert resp.status_code == 403 + upsert.assert_not_awaited(), "a shared-key holder wrote earnings" + + def test_a_confirmed_worker_is_accepted(self, client): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + ): + resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "2026-01-01"}]) + assert resp.status_code == 200 + + def test_the_refusal_says_how_to_proceed(self, client): + """ "403" alone leaves a client with no way forward.""" + with patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="enroll"): + resp = _import(client) + assert "heartbeat" in resp.json()["detail"].lower() + + def test_a_bad_key_is_still_rejected_by_the_shared_auth(self, client): + """The endpoint must not have its own weaker auth path.""" + from fastapi import HTTPException + + with patch( + "app.main._authenticate_worker_heartbeat", + new_callable=AsyncMock, + side_effect=HTTPException(status_code=401, detail="bad key"), + ): + resp = _import(client) + assert resp.status_code == 401 + + +class TestTheSourceComesFromAuthentication: + def test_readings_are_stored_under_the_authenticated_client(self, client): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, cid="desktop-a", readings=[{"slug": "grass", "balance": 2.0, "date": "2026-01-01"}]) + assert resp.status_code == 200 + assert upsert.await_args.kwargs["source"] == "desktop-a" + assert resp.json()["source"] == "desktop-a" + + def test_a_body_supplied_source_cannot_override_it(self, client): + """The model carries no source field, so a hostile body is inert. + + Asserted by SENDING one: a field the model ignores is the defence, and + a future model change that started honouring it would fail here. + """ + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = client.post( + "/api/workers/earnings-import", + json={ + "client_id": "desktop-a", + "source": "server", + "readings": [{"slug": "grass", "balance": 2.0, "date": "2026-01-01", "source": "server"}], + }, + headers={"Authorization": "Bearer k"}, + ) + assert resp.status_code == 200 + assert upsert.await_args.kwargs["source"] == "desktop-a", "a client overwrote the server's own series" + + def test_a_missing_client_id_is_refused(self, client): + resp = _import(client, cid=" ") + assert resp.status_code == 400 + + +class TestOnlyKnownPlatformsAreStored: + def test_an_unknown_slug_is_skipped_not_stored(self, client): + """It would create a platform the catalog cannot name, render, or ever + collect for again.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import( + client, + readings=[ + {"slug": "grass", "balance": 1.0, "date": "2026-01-01"}, + {"slug": "not-a-real-service", "balance": 9.0, "date": "2026-01-01"}, + ], + ) + body = resp.json() + assert body["imported"] == 1 + assert body["skipped"] == ["not-a-real-service"] + assert upsert.await_count == 1 + + def test_the_skips_are_reported_rather_than_swallowed(self, client): + """A silent drop looks identical to a successful import.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + ): + resp = _import(client, readings=[{"slug": "nope", "balance": 1.0, "date": "2026-01-01"}]) + assert resp.json()["skipped"] == ["nope"] + assert resp.json()["imported"] == 0 + + def test_a_blank_slug_is_skipped(self, client): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=[{"slug": " ", "balance": 1.0, "date": "2026-01-01"}]) + assert resp.json()["imported"] == 0 + upsert.assert_not_awaited() + + +class TestTheReadingsSurviveIntact: + def test_currency_and_rate_are_passed_through(self, client): + """Dropping the rate would make a non-USD balance unpriceable later — + the reason the column exists at all.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + _import( + client, + readings=[ + {"slug": "mysterium", "balance": 3.0, "date": "2026-01-01", "currency": "myst", "fx_rate_usd": 0.4} + ], + ) + kwargs = upsert.await_args.kwargs + assert kwargs["currency"] == "MYST", "currency was not normalised" + assert kwargs["fx_rate_usd"] == 0.4 + + def test_an_absent_rate_stays_none_rather_than_becoming_zero(self, client): + """None is UNKNOWN. A 0.0 rate would price the whole reading at nothing.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + _import(client, readings=[{"slug": "grass", "balance": 3.0, "date": "2026-01-01"}]) + assert upsert.await_args.kwargs["fx_rate_usd"] is None + + def test_an_empty_import_is_accepted_and_writes_nothing(self, client): + """A client with no history to push is a normal case, not an error.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=[]) + assert resp.status_code == 200 + assert resp.json()["imported"] == 0 + upsert.assert_not_awaited() From 348287084f3f1547040e868db3913cf3a3cd82d7 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:28:44 +0200 Subject: [PATCH 2/5] fix(fleet): bound the import body and validate the reading date Two ways the new endpoint could be handed input it would store without complaint. The date was free text. Both delta readers ORDER BY it, so a reading dated 2026-1-2 or 01/02/2026 sorts into the wrong place in its own series and the readings either side then difference against the wrong neighbour. It fails silently, only for the client that sent it, and only in the earned figure -- never in the balance the dashboard shows. It is now required to be YYYY-MM-DD and a real calendar day, so 2026-02-30 is refused rather than stored. The readings list was unbounded. One authenticated client could hand the server an arbitrarily large body to parse and then write row by row; a single compromised worker is enough. Capped at 2000, comfortably above an honest import (the server keeps 400 days and a client chunks at 1000). Validation runs before authentication, so a malformed body cannot be used to probe which client ids exist. Refs: CashPilot-Desktop-xjr --- app/main.py | 30 +++++++++++- docs/fleet.md | 6 +++ tests/test_beads_batch_63.py | 93 ++++++++++++++++++++++++++++++++++++ 3 files changed, 127 insertions(+), 2 deletions(-) diff --git a/app/main.py b/app/main.py index 28b2f9f5..fbf107d0 100644 --- a/app/main.py +++ b/app/main.py @@ -25,7 +25,7 @@ from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel +from pydantic import BaseModel, Field, field_validator from starlette.middleware.base import BaseHTTPMiddleware from app import ( @@ -3713,6 +3713,13 @@ async def _earnings_for_worker(body: WorkerHeartbeat, days: int = 30) -> dict[st } +#: The exact shape ``upsert_earnings`` writes and both delta readers ORDER BY. +#: A date in any other shape would sort into the wrong place in its own series, +#: so the readings either side of it difference against the wrong neighbour -- +#: silently, and only for the client that sent it. +_ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + class EarningsReading(BaseModel): """One historical balance reading from a paired client.""" @@ -3722,6 +3729,19 @@ class EarningsReading(BaseModel): currency: str = "USD" fx_rate_usd: float | None = None + @field_validator("date") + @classmethod + def _iso_date(cls, value: str) -> str: + value = (value or "").strip() + if not _ISO_DATE.match(value): + raise ValueError("date must be YYYY-MM-DD") + # Reject 2026-02-30 and friends: the pattern above only proves shape. + try: + datetime.strptime(value, "%Y-%m-%d") # noqa: DTZ007 -- a calendar date, not an instant + except ValueError as exc: + raise ValueError("date must be a real calendar date") from exc + return value + class EarningsImport(BaseModel): """A client pushing the history it collected before it was paired. @@ -3733,7 +3753,13 @@ class EarningsImport(BaseModel): """ client_id: str - readings: list[EarningsReading] = [] + #: Bounded so one authenticated client cannot hand the server an + #: arbitrarily large body to parse and then write row by row. 2000 is + #: comfortably above a real import -- the server keeps 400 days, and a + #: client chunks anything larger -- while staying a body the process can + #: hold. An unbounded list here is a denial of service that needs only one + #: compromised worker. + readings: list[EarningsReading] = Field(default_factory=list, max_length=2000) @app.post("/api/workers/earnings-import") diff --git a/docs/fleet.md b/docs/fleet.md index edb2974e..97e5f8a5 100644 --- a/docs/fleet.md +++ b/docs/fleet.md @@ -68,6 +68,12 @@ Three things about it are deliberate: Re-sending the same day updates it rather than appending, so a retried or repeated import is safe. +`date` must be `YYYY-MM-DD` and a real calendar day. It is not free text: both +delta readers order by it, so a differently shaped date sorts into the wrong +place in its own series and the readings either side then difference against the +wrong neighbour — silently, and only in the earned figure. A request may carry at +most **2000 readings**; send larger histories in chunks. + ## Setting Up the Fleet ### Main server (UI + local worker) diff --git a/tests/test_beads_batch_63.py b/tests/test_beads_batch_63.py index 1bbca4d2..24106bd9 100644 --- a/tests/test_beads_batch_63.py +++ b/tests/test_beads_batch_63.py @@ -200,3 +200,96 @@ def test_an_empty_import_is_accepted_and_writes_nothing(self, client): assert resp.status_code == 200 assert resp.json()["imported"] == 0 upsert.assert_not_awaited() + + +class TestTheDateIsAValidCalendarDay: + """Both delta readers ``ORDER BY ... date``, so the date is not free text. + + A reading dated ``01/02/2026`` or ``2026-1-2`` sorts into the wrong place in + its own series, and the readings either side of it then difference against + the wrong neighbour. It fails silently, only for the client that sent it, + and only in the earned figure — never in the balance the dashboard shows. + """ + + @pytest.mark.parametrize( + "bad", + [ + "01/02/2026", # a different convention entirely + "2026-1-2", # unpadded: sorts after "2026-11-01" + "2026-01-01T00:00:00Z", # an instant, not a day + "yesterday", + "", + "2026-02-30", # right shape, no such day + "2026-13-01", + ], + ) + def test_a_malformed_date_is_refused(self, client, bad): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": bad}]) + assert resp.status_code == 422, f"{bad!r} was accepted" + upsert.assert_not_awaited() + + @pytest.mark.parametrize("good", ["2026-01-01", "2024-02-29", "2026-12-31"]) + def test_a_real_day_is_accepted(self, client, good): + """A validator that rejects everything is as useless as none at all — + 2024-02-29 exists and must survive.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": good}]) + assert resp.status_code == 200 + assert upsert.await_args.kwargs["date"] == good + + def test_the_date_is_refused_before_authentication_is_even_checked(self, client): + """Body validation runs first, so a malformed date cannot be used to + probe which client ids exist.""" + with patch( + "app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok" + ) as authenticate: + _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "nope"}]) + authenticate.assert_not_awaited() + + +class TestTheBodyIsBounded: + """One authenticated client must not be able to hand the server an + arbitrarily large body to parse and then write row by row.""" + + @staticmethod + def _readings(n: int) -> list[dict]: + return [{"slug": "grass", "balance": float(i), "date": "2026-01-01"} for i in range(n)] + + def test_an_oversized_import_is_refused(self, client): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=self._readings(2001)) + assert resp.status_code == 422 + upsert.assert_not_awaited(), "the server wrote rows from a body it should have refused" + + def test_a_real_sized_import_still_fits(self): + """The cap must clear an honest import, or clients can never pair. + + 400 days of retention against the whole catalog is the worst case a + client would ever chunk against, so the bound is checked against the + REAL catalog size rather than a guess. + """ + from app import catalog + + chunk = 1000 # what a client sends per request + assert chunk <= 2000 + assert len(catalog.get_services()) >= 40, "catalog too small for this to mean anything" + + def test_a_thousand_readings_are_accepted(self, client): + """The chunk size a client actually uses, exercised end to end.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import(client, readings=self._readings(1000)) + assert resp.status_code == 200 + assert upsert.await_count == 1000 From 730b85df0c59b7894364add854a18b6e4dba8ec2 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:40:13 +0200 Subject: [PATCH 3/5] fix(compose): pin the examples to the 1.15 series v1.15.0 released and the example compose files still pinned 1.14, so anyone following the quickstart deployed a series behind. The pin test caught it -- it exists because a stale pin is what gave issue #188 a version with a first-run bug that had been fixed for months. Unrelated to this branch's change, but it fails CI on every branch until it is fixed. --- docker-compose.fleet.yml | 6 +++--- docker-compose.yml | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docker-compose.fleet.yml b/docker-compose.fleet.yml index 3f434d8a..55941241 100644 --- a/docker-compose.fleet.yml +++ b/docker-compose.fleet.yml @@ -10,7 +10,7 @@ # === Deploy on main server (e.g., server-a) === services: cashpilot-ui: - image: drumsergio/cashpilot:1.14 + image: drumsergio/cashpilot:1.15 pull_policy: always container_name: cashpilot-ui # Loopback by default (the UI commands the Docker-socket worker). Set @@ -38,7 +38,7 @@ services: - /tmp:size=64M cashpilot-worker: - image: drumsergio/cashpilot-worker:1.14 + image: drumsergio/cashpilot-worker:1.15 pull_policy: always container_name: cashpilot-worker # The worker exposes a Docker-socket-backed API (deploy/stop/remove any @@ -105,7 +105,7 @@ services: # # services: # cashpilot-worker: -# image: drumsergio/cashpilot-worker:1.14 +# image: drumsergio/cashpilot-worker:1.15 # pull_policy: always # container_name: cashpilot-worker # # Docker-socket-backed API = full host control. Bind a PRIVATE/VPN interface diff --git a/docker-compose.yml b/docker-compose.yml index 525d2519..79c7b4d7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,7 +21,7 @@ services: cashpilot-ui: - image: drumsergio/cashpilot:1.14 + image: drumsergio/cashpilot:1.15 pull_policy: always container_name: cashpilot-ui # Published on loopback by default: the dashboard can command the Docker-socket @@ -51,7 +51,7 @@ services: - /tmp:size=64M cashpilot-worker: - image: drumsergio/cashpilot-worker:1.14 + image: drumsergio/cashpilot-worker:1.15 pull_policy: always container_name: cashpilot-worker expose: From f938d6aabd91c7f833ab7aee7383ffbc1c30023f Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Thu, 6 Aug 2026 04:28:38 +0200 Subject: [PATCH 4/5] fix(fleet): a NaN balance was accepted, and rejecting one was a 500 Found by re-reading the endpoint rather than from a report. JSON has no NaN or Infinity. Python's parser accepts them anyway, so {"balance": NaN} was stored verbatim. One such reading poisons every delta taken from that series -- NaN - x is NaN, and every comparison against it is False, so the clamp silently misbehaves -- the account total becomes NaN, and serialising that back out emits a bare NaN that JSON.parse rejects. A single bad reading from one client breaks the dashboard for everyone. Both float fields now refuse non-finite values. That exposed a second problem underneath: FastAPI's 422 body echoes the offending input, so the REJECTION could not be serialised either and the client got a 500 for what is squarely a bad request. A RequestValidationError handler now renders non-finite floats as their names -- keeping the message diagnostic rather than dropping the field -- and fixes the whole class instead of the one endpoint that takes a float today. My first version of that handler broke every OTHER validation error: a custom validator's error carries the raised ValueError OBJECT in ctx, which is not serialisable, and skipping jsonable_encoder turned each one into a 500. Caught by the date tests, and now pinned by its own regression test. Also: the skipped list is deduplicated. A client pushing 400 days of a platform this server does not know got the same name back 400 times -- a response that grows with the request, echoing client-supplied strings, and saying nothing the set does not. A negative control showed the sanitiser's bool guard was dead (bool subclasses int, not float), so it is gone along with the comment that justified it incorrectly. Refs: CashPilot-Desktop-xjr --- app/main.py | 76 ++++++++++++-- docs/fleet.md | 9 ++ tests/test_beads_batch_63.py | 188 +++++++++++++++++++++++++++++++++++ 3 files changed, 267 insertions(+), 6 deletions(-) diff --git a/app/main.py b/app/main.py index fbf107d0..9f69fac1 100644 --- a/app/main.py +++ b/app/main.py @@ -11,6 +11,7 @@ import hmac import json import logging +import math import os import re import secrets @@ -23,6 +24,8 @@ from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import FastAPI, HTTPException, Request +from fastapi.encoders import jsonable_encoder +from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field, field_validator @@ -984,6 +987,51 @@ async def dispatch(self, request, call_next): app.add_middleware(_SecurityHeadersMiddleware) + +@app.exception_handler(RequestValidationError) +async def _validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse: + """FastAPI's 422 body echoes the offending input, and some inputs cannot be + encoded — so the rejection itself became a 500. + + JSON has no ``NaN`` or ``Infinity``. Python's parser accepts them anyway, so + a client can put one in a float field; pydantic correctly rejects it, and + then the default handler tries to serialise ``{"input": nan}`` and Starlette's + encoder raises "Out of range float values are not JSON compliant". The client + gets an opaque 500 for what is squarely a bad request, and the log fills with + a traceback that names the encoder rather than the cause. + + Non-finite floats are therefore rendered as their names. That keeps the + message diagnostic ("we saw NaN") instead of dropping the field, and it fixes + the whole class rather than the one endpoint that happens to take a float + today. Every other error is passed through byte-for-byte, so the response + shape callers already parse is unchanged. + """ + # jsonable_encoder FIRST, exactly as FastAPI's own handler does: a custom + # validator's error carries the raised ValueError OBJECT in ctx, which is not + # serialisable either. Encoding then sanitising fixes both without changing + # the body for any error that was already fine. + return JSONResponse(status_code=422, content={"detail": _json_safe(jsonable_encoder(exc.errors()))}) + + +def _json_safe(value: Any) -> Any: + """Recursively replace values the JSON encoder refuses — currently only + non-finite floats. + + Deliberately narrow. ``bool`` needs no special case: it subclasses ``int``, + not ``float``, so the check below never sees it. An earlier version guarded + for it anyway, with a comment that was simply wrong; a negative control + showed removing the guard changed nothing, so it went. + (``test_booleans_survive_as_booleans`` still pins the guarantee.) + """ + if isinstance(value, dict): + return {k: _json_safe(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_json_safe(v) for v in value] + if isinstance(value, float) and not math.isfinite(value): + return repr(value) # "nan", "inf", "-inf" + return value + + # Static files app.mount("/static", StaticFiles(directory="app/static"), name="static") @@ -3724,10 +3772,17 @@ class EarningsReading(BaseModel): """One historical balance reading from a paired client.""" slug: str - balance: float + #: allow_inf_nan=False because JSON's `NaN` and `Infinity` -- which Python's + #: parser accepts even though the spec does not -- are otherwise stored + #: verbatim. One NaN balance poisons every delta taken from that series + #: (NaN - x is NaN, and every comparison against it is False, so the clamp + #: silently misbehaves), the account total becomes NaN, and serialising that + #: back out emits a bare `NaN` that JSON.parse rejects. So a single bad + #: reading from one client breaks the dashboard for everyone. + balance: float = Field(allow_inf_nan=False) date: str currency: str = "USD" - fx_rate_usd: float | None = None + fx_rate_usd: float | None = Field(default=None, allow_inf_nan=False) @field_validator("date") @classmethod @@ -3798,13 +3853,17 @@ async def api_worker_earnings_import(request: Request, body: EarningsImport) -> known = {s["slug"] for s in catalog.get_services()} written = 0 - skipped: list[str] = [] + # DISTINCT slugs, not one entry per skipped reading. A client pushing 400 + # days of a platform this server does not know would otherwise get the same + # name back 400 times -- a response that grows with the request, echoing + # client-supplied strings, and tells the reader nothing the set does not. + skipped: set[str] = set() for reading in body.readings: slug = (reading.slug or "").strip() # An unknown slug is dropped rather than stored: it would create a # platform the catalog cannot render, name or ever collect for again. if not slug or slug not in known: - skipped.append(slug or "(blank)") + skipped.add(slug or "(blank)") continue await database.upsert_earnings( platform=slug, @@ -3816,8 +3875,13 @@ async def api_worker_earnings_import(request: Request, body: EarningsImport) -> ) written += 1 - logger.info("Imported %d earnings reading(s) from worker '%s' (%d skipped)", written, cid, len(skipped)) - return {"status": "ok", "imported": written, "skipped": skipped, "source": cid} + logger.info( + "Imported %d earnings reading(s) from worker '%s' (%d unknown platform(s) skipped)", + written, + cid, + len(skipped), + ) + return {"status": "ok", "imported": written, "skipped": sorted(skipped), "source": cid} @app.post("/api/workers/heartbeat") diff --git a/docs/fleet.md b/docs/fleet.md index 97e5f8a5..475bcbb6 100644 --- a/docs/fleet.md +++ b/docs/fleet.md @@ -74,6 +74,15 @@ place in its own series and the readings either side then difference against the wrong neighbour — silently, and only in the earned figure. A request may carry at most **2000 readings**; send larger histories in chunks. +`balance` and `fx_rate_usd` must be finite. JSON has no `NaN` or `Infinity`, but +Python's parser accepts them, and one `NaN` balance poisons every earned figure +taken from that series — every comparison against it is false, so the clamp +misbehaves silently and the account total becomes `NaN`. They are rejected with +`422`. + +Unknown slugs come back in `skipped` as a **distinct, sorted** list, so a year of +one unrecognised platform is reported once rather than 400 times. + ## Setting Up the Fleet ### Main server (UI + local worker) diff --git a/tests/test_beads_batch_63.py b/tests/test_beads_batch_63.py index 24106bd9..b004b2a7 100644 --- a/tests/test_beads_batch_63.py +++ b/tests/test_beads_batch_63.py @@ -202,6 +202,194 @@ def test_an_empty_import_is_accepted_and_writes_nothing(self, client): upsert.assert_not_awaited() +class TestABalanceMustBeARealNumber: + """JSON has no NaN or Infinity. Python's parser accepts them anyway. + + Found by re-reading the endpoint rather than by a report: ``{"balance": NaN}`` + was accepted and stored verbatim. One such reading poisons every delta taken + from that series — ``NaN - x`` is ``NaN``, and every comparison against it is + False, so the clamp silently misbehaves — the account total becomes ``NaN``, + and serialising that back out emits a bare ``NaN`` that ``JSON.parse`` + rejects. A single bad reading from one client breaks the dashboard for + everyone. + """ + + @staticmethod + def _raw(client, body: str): + return client.post( + "/api/workers/earnings-import", + content=body, + headers={"Authorization": "Bearer k", "Content-Type": "application/json"}, + ) + + @pytest.mark.parametrize("literal", ["NaN", "Infinity", "-Infinity"]) + def test_a_non_finite_balance_is_refused(self, client, literal): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = self._raw( + client, + f'{{"client_id":"d","readings":[{{"slug":"grass","balance":{literal},"date":"2026-01-01"}}]}}', + ) + assert resp.status_code == 422, f"{literal} balance was accepted" + upsert.assert_not_awaited() + + @pytest.mark.parametrize("literal", ["NaN", "Infinity"]) + def test_a_non_finite_rate_is_refused(self, client, literal): + """An infinite rate prices the balance at infinity, which is worse than + pricing it at nothing.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = self._raw( + client, + f'{{"client_id":"d","readings":[{{"slug":"grass","balance":1.0,"date":"2026-01-01",' + f'"fx_rate_usd":{literal}}}]}}', + ) + assert resp.status_code == 422, f"{literal} rate was accepted" + upsert.assert_not_awaited() + + def test_ordinary_numbers_still_pass(self, client): + """A guard that rejects everything is as useless as none at all. Zero and + a negative are both legitimate: a provider balance can be zero, and a + clawback can take it negative.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + ): + resp = _import( + client, + readings=[ + {"slug": "grass", "balance": 0.0, "date": "2026-01-01"}, + {"slug": "honeygain", "balance": -1.25, "date": "2026-01-02", "fx_rate_usd": 0.0001}, + {"slug": "storj", "balance": 1e9, "date": "2026-01-03"}, + ], + ) + assert resp.status_code == 200 + assert upsert.await_count == 3 + + +class TestTheRejectionItselfCanBeSerialised: + """FastAPI's 422 body echoes the offending input, and some inputs cannot be + encoded — so the rejection became a 500. + + Both halves were found the same way: by driving the real request rather than + reading the code. ``NaN`` made the default handler raise "Out of range float + values are not JSON compliant"; my FIRST attempt at a handler then broke every + OTHER validation error, because a custom validator's error carries the raised + ``ValueError`` object in ``ctx`` and that is not serialisable either. The + second test below is the regression that caught it. + """ + + def test_a_non_finite_value_yields_a_bad_request_not_a_server_error(self, client): + with patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"): + resp = client.post( + "/api/workers/earnings-import", + content='{"client_id":"d","readings":[{"slug":"grass","balance":NaN,"date":"2026-01-01"}]}', + headers={"Authorization": "Bearer k", "Content-Type": "application/json"}, + ) + assert resp.status_code == 422 + resp.json() # must parse: a bare NaN in the body would make this raise + + def test_an_ordinary_validation_error_is_unchanged(self, client): + """The regression. A custom validator raises ValueError, which lands in + `ctx` as an OBJECT; a handler that skips jsonable_encoder turns every one + of those into a 500.""" + with patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"): + resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "yesterday"}]) + assert resp.status_code == 422 + detail = resp.json()["detail"] + assert isinstance(detail, list) and detail, detail + assert "date" in str(detail), "the error no longer says which field was wrong" + + def test_a_missing_field_still_reports_normally(self, client): + """The plainest possible validation error, to catch a handler that only + works for the exotic cases it was written for.""" + resp = client.post( + "/api/workers/earnings-import", + json={"readings": []}, + headers={"Authorization": "Bearer k"}, + ) + assert resp.status_code == 422 + assert "client_id" in str(resp.json()["detail"]) + + +class TestJsonSafe: + """The sanitiser on its own, so its edge cases are not only reachable through + a request that happens to hit them.""" + + @staticmethod + def _safe(value): + from app.main import _json_safe + + return _json_safe(value) + + def test_non_finite_floats_become_their_names(self): + assert self._safe(float("nan")) == "nan" + assert self._safe(float("inf")) == "inf" + assert self._safe(float("-inf")) == "-inf" + + def test_ordinary_numbers_are_untouched(self): + """A sanitiser that rewrites healthy values corrupts every error body it + touches.""" + for value in (0.0, -1.5, 1e300, 42): + assert self._safe(value) == value + + def test_booleans_survive_as_booleans(self): + """bool is a subclass of int. A guard written for numbers that forgets + that turns True into 1 in every error body.""" + assert self._safe(True) is True + assert self._safe(False) is False + + def test_it_recurses_into_nested_structures(self): + """The value is buried at errors()[0]["input"], inside a list of dicts.""" + got = self._safe([{"input": float("nan"), "loc": ["body", 0, "balance"], "ok": 1.5}]) + assert got == [{"input": "nan", "loc": ["body", 0, "balance"], "ok": 1.5}] + + def test_strings_and_none_pass_through(self): + assert self._safe(None) is None + assert self._safe("nan") == "nan" + + +class TestTheSkipReportIsBounded: + def test_a_repeated_unknown_slug_is_reported_once(self, client): + """A client pushing 400 days of a platform this server does not know + would otherwise get the same name back 400 times: a response that grows + with the request, echoing client-supplied strings, and saying nothing the + set does not.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + ): + resp = _import( + client, + readings=[ + {"slug": "not-a-real-service", "balance": float(i), "date": f"2026-01-{i + 1:02d}"} + for i in range(20) + ], + ) + assert resp.json()["skipped"] == ["not-a-real-service"] + assert resp.json()["imported"] == 0 + + def test_several_distinct_unknowns_are_all_reported(self, client): + """Deduplicating must not become "report only the first".""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + ): + resp = _import( + client, + readings=[ + {"slug": "gone-one", "balance": 1.0, "date": "2026-01-01"}, + {"slug": "gone-two", "balance": 1.0, "date": "2026-01-01"}, + {"slug": "gone-one", "balance": 2.0, "date": "2026-01-02"}, + ], + ) + assert resp.json()["skipped"] == ["gone-one", "gone-two"], "a distinct unknown platform was swallowed" + + class TestTheDateIsAValidCalendarDay: """Both delta readers ``ORDER BY ... date``, so the date is not free text. From 7d6bc4005040462b42d7d49b6a6569e85347286d Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:15:29 +0200 Subject: [PATCH 5/5] perf(fleet): import in one transaction, and roll back when it fails A thousand-reading import committed a thousand times. Every commit is an fsync that takes SQLite's write lock, so one import serialised a thousand disk syncs against this server's own collector and request latency tracked sync cost rather than row count. upsert_earnings_many does the same upsert with executemany and one commit. Only half the reported cause was real, and the distinction matters for anyone reading this later: _get_db hands out a borrowed handle on a SHARED per-loop connection whose close() is a documented no-op, so the loop was never opening and closing a thousand connections. It was committing a thousand times. Batching then introduced a bug of its own, which the tests caught: the connection is shared and outlives the request, so a failed batch left an abandoned transaction holding the write lock and the NEXT write blocked for twelve seconds before timing out. It now rolls back. Two more from the same review: * docs/fleet.md gains the Authorization header and every status the endpoint can answer with. A reader integrating against that page alone could not previously construct a valid request. * Two assertions were tuple expressions -- `mock.assert_not_awaited(), "why"` builds a tuple and discards the message, so a red build showed the mock's generic text instead of the reason. Fixed here and in test_optional_runtime.py. The review said ruff's B018 catches that pattern when bugbear is enabled. Bugbear IS enabled here and B018 is not ignored, and ruff 0.15.14 passes it clean -- checked against a minimal probe rather than assumed. So nothing in CI would have caught a recurrence, and there is now an AST-based test that does. Structural, not a string search, because a string search would match the pattern inside its own docstring. One negative control PASSED, which meant the test was wrong: the transactionality test raised while BUILDING the rows, before any SQL ran, so it proved only that the row build validates first and it passed against a writer that committed after every row. It now fails inside the statement, and both it and the wedged-connection test fail under their controls. Reported by CodeRabbit on PR #256. --- app/database.py | 74 ++++++++ app/main.py | 28 ++- docs/fleet.md | 16 +- tests/test_beads_batch_63.py | 315 +++++++++++++++++++++++++++++---- tests/test_optional_runtime.py | 2 +- 5 files changed, 388 insertions(+), 47 deletions(-) diff --git a/app/database.py b/app/database.py index a49772ea..2d9a00af 100644 --- a/app/database.py +++ b/app/database.py @@ -8,10 +8,12 @@ from __future__ import annotations import asyncio +import contextlib import json import logging import math import os +from collections.abc import Mapping, Sequence from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -921,6 +923,78 @@ async def upsert_earnings( await db.close() +async def upsert_earnings_many(readings: Sequence[Mapping[str, Any]]) -> int: + """Upsert many readings in ONE transaction. Returns how many were written. + + Same statement and same conflict rules as :func:`upsert_earnings` — this is + purely about how often the work is committed. + + WHY IT EXISTS + ------------- + A client importing its pre-pairing history sends up to a thousand readings + per request, and calling ``upsert_earnings`` in a loop commits once per row. + Every commit is an fsync, and every one of them takes SQLite's write lock — + so a single import serialised a thousand disk syncs against the server's own + collector, and request latency tracked disk sync cost rather than row count. + + (The connection is NOT the problem, despite appearances: ``_get_db`` hands + out a borrowed handle on a shared per-loop connection whose ``close()`` is a + documented no-op, so the loop was never opening and closing a thousand + connections. It was committing a thousand times.) + + ONE TRANSACTION ALSO MEANS ALL-OR-NOTHING, which is the behaviour to want + here: a failure part-way through leaves the caller's history exactly as it + was rather than half-applied, and the import is idempotent so retrying costs + a round trip. + """ + rows = [ + ( + r["platform"], + float(r["balance"]), + (r.get("currency") or "USD"), + r.get("date") or datetime.now(UTC).strftime("%Y-%m-%d"), + r.get("fx_rate_usd"), + (r.get("source") or "server"), + ) + for r in readings + ] + if not rows: + # No statement, no commit. An empty import is a normal case, and taking + # the write lock to do nothing would still block the collector. + return 0 + + db = await _get_db() + try: + try: + await db.executemany( + """ + INSERT INTO earnings (platform, balance, currency, date, fx_rate_usd, source) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(platform, source, date) DO UPDATE SET + balance = excluded.balance, + currency = excluded.currency, + fx_rate_usd = COALESCE(excluded.fx_rate_usd, earnings.fx_rate_usd), + created_at = datetime('now') + WHERE earnings.balance != excluded.balance + OR earnings.fx_rate_usd IS NULL + """, + rows, + ) + await db.commit() + except Exception: + # ROLL BACK, do not merely propagate. The connection is SHARED and + # outlives the request, so an abandoned transaction keeps SQLite's + # write lock and every later write on this loop -- including this + # server's own collector -- blocks until it times out. Measured: the + # next write took twelve seconds before this was added. + with contextlib.suppress(Exception): + await db.rollback() + raise + finally: + await db.close() + return len(rows) + + async def get_earnings_summary() -> list[dict[str, Any]]: """Return the latest balance for each platform.""" db = await _get_db() diff --git a/app/main.py b/app/main.py index 9f69fac1..85e28979 100644 --- a/app/main.py +++ b/app/main.py @@ -3852,12 +3852,12 @@ async def api_worker_earnings_import(request: Request, body: EarningsImport) -> ) known = {s["slug"] for s in catalog.get_services()} - written = 0 # DISTINCT slugs, not one entry per skipped reading. A client pushing 400 # days of a platform this server does not know would otherwise get the same # name back 400 times -- a response that grows with the request, echoing # client-supplied strings, and tells the reader nothing the set does not. skipped: set[str] = set() + rows: list[dict[str, Any]] = [] for reading in body.readings: slug = (reading.slug or "").strip() # An unknown slug is dropped rather than stored: it would create a @@ -3865,15 +3865,25 @@ async def api_worker_earnings_import(request: Request, body: EarningsImport) -> if not slug or slug not in known: skipped.add(slug or "(blank)") continue - await database.upsert_earnings( - platform=slug, - balance=float(reading.balance), - currency=(reading.currency or "USD").upper(), - date=reading.date, - fx_rate_usd=reading.fx_rate_usd, - source=cid, + rows.append( + { + "platform": slug, + "balance": float(reading.balance), + "currency": (reading.currency or "USD").upper(), + "date": reading.date, + "fx_rate_usd": reading.fx_rate_usd, + "source": cid, + } ) - written += 1 + + # ONE transaction for the whole batch. Writing row by row committed once per + # reading, and every commit is an fsync that takes SQLite's write lock -- so + # a thousand-reading import serialised a thousand disk syncs against this + # server's own collector, and latency tracked sync cost rather than row + # count. It also makes the import all-or-nothing: a failure part-way leaves + # the client's history exactly as it was rather than half-applied, and the + # import is idempotent so retrying costs a round trip. (CodeRabbit, PR #256.) + written = await database.upsert_earnings_many(rows) logger.info( "Imported %d earnings reading(s) from worker '%s' (%d unknown platform(s) skipped)", diff --git a/docs/fleet.md b/docs/fleet.md index 475bcbb6..ba07f02c 100644 --- a/docs/fleet.md +++ b/docs/fleet.md @@ -41,7 +41,11 @@ provider account by itself — CashPilot Desktop, typically, before it was paire hand that history to the UI, so the fleet view shows the complete picture rather than starting from the day of pairing. -```json +```http +POST /api/workers/earnings-import +Authorization: Bearer +Content-Type: application/json + { "client_id": "desktop-macbook", "readings": [ @@ -51,6 +55,16 @@ than starting from the day of pairing. } ``` +It answers: + +| Status | Meaning | +|---|---| +| `200` | `{"status": "ok", "imported": N, "skipped": [...], "source": ""}` | +| `400` | `client_id` was missing or blank | +| `401` | the key is wrong or revoked | +| `403` | this worker is not fully enrolled yet — send a heartbeat first, then retry | +| `422` | the body was rejected: a date that is not a real `YYYY-MM-DD` day, a non-finite `balance` or `fx_rate_usd`, or more than 2000 readings | + Three things about it are deliberate: - **Each client's readings are stored under their own source, not merged with the diff --git a/tests/test_beads_batch_63.py b/tests/test_beads_batch_63.py index b004b2a7..14e799c1 100644 --- a/tests/test_beads_batch_63.py +++ b/tests/test_beads_batch_63.py @@ -18,11 +18,13 @@ from __future__ import annotations +import asyncio from unittest.mock import AsyncMock, patch import pytest from fastapi.testclient import TestClient +from app import database from app.main import app @@ -34,6 +36,24 @@ def client(): return TestClient(app, raise_server_exceptions=False) +async def _count(rows): + """Stand in for the real batched write, returning what it returns.""" + return len(rows) + + +def _rows(upsert) -> list[dict]: + """The rows handed to the ONE batched write, or [] if it never ran. + + The endpoint used to call upsert_earnings once per reading; it now hands the + whole batch to upsert_earnings_many in a single transaction, so "did it write + this?" is a question about the rows, not about the call count. + """ + if not upsert.await_args_list: + return [] + assert len(upsert.await_args_list) == 1, "the batch was split into several transactions" + return list(upsert.await_args_list[0].args[0]) + + def _import(client, *, cid="desktop-a", readings=None, token="worker-own-key"): return client.post( "/api/workers/earnings-import", @@ -52,16 +72,16 @@ def test_a_shared_key_holder_is_refused(self, client, state): # client_id they cared to name -- including 'server'. with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value=state), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "2026-01-01"}]) assert resp.status_code == 403 - upsert.assert_not_awaited(), "a shared-key holder wrote earnings" + assert _rows(upsert) == [], "a shared-key holder wrote earnings" def test_a_confirmed_worker_is_accepted(self, client): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count), ): resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "2026-01-01"}]) assert resp.status_code == 200 @@ -89,11 +109,11 @@ class TestTheSourceComesFromAuthentication: def test_readings_are_stored_under_the_authenticated_client(self, client): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, cid="desktop-a", readings=[{"slug": "grass", "balance": 2.0, "date": "2026-01-01"}]) assert resp.status_code == 200 - assert upsert.await_args.kwargs["source"] == "desktop-a" + assert _rows(upsert)[0]["source"] == "desktop-a" assert resp.json()["source"] == "desktop-a" def test_a_body_supplied_source_cannot_override_it(self, client): @@ -104,7 +124,7 @@ def test_a_body_supplied_source_cannot_override_it(self, client): """ with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = client.post( "/api/workers/earnings-import", @@ -116,7 +136,7 @@ def test_a_body_supplied_source_cannot_override_it(self, client): headers={"Authorization": "Bearer k"}, ) assert resp.status_code == 200 - assert upsert.await_args.kwargs["source"] == "desktop-a", "a client overwrote the server's own series" + assert _rows(upsert)[0]["source"] == "desktop-a", "a client overwrote the server's own series" def test_a_missing_client_id_is_refused(self, client): resp = _import(client, cid=" ") @@ -129,7 +149,7 @@ def test_an_unknown_slug_is_skipped_not_stored(self, client): collect for again.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import( client, @@ -141,13 +161,13 @@ def test_an_unknown_slug_is_skipped_not_stored(self, client): body = resp.json() assert body["imported"] == 1 assert body["skipped"] == ["not-a-real-service"] - assert upsert.await_count == 1 + assert len(_rows(upsert)) == 1 def test_the_skips_are_reported_rather_than_swallowed(self, client): """A silent drop looks identical to a successful import.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count), ): resp = _import(client, readings=[{"slug": "nope", "balance": 1.0, "date": "2026-01-01"}]) assert resp.json()["skipped"] == ["nope"] @@ -156,11 +176,11 @@ def test_the_skips_are_reported_rather_than_swallowed(self, client): def test_a_blank_slug_is_skipped(self, client): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=[{"slug": " ", "balance": 1.0, "date": "2026-01-01"}]) assert resp.json()["imported"] == 0 - upsert.assert_not_awaited() + assert _rows(upsert) == [] class TestTheReadingsSurviveIntact: @@ -169,7 +189,7 @@ def test_currency_and_rate_are_passed_through(self, client): the reason the column exists at all.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): _import( client, @@ -177,29 +197,29 @@ def test_currency_and_rate_are_passed_through(self, client): {"slug": "mysterium", "balance": 3.0, "date": "2026-01-01", "currency": "myst", "fx_rate_usd": 0.4} ], ) - kwargs = upsert.await_args.kwargs - assert kwargs["currency"] == "MYST", "currency was not normalised" - assert kwargs["fx_rate_usd"] == 0.4 + row = _rows(upsert)[0] + assert row["currency"] == "MYST", "currency was not normalised" + assert row["fx_rate_usd"] == 0.4 def test_an_absent_rate_stays_none_rather_than_becoming_zero(self, client): """None is UNKNOWN. A 0.0 rate would price the whole reading at nothing.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): _import(client, readings=[{"slug": "grass", "balance": 3.0, "date": "2026-01-01"}]) - assert upsert.await_args.kwargs["fx_rate_usd"] is None + assert _rows(upsert)[0]["fx_rate_usd"] is None def test_an_empty_import_is_accepted_and_writes_nothing(self, client): """A client with no history to push is a normal case, not an error.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=[]) assert resp.status_code == 200 assert resp.json()["imported"] == 0 - upsert.assert_not_awaited() + assert _rows(upsert) == [] class TestABalanceMustBeARealNumber: @@ -226,14 +246,14 @@ def _raw(client, body: str): def test_a_non_finite_balance_is_refused(self, client, literal): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = self._raw( client, f'{{"client_id":"d","readings":[{{"slug":"grass","balance":{literal},"date":"2026-01-01"}}]}}', ) assert resp.status_code == 422, f"{literal} balance was accepted" - upsert.assert_not_awaited() + assert _rows(upsert) == [] @pytest.mark.parametrize("literal", ["NaN", "Infinity"]) def test_a_non_finite_rate_is_refused(self, client, literal): @@ -241,7 +261,7 @@ def test_a_non_finite_rate_is_refused(self, client, literal): pricing it at nothing.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = self._raw( client, @@ -249,7 +269,7 @@ def test_a_non_finite_rate_is_refused(self, client, literal): f'"fx_rate_usd":{literal}}}]}}', ) assert resp.status_code == 422, f"{literal} rate was accepted" - upsert.assert_not_awaited() + assert _rows(upsert) == [] def test_ordinary_numbers_still_pass(self, client): """A guard that rejects everything is as useless as none at all. Zero and @@ -257,7 +277,7 @@ def test_ordinary_numbers_still_pass(self, client): clawback can take it negative.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import( client, @@ -268,7 +288,7 @@ def test_ordinary_numbers_still_pass(self, client): ], ) assert resp.status_code == 200 - assert upsert.await_count == 3 + assert len(_rows(upsert)) == 3 class TestTheRejectionItselfCanBeSerialised: @@ -361,7 +381,7 @@ def test_a_repeated_unknown_slug_is_reported_once(self, client): set does not.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count), ): resp = _import( client, @@ -377,7 +397,7 @@ def test_several_distinct_unknowns_are_all_reported(self, client): """Deduplicating must not become "report only the first".""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count), ): resp = _import( client, @@ -414,11 +434,11 @@ class TestTheDateIsAValidCalendarDay: def test_a_malformed_date_is_refused(self, client, bad): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": bad}]) assert resp.status_code == 422, f"{bad!r} was accepted" - upsert.assert_not_awaited() + assert _rows(upsert) == [] @pytest.mark.parametrize("good", ["2026-01-01", "2024-02-29", "2026-12-31"]) def test_a_real_day_is_accepted(self, client, good): @@ -426,11 +446,11 @@ def test_a_real_day_is_accepted(self, client, good): 2024-02-29 exists and must survive.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": good}]) assert resp.status_code == 200 - assert upsert.await_args.kwargs["date"] == good + assert _rows(upsert)[0]["date"] == good def test_the_date_is_refused_before_authentication_is_even_checked(self, client): """Body validation runs first, so a malformed date cannot be used to @@ -453,11 +473,11 @@ def _readings(n: int) -> list[dict]: def test_an_oversized_import_is_refused(self, client): with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=self._readings(2001)) assert resp.status_code == 422 - upsert.assert_not_awaited(), "the server wrote rows from a body it should have refused" + assert _rows(upsert) == [], "the server wrote rows from a body it should have refused" def test_a_real_sized_import_still_fits(self): """The cap must clear an honest import, or clients can never pair. @@ -476,8 +496,231 @@ def test_a_thousand_readings_are_accepted(self, client): """The chunk size a client actually uses, exercised end to end.""" with ( patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), - patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as upsert, + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, ): resp = _import(client, readings=self._readings(1000)) assert resp.status_code == 200 - assert upsert.await_count == 1000 + assert len(_rows(upsert)) == 1000 + + +class TestNoAssertionHidesItsOwnMessage: + """``mock.assert_not_awaited(), "why"`` is a TUPLE, not an assertion. + + Python evaluates the call, builds a two-element tuple with the string, and + throws it away. The mock still raises when the expectation fails, so the test + is not broken — but the message that explains WHY never appears, and the + reader of a red build gets ``AssertionError: Expected 'x' to not have been + awaited`` instead of "a shared-key holder wrote earnings". + + CodeRabbit found two of these in this file and said Ruff's ``B018`` catches + the pattern when bugbear is enabled. Bugbear IS enabled here (``B`` is in + ``select``, and ``B018`` is not ignored) and ruff 0.15.14 passes it clean — + checked against a minimal probe rather than assumed. So nothing in CI would + catch a recurrence, which is what this test is for. + + Structural, over the AST: a string search would match the pattern inside this + docstring, which is the specific way a test can end up matching its own prose. + """ + + def test_no_test_file_discards_a_message_into_a_tuple(self): + import ast + from pathlib import Path + + offenders: list[str] = [] + tests_dir = Path(__file__).resolve().parent + files = sorted(tests_dir.glob("test_*.py")) + # A silent empty scan would make this vacuously true forever. + assert len(files) >= 20, f"only found {len(files)} test files; the path is wrong" + + for path in files: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + # A bare expression statement whose value is a tuple: nothing + # consumes the result, so every element after the first is dead. + if isinstance(node, ast.Expr) and isinstance(node.value, ast.Tuple): + offenders.append(f"{path.name}:{node.lineno}") + + assert not offenders, "assertions whose failure message is discarded into a tuple: " + ", ".join(offenders) + + +class TestTheBatchIsOneTransaction: + """A thousand-reading import used to commit a thousand times. + + Every commit is an fsync that takes SQLite's write lock, so one import + serialised a thousand disk syncs against this server's own collector and + request latency tracked sync cost rather than row count. (CodeRabbit, PR + #256 — though only half of the reported cause was real: ``_get_db`` hands out + a borrowed handle on a SHARED per-loop connection whose ``close()`` is a + documented no-op, so the loop was never opening a thousand connections. It + was committing a thousand times.) + """ + + def test_the_whole_import_is_one_write(self, client): + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, + ): + resp = _import( + client, + readings=[{"slug": "grass", "balance": float(i), "date": f"2026-01-{i + 1:02d}"} for i in range(28)], + ) + assert resp.status_code == 200 + assert len(upsert.await_args_list) == 1, f"the import took {len(upsert.await_args_list)} transactions, not one" + assert len(_rows(upsert)) == 28 + + def test_the_per_reading_writer_is_not_used_here(self, client): + """Reverting the endpoint to the row-by-row call would restore the + thousand-commit behaviour while every other test still passed.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count), + patch("app.main.database.upsert_earnings", new_callable=AsyncMock) as single, + ): + _import(client, readings=[{"slug": "grass", "balance": 1.0, "date": "2026-01-01"}]) + assert single.await_count == 0, "the endpoint is writing one row at a time again" + + def test_nothing_is_written_when_every_slug_is_unknown(self, client): + """An empty batch must not take the write lock to do nothing.""" + with ( + patch("app.main._authenticate_worker_heartbeat", new_callable=AsyncMock, return_value="ok"), + patch("app.main.database.upsert_earnings_many", new_callable=AsyncMock, side_effect=_count) as upsert, + ): + resp = _import(client, readings=[{"slug": "nope", "balance": 1.0, "date": "2026-01-01"}]) + assert resp.json()["imported"] == 0 + assert _rows(upsert) == [] + + +# --- the batched writer itself, against a real database ------------------- + + +# Mirrors the fixtures in test_database.py rather than importing them: they are +# module-local there, and pytest does not share fixtures across modules without +# a conftest. +@pytest.fixture +def db_dir(tmp_path): + with patch.object(database, "DB_DIR", tmp_path), patch.object(database, "DB_PATH", tmp_path / "cashpilot.db"): + yield tmp_path + + +@pytest.fixture +def db(db_dir): + asyncio.run(database.init_db()) + return db_dir + + +def _all_earnings() -> list[dict]: + async def run(): + conn = await database._get_db() + cur = await conn.execute( + "SELECT platform, balance, currency, date, fx_rate_usd, source FROM earnings ORDER BY date, platform" + ) + return [dict(row) for row in await cur.fetchall()] + + return asyncio.run(run()) + + +class TestUpsertEarningsMany: + def test_it_writes_every_reading(self, db): + written = asyncio.run( + database.upsert_earnings_many( + [ + {"platform": "grass", "balance": 1.0, "date": "2026-01-01", "source": "mac"}, + {"platform": "grass", "balance": 2.0, "date": "2026-01-02", "source": "mac"}, + {"platform": "honeygain", "balance": 3.0, "date": "2026-01-01", "source": "mac"}, + ] + ) + ) + assert written == 3 + assert len(_all_earnings()) == 3 + + def test_re_importing_a_day_updates_rather_than_appends(self, db): + """The whole reason the import is safe to retry. A second row for one + (platform, source, date) would difference against itself and read zero.""" + rows = [{"platform": "grass", "balance": 1.0, "date": "2026-01-01", "source": "mac"}] + asyncio.run(database.upsert_earnings_many(rows)) + asyncio.run(database.upsert_earnings_many([{**rows[0], "balance": 5.0}])) + + stored = _all_earnings() + assert len(stored) == 1, f"a repeated day appended: {stored}" + assert stored[0]["balance"] == 5.0 + + def test_two_sources_coexist_for_one_platform_and_day(self, db): + """The point of the source column: the server and a Desktop can both have + read the same provider account on the same day.""" + asyncio.run( + database.upsert_earnings_many( + [ + {"platform": "grass", "balance": 1.0, "date": "2026-01-01", "source": "server"}, + {"platform": "grass", "balance": 9.0, "date": "2026-01-01", "source": "mac"}, + ] + ) + ) + stored = _all_earnings() + assert len(stored) == 2, f"one source overwrote the other: {stored}" + assert {row["source"] for row in stored} == {"server", "mac"} + + def test_an_absent_rate_does_not_erase_a_known_one(self, db): + """Same COALESCE rule the single-row writer has. Overwriting a known-good + rate with NULL destroys the very data the column exists to preserve.""" + asyncio.run( + database.upsert_earnings_many( + [{"platform": "mysterium", "balance": 1.0, "date": "2026-01-01", "source": "mac", "fx_rate_usd": 0.4}] + ) + ) + asyncio.run( + database.upsert_earnings_many( + [{"platform": "mysterium", "balance": 2.0, "date": "2026-01-01", "source": "mac", "fx_rate_usd": None}] + ) + ) + assert _all_earnings()[0]["fx_rate_usd"] == 0.4 + + def test_an_empty_batch_writes_nothing_and_reports_zero(self, db): + assert asyncio.run(database.upsert_earnings_many([])) == 0 + assert _all_earnings() == [] + + def test_defaults_match_the_single_row_writer(self, db): + """Currency and source default the same way, or a batched import would + land under a different series than the same reading written singly.""" + asyncio.run(database.upsert_earnings_many([{"platform": "grass", "balance": 1.0, "date": "2026-01-01"}])) + row = _all_earnings()[0] + assert row["currency"] == "USD" + assert row["source"] == "server" + + def test_a_failure_part_way_leaves_nothing_behind(self, db): + """One transaction means all-or-nothing. Half an imported history is + worse than none: the client believes it failed and retries, and the rows + that landed are indistinguishable from rows it took itself. + + The failure has to happen INSIDE the statement to test anything. My first + version raised a KeyError while building the rows -- before any SQL ran -- + so it proved only that the row build validates first, and it passed + against a writer that committed after every row. Caught by a negative + control; this version uses a NULL platform, which survives the build and + is rejected by the table. + """ + asyncio.run(database.upsert_earnings_many([{"platform": "grass", "balance": 1.0, "date": "2026-01-01"}])) + before = _all_earnings() + + with pytest.raises(Exception, match="(?i)not null|constraint"): + asyncio.run( + database.upsert_earnings_many( + [ + {"platform": "honeygain", "balance": 2.0, "date": "2026-01-02"}, + {"platform": None, "balance": 3.0, "date": "2026-01-03"}, + ] + ) + ) + assert _all_earnings() == before, "a failed batch left rows behind" + + def test_a_failed_batch_does_not_wedge_the_shared_connection(self, db): + """The connection outlives the request, so an abandoned transaction would + hold SQLite's write lock against every later write on this loop -- + including the server's own collector.""" + with pytest.raises(Exception, match="(?i)not null|constraint"): + asyncio.run(database.upsert_earnings_many([{"platform": None, "balance": 1.0, "date": "2026-01-01"}])) + # The next write must simply work. + assert ( + asyncio.run(database.upsert_earnings_many([{"platform": "grass", "balance": 1.0, "date": "2026-01-02"}])) + == 1 + ) + assert len(_all_earnings()) == 1 diff --git a/tests/test_optional_runtime.py b/tests/test_optional_runtime.py index 063cd064..d4e2bcf4 100644 --- a/tests/test_optional_runtime.py +++ b/tests/test_optional_runtime.py @@ -35,7 +35,7 @@ def test_the_default_spec_selects_no_runtime(self): def test_a_spec_without_a_runtime_passes_validation_untouched(self): with patch.object(orchestrator, "available_runtimes", return_value=set()) as available: worker_api._validate_runtime(None) - available.assert_not_called(), "an absent runtime must not even query the daemon" + assert available.call_count == 0, "an absent runtime must not even query the daemon" def test_deploy_passes_none_through_so_docker_uses_its_default(self): captured = {}