diff --git a/src/confluent_sql/__init__.py b/src/confluent_sql/__init__.py index aebd346..47faff5 100644 --- a/src/confluent_sql/__init__.py +++ b/src/confluent_sql/__init__.py @@ -58,6 +58,7 @@ TableflowPhase, TableflowTopic, TableflowTopicConfig, + TableflowTopicSpec, TableFormat, ) from .types import PropertiesDict, PropertiesMapping, SqlNone, YearMonthInterval @@ -103,6 +104,7 @@ "TableflowPhase", "TableflowTopic", "TableflowTopicConfig", + "TableflowTopicSpec", "ManagedStorage", "ByobAwsStorage", "AzureAdlsStorage", diff --git a/src/confluent_sql/connection.py b/src/confluent_sql/connection.py index bc63c85..981f38b 100644 --- a/src/confluent_sql/connection.py +++ b/src/confluent_sql/connection.py @@ -53,6 +53,7 @@ TableflowTopicConfig, TableFormat, build_create_payload, + build_update_payload, normalize_table_formats, ) from .types import PropertiesDict, RowPythonTypes, StrAnyDict @@ -2205,6 +2206,97 @@ def get_tableflow(self, table_name: str) -> TableflowTopic: ) from e return TableflowTopic.from_response(response.json()) + def update_tableflow( + self, + table_name: str, + *, + table_formats: TableFormat | Collection[TableFormat] | None = None, + config: TableflowTopicConfig | None = None, + wait_for_running: bool = True, + timeout: float = 300, + ) -> TableflowTopic: + """Update an already-enabled Tableflow topic's `table_formats`/`config`. + + `PATCH /tableflow/v1/tableflow-topics/{display_name}`. `table_formats`/`config` are the + only fields updatable this way -- `storage`/`display_name` are `x-immutable`; a caller + needing to change either has no in-place path and must recreate the topic instead. + + `None` means "leave unchanged," for this method's own two arguments and for each of + `config`'s own sub-fields (same convention `config.to_spec()` already uses for create) -- + not "clear it": none of `retention_ms`/`data_retention_ms`/`error_handling` can actually + be unset via this API (each has a server-enforced default, and the server rejects an + explicit null for any of them), so there's no supported way to request a delete, and this + method doesn't attempt one. This method itself doesn't diff against the topic's current + live state -- that's the caller's job (comparing against `get_tableflow`'s response); + it just sends whatever `table_formats`/`config` it's given, same as `enable_tableflow`. + + Args: + table_name: The Flink table / Kafka topic name (the {display_name} path segment). + table_formats: New table_formats, replacing the full list -- or None to leave + unchanged. + config: New topic-level config -- or None to leave unchanged. Only fields actually + set on it (not None) are changed; the rest are left as-is. + wait_for_running: If True (default), poll until the topic reaches RUNNING, raising on + FAILED. If False, return as soon as the update is accepted. + timeout: Maximum seconds to wait when wait_for_running is True. + + Returns: + The TableflowTopic -- RUNNING when waited (the default), otherwise the just-accepted + topic (typically PENDING while formats/config changes roll out). + + Raises: + InterfaceError: If table_formats and config are both None (nothing to update), or if + table_formats is empty or names an unknown format. + ProgrammingError: If no control-plane credential is available, or the cluster id + can't be resolved without a global key. + TableflowTopicNotFoundError: If Tableflow is not enabled for the topic (HTTP 404). + OperationalError: On other API errors, on FAILED during a wait, or on wait timeout. + """ + # Validate/normalize before any network work (including the possibly CMK-resolving + # cluster-id lookup below), same as enable_tableflow. + wire_formats = normalize_table_formats(table_formats) if table_formats is not None else None + config_spec = config.to_spec() if config is not None else None + if not wire_formats and not config_spec: + raise InterfaceError("update_tableflow requires at least one of table_formats/config") + + payload = build_update_payload( + table_formats=wire_formats, + config_spec=config_spec, + environment_id=self.environment_id, + kafka_cluster_id=self._resolve_kafka_cluster_id(), + ) + logger.info(f"Updating Tableflow for table '{table_name}'") + response = self._tableflow_request( + f"{self._TABLEFLOW_TOPICS_PATH}/{table_name}", + method="PATCH", + json=payload, + raise_for_status=False, + ) + try: + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + raise TableflowTopicNotFoundError( + f"Tableflow is not enabled for table '{table_name}'", table_name=table_name + ) from e + try: + res = e.response.json() + errors = res.get("errors", []) + details = "; ".join(err["detail"] for err in errors if err.get("detail")) + except Exception: + details = "" + if not details: + details = "no more details" + raise OperationalError( + f"Error updating Tableflow topic: {details}", + http_status_code=e.response.status_code, + ) from e + + topic = TableflowTopic.from_response(response.json()) + if wait_for_running: + return self._wait_for_tableflow_running(topic, timeout) + return topic + def disable_tableflow( self, table_name: str, diff --git a/src/confluent_sql/tableflow.py b/src/confluent_sql/tableflow.py index fa84811..b7dcb2d 100644 --- a/src/confluent_sql/tableflow.py +++ b/src/confluent_sql/tableflow.py @@ -19,6 +19,46 @@ from .types import StrAnyDict +class Fields: + """Wire field names for the Tableflow Topic API + + Plain string class attributes, not an Enum: `StrEnum` needs Python 3.11+ (newer than this + package's floor, 3.10), and the older `class X(str, Enum)` mixin (used above for `TableFormat`/ + `TableflowPhase`) returns the qualified member name from `str()`/default formatting rather + than the plain value unless a caller remembers `.value` -- a real risk given how pervasively + these get used directly as dict keys, in f-strings, and in JSON payloads. Plain strings have + no such gotcha. + """ + + ID = "id" + DISPLAY_NAME = "display_name" + STORAGE = "storage" + TABLE_FORMATS = "table_formats" + ENVIRONMENT = "environment" + KAFKA_CLUSTER = "kafka_cluster" + CONFIG = "config" + SUSPENDED = "suspended" + + RETENTION_MS = "retention_ms" + DATA_RETENTION_MS = "data_retention_ms" + ERROR_HANDLING = "error_handling" + # Deliberately unmodeled by TableflowTopicConfig (deprecated/read-only), but still + # referenced by name by a caller masking them out of a diff. + ENABLE_COMPACTION = "enable_compaction" + ENABLE_PARTITIONING = "enable_partitioning" + RECORD_FAILURE_STRATEGY = "record_failure_strategy" + + KIND = "kind" + BUCKET_NAME = "bucket_name" + PROVIDER_INTEGRATION_ID = "provider_integration_id" + STORAGE_ACCOUNT_NAME = "storage_account_name" + CONTAINER_NAME = "container_name" + TABLE_PATH = "table_path" + + MODE = "mode" + TARGET = "target" + + class TableFormat(str, Enum): """A concrete table format a Tableflow topic materializes to. @@ -104,7 +144,7 @@ class TableflowStorage: def to_spec(self) -> StrAnyDict: """Render the writable storage fields to the wire `spec.storage` object.""" - return {"kind": self.kind} + return {Fields.KIND: self.kind} @dataclass(frozen=True) @@ -125,9 +165,9 @@ class ByobAwsStorage(TableflowStorage): def to_spec(self) -> StrAnyDict: return { - "kind": self.kind, - "bucket_name": self.bucket_name, - "provider_integration_id": self.provider_integration_id, + Fields.KIND: self.kind, + Fields.BUCKET_NAME: self.bucket_name, + Fields.PROVIDER_INTEGRATION_ID: self.provider_integration_id, } @@ -143,10 +183,10 @@ class AzureAdlsStorage(TableflowStorage): def to_spec(self) -> StrAnyDict: return { - "kind": self.kind, - "storage_account_name": self.storage_account_name, - "container_name": self.container_name, - "provider_integration_id": self.provider_integration_id, + Fields.KIND: self.kind, + Fields.STORAGE_ACCOUNT_NAME: self.storage_account_name, + Fields.CONTAINER_NAME: self.container_name, + Fields.PROVIDER_INTEGRATION_ID: self.provider_integration_id, } @@ -156,19 +196,19 @@ def storage_from_spec(data: StrAnyDict) -> TableflowStorage: Captures only the writable fields; server-assigned read-only fields (`table_path`, `bucket_region`, `storage_region`) remain available on the topic's raw spec dict. """ - kind = data.get("kind") + kind = data.get(Fields.KIND) if kind == ManagedStorage.kind: return ManagedStorage() if kind == ByobAwsStorage.kind: return ByobAwsStorage( - bucket_name=data["bucket_name"], - provider_integration_id=data["provider_integration_id"], + bucket_name=data[Fields.BUCKET_NAME], + provider_integration_id=data[Fields.PROVIDER_INTEGRATION_ID], ) if kind == AzureAdlsStorage.kind: return AzureAdlsStorage( - storage_account_name=data["storage_account_name"], - container_name=data["container_name"], - provider_integration_id=data["provider_integration_id"], + storage_account_name=data[Fields.STORAGE_ACCOUNT_NAME], + container_name=data[Fields.CONTAINER_NAME], + provider_integration_id=data[Fields.PROVIDER_INTEGRATION_ID], ) raise OperationalError(f"Wacky -- unknown Tableflow storage kind '{kind}' in response") @@ -181,7 +221,7 @@ class TableflowErrorHandling: def to_spec(self) -> StrAnyDict: """Render to the wire `error_handling` object.""" - return {"mode": self.mode} + return {Fields.MODE: self.mode} @dataclass(frozen=True) @@ -207,7 +247,23 @@ class TableflowErrorHandlingLog(TableflowErrorHandling): target: str = "error_log" def to_spec(self) -> StrAnyDict: - return {"mode": self.mode, "target": self.target} + return {Fields.MODE: self.mode, Fields.TARGET: self.target} + + +def error_handling_from_spec(data: StrAnyDict) -> TableflowErrorHandling: + """Parse a response `config.error_handling` object into its typed error-handling class. + + Mirrors `storage_from_spec` -- only the mode-to-class dispatch is ours, and every mode's + field shape already round-trips through its own dataclass. + """ + mode = data.get(Fields.MODE) + if mode == TableflowErrorHandlingSuspend.mode: + return TableflowErrorHandlingSuspend() + if mode == TableflowErrorHandlingSkip.mode: + return TableflowErrorHandlingSkip() + if mode == TableflowErrorHandlingLog.mode: + return TableflowErrorHandlingLog(target=data.get(Fields.TARGET, "error_log")) + raise OperationalError(f"Wacky -- unknown Tableflow error-handling mode '{mode}' in response") @dataclass(frozen=True) @@ -219,18 +275,40 @@ class TableflowTopicConfig: `enable_compaction`/`enable_partitioning` flags are deliberately omitted. """ - retention_ms: str | int | None = None - data_retention_ms: str | int | None = None + retention_ms: int | None = None + data_retention_ms: int | None = None error_handling: TableflowErrorHandling | None = None + @classmethod + def from_spec(cls, data: StrAnyDict) -> TableflowTopicConfig: + """Parse a response `spec.config` object, dropping anything not formally modeled.""" + error_handling_conf = data.get(Fields.ERROR_HANDLING) + return cls( + retention_ms=optional_int_from_str(data.get(Fields.RETENTION_MS)), + data_retention_ms=optional_int_from_str(data.get(Fields.DATA_RETENTION_MS)), + error_handling=( + error_handling_from_spec(error_handling_conf) + if error_handling_conf is not None + else None + ), + ) + def to_spec(self) -> StrAnyDict: + """Render to the wire `config` object. + + `retention_ms`/`data_retention_ms` accept `int` here for caller convenience, but the API + schema types both as `string` (`format: int64`) on every request and response -- so + that's what's actually sent, even when constructed with an `int`. Without this, a value + that's genuinely unchanged could look different across a create/update payload and a GET + response purely from Python's `int`/`str` distinction, not a real difference on the wire. + """ spec: StrAnyDict = {} if self.retention_ms is not None: - spec["retention_ms"] = self.retention_ms + spec[Fields.RETENTION_MS] = str(self.retention_ms) if self.data_retention_ms is not None: - spec["data_retention_ms"] = self.data_retention_ms + spec[Fields.DATA_RETENTION_MS] = str(self.data_retention_ms) if self.error_handling is not None: - spec["error_handling"] = self.error_handling.to_spec() + spec[Fields.ERROR_HANDLING] = self.error_handling.to_spec() return spec @@ -250,16 +328,50 @@ def build_create_payload( empty config is omitted entirely. """ spec: StrAnyDict = { - "display_name": table_name, - "storage": storage.to_spec(), - "table_formats": table_formats, - "environment": {"id": environment_id}, - "kafka_cluster": {"id": kafka_cluster_id}, + Fields.DISPLAY_NAME: table_name, + Fields.STORAGE: storage.to_spec(), + Fields.TABLE_FORMATS: table_formats, + Fields.ENVIRONMENT: {Fields.ID: environment_id}, + Fields.KAFKA_CLUSTER: {Fields.ID: kafka_cluster_id}, } if config is not None: config_spec = config.to_spec() if config_spec: - spec["config"] = config_spec + spec[Fields.CONFIG] = config_spec + return {"spec": spec} + + +def build_update_payload( + *, + table_formats: list[str] | None, + config_spec: StrAnyDict | None, + environment_id: str, + kafka_cluster_id: str, +) -> StrAnyDict: + """Assemble the `PATCH /tableflow/v1/tableflow-topics/{display_name}` request body. + + `table_formats`/`config_spec` are the only fields updatable via this API (`storage`/ + `display_name` are `x-immutable`; `suspended` isn't part of `tableflow`'s config surface) -- + `None` (or an empty `config_spec`) means "leave unchanged," so it's omitted from the body + entirely rather than sent as `null`. Diffing to decide what's actually changing -- + comparing against a real `get_tableflow` response -- is the caller's job, not this driver's; + this function (and `Connection.update_tableflow`) just assembles what it's given, same as + every other Tableflow request-building function here. + + `environment` and `kafka_cluster` are both required routing/identity keys on this endpoint + (the path only carries `display_name`, which isn't unique on its own) -- not values being + changed. The API spec only marks `environment` required in the PATCH request schema, but + that's wrong in practice: `kafka_cluster` is required here too, the same as it is for + GET/DELETE. + """ + spec: StrAnyDict = { + Fields.ENVIRONMENT: {Fields.ID: environment_id}, + Fields.KAFKA_CLUSTER: {Fields.ID: kafka_cluster_id}, + } + if table_formats is not None: + spec[Fields.TABLE_FORMATS] = table_formats + if config_spec: + spec[Fields.CONFIG] = config_spec return {"spec": spec} @@ -303,17 +415,16 @@ def from_response(cls, data: StrAnyDict) -> TableflowTopicStatus: @dataclass class TableflowTopicSpec: - """Parsed topic spec; `table_formats` and `storage` are typed, `config` retained raw. - - The raw spec dict is kept (mirroring `Statement`). Config is left as a dict because its - response carries read-only fields (`enable_compaction`, `enable_partitioning`) the writable - `TableflowTopicConfig` doesn't model. + """Parsed topic spec, in the same shape whether it came from a real GET/create response or + was assembled locally to represent a desired state -- `table_formats`/`storage`/`config` are + all typed either way. The raw spec dict is kept (mirroring `Statement`) for anything not + formally modeled here. """ display_name: str table_formats: list[TableFormat] storage: TableflowStorage - config: StrAnyDict | None + config: TableflowTopicConfig | None environment_id: str | None kafka_cluster_id: str | None suspended: bool @@ -321,14 +432,15 @@ class TableflowTopicSpec: @classmethod def from_response(cls, data: StrAnyDict) -> TableflowTopicSpec: + config_data = data.get(Fields.CONFIG) return cls( - display_name=data["display_name"], - table_formats=[TableFormat(fmt) for fmt in data.get("table_formats", [])], - storage=storage_from_spec(data["storage"]), - config=data.get("config"), - environment_id=(data.get("environment") or {}).get("id"), - kafka_cluster_id=(data.get("kafka_cluster") or {}).get("id"), - suspended=bool(data.get("suspended", False)), + display_name=data[Fields.DISPLAY_NAME], + table_formats=[TableFormat(fmt) for fmt in data.get(Fields.TABLE_FORMATS, [])], + storage=storage_from_spec(data[Fields.STORAGE]), + config=TableflowTopicConfig.from_spec(config_data) if config_data is not None else None, + environment_id=(data.get(Fields.ENVIRONMENT) or {}).get(Fields.ID), + kafka_cluster_id=(data.get(Fields.KAFKA_CLUSTER) or {}).get(Fields.ID), + suspended=bool(data.get(Fields.SUSPENDED, False)), raw=data, ) @@ -360,4 +472,12 @@ def from_response(cls, response: StrAnyDict) -> TableflowTopic: metadata = response.get("metadata", {}) except KeyError as e: raise OperationalError(f"Error parsing Tableflow topic response, missing {e}.") from e + except (ValueError, TypeError) as e: + raise OperationalError(f"Error parsing Tableflow topic response: {e}") from e return cls(spec=spec, status=status, metadata=metadata) + + +def optional_int_from_str(s: str | None) -> int | None: + if s is None: + return None + return int(s) diff --git a/tests/unit/test_tableflow_connection_unit.py b/tests/unit/test_tableflow_connection_unit.py index 178c729..1a03ce2 100644 --- a/tests/unit/test_tableflow_connection_unit.py +++ b/tests/unit/test_tableflow_connection_unit.py @@ -15,6 +15,7 @@ ProgrammingError, TableflowPhase, TableflowTopicAlreadyExistsError, + TableflowTopicConfig, TableflowTopicNotFoundError, TableFormat, ) @@ -523,3 +524,146 @@ def test_wait_for_removal_times_out(self, mocker) -> None: ) with pytest.raises(OperationalError, match="was not removed within"): conn.disable_tableflow("orders", wait_for_removal=True, timeout=1) + + +class TestUpdateTableflow: + """update_tableflow request shaping, the empty-update guard, 404 mapping, error-detail + extraction, and the wait-for-running behavior.""" + + def test_table_formats_only_reaches_body_without_config(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_ok_response(_topic_body(), status_code=202)) + conn.update_tableflow( + "orders", + table_formats={TableFormat.DELTA, TableFormat.ICEBERG}, + wait_for_running=False, # this test asserts request shape, not the wait loop + ) + args, kwargs = conn._tableflow_request.call_args + assert args[0] == "/tableflow/v1/tableflow-topics/orders" + assert kwargs["method"] == "PATCH" + assert kwargs["json"]["spec"]["table_formats"] == ["ICEBERG", "DELTA"] + assert kwargs["json"]["spec"]["kafka_cluster"] == {"id": "lkc-1"} + assert "config" not in kwargs["json"]["spec"] + + def test_config_only_reaches_body_without_table_formats(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_ok_response(_topic_body(), status_code=202)) + conn.update_tableflow( + "orders", + config=TableflowTopicConfig(retention_ms=604800000), + wait_for_running=False, + ) + _, kwargs = conn._tableflow_request.call_args + assert kwargs["json"]["spec"]["config"] == {"retention_ms": "604800000"} + assert "table_formats" not in kwargs["json"]["spec"] + + def test_neither_argument_raises_interface_error(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(side_effect=AssertionError("should not make a request")) + with pytest.raises(InterfaceError, match="at least one of table_formats/config"): + conn.update_tableflow("orders") + + def test_empty_config_and_no_formats_raises_interface_error(self) -> None: + # A config with nothing actually set on it renders to an empty spec -- same as + # omitting config entirely, so this must be treated as no update at all. + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(side_effect=AssertionError("should not make a request")) + with pytest.raises(InterfaceError, match="at least one of table_formats/config"): + conn.update_tableflow("orders", config=TableflowTopicConfig()) + + def test_404_raises_not_found(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_error_response(404)) + with pytest.raises(TableflowTopicNotFoundError) as exc: + conn.update_tableflow("orders", table_formats=TableFormat.ICEBERG) + assert exc.value.table_name == "orders" + + def test_other_error_status_extracts_detail(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + response = Mock() + response.status_code = 422 + + def _raise() -> None: + inner = Mock() + inner.status_code = 422 + inner.json.return_value = {"errors": [{"detail": "bad retention_ms"}]} + raise httpx.HTTPStatusError("boom", request=Mock(), response=inner) + + response.raise_for_status = _raise + conn._tableflow_request = Mock(return_value=response) + with pytest.raises(OperationalError, match="bad retention_ms") as exc: + conn.update_tableflow("orders", table_formats=TableFormat.ICEBERG) + assert exc.value.http_status_code == 422 + + def test_other_error_status_falls_back_when_errors_list_empty(self) -> None: + # A parseable body with an empty (or detail-less) errors list must still fall back to + # "no more details" -- not silently produce a message with nothing after the colon. + conn = _connect(database_kafka_cluster_id="lkc-1") + response = Mock() + response.status_code = 422 + + def _raise() -> None: + inner = Mock() + inner.status_code = 422 + inner.json.return_value = {"errors": []} + raise httpx.HTTPStatusError("boom", request=Mock(), response=inner) + + response.raise_for_status = _raise + conn._tableflow_request = Mock(return_value=response) + with pytest.raises(OperationalError, match="no more details") as exc: + conn.update_tableflow("orders", table_formats=TableFormat.ICEBERG) + assert exc.value.http_status_code == 422 + + def test_other_error_status_falls_back_when_body_unparseable(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_error_response(500)) + with pytest.raises(OperationalError, match="no more details") as exc: + conn.update_tableflow("orders", table_formats=TableFormat.ICEBERG) + assert exc.value.http_status_code == 500 + + def test_blocks_for_running_by_default(self, mocker) -> None: + # No wait_for_running argument -> default (True) must poll to RUNNING, not return PENDING. + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_ok_response(_topic_body(), status_code=202)) + mocker.patch("confluent_sql.connection.sleep_with_backoff", return_value=iter([None])) + conn.get_tableflow = Mock( # type: ignore[method-assign] + return_value=TableflowTopic.from_response(_topic_body(phase="RUNNING")) + ) + topic = conn.update_tableflow("orders", table_formats=TableFormat.ICEBERG) + assert topic.phase is TableflowPhase.RUNNING + conn.get_tableflow.assert_called() + + def test_wait_for_running_raises_on_failed(self, mocker) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_ok_response(_topic_body(), status_code=202)) + failed = _topic_body(phase="FAILED") + failed["status"]["error_message"] = "schema boom" + failed["status"]["failing_table_formats"] = [ + {"format": "ICEBERG", "error_message": "bad schema"} + ] + mocker.patch("confluent_sql.connection.sleep_with_backoff", return_value=iter([None])) + conn.get_tableflow = Mock(return_value=TableflowTopic.from_response(failed)) # type: ignore[method-assign] + with pytest.raises(OperationalError, match="schema boom"): + conn.update_tableflow( + "orders", table_formats=TableFormat.ICEBERG, wait_for_running=True + ) + + def test_wait_for_running_times_out(self, mocker) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock(return_value=_ok_response(_topic_body(), status_code=202)) + mocker.patch("confluent_sql.connection.sleep_with_backoff", return_value=iter([])) + with pytest.raises(OperationalError, match="did not reach RUNNING within"): + conn.update_tableflow( + "orders", table_formats=TableFormat.ICEBERG, wait_for_running=True, timeout=1 + ) + + def test_wait_for_running_returns_immediately_if_already_running(self) -> None: + conn = _connect(database_kafka_cluster_id="lkc-1") + conn._tableflow_request = Mock( + return_value=_ok_response(_topic_body(phase="RUNNING"), status_code=202) + ) + conn.get_tableflow = Mock(side_effect=AssertionError("should not poll")) # type: ignore[method-assign] + topic = conn.update_tableflow( + "orders", table_formats=TableFormat.ICEBERG, wait_for_running=True + ) + assert topic.phase is TableflowPhase.RUNNING diff --git a/tests/unit/test_tableflow_unit.py b/tests/unit/test_tableflow_unit.py index b7d22ac..0cedba5 100644 --- a/tests/unit/test_tableflow_unit.py +++ b/tests/unit/test_tableflow_unit.py @@ -187,14 +187,18 @@ def test_empty(self) -> None: assert TableflowTopicConfig().to_spec() == {} def test_retention_only(self) -> None: - assert TableflowTopicConfig(retention_ms="604800000").to_spec() == { + # int in, str out: the API schema types retention_ms as string (int64, string-encoded + # to dodge JS/IEEE-754 double precision loss) on every request, even when this dataclass + # is constructed with a plain int for caller convenience -- confirmed against the live + # API, which rejects a non-string value outright. + assert TableflowTopicConfig(retention_ms=604800000).to_spec() == { "retention_ms": "604800000" } def test_all_fields(self) -> None: config = TableflowTopicConfig( - retention_ms="604800000", - data_retention_ms="2592000000", + retention_ms=604800000, + data_retention_ms=2592000000, error_handling=TableflowErrorHandlingLog(target="dlq"), ) assert config.to_spec() == { @@ -203,6 +207,18 @@ def test_all_fields(self) -> None: "error_handling": {"mode": "LOG", "target": "dlq"}, } + def test_from_spec_parses_wire_strings_to_int(self) -> None: + # The inverse of the int-to-string encoding above: a real GET/create response always + # has these as strings on the wire, and from_spec must parse them back to int so a + # config round-tripped through from_spec/to_spec compares equal to one built directly. + config = TableflowTopicConfig.from_spec( + {"retention_ms": "604800000", "data_retention_ms": "2592000000"} + ) + assert config == TableflowTopicConfig(retention_ms=604800000, data_retention_ms=2592000000) + + def test_from_spec_empty(self) -> None: + assert TableflowTopicConfig.from_spec({}) == TableflowTopicConfig() + class TestBuildCreatePayload: """The POST body assembles spec from the wire formats, storage, config, and connection ids.""" @@ -231,7 +247,7 @@ def test_both_formats_and_config(self) -> None: table_name="orders", table_formats=["ICEBERG", "DELTA"], storage=ManagedStorage(), - config=TableflowTopicConfig(retention_ms="604800000"), + config=TableflowTopicConfig(retention_ms=604800000), environment_id="env-1", kafka_cluster_id="lkc-1", ) @@ -293,7 +309,9 @@ def test_parses_spec_and_status(self) -> None: assert topic.spec.environment_id == "env-1" assert topic.spec.kafka_cluster_id == "lkc-1" assert topic.spec.suspended is False - assert topic.spec.config == {"retention_ms": "604800000", "enable_compaction": True} + # enable_compaction is deprecated/read-only and deliberately not modeled -- dropped, + # not retained. + assert topic.spec.config == TableflowTopicConfig(retention_ms=604800000) assert topic.status.write_mode == "APPEND" assert topic.phase is TableflowPhase.RUNNING assert topic.status.phase is TableflowPhase.RUNNING @@ -320,3 +338,12 @@ def test_missing_required_section_raises(self) -> None: del response["status"] with pytest.raises(OperationalError, match="missing 'status'"): TableflowTopic.from_response(response) + + def test_malformed_retention_ms_raises_operational_error(self) -> None: + # optional_int_from_str's int(s) raises ValueError on a non-numeric wire value -- this + # must surface as the same OperationalError every other malformed-response case does, + # not leak the raw ValueError past TableflowTopic.from_response. + response = _topic_response() + response["spec"]["config"] = {"retention_ms": "not-a-number"} + with pytest.raises(OperationalError, match="Error parsing Tableflow topic response"): + TableflowTopic.from_response(response)