From 1de57d3d174a23ef75f529243ceaa944acece35a Mon Sep 17 00:00:00 2001 From: jallegri <18682329+jallegri@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:56:43 -0300 Subject: [PATCH] fix(management): align retention and continuous-query contracts --- CHANGELOG.md | 3 + src/arc_client/__init__.py | 2 + src/arc_client/_sentinels.py | 12 + .../management/async_continuous_query.py | 128 ++++++++-- src/arc_client/management/async_retention.py | 85 +++++-- src/arc_client/management/continuous_query.py | 128 ++++++++-- src/arc_client/management/retention.py | 85 +++++-- src/arc_client/models/continuous_query.py | 11 +- src/arc_client/models/retention.py | 16 +- tests/unit/http/test_cq_contracts.py | 145 +++++++++++ tests/unit/http/test_retention_contracts.py | 229 ++++++++++++++++++ 11 files changed, 761 insertions(+), 83 deletions(-) create mode 100644 src/arc_client/_sentinels.py create mode 100644 tests/unit/http/test_cq_contracts.py create mode 100644 tests/unit/http/test_retention_contracts.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ab5948b..4baaefc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,9 @@ plus bug fixes and the first real HTTP-layer test coverage. ### Fixed +- Retention and continuous-query clients accept Arc's list/null/flat response + shapes. Sparse updates remain sparse; explicit full replacements support + released servers, and CLEAR requests explicit null values on newer servers. - Buffered close waits for active writes, retains unacknowledged batches and surfaces failures without replay. Typed errors retain structured diagnostics. diff --git a/src/arc_client/__init__.py b/src/arc_client/__init__.py index fcbf45b..0118e42 100644 --- a/src/arc_client/__init__.py +++ b/src/arc_client/__init__.py @@ -5,6 +5,7 @@ # failure during package initialization. __version__ = "0.2.0" +from arc_client._sentinels import CLEAR from arc_client.async_client import AsyncArcClient from arc_client.client import ArcClient from arc_client.compression import Compression @@ -27,6 +28,7 @@ ) __all__ = [ + "CLEAR", # Clients "ArcClient", "AsyncArcClient", diff --git a/src/arc_client/_sentinels.py b/src/arc_client/_sentinels.py new file mode 100644 index 0000000..4cf0dce --- /dev/null +++ b/src/arc_client/_sentinels.py @@ -0,0 +1,12 @@ +"""Explicit clearing for nullable fields in partial updates.""" + +from enum import Enum + + +class ClearType(Enum): + """Marker for an explicit JSON null, distinct from an omitted argument.""" + + CLEAR = "CLEAR" + + +CLEAR = ClearType.CLEAR diff --git a/src/arc_client/management/async_continuous_query.py b/src/arc_client/management/async_continuous_query.py index fc3e1fb..e07e5fd 100644 --- a/src/arc_client/management/async_continuous_query.py +++ b/src/arc_client/management/async_continuous_query.py @@ -5,8 +5,9 @@ from datetime import datetime from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.async_http import AsyncHTTPClient from arc_client.models.continuous_query import ( ContinuousQuery, @@ -34,6 +35,8 @@ async def create( retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: bool = True, + *, + tag_columns: Optional[List[str]] = None, ) -> ContinuousQuery: """Create a new continuous query.""" payload = { @@ -47,13 +50,14 @@ async def create( "retention_days": retention_days, "delete_source_after_days": delete_source_after_days, "is_active": is_active, + "tag_columns": tag_columns, } try: response = await self._http.post("/api/v1/continuous_queries", json=payload) data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,8 +76,21 @@ async def list( response = await self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) - return [ContinuousQuery(**q) for q in queries] + if data is None: + queries = [] + elif isinstance(data, list): + queries = data + elif isinstance(data, dict) and "queries" in data: + queries = data["queries"] + if queries is None: + queries = [] + else: + raise ArcError("Malformed continuous query list: expected an array") + if not isinstance(queries, list): + raise ArcError("Malformed continuous query list: expected an array") + return [ContinuousQuery.from_response(q) for q in queries] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +104,7 @@ async def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -97,30 +114,85 @@ async def update( self, query_id: int, name: Optional[str] = None, - description: Optional[str] = None, + description: Union[str, ClearType, None] = None, query: Optional[str] = None, interval: Optional[str] = None, - retention_days: Optional[int] = None, - delete_source_after_days: Optional[int] = None, + retention_days: Union[int, ClearType, None] = None, + delete_source_after_days: Union[int, ClearType, None] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, + tag_columns: Union[List[str], ClearType, None] = None, ) -> ContinuousQuery: - """Update a continuous query.""" + """Send a partial update without reading or replacing unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for nullable fields. Servers with full-replacement + PUT semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "description": description, + "query": query, + "interval": interval, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "tag_columns": tag_columns, + } + nullable = {"description", "tag_columns", "retention_days", "delete_source_after_days"} payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key not in nullable: + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return await self._put_definition(query_id, payload) + + async def replace( + self, + query_id: int, + *, + name: str, + database: str, + source_measurement: str, + destination_measurement: str, + query: str, + interval: str, + description: Optional[str], + tag_columns: Optional[List[str]], + retention_days: Optional[int], + delete_source_after_days: Optional[int], + is_active: bool, + ) -> ContinuousQuery: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "description": description, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + return await self._put_definition(query_id, payload) + async def _put_definition(self, query_id: int, payload: dict[str, Any]) -> ContinuousQuery: try: response = await self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) data = response.json() @@ -129,8 +201,8 @@ async def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) - except (ArcNotFoundError, ArcError): + return ContinuousQuery.from_response(data) + except ArcError: raise except Exception as e: raise ArcError(f"Failed to update continuous query: {e}") from e @@ -186,6 +258,12 @@ async def get_executions(self, query_id: int, limit: int = 50) -> List[CQExecuti ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed CQ history: expected an array") return [CQExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get CQ executions: {e}") from e diff --git a/src/arc_client/management/async_retention.py b/src/arc_client/management/async_retention.py index 225a8d2..4229b08 100644 --- a/src/arc_client/management/async_retention.py +++ b/src/arc_client/management/async_retention.py @@ -2,10 +2,11 @@ from __future__ import annotations -from typing import Any, List, Optional +from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.async_http import AsyncHTTPClient from arc_client.models.retention import ( ExecuteRetentionResponse, @@ -55,8 +56,14 @@ async def list(self) -> List[RetentionPolicy]: try: response = await self._http.get("/api/v1/retention") data = response.json() - policies = data.get("policies", []) + policies = data.get("policies", data) if isinstance(data, dict) else data + if policies is None: + policies = [] + if not isinstance(policies, list): + raise ArcError("Malformed retention policy list: expected an array") return [RetentionPolicy(**p) for p in policies] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list retention policies: {e}") from e @@ -81,23 +88,63 @@ async def update( policy_id: int, name: Optional[str] = None, retention_days: Optional[int] = None, - measurement: Optional[str] = None, + measurement: Union[str, ClearType, None] = None, buffer_days: Optional[int] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, ) -> RetentionPolicy: - """Update a retention policy.""" + """Send a partial update without reading unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for measurement. Servers with full-replacement PUT + semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + "database": database, + } payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if retention_days is not None: - payload["retention_days"] = retention_days - if measurement is not None: - payload["measurement"] = measurement - if buffer_days is not None: - payload["buffer_days"] = buffer_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key != "measurement": + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return await self._put_definition(policy_id, payload) + async def replace( + self, + policy_id: int, + *, + name: str, + database: str, + retention_days: int, + measurement: Optional[str], + buffer_days: int, + is_active: bool, + ) -> RetentionPolicy: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + } + return await self._put_definition(policy_id, payload) + + async def _put_definition(self, policy_id: int, payload: dict[str, Any]) -> RetentionPolicy: try: response = await self._http.put(f"/api/v1/retention/{policy_id}", json=payload) data = response.json() @@ -136,6 +183,8 @@ async def execute( response = await self._http.post(f"/api/v1/retention/{policy_id}/execute", json=payload) data = response.json() return ExecuteRetentionResponse(**data) + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to execute retention policy: {e}") from e @@ -148,6 +197,12 @@ async def get_executions(self, policy_id: int, limit: int = 50) -> List[Retentio ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed retention execution history: expected an array") return [RetentionExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get retention executions: {e}") from e diff --git a/src/arc_client/management/continuous_query.py b/src/arc_client/management/continuous_query.py index 6f0196a..601ae6e 100644 --- a/src/arc_client/management/continuous_query.py +++ b/src/arc_client/management/continuous_query.py @@ -5,8 +5,9 @@ from datetime import datetime from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.continuous_query import ( ContinuousQuery, @@ -34,6 +35,8 @@ def create( retention_days: Optional[int] = None, delete_source_after_days: Optional[int] = None, is_active: bool = True, + *, + tag_columns: Optional[List[str]] = None, ) -> ContinuousQuery: """Create a new continuous query.""" payload = { @@ -47,13 +50,14 @@ def create( "retention_days": retention_days, "delete_source_after_days": delete_source_after_days, "is_active": is_active, + "tag_columns": tag_columns, } try: response = self._http.post("/api/v1/continuous_queries", json=payload) data = response.json() if not data.get("success", True): raise ArcError(data.get("error", "Failed to create CQ")) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except ArcError: raise except Exception as e: @@ -72,8 +76,21 @@ def list( response = self._http.get("/api/v1/continuous_queries", params=params) data = response.json() - queries = data.get("queries", []) - return [ContinuousQuery(**q) for q in queries] + if data is None: + queries = [] + elif isinstance(data, list): + queries = data + elif isinstance(data, dict) and "queries" in data: + queries = data["queries"] + if queries is None: + queries = [] + else: + raise ArcError("Malformed continuous query list: expected an array") + if not isinstance(queries, list): + raise ArcError("Malformed continuous query list: expected an array") + return [ContinuousQuery.from_response(q) for q in queries] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list continuous queries: {e}") from e @@ -87,7 +104,7 @@ def get(self, query_id: int) -> ContinuousQuery: if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) + return ContinuousQuery.from_response(data) except (ArcNotFoundError, ArcError): raise except Exception as e: @@ -97,30 +114,85 @@ def update( self, query_id: int, name: Optional[str] = None, - description: Optional[str] = None, + description: Union[str, ClearType, None] = None, query: Optional[str] = None, interval: Optional[str] = None, - retention_days: Optional[int] = None, - delete_source_after_days: Optional[int] = None, + retention_days: Union[int, ClearType, None] = None, + delete_source_after_days: Union[int, ClearType, None] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, + source_measurement: Optional[str] = None, + destination_measurement: Optional[str] = None, + tag_columns: Union[List[str], ClearType, None] = None, ) -> ContinuousQuery: - """Update a continuous query.""" + """Send a partial update without reading or replacing unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for nullable fields. Servers with full-replacement + PUT semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "description": description, + "query": query, + "interval": interval, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "tag_columns": tag_columns, + } + nullable = {"description", "tag_columns", "retention_days", "delete_source_after_days"} payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if description is not None: - payload["description"] = description - if query is not None: - payload["query"] = query - if interval is not None: - payload["interval"] = interval - if retention_days is not None: - payload["retention_days"] = retention_days - if delete_source_after_days is not None: - payload["delete_source_after_days"] = delete_source_after_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key not in nullable: + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return self._put_definition(query_id, payload) + + def replace( + self, + query_id: int, + *, + name: str, + database: str, + source_measurement: str, + destination_measurement: str, + query: str, + interval: str, + description: Optional[str], + tag_columns: Optional[List[str]], + retention_days: Optional[int], + delete_source_after_days: Optional[int], + is_active: bool, + ) -> ContinuousQuery: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "source_measurement": source_measurement, + "destination_measurement": destination_measurement, + "query": query, + "interval": interval, + "description": description, + "tag_columns": tag_columns, + "retention_days": retention_days, + "delete_source_after_days": delete_source_after_days, + "is_active": is_active, + } + return self._put_definition(query_id, payload) + def _put_definition(self, query_id: int, payload: dict[str, Any]) -> ContinuousQuery: try: response = self._http.put(f"/api/v1/continuous_queries/{query_id}", json=payload) data = response.json() @@ -129,8 +201,8 @@ def update( if "not found" in error.lower(): raise ArcNotFoundError(f"CQ {query_id} not found") raise ArcError(error) - return ContinuousQuery(**data.get("query", data)) - except (ArcNotFoundError, ArcError): + return ContinuousQuery.from_response(data) + except ArcError: raise except Exception as e: raise ArcError(f"Failed to update continuous query: {e}") from e @@ -186,6 +258,12 @@ def get_executions(self, query_id: int, limit: int = 50) -> List[CQExecution]: ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed CQ history: expected an array") return [CQExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get CQ executions: {e}") from e diff --git a/src/arc_client/management/retention.py b/src/arc_client/management/retention.py index f7cf361..fb0128a 100644 --- a/src/arc_client/management/retention.py +++ b/src/arc_client/management/retention.py @@ -2,10 +2,11 @@ from __future__ import annotations -from typing import Any, List, Optional +from typing import Any, List, Optional, Union +from arc_client._sentinels import CLEAR, ClearType from arc_client.config import ClientConfig -from arc_client.exceptions import ArcError, ArcNotFoundError +from arc_client.exceptions import ArcError, ArcNotFoundError, ArcValidationError from arc_client.http.sync_http import SyncHTTPClient from arc_client.models.retention import ( ExecuteRetentionResponse, @@ -55,8 +56,14 @@ def list(self) -> List[RetentionPolicy]: try: response = self._http.get("/api/v1/retention") data = response.json() - policies = data.get("policies", []) + policies = data.get("policies", data) if isinstance(data, dict) else data + if policies is None: + policies = [] + if not isinstance(policies, list): + raise ArcError("Malformed retention policy list: expected an array") return [RetentionPolicy(**p) for p in policies] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to list retention policies: {e}") from e @@ -81,23 +88,63 @@ def update( policy_id: int, name: Optional[str] = None, retention_days: Optional[int] = None, - measurement: Optional[str] = None, + measurement: Union[str, ClearType, None] = None, buffer_days: Optional[int] = None, is_active: Optional[bool] = None, + *, + database: Optional[str] = None, ) -> RetentionPolicy: - """Update a retention policy.""" + """Send a partial update without reading unrelated fields. + + None omits an argument, preserving existing caller behavior. CLEAR + sends JSON null for measurement. Servers with full-replacement PUT + semantics (including Arc 26.09.3) require replace() instead. + """ + values = { + "name": name, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + "database": database, + } payload: dict[str, Any] = {} - if name is not None: - payload["name"] = name - if retention_days is not None: - payload["retention_days"] = retention_days - if measurement is not None: - payload["measurement"] = measurement - if buffer_days is not None: - payload["buffer_days"] = buffer_days - if is_active is not None: - payload["is_active"] = is_active + for key, value in values.items(): + if value is CLEAR: + if key != "measurement": + raise ArcValidationError(f"{key} may not be cleared") + payload[key] = None + elif value is not None: + payload[key] = value + return self._put_definition(policy_id, payload) + def replace( + self, + policy_id: int, + *, + name: str, + database: str, + retention_days: int, + measurement: Optional[str], + buffer_days: int, + is_active: bool, + ) -> RetentionPolicy: + """Replace all mutable fields with one PUT on old or new servers. + + Every field must be explicit. This operation intentionally overwrites + the definition; it does not fetch or merge a potentially stale copy. + """ + payload = { + "name": name, + "database": database, + "retention_days": retention_days, + "measurement": measurement, + "buffer_days": buffer_days, + "is_active": is_active, + } + return self._put_definition(policy_id, payload) + + def _put_definition(self, policy_id: int, payload: dict[str, Any]) -> RetentionPolicy: try: response = self._http.put(f"/api/v1/retention/{policy_id}", json=payload) data = response.json() @@ -136,6 +183,8 @@ def execute( response = self._http.post(f"/api/v1/retention/{policy_id}/execute", json=payload) data = response.json() return ExecuteRetentionResponse(**data) + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to execute retention policy: {e}") from e @@ -148,6 +197,12 @@ def get_executions(self, policy_id: int, limit: int = 50) -> List[RetentionExecu ) data = response.json() executions = data.get("executions", []) + if executions is None: + executions = [] + if not isinstance(executions, list): + raise ArcError("Malformed retention execution history: expected an array") return [RetentionExecution(**e) for e in executions] + except ArcError: + raise except Exception as e: raise ArcError(f"Failed to get retention executions: {e}") from e diff --git a/src/arc_client/models/continuous_query.py b/src/arc_client/models/continuous_query.py index 04909d2..de435ff 100644 --- a/src/arc_client/models/continuous_query.py +++ b/src/arc_client/models/continuous_query.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Optional +from typing import Any, List, Optional from pydantic import BaseModel @@ -18,6 +18,7 @@ class ContinuousQuery(BaseModel): destination_measurement: str query: str interval: str + tag_columns: Optional[List[str]] = None retention_days: Optional[int] = None delete_source_after_days: Optional[int] = None is_active: bool = True @@ -28,6 +29,14 @@ class ContinuousQuery(BaseModel): created_at: Optional[str] = None updated_at: Optional[str] = None + @classmethod + def from_response(cls, data: dict[str, Any]) -> ContinuousQuery: + """Accept flat definitions and legacy mapping envelopes, never SQL text.""" + if not isinstance(data, dict): + raise ValueError("Expected a continuous query object") + nested = data.get("query") + return cls(**nested) if isinstance(nested, dict) else cls(**data) + class ExecuteCQResponse(BaseModel): """Response from executing a continuous query.""" diff --git a/src/arc_client/models/retention.py b/src/arc_client/models/retention.py index 4fbfc45..ecfeb0f 100644 --- a/src/arc_client/models/retention.py +++ b/src/arc_client/models/retention.py @@ -2,9 +2,9 @@ from __future__ import annotations -from typing import Optional +from typing import Any, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class RetentionPolicy(BaseModel): @@ -31,12 +31,24 @@ class ExecuteRetentionResponse(BaseModel): policy_name: str deleted_count: int = 0 files_deleted: int = 0 + skipped_files: int = 0 + skipped_reason: str = "" + cold_files_deleted: int = 0 + cold_rows_deleted: int = 0 + cold_files_skipped: int = 0 + cold_skipped_reason: str = "" execution_time_ms: float = 0.0 dry_run: bool = False cutoff_date: Optional[str] = None affected_measurements: list[str] = Field(default_factory=list) error: Optional[str] = None + @field_validator("affected_measurements", mode="before") + @classmethod + def normalize_measurements(cls, value: Any) -> Any: + """Treat a nil Go slice as empty, preserving validation for other values.""" + return [] if value is None else value + class RetentionExecution(BaseModel): """A retention execution history record.""" diff --git a/tests/unit/http/test_cq_contracts.py b/tests/unit/http/test_cq_contracts.py new file mode 100644 index 0000000..5fe63e5 --- /dev/null +++ b/tests/unit/http/test_cq_contracts.py @@ -0,0 +1,145 @@ +"""Continuous-query wire contracts shared by sync and async public clients.""" + +import inspect +import json +from contextlib import AsyncExitStack + +import pytest + +from arc_client import CLEAR, ArcClient, AsyncArcClient +from arc_client.exceptions import ArcPermissionError, ArcValidationError + +DEFINITION = { + "id": 7, + "name": "rollup", + "database": "metrics", + "source_measurement": "cpu", + "destination_measurement": "cpu_hour", + "query": "SELECT avg(value) FROM metrics.cpu", + "interval": "1h", + "description": "keep", + "tag_columns": ["host"], + "retention_days": 30, + "delete_source_after_days": 60, + "is_active": True, +} + + +async def call(client, method, *args, **kwargs): + result = getattr(client.continuous_queries, method)(*args, **kwargs) + return await result if inspect.isawaitable(result) else result + + +@pytest.fixture(params=[False, True], ids=["sync", "async"]) +def async_mode(request): + return request.param + + +async def enter_client(stack, async_mode): + if async_mode: + return await stack.enter_async_context(AsyncArcClient(token="test", max_retries=0)) + return stack.enter_context(ArcClient(token="test", max_retries=0)) + + +@pytest.mark.parametrize("body", [None, [], {"queries": None}, {"queries": []}]) +async def test_empty_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response( + method="GET", + content=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "list") == [] + + +async def test_empty_execution_history(httpx_mock, async_mode): + httpx_mock.add_response(method="GET", json={"executions": None}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "get_executions", 7) == [] + + +async def test_create_flat_definition_and_tags(httpx_mock, async_mode): + httpx_mock.add_response(method="POST", json=DEFINITION) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + args = { + k: DEFINITION[k] + for k in ( + "name", + "database", + "source_measurement", + "destination_measurement", + "query", + "interval", + "tag_columns", + ) + } + result = await call(client, "create", **args) + assert result.query == DEFINITION["query"] + assert result.tag_columns == ["host"] + assert json.loads(httpx_mock.get_request().content)["tag_columns"] == ["host"] + + +async def test_sparse_update_no_get_preserves_false_zero_and_clear(httpx_mock, async_mode): + httpx_mock.add_response( + method="PUT", json={**DEFINITION, "is_active": False, "description": None} + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call( + client, + "update", + 7, + is_active=False, + description=CLEAR, + retention_days=0, + tag_columns=[], + database=None, + ) + assert not result.is_active + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert json.loads(requests[0].content) == { + "is_active": False, + "description": None, + "retention_days": 0, + "tag_columns": [], + } + + +async def test_replace_sends_complete_definition_once(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json=DEFINITION) + args = {k: v for k, v in DEFINITION.items() if k != "id"} + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "replace", 7, **args) + assert result.tag_columns == ["host"] + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert json.loads(requests[0].content) == args + + +async def test_none_omits_existing_arguments(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json=DEFINITION) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + await call(client, "update", 7, description=None, retention_days=None) + assert json.loads(httpx_mock.get_request().content) == {} + + +async def test_permissions_remain_typed(httpx_mock, async_mode): + httpx_mock.add_response(status_code=403, json={"error": "Permission denied: read required"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcPermissionError): + await call(client, "list") + + +async def test_clear_nonnullable_rejected_before_request(async_mode): + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcValidationError): + await call(client, "update", 7, name=CLEAR) diff --git a/tests/unit/http/test_retention_contracts.py b/tests/unit/http/test_retention_contracts.py new file mode 100644 index 0000000..ba73606 --- /dev/null +++ b/tests/unit/http/test_retention_contracts.py @@ -0,0 +1,229 @@ +"""Retention wire contracts exercised through both public clients.""" + +import inspect +import json +from contextlib import AsyncExitStack + +import pytest + +from arc_client import CLEAR, ArcClient, AsyncArcClient +from arc_client.exceptions import ArcError, ArcPermissionError, ArcValidationError + +DEFINITION = { + "id": 7, + "name": "cleanup", + "database": "metrics", + "measurement": "cpu", + "retention_days": 30, + "buffer_days": 7, + "is_active": True, +} +REPLACEMENT = {key: value for key, value in DEFINITION.items() if key != "id"} + + +async def call(client, method, *args, **kwargs): + result = getattr(client.retention, method)(*args, **kwargs) + return await result if inspect.isawaitable(result) else result + + +@pytest.fixture(params=[False, True], ids=["sync", "async"]) +def async_mode(request): + return request.param + + +async def enter_client(stack, async_mode): + if async_mode: + return await stack.enter_async_context(AsyncArcClient(token="test", max_retries=0)) + return stack.enter_context(ArcClient(token="test", max_retries=0)) + + +@pytest.mark.parametrize("body", [None, [], {"policies": None}, {"policies": []}]) +async def test_empty_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response(content=json.dumps(body).encode()) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "list") == [] + + +@pytest.mark.parametrize("body", [[DEFINITION], {"policies": [DEFINITION]}]) +async def test_populated_list_shapes(httpx_mock, async_mode, body): + httpx_mock.add_response(json=body) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert (await call(client, "list"))[0].measurement == "cpu" + + +@pytest.mark.parametrize("body", [False, 0, "", {}, {"policies": False}, {"policies": {}}]) +async def test_malformed_lists_rejected(httpx_mock, async_mode, body): + httpx_mock.add_response(json=body) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "list") + + +@pytest.mark.parametrize("executions", [None, []]) +async def test_empty_history(httpx_mock, async_mode, executions): + httpx_mock.add_response(json={"executions": executions}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + assert await call(client, "get_executions", 7, limit=3) == [] + assert httpx_mock.get_request().url.params["limit"] == "3" + + +@pytest.mark.parametrize("executions", [False, 0, "", {}]) +async def test_malformed_history_rejected(httpx_mock, async_mode, executions): + httpx_mock.add_response(json={"executions": executions}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "get_executions", 7) + + +@pytest.mark.parametrize("measurement", [CLEAR, ""]) +async def test_sparse_update_single_put(httpx_mock, async_mode, measurement): + httpx_mock.add_response(method="PUT", json={**DEFINITION, "is_active": False}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call( + client, + "update", + 7, + name="", + retention_days=0, + measurement=measurement, + buffer_days=0, + is_active=False, + database="other", + ) + assert result.is_active is False + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert requests[0].url.path == "/api/v1/retention/7" + assert json.loads(requests[0].content) == { + "name": "", + "retention_days": 0, + "measurement": None if measurement is CLEAR else "", + "buffer_days": 0, + "is_active": False, + "database": "other", + } + + +async def test_none_omits_and_positional_arguments_stay_compatible(httpx_mock, async_mode): + httpx_mock.add_response(method="PUT", json={"policy": DEFINITION}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + await call(client, "update", 7, None, None, None, None, None, database=None) + assert json.loads(httpx_mock.get_request().content) == {} + + +@pytest.mark.parametrize( + "field", ["name", "database", "retention_days", "buffer_days", "is_active"] +) +async def test_clear_nonnullable_rejected_before_http(httpx_mock, async_mode, field): + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcValidationError): + await call(client, "update", 7, **{field: CLEAR}) + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("measurement", [None, "cpu"]) +async def test_replace_complete_single_put(httpx_mock, async_mode, measurement): + values = {**REPLACEMENT, "measurement": measurement, "buffer_days": 0, "is_active": False} + httpx_mock.add_response(method="PUT", json={"id": 7, **values}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "replace", 7, **values) + assert result.measurement == measurement + requests = httpx_mock.get_requests() + assert len(requests) == 1 + assert requests[0].method == "PUT" + assert json.loads(requests[0].content) == values + + +@pytest.mark.parametrize("field", list(REPLACEMENT)) +async def test_replace_requires_every_field(httpx_mock, async_mode, field): + values = {key: value for key, value in REPLACEMENT.items() if key != field} + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(TypeError): + await call(client, "replace", 7, **values) + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize( + "method,args,kwargs", + [ + ("create", (), REPLACEMENT), + ("get", (7,), {}), + ("list", (), {}), + ("update", (7,), {"is_active": False}), + ("replace", (7,), REPLACEMENT), + ("delete", (7,), {}), + ("execute", (7,), {}), + ("get_executions", (7,), {}), + ], +) +async def test_permissions_remain_typed_without_fallback( + httpx_mock, async_mode, method, args, kwargs +): + httpx_mock.add_response(status_code=403, json={"error": "Permission denied: read required"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcPermissionError): + await call(client, method, *args, **kwargs) + assert len(httpx_mock.get_requests()) == 1 + + +async def test_execution_diagnostics_and_null_measurements(httpx_mock, async_mode): + diagnostics = { + "skipped_files": 2, + "skipped_reason": "unaddressable files", + "cold_files_deleted": 3, + "cold_rows_deleted": 42, + "cold_files_skipped": 4, + "cold_skipped_reason": "cold tier unavailable", + } + httpx_mock.add_response( + json={ + "policy_id": 7, + "policy_name": "cleanup", + "affected_measurements": None, + "dry_run": True, + **diagnostics, + } + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "execute", 7) + assert result.affected_measurements == [] + assert {field: getattr(result, field) for field in diagnostics} == diagnostics + assert json.loads(httpx_mock.get_request().content) == {"dry_run": True, "confirm": False} + + +async def test_older_execution_defaults(httpx_mock, async_mode): + httpx_mock.add_response(json={"policy_id": 7, "policy_name": "cleanup"}) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + result = await call(client, "execute", 7) + assert result.skipped_files == result.cold_files_deleted == result.cold_rows_deleted == 0 + assert result.cold_files_skipped == 0 + assert result.skipped_reason == result.cold_skipped_reason == "" + + +@pytest.mark.parametrize("measurements", [False, 0, "", {}]) +async def test_malformed_measurements_rejected(httpx_mock, async_mode, measurements): + httpx_mock.add_response( + json={ + "policy_id": 7, + "policy_name": "cleanup", + "affected_measurements": measurements, + } + ) + async with AsyncExitStack() as stack: + client = await enter_client(stack, async_mode) + with pytest.raises(ArcError): + await call(client, "execute", 7)