This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8067-950ee0373a5525d3ece4918c2d4b8ccee3f2e390 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 5e6d925ee875b7286ff591e87c461e13d8cb701e Author: Xinyuan Lin <[email protected]> AuthorDate: Sat Aug 29 07:13:22 2026 +0000 test(pyamber): close the schema, table, proxy and network-sender gaps (#8067) ### What changes were proposed in this PR? Seven pyamber files, five existing specs extended and two new ones added. **+15 fully-covered lines; six of the seven files reach 100%.** | File | Missed before | After | |---|---|---| | `handlers/control/open_executor_handler.py` | 2 (67%) | **0 — 100%** | | `models/schema/schema.py` | 3 (94%) | **0 — 100%** | | `models/schema/attribute_type.py` | 2 (96%) | **0 — 100%** | | `models/table.py` | 2 (95%) | **0 — 100%** | | `proxy/proxy_client.py` | 2 (94%) | **0 — 100%** | | `runnables/network_sender.py` | 2 (96%) | **0 — 100%** | | `storage/iceberg/iceberg_utils.py` | 4 local / 2 CI | 2 left | **No file here is worth a PR alone** — each carries two or three lines. The bundle total is the point, and it is stated that way rather than dressed up per file. `iceberg_utils.py` closes lines 239 and 275; its remaining 302–303 are **already green in CI** and worth zero. The local figure of 4 missed is a local-environment artefact, which is why the honest CI-equivalent gain is +15 and not +17. Tests across the seven files go **37 → 89**. ### A measurement trap worth recording This machine cannot run the Iceberg-backed tests — the postgres `texera` role cannot read `iceberg_tables` — so a raw local coverage run **under-reports**. An earlier assessment of one of these files measured 12 missed lines locally where Codecov showed 2, wrong by 6×. Every target here was sized from Codecov and the local numbers were used only for the delta, never for the ceiling. ### Dead production code found, and deliberately not pinned `attribute_type.py:96-97` normalises a trailing `Z` before calling `datetime.fromisoformat`. **Deleting the whole block leaves the suite byte-identical.** That is not a coverage gap — `fromisoformat` has accepted `Z` since Python 3.11, and the CI matrix is 3.11/3.12/3.13, so the block is dead on every interpreter the project runs. A 10-case differential probe confirmed it. No test was written to pin it. The test that had been named after that branch was **renamed and annotated**, so nothing in the suite claims to constrain a branch that cannot be reached. ### Verification **27 mutations, 26 killed, 1 equivalent** — the equivalent one being the dead block above. Each was applied one at a time against a tree verified hash-equal to a snapshot, reverted from that snapshot rather than from git. Two adversarial reviewers returned nine findings against the first draft; all repaired, adding six further tests. Measured with `--cov=src/main/python` under the exact marker filter CI uses (`-m "not integration"`), no name filter, identical on both sides. The before-state restored each tracked spec **by exact single file path** via `git show HEAD:<path>` and moved the two new specs aside; all seven were then restored from a scratch snapshot and md5-verified. Full pyamber suite: `5 failed, 1269 passed, 7 errors`, with the `FAILED`/`ERROR` set **identical by identity** to main's 12 entries — no regression. `ruff check` and `ruff format --check` pass on CI's exact scope (`src/main/python src/test/python`). Both new spec files carry the Apache licence header. No production file is touched; `git diff -- 'amber/src/main/*'` is empty. ### Any related issues, documentation, discussions? Closes #8065 ### How was this PR tested? ``` cd amber && python -m pytest -m "not integration" -q src/test/python/core/models src/test/python/core/proxy src/test/python/core/runnables/test_network_sender.py src/test/python/core/architecture/handlers/control/test_open_executor_handler.py src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py ``` ``` 89 passed, 1 warning in 1.73s ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --------- Signed-off-by: Xinyuan Lin <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../handlers/control/test_open_executor_handler.py | 51 ++++++++++ .../core/models/schema/test_attribute_type.py | 94 +++++++++++++++++ .../test/python/core/models/schema/test_schema.py | 68 +++++++++++++ amber/src/test/python/core/models/test_table.py | 80 +++++++++++++++ .../test/python/core/proxy/test_proxy_client.py | 110 ++++++++++++++++++++ .../python/core/runnables/test_network_sender.py | 87 +++++++++++++++- .../storage/iceberg/test_iceberg_utils_catalog.py | 113 ++++++++++++++++++++- 7 files changed, 600 insertions(+), 3 deletions(-) diff --git a/amber/src/test/python/core/architecture/handlers/control/test_open_executor_handler.py b/amber/src/test/python/core/architecture/handlers/control/test_open_executor_handler.py new file mode 100644 index 0000000000..5f6e75da22 --- /dev/null +++ b/amber/src/test/python/core/architecture/handlers/control/test_open_executor_handler.py @@ -0,0 +1,51 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import asyncio +from types import SimpleNamespace +from unittest.mock import MagicMock + +from core.architecture.handlers.control.open_executor_handler import ( + OpenExecutorHandler, +) +from proto.org.apache.texera.amber.engine.architecture.rpc import ( + EmptyReturn, + EmptyRequest, +) + + +def make_handler() -> OpenExecutorHandler: + """Wire a handler with a SimpleNamespace context exposing executor_manager.""" + executor_manager = MagicMock() + context = SimpleNamespace(executor_manager=executor_manager) + return OpenExecutorHandler(context) + + +class TestOpenExecutorHandler: + def test_opens_the_current_executor(self): + handler = make_handler() + asyncio.run(handler.open_executor(EmptyRequest())) + executor = handler.context.executor_manager.executor + executor.open.assert_called_once_with() + # `open` is the only lifecycle call this handler is allowed to make; + # pin that it does not also close or otherwise disturb the executor. + executor.close.assert_not_called() + + def test_returns_empty_return(self): + handler = make_handler() + result = asyncio.run(handler.open_executor(EmptyRequest())) + assert isinstance(result, EmptyReturn) diff --git a/amber/src/test/python/core/models/schema/test_attribute_type.py b/amber/src/test/python/core/models/schema/test_attribute_type.py new file mode 100644 index 0000000000..3f737dbb2c --- /dev/null +++ b/amber/src/test/python/core/models/schema/test_attribute_type.py @@ -0,0 +1,94 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import datetime + +import pytest + +from core.models.schema.attribute_type import ( + AttributeType, + FROM_STRING_PARSER_MAPPING, +) + +# Go through the dispatch table rather than the private helpers: this is the +# entry point production code uses when it materializes a string column into +# a typed field, so the tests pin the reachable behaviour rather than an +# implementation detail. +parse_bool = FROM_STRING_PARSER_MAPPING[AttributeType.BOOL] +parse_timestamp = FROM_STRING_PARSER_MAPPING[AttributeType.TIMESTAMP] + +EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) + + +class TestParseBool: + @pytest.mark.parametrize("empty", [None, "", " ", "\t\n"]) + def test_an_absent_value_parses_as_false(self, empty): + # Every "empty" spelling short-circuits to False before the + # true/false/numeric ladder is consulted. Covering several spellings + # keeps the assertion from being dodged by a narrower emptiness test. + assert parse_bool(empty) is False + + @pytest.mark.parametrize( + "text, expected", + [("true", True), ("TRUE", True), (" True ", True), ("false", False)], + ) + def test_literal_spellings_are_case_and_space_insensitive(self, text, expected): + assert parse_bool(text) is expected + + @pytest.mark.parametrize("text, expected", [("1", True), ("0", False)]) + def test_a_numeric_value_is_compared_against_zero(self, text, expected): + assert parse_bool(text) is expected + + def test_a_non_numeric_non_literal_value_is_rejected(self): + with pytest.raises(ValueError): + parse_bool("maybe") + + +class TestParseTimestamp: + @pytest.mark.parametrize("empty", [None, "", " ", "\t\n"]) + def test_an_absent_value_parses_as_the_utc_epoch(self, empty): + parsed = parse_timestamp(empty) + # Assert the exact instant *and* the tzinfo: a naive 1970-01-01 would + # compare unequal here, so dropping the timezone is caught too. + assert parsed == EPOCH + assert parsed.tzinfo == datetime.timezone.utc + assert (parsed.year, parsed.month, parsed.day) == (1970, 1, 1) + + def test_a_zulu_suffix_yields_a_utc_aware_instant(self): + # Named for the observable outcome, not for the code that produces it. + # `_parse_timestamp` rewrites a trailing "Z" into "+00:00" before + # calling `fromisoformat`, but `fromisoformat` has accepted "Z" itself + # since Python 3.11 and the pyamber CI matrix is 3.11/3.12/3.13 -- so + # deleting that rewrite leaves the entire suite green, and this test + # must not be credited with pinning it. What it does pin is the + # resulting instant and its offset (and, uniquely in this file, the + # "+00:00" constant, should the rewrite ever run on an older runtime). + assert parse_timestamp("2024-05-06T07:08:09Z") == datetime.datetime( + 2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc + ) + + def test_a_naive_value_is_assumed_to_be_utc(self): + assert parse_timestamp("2024-05-06T07:08:09") == datetime.datetime( + 2024, 5, 6, 7, 8, 9, tzinfo=datetime.timezone.utc + ) + + def test_an_explicit_offset_is_preserved(self): + parsed = parse_timestamp("2024-05-06T07:08:09+02:00") + assert parsed.utcoffset() == datetime.timedelta(hours=2) + assert parsed == datetime.datetime( + 2024, 5, 6, 5, 8, 9, tzinfo=datetime.timezone.utc + ) diff --git a/amber/src/test/python/core/models/schema/test_schema.py b/amber/src/test/python/core/models/schema/test_schema.py index 60e4c848a5..0842024224 100644 --- a/amber/src/test/python/core/models/schema/test_schema.py +++ b/amber/src/test/python/core/models/schema/test_schema.py @@ -154,3 +154,71 @@ class TestSchema: assert round_trip_schema.get_attr_type("field1") == AttributeType.STRING assert round_trip_schema.get_attr_type("field2") == AttributeType.LARGE_BINARY assert round_trip_schema.get_attr_type("field3") == AttributeType.INT + + @pytest.mark.parametrize( + "other", ["not a schema", None, 42, {"field-1": "STRING"}, ["field-1"]] + ) + def test_comparing_against_a_non_schema_is_false_not_an_error(self, schema, other): + # `__eq__` guards on isinstance before touching `as_key_value_pairs`, + # so a foreign operand must compare unequal rather than raise. Assert + # the boolean explicitly (and both directions) so a guard that returned + # True would be caught. + assert (schema == other) is False + assert schema != other + + def test_a_schema_equals_only_a_schema_with_the_same_ordered_pairs(self, schema): + # `__eq__` is defined in terms of `as_key_value_pairs`, and both + # operands below are built by replaying that same accessor -- so a + # corrupted accessor moves both sides together and `schema == same` + # would stay green on its own. Guard each constructed operand against a + # literal expectation first, read back through the *independent* + # `get_attr_names` accessor, so a degenerate fixture fails here instead + # of sailing through the equality claim. + same = Schema() + for name, attr_type in schema.as_key_value_pairs(): + same.add(name, attr_type) + assert same.get_attr_names() == [f"field-{i}" for i in range(1, 8)] + assert schema == same + + reordered = Schema() + for name, attr_type in reversed(schema.as_key_value_pairs()): + reordered.add(name, attr_type) + # ... and this one really is a *reordering* (same names, reverse order), + # not an empty or truncated schema that would compare unequal for a + # reason that has nothing to do with ordering. + assert reordered.get_attr_names() == [f"field-{i}" for i in range(7, 0, -1)] + assert (schema == reordered) is False + + def test_a_partial_schema_keeps_the_order_of_the_requested_names(self, schema): + # `get_partial_schema` documents that it preserves "the order specified + # by the attribute names", and Schema equality is order-sensitive (see + # the test above), yet nothing pinned that promise. It matters: the sole + # caller, `Tuple.get_partial_tuple`, builds its field values in + # `attribute_names` order and takes its schema from here, so a + # reordering would make a Tuple's data and its schema silently disagree. + # + # The requested names are deliberately out of the source order, and the + # assertion is on the ordered pair list rather than on membership -- + # both are what make this non-vacuous. + partial = schema.get_partial_schema(["field-5", "field-2"]) + assert partial.as_key_value_pairs() == [ + ("field-5", AttributeType.BOOL), + ("field-2", AttributeType.INT), + ] + + def test_str_renders_each_attribute_with_a_zero_based_index_and_type(self): + rendered = Schema( + raw_schema={"field-1": "STRING", "field-2": "INTEGER", "field-3": "BOOLEAN"} + ) + # Pin the whole rendering: the bracketed header/footer, the ",\n" + # separator, the zero-based index, and the name-then-type ordering. + assert str(rendered) == ( + "Schema[\n" + "(0)'field-1' -> AttributeType.STRING,\n" + "(1)'field-2' -> AttributeType.INT,\n" + "(2)'field-3' -> AttributeType.BOOL\n" + "]" + ) + + def test_str_of_an_empty_schema_still_renders_the_brackets(self): + assert str(Schema()) == "Schema[\n\n]" diff --git a/amber/src/test/python/core/models/test_table.py b/amber/src/test/python/core/models/test_table.py index 368220779d..80a985e5d6 100644 --- a/amber/src/test/python/core/models/test_table.py +++ b/amber/src/test/python/core/models/test_table.py @@ -16,9 +16,11 @@ # under the License. import datetime +import numpy import pandas import pickle import pytest +import re from pandas import RangeIndex from core.models import Table, Tuple @@ -142,3 +144,81 @@ class TestTable: def test_validation_of_schema(self): with pytest.raises(AssertionError): Table([{"text": "hello"}, {"book": "harry"}]) + + @pytest.mark.parametrize( + "table_like", [42, "hello", None, {"field1": [1, 2]}, (1, 2), b"bytes"] + ) + def test_an_unsupported_tablelike_is_rejected(self, table_like): + # Only Table / DataFrame / list reach a constructor; anything else must + # be refused with a message naming the offending type, rather than + # falling through into `super().__init__` with an unbound frame. + # + # Match the *whole* rendered message, interpolation included: a prefix + # match would leave `{type(table_like)}` -- the only non-constant part + # of that line -- unpinned, and would also make all six parametrized + # cases assert the identical string. + expected = ( + "^" + re.escape(f"unsupported tablelike type {type(table_like)}") + "$" + ) + with pytest.raises(TypeError, match=expected): + Table(table_like) + + @pytest.fixture + def comparable_frame(self): + # Deliberately free of all-None columns: elementwise `None == None` + # is False, which would make the comparison below fail for reasons + # unrelated to the branch under test. + return pandas.DataFrame( + {"field1": [1, 2], "field2": ["hello", "world"], "field3": [2.3, 0.0]}, + columns=["field1", "field2", "field3"], + ) + + def test_comparing_to_an_equal_data_frame_reports_equal(self, comparable_frame): + # A non-Table operand takes the `super().__eq__` branch. Reduce with + # `numpy.all` rather than asserting the result's *shape*: today that + # branch yields a per-column Series (see the characterization test + # below), and hard-asserting that here would cement the very mismatch + # between the code and its `-> bool` annotation. + table = Table(comparable_frame) + assert numpy.all(table == comparable_frame) + + def test_comparing_to_a_differing_data_frame_reports_unequal( + self, comparable_frame + ): + table = Table(comparable_frame) + differing = comparable_frame.copy() + differing.loc[0, "field2"] = "goodbye" + # One cell of one column differs. Reducing the whole comparison must + # therefore be falsy -- which is also what distinguishes the real + # `.all()` reduction from an `.any()` one (under `.any()` every column + # has at least one matching row, so the reduction would come back True). + assert not numpy.all(table == differing) + + def test_comparing_to_a_data_frame_currently_yields_a_per_column_series( + self, comparable_frame + ): + # CHARACTERIZATION, not a contract. `Table.__eq__` is annotated + # `-> bool`, but its non-Table branch returns + # `super().__eq__(other).all()`, which for a DataFrame operand reduces + # only over rows and leaves a Series indexed by column name. A bare + # `assert table == frame` therefore raises "truth value of a Series is + # ambiguous". This test records today's shape so that narrowing the + # branch to a real bool surfaces here deliberately, with a name that + # says so, rather than silently through the behavioural tests above. + table = Table(comparable_frame) + comparison = table == comparable_frame + assert isinstance(comparison, pandas.Series) + assert list(comparison.index) == ["field1", "field2", "field3"] + + def test_two_tables_with_differing_rows_are_not_equal(self, comparable_frame): + # The Table-vs-Table arm had only positive coverage: every pre-existing + # test in this file compares two *equal* Tables, so replacing the arm's + # body with `return True` survived the whole suite. One negative case + # closes that. + # + # Deliberately NOT pinned here: `zip` truncates to the shorter operand, + # so `Table(frame.head(1)) == Table(frame)` is True today. That is a + # defect, not a contract, and asserting it would cement it. + differing = comparable_frame.copy() + differing.loc[0, "field2"] = "goodbye" + assert (Table(comparable_frame) == Table(differing)) is False diff --git a/amber/src/test/python/core/proxy/test_proxy_client.py b/amber/src/test/python/core/proxy/test_proxy_client.py index 891c2fda75..0c50393c74 100644 --- a/amber/src/test/python/core/proxy/test_proxy_client.py +++ b/amber/src/test/python/core/proxy/test_proxy_client.py @@ -20,6 +20,7 @@ from pandas import DataFrame from pyarrow import ArrowNotImplementedError, Table from queue import Queue +import core.proxy.proxy_client as proxy_client_module from core.proxy.proxy_client import ProxyClient from core.proxy.proxy_server import ProxyServer @@ -153,3 +154,112 @@ class TestProxyClient: assert data_queue.qsize() == 4 for i, row in data_table.to_pandas().iterrows(): assert data_queue.get().equals(row) + + def test_a_handshake_port_is_announced_to_the_server_on_connect(self, server): + # The Java side learns which port the Python proxy *server* listens on + # through a "handshake" action issued while the client connects. + # ProxyServer does not register "handshake" itself, so the test + # registers a capturing stand-in for the Java handler. + received = [] + server.register("handshake", lambda payload: received.append(payload) or "ok") + + client = ProxyClient(handshake_port=6789) + try: + # The payload must be the handshake port rendered as UTF-8 digits, + # not the client's timeout or the data port it dialled. (The action + # *name* is pinned separately, by the test below -- see why there.) + assert received == [b"6789"] + finally: + client.close() + + def test_the_handshake_call_names_the_handshake_action(self, monkeypatch): + # The end-to-end test above cannot pin the action *name*. Substituting + # any other registered name makes construction blow up server-side (the + # built-in `heartbeat`/`shutdown` handlers take no argument, while + # `control`/`actor` try to deserialize the payload), so the test dies + # during `ProxyClient(...)` and its `received` assertion never runs -- + # the name is pinned by an accident of the fixture, not by an + # assertion. Stub the call at the client boundary instead, so the name + # reaches an assertion rather than a server-side crash. + seen = [] + monkeypatch.setattr( + ProxyClient, + "call_action", + lambda self, name, payload=bytes(), options=None: seen.append( + (name, payload) + ), + ) + + client = ProxyClient(handshake_port=6789) + try: + assert seen == [("handshake", b"6789")] + finally: + client.close() + + def test_no_handshake_is_sent_when_no_handshake_port_is_given(self, server): + # Companion negative case: without it, unconditionally calling + # `_handshake` would still satisfy the test above. + received = [] + server.register("handshake", lambda payload: received.append(payload) or "ok") + + client = ProxyClient() + try: + assert received == [] + # ... and the client is still usable, i.e. skipping the handshake + # is a real branch rather than a failed connect. + assert client.call_action("heartbeat") == b"ack" + finally: + client.close() + + @staticmethod + def _record_call_options(monkeypatch): + """ + Record every `FlightCallOptions(...)` `call_action` manufactures. + + A plain factory function rather than a subclass: `FlightCallOptions` is + a Cython cdef class. It still delegates to the real constructor, so the + RPC underneath stays a real one. + """ + real = proxy_client_module.FlightCallOptions + captured = [] + + def recording_factory(**kwargs): + captured.append(kwargs) + return real(**kwargs) + + monkeypatch.setattr(proxy_client_module, "FlightCallOptions", recording_factory) + return captured, real + + def test_the_configured_timeout_is_applied_when_no_options_are_supplied( + self, server, monkeypatch + ): + # This is the only place the client's configured timeout ever reaches + # an RPC, and nothing constrained it: dropping the argument entirely + # (`FlightCallOptions()`) survived the whole suite. Without a timeout a + # hung Java-side server would block the Python worker forever instead + # of raising. + captured, _ = self._record_call_options(monkeypatch) + + # A non-default timeout, so the assertion pins the value *flowing from + # the constructor* rather than a hard-coded 1000. + client = ProxyClient(timeout=1234) + try: + assert client.call_action("heartbeat") == b"ack" + finally: + client.close() + + assert captured == [{"timeout": 1234}] + + def test_caller_supplied_options_are_used_as_is(self, server, monkeypatch): + # The companion arm: when the caller hands in an options object, + # `call_action` must not manufacture its own and discard it. + captured, real = self._record_call_options(monkeypatch) + + client = ProxyClient(timeout=1234) + try: + options = real(timeout=99) + assert client.call_action("heartbeat", options=options) == b"ack" + finally: + client.close() + + assert captured == [] diff --git a/amber/src/test/python/core/runnables/test_network_sender.py b/amber/src/test/python/core/runnables/test_network_sender.py index 529cd19d33..bca9c52a58 100644 --- a/amber/src/test/python/core/runnables/test_network_sender.py +++ b/amber/src/test/python/core/runnables/test_network_sender.py @@ -17,11 +17,39 @@ import pytest import threading +from contextlib import contextmanager from time import sleep -from core.models.internal_queue import InternalQueue +from loguru import logger + +from core.models.internal_queue import InternalQueue, InternalQueueElement +from core.models.payload import DataFrame from core.runnables.network_receiver import NetworkReceiver from core.runnables.network_sender import NetworkSender +from proto.org.apache.texera.amber.core import ( + ActorVirtualIdentity, + ChannelIdentity, +) + + +@contextmanager +def muted_catch_logs(): + """ + `NetworkSender._send_data` is wrapped in `@logger.catch(reraise=True)`, + which logs the whole traceback at ERROR before re-raising. CI runs pytest + with `-s` and `LOGURU_LEVEL=WARNING`, so an expected failure would + otherwise dump a traceback into the build log and read as a real error. + + `logger.catch` attributes the record to the *caller's* module rather than + to the decorated function's module, so the name to silence is this test + module's own `__name__` (which varies with pytest's import mode -- hence + `__name__` rather than a literal). + """ + logger.disable(__name__) + try: + yield + finally: + logger.enable(__name__) class TestNetworkSender: @@ -67,3 +95,60 @@ class TestNetworkSender: assert not network_sender_thread.is_alive() network_receiver_thread.join() network_sender_thread.join() + + @pytest.fixture + def channel_id(self): + worker_id = ActorVirtualIdentity(name="test") + return ChannelIdentity(worker_id, worker_id, False) + + @pytest.mark.timeout(5) + def test_receive_rejects_an_element_that_is_neither_data_control_nor_ecm( + self, network_sender, channel_id + ): + # A plain subclass of InternalQueueElement is neither DataElement nor + # DCMElement nor ECMElement, so it walks the whole dispatch chain and + # falls off the end. The sender must refuse it loudly rather than drop + # it silently. + # + # The stable `__repr__` lets the matcher pin the *interpolated* entry + # as well, mirroring the payload test below: matching only the constant + # prefix would leave `{next_entry}` -- the sole non-constant part of + # that line -- unconstrained. + class UnknownEntry(InternalQueueElement): + def __repr__(self): + return "<unknown-entry>" + + unknown = UnknownEntry(tag=channel_id) + with pytest.raises(TypeError, match="Unexpected entry <unknown-entry>"): + network_sender.receive(unknown) + + @pytest.mark.timeout(5) + def test_send_data_rejects_a_payload_that_is_neither_dataframe_nor_stateframe( + self, network_sender, channel_id + ): + class NotAPayload: + def __repr__(self): + return "<not-a-payload>" + + with muted_catch_logs(): + with pytest.raises(TypeError, match="Unexpected payload <not-a-payload>"): + network_sender._send_data(channel_id, NotAPayload()) + + @pytest.mark.timeout(5) + def test_receive_routes_a_data_element_to_send_data( + self, network_sender, channel_id, monkeypatch + ): + # Guards the dispatch chain above: without a positive case, swapping + # the DataElement arm for the else-arm would still leave the negative + # test green. + from core.models.internal_queue import DataElement + + seen = [] + monkeypatch.setattr( + network_sender, + "_send_data", + lambda to, payload: seen.append((to, payload)), + ) + payload = DataFrame(frame=None) + network_sender.receive(DataElement(tag=channel_id, payload=payload)) + assert seen == [(channel_id, payload)] diff --git a/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py b/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py index a387a51597..c3fb6e3541 100644 --- a/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py +++ b/amber/src/test/python/core/storage/iceberg/test_iceberg_utils_catalog.py @@ -15,10 +15,16 @@ # specific language governing permissions and limitations # under the License. -from unittest.mock import patch +from unittest.mock import MagicMock, patch + +from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC from core.storage.iceberg import iceberg_utils -from core.storage.iceberg.iceberg_utils import create_postgres_catalog +from core.storage.iceberg.iceberg_utils import ( + create_postgres_catalog, + create_rest_catalog, + create_table, +) class TestCreatePostgresCatalog: @@ -255,3 +261,106 @@ class TestCreatePostgresCatalogWindowsLocal: assert kwargs["warehouse"] == "" assert "py-io-impl" not in kwargs + + +class TestCreateRestCatalog: + """ + `create_rest_catalog` is the REST counterpart of `create_postgres_catalog`. + + Unlike the postgres path it deliberately performs *no* Windows-local + normalization: `warehouse_name` is a logical warehouse identifier the REST + server resolves, and its I/O goes through S3FileIO rather than the local + filesystem. `load_catalog` is patched so these run without a REST server. + """ + + def _make(self, catalog_name, warehouse_name, rest_uri): + with patch.object(iceberg_utils, "load_catalog") as mock_load_catalog: + catalog = create_rest_catalog( + catalog_name=catalog_name, + warehouse_name=warehouse_name, + rest_uri=rest_uri, + ) + assert mock_load_catalog.call_count == 1 + assert catalog is mock_load_catalog.return_value + return mock_load_catalog.call_args + + def test_catalog_name_uri_and_warehouse_are_forwarded_by_value(self): + args, kwargs = self._make( + "texera_iceberg", "texera-warehouse", "http://lakekeeper:8181/catalog"; + ) + # Assert every value, not just `type`: swapping `uri` and `warehouse` + # would otherwise go unnoticed. + assert args == ("texera_iceberg",) + assert kwargs == { + "type": "rest", + "uri": "http://lakekeeper:8181/catalog";, + "warehouse": "texera-warehouse", + } + + def test_a_windows_style_warehouse_name_is_not_normalized(self): + """ + The postgres path rewrites `C:\\...` into a `file:///` URI; the REST + path must not, because the identifier is resolved server-side. + """ + _, kwargs = self._make( + "texera_iceberg", "C:\\Users\\texera\\warehouse", "http://localhost:8181"; + ) + assert kwargs["warehouse"] == "C:\\Users\\texera\\warehouse" + assert "py-io-impl" not in kwargs + + +class TestCreateTable: + """ + `create_table` against a mocked `Catalog`: no postgres, no REST server, + no object store. + """ + + NAMESPACE = "ns" + NAME = "tbl" + IDENTIFIER = "ns.tbl" + + def _catalog(self, table_exists): + catalog = MagicMock() + catalog.table_exists.return_value = table_exists + return catalog + + def _create(self, catalog, override_if_exists): + schema = MagicMock(name="schema") + table = create_table( + catalog=catalog, + table_namespace=self.NAMESPACE, + table_name=self.NAME, + table_schema=schema, + override_if_exists=override_if_exists, + ) + return table, schema + + def test_an_existing_table_is_dropped_when_override_is_requested(self): + catalog = self._catalog(table_exists=True) + table, schema = self._create(catalog, override_if_exists=True) + + catalog.create_namespace_if_not_exists.assert_called_once_with(self.NAMESPACE) + catalog.table_exists.assert_called_once_with(self.IDENTIFIER) + catalog.drop_table.assert_called_once_with(self.IDENTIFIER) + catalog.create_table.assert_called_once_with( + identifier=self.IDENTIFIER, + schema=schema, + partition_spec=UNPARTITIONED_PARTITION_SPEC, + ) + assert table is catalog.create_table.return_value + + def test_an_existing_table_is_kept_when_override_is_not_requested(self): + # Companion to the case above: without it, relaxing the guard from + # `and` to `or` would survive. + catalog = self._catalog(table_exists=True) + self._create(catalog, override_if_exists=False) + + catalog.drop_table.assert_not_called() + catalog.create_table.assert_called_once() + + def test_a_missing_table_is_never_dropped_even_when_override_is_requested(self): + catalog = self._catalog(table_exists=False) + self._create(catalog, override_if_exists=True) + + catalog.drop_table.assert_not_called() + catalog.create_table.assert_called_once()
