diff --git a/src/production_analytics/enlyze/gateway.py b/src/production_analytics/enlyze/gateway.py index 47077f1..da76a0d 100644 --- a/src/production_analytics/enlyze/gateway.py +++ b/src/production_analytics/enlyze/gateway.py @@ -68,52 +68,68 @@ class EnlyzeApiGateway: raise ValueError(f"{name} must be timezone-aware") if end < start: raise ValueError("end must not precede start") - response = self._client.post_json( - "/v2/timeseries", - { - "machine": machine_id, - "start": start.astimezone(UTC).isoformat(), - "end": end.astimezone(UTC).isoformat(), - "variables": [ - {"uuid": rate_variable_id}, - {"uuid": gate_variable_id}, - ], - }, - ) - - try: - data = response.body["data"] - columns = data["columns"] - if not isinstance(columns, list): - raise ValueError("columns must be a list") - for required in ("time", rate_variable_id, gate_variable_id): - if columns.count(required) != 1: - raise ValueError(f"required column {required!r} must occur exactly once") - time_index = columns.index("time") - rate_index = columns.index(rate_variable_id) - gate_index = columns.index(gate_variable_id) - if not isinstance(data["records"], list): - raise ValueError("records must be a list") - samples = [] - for index, record in enumerate(data["records"]): - try: - if not isinstance(record, list) or len(record) != len(columns): - raise ValueError("record must match columns") - rate, gate = record[rate_index], record[gate_index] - for value in (rate, gate): - if isinstance(value, bool) or not isinstance(value, (int, float)): - raise ValueError("rate and gate must be numeric") - if not isfinite(value): - raise ValueError("rate and gate must be finite") - samples.append(MaterialSample( - timestamp=_parse_timestamp(record[time_index]), - material_rate_kg_per_hour=float(rate), gate_value=float(gate), - )) - except (TypeError, ValueError) as exc: - raise ValueError(f"malformed record {index}: {exc}") from exc - return samples - except (KeyError, TypeError, ValueError) as exc: - raise ValueError(f"Invalid timeseries response: {exc}") from exc + request_body = { + "machine": machine_id, + "start": start.astimezone(UTC).isoformat(), + "end": end.astimezone(UTC).isoformat(), + "variables": [ + {"uuid": rate_variable_id}, + {"uuid": gate_variable_id}, + ], + } + samples = [] + followed_cursors: set[str] = set() + page = 1 + while True: + response = self._client.post_json("/v2/timeseries", request_body) + try: + data = response.body["data"] + columns = data["columns"] + if not isinstance(columns, list): + raise ValueError("columns must be a list") + for required in ("time", rate_variable_id, gate_variable_id): + if columns.count(required) != 1: + raise ValueError(f"required column {required!r} must occur exactly once") + time_index = columns.index("time") + rate_index = columns.index(rate_variable_id) + gate_index = columns.index(gate_variable_id) + if not isinstance(data["records"], list): + raise ValueError("records must be a list") + for index, record in enumerate(data["records"]): + try: + if not isinstance(record, list) or len(record) != len(columns): + raise ValueError("record must match columns") + rate, gate = record[rate_index], record[gate_index] + for value in (rate, gate): + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("rate and gate must be numeric") + if not isfinite(value): + raise ValueError("rate and gate must be finite") + samples.append(MaterialSample( + timestamp=_parse_timestamp(record[time_index]), + material_rate_kg_per_hour=float(rate), gate_value=float(gate), + )) + except (TypeError, ValueError) as exc: + raise ValueError(f"malformed record {index}: {exc}") from exc + next_cursor = None + if "metadata" in response.body: + metadata = response.body["metadata"] + if not isinstance(metadata, dict) or "next_cursor" not in metadata: + raise ValueError("metadata must be an object with a next_cursor field") + next_cursor = metadata["next_cursor"] + if next_cursor is not None and ( + not isinstance(next_cursor, str) or not next_cursor + ): + raise ValueError("next_cursor must be null or a non-empty string") + if next_cursor is None: + return samples + if next_cursor in followed_cursors: + raise ValueError("next_cursor has already been followed") + followed_cursors.add(next_cursor) + request_body = {**request_body, "cursor": next_cursor} + page += 1 + except (KeyError, TypeError, ValueError) as exc: + raise ValueError(f"Invalid timeseries response on page {page}: {exc}") from exc def _parse_timestamp(value: str) -> datetime: diff --git a/tests/test_enlyze_gateway.py b/tests/test_enlyze_gateway.py index 16b121e..2a05166 100644 --- a/tests/test_enlyze_gateway.py +++ b/tests/test_enlyze_gateway.py @@ -202,3 +202,91 @@ def test_malformed_run_fails_clearly(item) -> None: client.get.return_value.body = {"data": [item]} with pytest.raises(ValueError, match="Invalid production-run response"): EnlyzeApiGateway(client).get_open_production_run("m") + + +def test_timeseries_with_null_cursor(timeseries) -> None: + gateway, client, args = timeseries + client.post_json.return_value.body["metadata"] = {"next_cursor": None} + client.post_json.return_value.body["data"]["records"] = [ + ["2026-09-03T00:00:00Z", 10, 2], + ] + samples = gateway.get_material_samples(**args) + assert len(samples) == 1 + assert samples[0].material_rate_kg_per_hour == 10 + client.post_json.assert_called_once() + + +def test_timeseries_pagination_resolves_columns_per_page(timeseries) -> None: + gateway, client, args = timeseries + client.post_json.side_effect = [ + Mock(body={ + "data": {"columns": ["time", "rate", "gate"], + "records": [["2026-09-03T00:00:00Z", 10, 2]]}, + "metadata": {"next_cursor": "continuation"}, + }), + Mock(body={ + "data": {"columns": ["gate", "time", "rate"], + "records": [[3, "2026-09-03T01:00:00Z", 20]]}, + "metadata": {"next_cursor": None}, + }), + ] + samples = gateway.get_material_samples(**args) + assert [(s.timestamp, s.material_rate_kg_per_hour, s.gate_value) for s in samples] == [ + (datetime(2026, 9, 3, tzinfo=UTC), 10, 2), + (datetime(2026, 9, 3, 1, tzinfo=UTC), 20, 3), + ] + assert client.post_json.call_count == 2 + first, second = client.post_json.call_args_list + assert first.args == ("/v2/timeseries", { + "machine": "m", "start": args["start"].isoformat(), "end": args["end"].isoformat(), + "variables": [{"uuid": "rate"}, {"uuid": "gate"}], + }) + assert second.args == ("/v2/timeseries", {**first.args[1], "cursor": "continuation"}) + + +@pytest.mark.parametrize("metadata", [None, [], "bad", {}, + {"next_cursor": ""}, {"next_cursor": 1}, {"next_cursor": False}, + {"next_cursor": []}, {"next_cursor": {}}]) +def test_invalid_pagination_metadata(timeseries, metadata) -> None: + gateway, client, args = timeseries + client.post_json.return_value.body["metadata"] = metadata + with pytest.raises(ValueError, match="Invalid timeseries response.*next_cursor"): + gateway.get_material_samples(**args) + client.post_json.assert_called_once() + + +@pytest.mark.parametrize("cursors", [["a", "a"], ["a", "b", "a"]]) +def test_repeated_pagination_cursor(timeseries, cursors) -> None: + gateway, client, args = timeseries + data = client.post_json.return_value.body["data"] + client.post_json.side_effect = [ + Mock(body={"data": data, "metadata": {"next_cursor": cursor}}) for cursor in cursors + ] + with pytest.raises(ValueError, match="next_cursor has already been followed"): + gateway.get_material_samples(**args) + assert client.post_json.call_count == len(cursors) + + +@pytest.mark.parametrize("data, error", [ + ({"columns": ["time", "rate", "gate"], "records": [[]]}, "malformed record 0"), + ({"columns": ["time", "rate", "rate", "gate"], "records": []}, "required column"), + ({"columns": ["time", "rate"], "records": []}, "required column"), + ({"columns": None, "records": []}, "columns must be a list"), + ({"columns": ["time", "rate", "gate"], "records": None}, "records must be a list"), + ({"columns": ["time", "rate", "gate"], + "records": [["2026-09-03T01:00:00", 10, 2]]}, "timezone-aware"), + ({"columns": ["time", "rate", "gate"], + "records": [["2026-09-03T01:00:00Z", float("nan"), 2]]}, "finite"), + ({"columns": ["time", "rate", "gate"], + "records": [["2026-09-03T01:00:00Z", 10, True]]}, "numeric"), +]) +def test_malformed_later_page(timeseries, data, error) -> None: + gateway, client, args = timeseries + first_body = client.post_json.return_value.body + first_body["metadata"] = {"next_cursor": "continuation"} + client.post_json.side_effect = [ + Mock(body=first_body), Mock(body={"data": data, "metadata": {"next_cursor": None}}), + ] + with pytest.raises(ValueError, match=f"Invalid timeseries response on page 2:.*{error}"): + gateway.get_material_samples(**args) + assert client.post_json.call_count == 2