diff --git a/CHANGELOG.md b/CHANGELOG.md index c9c3e892..8279d3b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Added +- **Bulk document upload, and waiting for it to be ingested** (mirroring + VH-Lab/NDI-matlab#1047). `ndi.cloud.upload.uploadDocumentCollection` now + uploads documents in bulk, as NDI-matlab does: each chunk + (`Cloud.Upload.Max_Document_Batch_Count`) is one zipped JSON array + (new `zip_documents_for_upload`) PUT to `getBulkUploadURL`, instead of one + `addDocument` request per document, which for a dataset of 100,000 + documents took hours. `NDI_CLOUD_UPLOAD_NO_ZIP=true` keeps the + one-at-a-time path. A bulk upload returns before the server has ingested + the zip and has no job to wait on, so afterwards it waits, with the new + `ndi.cloud.api.documents.waitForDocumentCount`, for the dataset to list the + uploaded documents (`wait_for_ingestion`, `ingestion_timeout`, default + 300 s; a timeout is a warning, and the result is in `report["ingestion"]`). + `upload_type` now says `'batch'` or `'serial'` truthfully. + - **Checking for, and installing, a newer NDI-python** (mirroring VH-Lab/NDI-matlab#1038). New `ndi.update`: `check()` compares this copy with the latest commit on main -- a git clone through `git fetch` (plus the diff --git a/src/ndi/cloud/api/documents.py b/src/ndi/cloud/api/documents.py index 519cd833..3367a295 100644 --- a/src/ndi/cloud/api/documents.py +++ b/src/ndi/cloud/api/documents.py @@ -10,10 +10,12 @@ from __future__ import annotations +import math import re +import time from typing import Annotated, Any -from pydantic import SkipValidation, validate_call +from pydantic import Field, SkipValidation, validate_call from ..client import APIResponse, CloudClient, _auto_client from ._validators import VALIDATE_CONFIG, CloudId, FilePath, PageNumber, PageSize, Scope @@ -182,6 +184,77 @@ def countDocuments(dataset_id: CloudId, *, client: _Client = None) -> int: return ds.get("documentCount", 0) +@_auto_client +@validate_call(config=VALIDATE_CONFIG) +def waitForDocumentCount( + dataset_id: CloudId, + expected_count: Annotated[int, Field(ge=0)], + *, + timeout: float = 300.0, + initial_interval: float = 1.0, + max_interval: float = 10.0, + backoff_factor: float = 2.0, + client: _Client = None, +) -> dict[str, Any]: + """Wait until a dataset lists at least *expected_count* documents. + + Repeatedly calls :func:`countDocuments` at exponentially growing + intervals until the count is at least *expected_count* or the overall + timeout elapses. A failed poll is retried, not treated as the end of + the wait. + + Use it after a bulk document upload: the upload returns once the zip is + stored, before the server has ingested the documents, so a listing made + straight away can miss some of them. Bulk document uploads have no job + id, so :func:`ndi.cloud.api.files.waitForAllBulkUploads` (which waits on + bulk FILE uploads) does not cover them. + + Args: + dataset_id: Cloud dataset ID. + expected_count: The number of documents the dataset should list. + timeout: Overall deadline in seconds. Default 300. + initial_interval: First sleep between polls (s). Default 1. + max_interval: Cap on the per-poll sleep (s). Default 10. + backoff_factor: Multiplier applied after each poll. Default 2. + + Returns: + Dict with ``state`` (``'complete'`` or ``'timeout'``), ``count`` + (the last count read; ``nan`` if no poll succeeded), + ``expectedCount`` and ``elapsed`` (seconds). ``state == 'complete'`` + is the success verdict. + + MATLAB equivalent: +cloud/+api/+documents/waitForDocumentCount.m + """ + start = time.monotonic() + interval = initial_interval + answer: dict[str, Any] = { + "state": "unknown", + "count": math.nan, + "expectedCount": expected_count, + "elapsed": 0.0, + } + while True: + try: + count = countDocuments(dataset_id, client=client) + if isinstance(count, (int, float)) and not isinstance(count, bool): + answer["count"] = count + if count >= expected_count: + answer["state"] = "complete" + answer["elapsed"] = time.monotonic() - start + return answer + except Exception: + # A failed poll is transient: keep the last good count and try + # again rather than giving up on a gateway blip. + pass + elapsed = time.monotonic() - start + if elapsed >= timeout: + answer["state"] = "timeout" + answer["elapsed"] = elapsed + return answer + time.sleep(min(interval, max_interval, timeout - elapsed)) + interval = min(interval * backoff_factor, max_interval) + + @_auto_client @validate_call(config=VALIDATE_CONFIG) def bulkFetch( diff --git a/src/ndi/cloud/api/ndi_matlab_python_bridge.yaml b/src/ndi/cloud/api/ndi_matlab_python_bridge.yaml index 91dfe0fa..f321b277 100644 --- a/src/ndi/cloud/api/ndi_matlab_python_bridge.yaml +++ b/src/ndi/cloud/api/ndi_matlab_python_bridge.yaml @@ -554,6 +554,49 @@ functions: decision_log: > Exact match. MATLAB also has documentCount.m which is a duplicate/alias. + - name: waitForDocumentCount + matlab_path: "+ndi/+cloud/+api/+documents/waitForDocumentCount.m" + matlab_last_sync_hash: "004585c" + python_path: "ndi/cloud/api/documents.py" + input_arguments: + - name: dataset_id + type_matlab: "string" + type_python: "CloudId" + - name: expected_count + type_matlab: "double" + type_python: "int" + - name: timeout + type_matlab: "double" + type_python: "float" + default: "300.0" + - name: initial_interval + type_matlab: "double" + type_python: "float" + default: "1.0" + - name: max_interval + type_matlab: "double" + type_python: "float" + default: "10.0" + - name: backoff_factor + type_matlab: "double" + type_python: "float" + default: "2.0" + - name: client + type_python: "_Client" + default: "None" + output_arguments: + - name: result + type_python: "dict[str, Any]" + decision_log: > + Ported from VH-Lab/NDI-matlab#1047 (matlab 004585c). Polls + countDocuments at exponentially growing intervals until the count is at + least expected_count or the timeout elapses; a failed poll is retried, + not fatal. Returns MATLAB's ANSWER struct (state 'complete' | 'timeout', + count -- nan if no poll succeeded -- expectedCount, elapsed); MATLAB's + leading b flag is state == 'complete', the same convention + waitForFileTierJob follows. WaitForDocumentCountMockTest ports over as + TestWaitForDocumentCount in tests/test_cloud_upload_bulk_documents.py. + - name: bulkUpload matlab_path: "N/A" python_path: "ndi/cloud/api/documents.py" diff --git a/src/ndi/cloud/ndi_matlab_python_bridge.yaml b/src/ndi/cloud/ndi_matlab_python_bridge.yaml index c35d774d..97e97d37 100644 --- a/src/ndi/cloud/ndi_matlab_python_bridge.yaml +++ b/src/ndi/cloud/ndi_matlab_python_bridge.yaml @@ -884,7 +884,7 @@ functions: # --- upload.py functions --- - name: uploadDocumentCollection matlab_path: "+ndi/+cloud/+upload/uploadDocumentCollection.m" - matlab_last_sync_hash: "7180cc8" + matlab_last_sync_hash: "004585c" python_path: "ndi/cloud/upload.py" input_arguments: - name: dataset_id @@ -900,6 +900,14 @@ functions: - name: max_chunk type_python: "int | None" default: "None" + - name: wait_for_ingestion + type_matlab: "logical" + type_python: "bool" + default: "True" + - name: ingestion_timeout + type_matlab: "double" + type_python: "float" + default: "300.0" - name: client type_python: "CloudClient | None" default: "None" @@ -908,6 +916,24 @@ functions: type_python: "dict[str, Any]" decision_log: > Exact match. + 2026-10-23: synchronized with NDI-matlab 004585c (VH-Lab/NDI-matlab#1047). + Until now this entry said "Exact match" but the port only had MATLAB's + SERIAL path: every document went up as its own addDocument request, + whatever NDI_CLOUD_UPLOAD_NO_ZIP said -- one request per document, hours + for a 100,000-document dataset. It now has MATLAB's default BATCH path: + each chunk is one JSON array zipped by zip_documents_for_upload and PUT + to getBulkUploadURL; NDI_CLOUD_UPLOAD_NO_ZIP=true keeps the serial path, + as in MATLAB. #1047's ingestion wait comes with it: a bulk upload + returns once the zip is stored, not ingested, and has no job id, so the + batch path reads countDocuments before uploading and afterwards calls + waitForDocumentCount for that count plus the documents in the stored + batches (wait_for_ingestion / ingestion_timeout mirror MATLAB's + waitForIngestion / ingestionTimeout). A timeout logs a warning and + leaves status alone; the result is in report["ingestion"]. + upload_type now reports 'batch' or 'serial' truthfully. Tests that model + the transport by patching addDocument run on the serial path through the + serial_document_upload fixture; the batch path is tested in + tests/test_cloud_upload_bulk_documents.py. 2026-10-17: NDI-matlab 7180cc8 (VH-Lab/NDI-matlab#1033) makes maxDocumentChunk default to the preference Cloud.Upload.Max_Document_Batch_Count (was Inf). max_chunk=None now reads the same preference; 0 or negative still means all at once. @@ -1877,10 +1903,34 @@ porting_deferred: - name: zip_documents_for_upload matlab_path: "+ndi/+cloud/+upload/+internal/zip_documents_for_upload.m" matlab_last_sync_hash: "c6aef552" - status: ported_differently + python_path: "ndi/cloud/upload.py" + input_arguments: + - name: documents + type_matlab: "cell" + type_python: "list[dict[str, Any]]" + - name: dataset_id + type_matlab: "char" + type_python: "str" + - name: target_folder + type_matlab: "string" + type_python: "str | Path | None" + default: "None" + output_arguments: + - name: zip_path + type_python: "Path" + - name: id_manifest + type_python: "list[str]" decision_log: > - MATLAB internal helper. Python uses zipForUpload which handles - this internally with zipfile module. + 2026-10-23: ported (was ported_differently). The old note said + zipForUpload covered this; it does not -- zipForUpload zips binary FILES + for getFileCollectionUploadURL, and nothing zipped DOCUMENTS, which is + why uploadDocumentCollection had no bulk path. Same archive as MATLAB: + one .json file holding the documents as a JSON array, encoded with + did.datastructures.json_encode_nan (MATLAB's jsonencodenan), the zip + named ..zip. Lives in ndi.cloud.upload rather than + an internal subpackage, next to its one caller. Python's own zip entry + name is random hex rather than ndi.file.temp_name's; the server reads + the archive's single .json, whatever it is called. # ======================================================================= # +ndi/+cloud/+ui - MATLAB's cloud dialogs. # diff --git a/src/ndi/cloud/upload.py b/src/ndi/cloud/upload.py index b13c4eff..17a93ecf 100644 --- a/src/ndi/cloud/upload.py +++ b/src/ndi/cloud/upload.py @@ -49,10 +49,20 @@ def uploadDocumentCollection( only_missing: bool = True, max_chunk: int | None = None, *, + wait_for_ingestion: bool = True, + ingestion_timeout: float = 300.0, client: CloudClient | None = None, ) -> dict[str, Any]: """Upload a list of document dicts to the cloud. + MATLAB equivalent: +ndi/+cloud/+upload/uploadDocumentCollection.m + + By default the documents go up in bulk: each chunk is written as one + JSON array, zipped (:func:`zip_documents_for_upload`) and PUT to the URL + from :func:`ndi.cloud.api.documents.getBulkUploadURL`. If the environment + variable ``NDI_CLOUD_UPLOAD_NO_ZIP`` is ``'true'``, they are sent one at + a time with :func:`ndi.cloud.api.documents.addDocument` instead. + Args: dataset_id: Cloud dataset ID. documents: List of document property dicts. @@ -60,15 +70,32 @@ def uploadDocumentCollection( max_chunk: Maximum documents per chunk. ``None`` (default) reads the preference ``Cloud.Upload.Max_Document_Batch_Count``; ``0`` or a negative value means all at once. + wait_for_ingestion: If True (default), a bulk upload does not return + until the server lists the uploaded documents, or until + *ingestion_timeout* elapses. A bulk upload returns once each zip + is stored, before the server has ingested it, so without this a + listing made straight afterwards can miss documents. Serial + uploads are not affected. + ingestion_timeout: Seconds to wait for ingestion. Default 300. A + timeout is logged as a warning and does not change ``status``. client: Authenticated cloud client (auto-created if omitted). Returns: - Report dict with ``upload_type``, ``manifest``, ``status``. + Report dict with ``upload_type`` (``'batch'`` or ``'serial'``), + ``total``, ``uploaded``, ``skipped``, ``manifest`` (ids of the + documents sent), ``status`` (``'ok'`` or ``'partial'``) and, on a + failure, ``errors``. When a bulk upload waited for ingestion, + ``ingestion`` holds the result of + :func:`ndi.cloud.api.documents.waitForDocumentCount`. """ + import os + from .api import documents as docs_api + serial = os.environ.get("NDI_CLOUD_UPLOAD_NO_ZIP", "").lower() in ("true", "1", "yes") + report: dict[str, Any] = { - "upload_type": "batch", + "upload_type": "serial" if serial else "batch", "total": len(documents), "uploaded": 0, "skipped": 0, @@ -108,21 +135,125 @@ def uploadDocumentCollection( if max_chunk and max_chunk > 0: chunks = [documents[i : i + max_chunk] for i in range(0, len(documents), max_chunk)] - for chunk in chunks: - for doc in chunk: - try: - docs_api.addDocument(dataset_id, doc, client=client) - report["uploaded"] += 1 - report["manifest"].append(document_id(doc)) - except Exception as exc: - report["status"] = "partial" - if report.get("errors") is None: - report["errors"] = [] - report["errors"].append(str(exc)) + def _fail(message: str) -> None: + report["status"] = "partial" + report.setdefault("errors", []).append(message) + + if serial: + for chunk in chunks: + for doc in chunk: + try: + docs_api.addDocument(dataset_id, doc, client=client) + report["uploaded"] += 1 + report["manifest"].append(document_id(doc)) + except Exception as exc: + _fail(str(exc)) + return report + + from .api import files as files_api + + # The server's count before this upload, so the ingestion wait below + # knows what count to expect. None skips the wait. + count_before: int | None = None + if wait_for_ingestion: + try: + count_before = int(docs_api.countDocuments(dataset_id, client=client)) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Could not read the document count of %s (%s); not waiting for ingestion", + dataset_id, + exc, + ) + + for n, chunk in enumerate(chunks, start=1): + zip_path: Path | None = None + try: + zip_path, id_manifest = zip_documents_for_upload(chunk, dataset_id) + url = docs_api.getBulkUploadURL(dataset_id, client=client) + if not url: + raise RuntimeError("no bulk upload URL returned") + files_api.putFiles(url, str(zip_path)) + report["uploaded"] += len(id_manifest) + report["manifest"].extend(id_manifest) + except Exception as exc: # noqa: BLE001 - reported per batch + _fail(f"Batch {n} ({len(chunk)} documents): {exc}") + finally: + if zip_path is not None and zip_path.exists(): + zip_path.unlink() + + # A stored zip is not yet an ingested one, and bulk document uploads have + # no job to wait on, so wait for the server's document count to take in + # every document from the batches that were stored. The server adds a new + # document for each one uploaded, even one whose id it already lists, so + # the count grows by exactly that many. + if count_before is not None and report["uploaded"] > 0: + expected = count_before + report["uploaded"] + try: + ingestion = docs_api.waitForDocumentCount( + dataset_id, expected, timeout=ingestion_timeout, client=client + ) + except Exception as exc: # noqa: BLE001 + ingestion = { + "state": "error", + "count": float("nan"), + "expectedCount": expected, + "elapsed": float("nan"), + "message": str(exc), + } + report["ingestion"] = ingestion + if ingestion.get("state") != "complete": + logger.warning( + "Dataset %s lists %s of the %d documents expected after this upload " + "(state: %s). The upload succeeded; the server may still be ingesting " + "the rest.", + dataset_id, + ingestion.get("count"), + expected, + ingestion.get("state"), + ) return report +def zip_documents_for_upload( + documents: list[dict[str, Any]], + dataset_id: str, + *, + target_folder: str | Path | None = None, +) -> tuple[Path, list[str]]: + """Write documents as one JSON array and zip it for a bulk upload. + + MATLAB equivalent: +ndi/+cloud/+upload/+internal/zip_documents_for_upload.m + + The archive holds a single ``.json`` file: the documents' properties as a + JSON array, encoded with :func:`did.datastructures.json_encode_nan` so NaN + and Infinity survive, as MATLAB's ``did.datastructures.jsonencodenan`` + does. The zip is named ``..zip`` so concurrent uploads + do not collide. + + Args: + documents: Document property dicts. + dataset_id: Cloud dataset ID (used in the zip's name). + target_folder: Where to write the zip. Defaults to the system's + temporary folder. + + Returns: + ``(zip_path, id_manifest)``: the zip's path, which the caller must + delete, and the ids of the documents in it, in order. + """ + import uuid + + from did.datastructures import json_encode_nan + + folder = Path(target_folder) if target_folder is not None else Path(tempfile.gettempdir()) + json_text = json_encode_nan(list(documents)) + id_manifest = [document_id(doc) for doc in documents] + zip_path = folder / f"{dataset_id}.{uuid.uuid4().hex}.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr(f"{uuid.uuid4().hex}.json", json_text) + return zip_path, id_manifest + + def zipForUpload( file_manifest: list[dict[str, Any]], dataset_id: str, diff --git a/src/ndi/mock/__init__.py b/src/ndi/mock/__init__.py index 2ebabede..9fbb675d 100644 --- a/src/ndi/mock/__init__.py +++ b/src/ndi/mock/__init__.py @@ -333,6 +333,43 @@ def clear_mock_docs(session: Any) -> None: session.database_rm(docs) +def _numeric_array(value: Any) -> np.ndarray | None: + """*value* as a NumPy array if it is numeric or boolean, else None. + + MATLAB's ``isnumeric(v) || islogical(v)``: Python numbers and booleans, + NumPy arrays and scalars, and (nested) lists of them, as a document's + JSON holds them. Text and ragged or mixed lists are not numeric. + """ + if isinstance(value, str) or value is None: + return None + try: + arr = np.asarray(value) + except (ValueError, TypeError): + return None + if arr.dtype.kind not in "biuf": + return None + return arr + + +def _mat2str(arr: np.ndarray) -> str: + """MATLAB's ``mat2str(arr, 6)`` for a numeric or boolean array of at most 2-D.""" + + def one(x: Any) -> str: + if arr.dtype.kind == "b": + return "true" if x else "false" + return f"{x:.6g}" + + if arr.ndim == 0: + return one(arr.item()) + if arr.ndim == 1: + arr = arr.reshape(1, -1) + if arr.size == 1: + return one(arr.item()) + if arr.size == 0: + return f"zeros({arr.shape[0]},{arr.shape[1]})" + return "[" + ";".join(" ".join(one(x) for x in row) for row in arr.tolist()) + "]" + + class ndi_mock_ctest: """Base class for calculator testing framework. @@ -404,7 +441,19 @@ def compare_mock_docs( if scope.lower() == "highsnr" and isinstance(docCompare, DocComparison): result = docCompare.compare(actual, expected) failures = [r for r in result.get("results", []) if not r.get("passed")] - report = [{"name": r["field"], **r} for r in failures] + # compare(actual, expected): value1 is the actual value, value2 the + # expected one. Named as MATLAB's docComparison report names them, + # so reportSummary can say how far out of tolerance each field was. + report = [ + { + "name": r["field"], + **r, + "actualValue": r.get("value1"), + "expectedValue": r.get("value2"), + "comment": r.get("detail", ""), + } + for r in failures + ] return bool(result["equal"]), report from ndi.fun.doc import diff @@ -475,18 +524,70 @@ def reportSummary(report: Any) -> str: MATLAB equivalent: the static ``ndi.mock.ctest.reportSummary``. Returns a LEADING-SPACE sentence, or ``''`` when there is nothing to say, so it can be appended to a message directly. + + When the report carries the compared values (``actualValue`` and + ``expectedValue``), each field is followed by a line giving the actual + value, the expected value, their absolute difference, and the + comparison's own ``comment``, so that a failure on CI says how far out + of tolerance it was and not only that it was. """ if not report: return "" if isinstance(report, str): return " " + report - if isinstance(report, dict) and "name" in report: - return f" Out of tolerance: {report['name']}." - if isinstance(report, (list, tuple)) and all( - isinstance(r, dict) and "name" in r for r in report + if isinstance(report, dict): + report = [report] + if not ( + isinstance(report, (list, tuple)) + and all(isinstance(r, dict) and "name" in r for r in report) ): - return " Out of tolerance: " + ", ".join(r["name"] for r in report) + "." - return "" + return "" + text = " Out of tolerance: " + ", ".join(str(r["name"]) for r in report) + "." + if not all("actualValue" in r and "expectedValue" in r for r in report): + return text + for r in report: + actual = r["actualValue"] + expected = r["expectedValue"] + text += ( + f"\n {r['name']}: actual {ndi_mock_ctest.valueString(actual)}" + f", expected {ndi_mock_ctest.valueString(expected)}" + ) + a, e = _numeric_array(actual), _numeric_array(expected) + if a is not None and e is not None and a.shape == e.shape: + difference = np.abs(a.astype(float) - e.astype(float)) + text += f", abs difference {ndi_mock_ctest.valueString(difference)}" + if r.get("comment"): + text += f" ({r['comment']})" + return text + + @staticmethod + def valueString(value: Any) -> str: + """Render a compared value compactly for a failure message. + + MATLAB equivalent: the static ``ndi.mock.ctest.valueString``. + + Numbers and booleans (scalars, lists, NumPy arrays) are written as + MATLAB's ``mat2str`` at 6 significant digits writes them -- ``3.14159``, + ``true``, ``[1 2;3 4]`` -- with arrays of more than two dimensions as a + row. Text is quoted, and anything else is shown as its size and type + (``[2 list]``, ``[dict]``). Renderings longer than 200 characters are + cut, since some compared fields hold whole fitted curves. + """ + max_length = 200 + arr = _numeric_array(value) + if arr is not None: + if arr.ndim > 2: + arr = arr.reshape(1, -1) + text = _mat2str(arr) + elif isinstance(value, str): + text = f"'{value}'" + elif hasattr(value, "__len__"): + text = f"[{len(value)} {type(value).__name__}]" + else: + text = f"[{type(value).__name__}]" + if len(text) > max_length: + text = text[:max_length] + " ..." + return text def load_mock_expected_output(self, number: int) -> dict | None: """Load expected output from file.""" diff --git a/src/ndi/mock/ndi_matlab_python_bridge.yaml b/src/ndi/mock/ndi_matlab_python_bridge.yaml index 2aaf117b..5ed02559 100644 --- a/src/ndi/mock/ndi_matlab_python_bridge.yaml +++ b/src/ndi/mock/ndi_matlab_python_bridge.yaml @@ -262,10 +262,24 @@ classes: - name: ctest type: class matlab_path: "src/ndi/+ndi/+mock/ctest.m" - matlab_last_sync_hash: "0bdc0ac79" + matlab_last_sync_hash: "0374dd2" python_path: "ndi/mock/__init__.py" python_class: "ndi_mock_ctest" decision_log: > + 2026-10-23: synchronized with NDI-matlab 0374dd2 (VH-Lab/NDI-matlab#1046). + reportSummary now follows each out-of-tolerance field with a line giving + the actual value, the expected value, their absolute difference (when + both are numeric and the same shape) and the comparison's comment, so a + failure says how far out it was; the mock data are unseeded and a failing + run cannot be reproduced. New static valueString renders a value as + mat2str(v,6) does (>2-D as a row), text quoted, anything else as its size + and type, cut at 200 characters. Python's DocComparison results name the + values value1/value2/detail; compare_mock_docs now also carries them as + MATLAB's actualValue/expectedValue/comment, which reportSummary reads. + For a non-numeric value Python prints [ ] ("[2 list]") where + MATLAB prints [ ] ("[1x2 cell]"). TestCtestReportSummary.m + ports over as TestReportSummarySaysHowFarOut in + tests/test_mock_against_matlab.py. Examined against NDI-matlab 0bdc0ac79, the file's last-touching commit. Six of MATLAB's methods had no counterpart, and the ones that did looked in the wrong place. PORTED: calc_path (MATLAB's is which(class(obj)) -- @@ -462,6 +476,20 @@ classes: DISCREPANCY NOTED: Not implemented in Python. Not critical for current testing infrastructure. + - name: valueString + kind: static + input_arguments: + - name: value + type_matlab: "any" + type_python: "Any" + output_arguments: + - name: str + type_python: "str" + decision_log: > + Ported from NDI-matlab 0374dd2 (VH-Lab/NDI-matlab#1046). See the + class decision_log for the one difference in how non-numeric values + are named. + not_applicable: - name: plot reason: "MATLAB GUI plotting. Python uses matplotlib." diff --git a/tests/conftest.py b/tests/conftest.py index 2065d458..67dbc1e8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -172,6 +172,19 @@ def fake_dataset(tmp_path): return FakeDataset(tmp_path) +@_pytest.fixture +def serial_document_upload(monkeypatch): + """Make uploadDocumentCollection send documents one at a time. + + uploadDocumentCollection uploads in bulk (zipped JSON batches) unless + ``NDI_CLOUD_UPLOAD_NO_ZIP`` is ``'true'``, as in MATLAB. Tests of the sync + logic built on it model the transport by patching ``addDocument`` -- to + see each document sent, or to fail one -- so they run on the serial path. + The bulk path has its own tests in tests/test_cloud_upload_bulk_documents.py. + """ + monkeypatch.setenv("NDI_CLOUD_UPLOAD_NO_ZIP", "true") + + # --------------------------------------------------------------------------- # Isolate DID's machine-global file cache for the whole test session. # diff --git a/tests/fixtures/lightsheet_cloud_fixture.json b/tests/fixtures/lightsheet_cloud_fixture.json index 1d1bb812..0b4802bd 100644 --- a/tests/fixtures/lightsheet_cloud_fixture.json +++ b/tests/fixtures/lightsheet_cloud_fixture.json @@ -1,8 +1,9 @@ { - "cloud_dataset_id": "6accf3a0044c77d93a6ebdb7", + "cloud_dataset_id": "6adb9d501dabe363f1268c3f", + "dataset_name": "Blob Test Dataset 20261023T134551 121", "user": "1", "environment": "prod", - "created_at": "2026-10-12T14:50:35Z", - "workflow_run": "https://github.com/Waltham-Data-Science/NDI-python/actions/runs/38265299422", - "notes": "Small synthetic lightsheet blob uploaded by ndi.test.cloud.lightsheet_blob_cloud_roundtrip (MATLAB), 12 documents / 15 files / 1.92 MB. Regenerate with the Build lightsheet blob fixture workflow. tests/test_pyramid_loader_integration.py reads this JSON by default when NDI_LIGHTSHEET_TEST_FIXTURE_ID is not set." + "created_at": "2026-10-23", + "workflow_run": "", + "notes": "Small synthetic lightsheet blob uploaded by ndi.test.cloud.lightsheet_blob_cloud_roundtrip (MATLAB). Rebuilt 2026-10-23 by running that routine by hand as test user 1 on prod, replacing 6accf3a0044c77d93a6ebdb7, whose coarsest-level chunk files had gone missing (test_the_coarsest_level_computes_and_is_not_all_zero read only fill_value). Regenerate with the Build lightsheet blob fixture workflow. tests/test_pyramid_loader_integration.py reads this JSON by default when NDI_LIGHTSHEET_TEST_FIXTURE_ID is not set." } diff --git a/tests/test_cloud_cluster_body_drift.py b/tests/test_cloud_cluster_body_drift.py index e751f35c..5c7ff63d 100644 --- a/tests/test_cloud_cluster_body_drift.py +++ b/tests/test_cloud_cluster_body_drift.py @@ -7,6 +7,8 @@ from __future__ import annotations +import pytest + class _Doc: def __init__(self, props): @@ -164,6 +166,7 @@ def test_a_very_short_answer_does_not_dump_every_id(self, monkeypatch, caplog): assert len(caplog.text.splitlines()) == 1 +@pytest.mark.usefixtures("serial_document_upload") class TestAFailedRemoteListingExplainsItself: """MATLAB counterpart: ``uploadDocumentCollection.m``, NDI-matlab ``e96801dee``. diff --git a/tests/test_cloud_sync_index_advancement.py b/tests/test_cloud_sync_index_advancement.py index 04155e1a..1b04f77a 100644 --- a/tests/test_cloud_sync_index_advancement.py +++ b/tests/test_cloud_sync_index_advancement.py @@ -18,6 +18,9 @@ from .conftest import FakeDataset, make_document +# These tests model the upload transport by patching addDocument. +pytestmark = pytest.mark.usefixtures("serial_document_upload") + @pytest.fixture(autouse=True) def _no_bulk_settling(monkeypatch): diff --git a/tests/test_cloud_upload_bulk_documents.py b/tests/test_cloud_upload_bulk_documents.py new file mode 100644 index 00000000..3f56ad1e --- /dev/null +++ b/tests/test_cloud_upload_bulk_documents.py @@ -0,0 +1,322 @@ +"""Bulk document upload and the ingestion wait -- no network required. + +MATLAB counterparts: +ndi/+cloud/+upload/uploadDocumentCollection.m (batch +path), +ndi/+cloud/+upload/+internal/zip_documents_for_upload.m, and +tests/+ndi/+unittest/WaitForDocumentCountMockTest.m. + +uploadDocumentCollection sends documents in bulk -- each chunk one zipped +JSON array PUT to a presigned URL -- rather than one request per document, +which for a dataset of 100,000 documents is the difference between minutes +and hours. A stored zip is not yet an ingested one, and a bulk document +upload has no job id to wait on, so afterwards it waits for the dataset's +document count to take the new documents in (VH-Lab/NDI-matlab#1047). +""" + +from __future__ import annotations + +import json +import math +import zipfile +from unittest.mock import MagicMock + +import pytest + +import ndi.cloud.upload as upload +from ndi.cloud.api import documents as docs_api +from ndi.cloud.api import files as files_api + +CLOUD_ID = "65a1b2c3d4e5f60718293a4b" + + +def _make_client() -> MagicMock: + client = MagicMock() + client.config.api_url = "https://api.ndi-cloud.com/v1" + return client + + +def _docs(*ids): + return [{"base": {"id": i}, "value": n} for n, i in enumerate(ids)] + + +class _ScriptedCount: + """countDocuments returning scripted values; an Exception entry is raised. + + The final entry repeats, so "never arrives" is one entry. + """ + + def __init__(self, script): + self.script = list(script) + self.calls = 0 + + def __call__(self, dataset_id, *, client=None): + self.calls += 1 + item = self.script[min(self.calls - 1, len(self.script) - 1)] + if isinstance(item, Exception): + raise item + return item + + +class _BulkServer: + """Stands in for getBulkUploadURL + putFiles and records each zip's documents.""" + + def __init__(self, fail_batches=()): + self.fail_batches = set(fail_batches) + self.batches: list[list[dict]] = [] + self.zip_paths: list[str] = [] + + def url(self, dataset_id, *, client=None): + return f"https://s3.example/bulk/{len(self.zip_paths) + 1}" + + def put(self, url, file_path, *args, **kwargs): + self.zip_paths.append(file_path) + with zipfile.ZipFile(file_path) as zf: + names = zf.namelist() + assert len(names) == 1 and names[0].endswith(".json") + self.batches.append(json.loads(zf.read(names[0]))) + if len(self.zip_paths) in self.fail_batches: + raise RuntimeError("S3 said no") + return True + + +@pytest.fixture() +def bulk(monkeypatch): + monkeypatch.delenv("NDI_CLOUD_UPLOAD_NO_ZIP", raising=False) + server = _BulkServer() + monkeypatch.setattr(docs_api, "getBulkUploadURL", server.url) + monkeypatch.setattr(files_api, "putFiles", server.put) + + def _no_single_adds(*a, **k): + raise AssertionError("a bulk upload must not send documents one at a time") + + monkeypatch.setattr(docs_api, "addDocument", _no_single_adds) + return server + + +# --------------------------------------------------------------------------- +# zip_documents_for_upload +# --------------------------------------------------------------------------- + + +class TestZipDocumentsForUpload: + def test_one_json_array_of_the_documents(self, tmp_path): + docs = _docs("a", "b") + zip_path, ids = upload.zip_documents_for_upload(docs, CLOUD_ID, target_folder=tmp_path) + with zipfile.ZipFile(zip_path) as zf: + (name,) = zf.namelist() + assert name.endswith(".json") + assert json.loads(zf.read(name)) == docs + assert ids == ["a", "b"] + + def test_named_for_the_dataset_and_unique(self, tmp_path): + p1, _ = upload.zip_documents_for_upload(_docs("a"), CLOUD_ID, target_folder=tmp_path) + p2, _ = upload.zip_documents_for_upload(_docs("a"), CLOUD_ID, target_folder=tmp_path) + assert p1.name.startswith(CLOUD_ID + ".") and p1.suffix == ".zip" + assert p1 != p2 + + def test_nan_and_infinity_survive(self, tmp_path): + docs = [{"base": {"id": "a"}, "x": float("nan"), "y": float("inf")}] + zip_path, _ = upload.zip_documents_for_upload(docs, CLOUD_ID, target_folder=tmp_path) + with zipfile.ZipFile(zip_path) as zf: + back = json.loads(zf.read(zf.namelist()[0])) + assert math.isnan(back[0]["x"]) and back[0]["y"] == math.inf + + +# --------------------------------------------------------------------------- +# uploadDocumentCollection -- the bulk path +# --------------------------------------------------------------------------- + + +class TestBulkUpload: + def test_chunks_go_up_as_zips_not_one_by_one(self, bulk, monkeypatch): + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([0, 5])) + docs = _docs("a", "b", "c", "d", "e") + report = upload.uploadDocumentCollection( + CLOUD_ID, docs, only_missing=False, max_chunk=2, client=_make_client() + ) + assert bulk.batches == [docs[0:2], docs[2:4], docs[4:5]] + assert report["upload_type"] == "batch" + assert report["status"] == "ok" + assert report["uploaded"] == 5 + assert report["manifest"] == ["a", "b", "c", "d", "e"] + + def test_the_zips_are_deleted(self, bulk, monkeypatch): + from pathlib import Path + + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([0, 2])) + upload.uploadDocumentCollection( + CLOUD_ID, _docs("a", "b"), only_missing=False, max_chunk=1, client=_make_client() + ) + assert len(bulk.zip_paths) == 2 + assert not any(Path(p).exists() for p in bulk.zip_paths) + + def test_a_failed_batch_is_partial_and_the_rest_still_go(self, bulk, monkeypatch): + bulk.fail_batches = {1} + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([0, 1])) + report = upload.uploadDocumentCollection( + CLOUD_ID, _docs("a", "b"), only_missing=False, max_chunk=1, client=_make_client() + ) + assert report["status"] == "partial" + assert report["uploaded"] == 1 + assert report["manifest"] == ["b"] + assert len(report["errors"]) == 1 and "S3 said no" in report["errors"][0] + + def test_the_wait_expects_the_count_before_plus_the_documents_stored(self, bulk, monkeypatch): + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([7])) + waits = [] + + def _wait(dataset_id, expected, *, timeout, client=None): + waits.append((expected, timeout)) + return {"state": "complete", "count": expected, "expectedCount": expected} + + monkeypatch.setattr(docs_api, "waitForDocumentCount", _wait) + report = upload.uploadDocumentCollection( + CLOUD_ID, + _docs("a", "b", "c"), + only_missing=False, + ingestion_timeout=42, + client=_make_client(), + ) + assert waits == [(10, 42)] + assert report["ingestion"]["state"] == "complete" + + def test_a_failed_batch_is_not_waited_for(self, bulk, monkeypatch): + bulk.fail_batches = {2} + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([4])) + waits = [] + monkeypatch.setattr( + docs_api, + "waitForDocumentCount", + lambda d, expected, **k: waits.append(expected) or {"state": "complete"}, + ) + upload.uploadDocumentCollection( + CLOUD_ID, _docs("a", "b"), only_missing=False, max_chunk=1, client=_make_client() + ) + assert waits == [5] + + def test_an_ingestion_timeout_warns_but_the_upload_stands(self, bulk, monkeypatch, caplog): + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([0])) + monkeypatch.setattr( + docs_api, + "waitForDocumentCount", + lambda d, expected, **k: {"state": "timeout", "count": 1, "expectedCount": expected}, + ) + with caplog.at_level("WARNING", logger="ndi.cloud.upload"): + report = upload.uploadDocumentCollection( + CLOUD_ID, _docs("a", "b"), only_missing=False, client=_make_client() + ) + assert report["status"] == "ok" + assert report["ingestion"]["state"] == "timeout" + assert "lists 1 of the 2 documents" in caplog.text + + def test_no_count_means_no_wait(self, bulk, monkeypatch, caplog): + monkeypatch.setattr(docs_api, "countDocuments", _ScriptedCount([RuntimeError("503")])) + monkeypatch.setattr( + docs_api, + "waitForDocumentCount", + lambda *a, **k: pytest.fail("must not wait without a starting count"), + ) + with caplog.at_level("WARNING", logger="ndi.cloud.upload"): + report = upload.uploadDocumentCollection( + CLOUD_ID, _docs("a"), only_missing=False, client=_make_client() + ) + assert report["uploaded"] == 1 + assert "ingestion" not in report + assert "503" in caplog.text + + def test_wait_for_ingestion_false_reads_no_count(self, bulk, monkeypatch): + counts = _ScriptedCount([0]) + monkeypatch.setattr(docs_api, "countDocuments", counts) + report = upload.uploadDocumentCollection( + CLOUD_ID, + _docs("a"), + only_missing=False, + wait_for_ingestion=False, + client=_make_client(), + ) + assert counts.calls == 0 + assert report["uploaded"] == 1 and "ingestion" not in report + + +class TestSerialSwitch: + def test_no_zip_sends_documents_one_at_a_time(self, monkeypatch): + monkeypatch.setenv("NDI_CLOUD_UPLOAD_NO_ZIP", "true") + sent = [] + monkeypatch.setattr( + docs_api, "addDocument", lambda d, doc, **k: sent.append(doc) or {"ok": True} + ) + monkeypatch.setattr( + docs_api, + "getBulkUploadURL", + lambda *a, **k: pytest.fail("a serial upload must not ask for a bulk URL"), + ) + docs = _docs("a", "b") + report = upload.uploadDocumentCollection( + CLOUD_ID, docs, only_missing=False, client=_make_client() + ) + assert sent == docs + assert report["upload_type"] == "serial" + assert report["uploaded"] == 2 + + +# --------------------------------------------------------------------------- +# waitForDocumentCount -- poll logic (WaitForDocumentCountMockTest.m) +# --------------------------------------------------------------------------- + + +def _wait(monkeypatch, script, expected, timeout=5.0): + counts = _ScriptedCount(script) + monkeypatch.setattr(docs_api, "countDocuments", counts) + answer = docs_api.waitForDocumentCount( + CLOUD_ID, + expected, + timeout=timeout, + initial_interval=0.01, + max_interval=0.01, + client=_make_client(), + ) + return answer, counts + + +class TestWaitForDocumentCount: + def test_count_already_reached_returns_at_once(self, monkeypatch): + answer, counts = _wait(monkeypatch, [5], 5) + assert answer["state"] == "complete" + assert answer["count"] == 5 and answer["expectedCount"] == 5 + assert counts.calls == 1 + + def test_count_above_target_is_success(self, monkeypatch): + answer, _ = _wait(monkeypatch, [7], 5) + assert answer["state"] == "complete" and answer["count"] == 7 + + def test_lagging_count_is_polled_until_it_arrives(self, monkeypatch): + answer, counts = _wait(monkeypatch, [0, 3, 5], 5) + assert answer["state"] == "complete" + assert counts.calls == 3 + + def test_count_that_never_arrives_times_out(self, monkeypatch): + answer, _ = _wait(monkeypatch, [4], 5, timeout=0.05) + assert answer["state"] == "timeout" + assert answer["count"] == 4, "the timeout should report the last count seen" + assert answer["elapsed"] >= 0.05 + + def test_an_api_error_is_retried_not_fatal(self, monkeypatch): + answer, counts = _wait(monkeypatch, [RuntimeError("gateway"), 5], 5) + assert answer["state"] == "complete" + assert counts.calls == 2 + + def test_only_api_errors_time_out_with_no_count(self, monkeypatch): + answer, counts = _wait(monkeypatch, [RuntimeError("gateway")], 5, timeout=0.05) + assert answer["state"] == "timeout" + assert math.isnan(answer["count"]) + assert counts.calls > 1 + + def test_zero_expected_is_immediately_met(self, monkeypatch): + answer, counts = _wait(monkeypatch, [0], 0) + assert answer["state"] == "complete" + assert counts.calls == 1 + + def test_a_negative_target_is_rejected(self, monkeypatch): + from pydantic import ValidationError + + with pytest.raises(ValidationError): + docs_api.waitForDocumentCount(CLOUD_ID, -1, client=_make_client()) diff --git a/tests/test_cloud_upload_file_batches.py b/tests/test_cloud_upload_file_batches.py index 069e4514..b3cfadc2 100644 --- a/tests/test_cloud_upload_file_batches.py +++ b/tests/test_cloud_upload_file_batches.py @@ -354,6 +354,7 @@ def _url(dataset_id, ids, client=None): download.downloadDocumentCollection(CLOUD_ID, ["d1", "d2", "d3"], client=_Client()) assert chunks == [["d1", "d2"], ["d3"]] + @pytest.mark.usefixtures("serial_document_upload") def test_upload_chunk_defaults_to_the_preference(self, monkeypatch): from ndi.cloud.api import documents as docs_api diff --git a/tests/test_mock_against_matlab.py b/tests/test_mock_against_matlab.py index 39543716..51ac8520 100644 --- a/tests/test_mock_against_matlab.py +++ b/tests/test_mock_against_matlab.py @@ -13,6 +13,7 @@ import tempfile from unittest.mock import MagicMock +import numpy as np import pytest from ndi.mock import ( @@ -219,6 +220,82 @@ def test_report_summary_names_the_out_of_tolerance_fields(self): ) +def _failure_report(name, actual, expected, comment): + """One report entry as compare_mock_docs builds it (MATLAB's failureReport).""" + return {"name": name, "actualValue": actual, "expectedValue": expected, "comment": comment} + + +class TestReportSummarySaysHowFarOut: + """Port of NDI-matlab TestCtestReportSummary.m (VH-Lab/NDI-matlab#1046). + + The failure message must say how far out of tolerance each field was, not + only that it was: mock data are not seeded, so a failing run cannot be + reproduced afterwards. + """ + + def test_values_and_difference_are_reported(self): + s = ndi_mock_ctest.reportSummary( + _failure_report("fit.p", [1, 2, 30], [1, 2, 10], "Abs difference > tolerance matrix") + ) + assert " Out of tolerance: fit.p." in s + assert "fit.p: actual [1 2 30], expected [1 2 10]" in s + assert "abs difference [0 0 20]" in s + assert "(Abs difference > tolerance matrix)" in s + + def test_each_failing_field_gets_its_own_line(self): + lines = ndi_mock_ctest.reportSummary( + [_failure_report("x", 1, 2, "m1"), _failure_report("y", 3, 5, "m2")] + ).splitlines() + assert len(lines) == 3 + assert "x: actual 1, expected 2, abs difference 1 (m1)" in lines[1] + assert "y: actual 3, expected 5, abs difference 2 (m2)" in lines[2] + + def test_no_difference_when_sizes_differ(self): + s = ndi_mock_ctest.reportSummary( + _failure_report("v", [1, 2], [1, 2, 3], "Dimension mismatch") + ) + assert "v: actual [1 2], expected [1 2 3]" in s + assert "abs difference" not in s + + def test_no_difference_for_text(self): + s = ndi_mock_ctest.reportSummary(_failure_report("t", "abc", "abd", "Strings differ")) + assert "t: actual 'abc', expected 'abd'" in s + assert "abs difference" not in s + + def test_empty_comment_is_omitted(self): + s = ndi_mock_ctest.reportSummary(_failure_report("n", 1, 2, "")) + assert s.endswith("abs difference 1") + + def test_value_string_formats(self): + vs = ndi_mock_ctest.valueString + assert vs(math.pi) == "3.14159" + assert vs(True) == "true" + assert vs(np.arange(1, 9).reshape(2, 2, 2)) == "[1 2 3 4 5 6 7 8]" + assert vs([[1, 2], [3, 4]]) == "[1 2;3 4]" + assert vs("s") == "'s'" + assert vs([1, "a"]) == "[2 list]" + + def test_long_values_are_cut(self): + s = ndi_mock_ctest.valueString(list(range(1, 1001))) + assert len(s) == 204 + assert s.endswith(" ...") + + def test_compare_mock_docs_report_carries_the_values(self): + from ndi.doc_comparison import DocComparison + + dc = DocComparison() + dc.add_comparison_parameter("response.mean", "abs_difference", tolerance=0.1) + expected = MagicMock() + expected.document_properties = {"response": {"mean": 1.0}} + actual = MagicMock() + actual.document_properties = {"response": {"mean": 2.0}} + match, report = ndi_mock_ctest().compare_mock_docs(expected, actual, "highSNR", dc) + assert match is False + assert "response.mean: actual 2, expected 1, abs difference 1" in ( + ndi_mock_ctest.reportSummary(report) + ) + + class TestCompareMockDocsUsesTolerances: """`compare_mock_docs` must honor the tolerance object MATLAB uses. diff --git a/tests/test_sync_binaries_and_deltas.py b/tests/test_sync_binaries_and_deltas.py index 5c71301f..556ed2f0 100644 --- a/tests/test_sync_binaries_and_deltas.py +++ b/tests/test_sync_binaries_and_deltas.py @@ -29,6 +29,9 @@ from .conftest import FakeDataset, make_document +# These tests model the upload transport by patching addDocument. +pytestmark = pytest.mark.usefixtures("serial_document_upload") + CLOUD_ID = "65a1b2c3d4e5f60718293a4b" diff --git a/tests/test_sync_binaries_gate_the_index.py b/tests/test_sync_binaries_gate_the_index.py index 11449101..982eaaf4 100644 --- a/tests/test_sync_binaries_gate_the_index.py +++ b/tests/test_sync_binaries_gate_the_index.py @@ -41,6 +41,9 @@ from .conftest import FakeDataset, make_document +# These tests model the upload transport by patching addDocument. +pytestmark = pytest.mark.usefixtures("serial_document_upload") + CLOUD_ID = "65a1b2c3d4e5f60718293a4b" diff --git a/tests/test_sync_returns_the_triple.py b/tests/test_sync_returns_the_triple.py index 61c607b8..c100587c 100644 --- a/tests/test_sync_returns_the_triple.py +++ b/tests/test_sync_returns_the_triple.py @@ -33,6 +33,9 @@ from .conftest import FakeDataset, make_document +# These tests model the upload transport by patching addDocument. +pytestmark = pytest.mark.usefixtures("serial_document_upload") + CLOUD_ID = "65a1b2c3d4e5f60718293a4b" ALL_MODES = ["uploadNew", "downloadNew", "mirrorToRemote", "mirrorFromRemote", "twoWaySync"] diff --git a/tests/test_sync_uploads_whole_documents.py b/tests/test_sync_uploads_whole_documents.py index 55b91cb2..55c18819 100644 --- a/tests/test_sync_uploads_whole_documents.py +++ b/tests/test_sync_uploads_whole_documents.py @@ -32,6 +32,9 @@ from .conftest import FakeDataset, make_document +# These tests model the upload transport by patching addDocument. +pytestmark = pytest.mark.usefixtures("serial_document_upload") + CLOUD_ID = "65a1b2c3d4e5f60718293a4b"