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-8337-7b5ee1461fce5f3c1b7ab9801222771ad7388c4b in repository https://gitbox.apache.org/repos/asf/texera.git
commit 562eacc6dac1336a9dea3eff0f449072d25539ef Author: Xinyuan Lin <[email protected]> AuthorDate: Wed Sep 2 07:51:17 2026 +0000 test(pyamber): close the last storage, manager and handler gaps (#8337) ### What changes were proposed in this PR? Five existing pyamber test modules extended and one added. **+8 fully-covered lines — all six files finish at 100% statement *and* 100% branch coverage.** | File | Codecov | Closed | |---|---|---| | `core/storage/document_factory.py` | 63/65 → **65/65** | 170, and the partial arm on 142 | | `pytexera/storage/large_binary_input_stream.py` | 54/56 → **56/56** | 103, and the partial arm on 102 | | `core/architecture/managers/state_manager.py` | 28/29 → **29/29** | 92 | | `core/util/stoppable/stoppable_queue_blocking_thread.py` | 38/39 → **39/39** | 74 | | `pytexera/udf/examples/echo_operator.py` | 7/8 → **8/8** | 28 | | `core/architecture/handlers/actorcommand/actor_handler_base.py` | 6/7 → **7/7** | 30 | Bundle total 196/204 → **204/204**. Six of the eight gained lines were never executed; the other two are lines whose second branch arm was never taken. `test_actor_handler_base.py` is new — that module had no test at all. No file here clears a meaningful bar alone; the bundle total is the point, and it is stated that way rather than dressed up per file. ### Verification **23 mutants, zero survivors.** Each applied one at a time against the repaired tests, and each verified at **full non-integration-suite scope** by comparing the exact set of failing test ids against the baseline run rather than just the count. **19 of the 23 are killed *uniquely* by their credited test**, proven by re-running the same mutant with only that test deselected and confirming the suite returns to the byte-identical baseline. Measured by running the exact CI command from `amber/` twice at identical whole-suite scope — no `-k` or `-z` filter either time, so no filter-attribution risk — once with the tracked specs restored from `HEAD` and the new test directory moved out of the tree, once with the bundle in place. Both `coverage.xml` files were parsed programmatically. No `logger.info`/`debug` body is claimed anywhere: every newly covered line is a plain statement or branch arm that executes identically under CI's `WARN` log level. Eight reviewer findings, all repaired — including one standalone test that was **deleted** because measurement showed it contributed zero: `actor_handler_base.py`'s baseline missed only line 30, so line 25 was already covered and the test pinning it added nothing. Full pyamber suite: `5 failed, 1295 passed, 7 errors`, with the `FAILED`/`ERROR` set **identical by identity** to main's known 12 entries — no regression. `ruff check` and `ruff format --check` pass on CI's exact scope. The new test file carries the Apache licence header. No production file is touched. ### Any related issues, documentation, discussions? Closes #8335 ### How was this PR tested? ``` cd amber && python -m pytest -m "not integration" -q ``` ``` 5 failed, 1295 passed, 1 deselected, 1 xfailed, 7 errors ``` The 5 failures and 7 errors are the pre-existing Iceberg/Windows set, identical to `main`; on CI, where the Iceberg catalog is available, they do not occur. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../actorcommand/test_actor_handler_base.py | 46 +++++++++++++ .../architecture/managers/test_state_manager.py | 15 +++++ .../python/core/storage/test_document_factory.py | 77 +++++++++++++++++++++- .../test_stoppable_queue_blocking_thread.py | 33 ++++++++++ .../storage/test_large_binary_input_stream.py | 17 +++++ .../pytexera/udf/examples/test_echo_operator.py | 13 ++++ 6 files changed, 198 insertions(+), 3 deletions(-) diff --git a/amber/src/test/python/core/architecture/handlers/actorcommand/test_actor_handler_base.py b/amber/src/test/python/core/architecture/handlers/actorcommand/test_actor_handler_base.py new file mode 100644 index 0000000000..fe769364de --- /dev/null +++ b/amber/src/test/python/core/architecture/handlers/actorcommand/test_actor_handler_base.py @@ -0,0 +1,46 @@ +# 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. + +from unittest.mock import MagicMock + +from core.architecture.handlers.actorcommand.actor_handler_base import ( + ActorCommandHandler, +) + + +class TestActorCommandHandler: + def test_the_base_handler_is_an_inert_no_op(self): + # ActorCommandHandler declares no abstract members, so an unfinished + # subclass -- or the base itself -- is instantiable and callable. The + # base __call__ must then do nothing at all: return None and touch + # neither of its two arguments, rather than half-handling the + # command. Both arguments are bound to names so a side effect on + # either is visible -- passing the command inline as an anonymous + # MagicMock hides every mutation that only writes to it. + handler = ActorCommandHandler() + command = MagicMock() + input_queue = MagicMock() + + result = handler(command, input_queue) + + assert result is None + input_queue.assert_not_called() + assert input_queue.method_calls == [] + assert command.mock_calls == [] + # Subclasses opt in by setting `cmd`; the base must not claim one, or + # the handler registry would dispatch every command to the no-op. + assert ActorCommandHandler.cmd is None diff --git a/amber/src/test/python/core/architecture/managers/test_state_manager.py b/amber/src/test/python/core/architecture/managers/test_state_manager.py index 902cdb7475..a787181e96 100644 --- a/amber/src/test/python/core/architecture/managers/test_state_manager.py +++ b/amber/src/test/python/core/architecture/managers/test_state_manager.py @@ -46,6 +46,21 @@ class TestStateManager: assert state_manager.confirm_state(state) state_manager.assert_state(state) + def test_get_current_state_reflects_the_latest_transition(self, state_manager): + # get_current_state must report the state itself, not the version + # counter that transit_to bumps alongside it. UNINITIALIZED/READY + # alone cannot tell the two apart -- their ordinals happen to equal + # the transition count -- so walk one step further to PAUSED, where + # the ordinal (3) and the version (2) diverge. + assert state_manager.get_current_state() == WorkerState.UNINITIALIZED + + state_manager.transit_to(WorkerState.READY) + assert state_manager.get_current_state() == WorkerState.READY + + state_manager.transit_to(WorkerState.PAUSED) + assert state_manager.get_current_state() == WorkerState.PAUSED + assert state_manager.get_current_state() != state_manager.get_state_version() + def test_it_raises_exception_when_transit_to_undefined_state(self, state_manager): state_manager.assert_state(WorkerState.UNINITIALIZED) for state in [WorkerState.READY, WorkerState.PAUSED]: diff --git a/amber/src/test/python/core/storage/test_document_factory.py b/amber/src/test/python/core/storage/test_document_factory.py index 654025006a..84e36a47fe 100644 --- a/amber/src/test/python/core/storage/test_document_factory.py +++ b/amber/src/test/python/core/storage/test_document_factory.py @@ -16,6 +16,7 @@ # under the License. from unittest.mock import MagicMock, patch +from urllib.parse import urlparse import pytest @@ -30,6 +31,9 @@ StorageConfig.ICEBERG_TABLE_RESULT_NAMESPACE = "test-result-ns" StorageConfig.ICEBERG_TABLE_STATE_NAMESPACE = "test-state-ns" VFS_URI = "vfs:///wid/0/eid/0/opid/test/main/0/0/result" +# The storage key VFS_URI sanitizes down to; the routing tests below assert +# that it is what actually reaches the iceberg layer. +VFS_URI_STORAGE_KEY = "wid_0_eid_0_opid_test_main_0_0_result" @pytest.fixture @@ -42,6 +46,19 @@ def _decode_returning(resource_type): return lambda _uri: VFSUriComponents(None, None, None, resource_type) +def test_sanitize_uri_path_unquotes_strips_the_warehouse_and_flattens(): + # Each of the four steps in sanitize_uri_path is load-bearing and none of + # them is observable through create/open/exists, which only ever assert + # the namespace -- so pin the storage key directly. The leading slash is + # stripped, an optional "wh/<warehouse>/" prefix is dropped so the key is + # warehouse-independent, percent escapes are decoded (urlparse, unlike + # java.net.URI.getPath, does not do it), and the remaining separators + # become underscores. + parsed = urlparse("vfs:///wh/w1/wid/0/my%20op/result/") + + assert DocumentFactory.sanitize_uri_path(parsed) == "wid_0_my op_result_" + + @patch("core.storage.document_factory.IcebergDocument") @patch("core.storage.document_factory.amber_schema_to_iceberg_schema") @patch("core.storage.document_factory.create_table") @@ -67,8 +84,29 @@ class TestCreateDocumentNamespaceRouting: DocumentFactory.create_document(VFS_URI, schema) - args, _ = mock_create_table.call_args + args, kwargs = mock_create_table.call_args assert args[1] == StorageConfig.ICEBERG_TABLE_RESULT_NAMESPACE + # A table left over from an earlier execution of the same operator + # must be overwritten, not silently reused with stale rows. + assert kwargs["override_if_exists"] is True + + def test_document_is_built_for_the_resolved_namespace_and_storage_key( + self, mock_vfs, _icb, _create_table, _amber_schema, mock_doc, schema + ): + mock_vfs.VFS_FILE_URI_SCHEME = "vfs" + mock_vfs.decode_uri.side_effect = _decode_returning(VFSResourceType.RESULT) + + document = DocumentFactory.create_document(VFS_URI, schema) + + # IcebergDocument[Tuple](namespace, storage_key, ...): the two leading + # positional arguments are both plain strings and transposing them + # would go unnoticed by every namespace-only assertion above. + constructor = mock_doc.__getitem__.return_value + assert constructor.call_args.args[:2] == ( + StorageConfig.ICEBERG_TABLE_RESULT_NAMESPACE, + VFS_URI_STORAGE_KEY, + ) + assert document is constructor.return_value def test_unsupported_resource_type_raises_value_error( self, mock_vfs, _icb, _create_table, _amber_schema, _doc, schema @@ -84,7 +122,12 @@ class TestCreateDocumentNamespaceRouting: def test_create_document_rejects_non_vfs_scheme(schema): - with pytest.raises(NotImplementedError, match="Unsupported URI scheme"): + # Match the per-site suffix, not the shared "Unsupported URI scheme" + # prefix: all three entry points raise NotImplementedError with that same + # leading text, so a bare prefix match cannot tell this raise site from + # the other two and would still pass if this branch delegated to one of + # them instead of raising on its own. + with pytest.raises(NotImplementedError, match="for creating the document"): DocumentFactory.create_document("file:///tmp/x", schema) @@ -112,6 +155,25 @@ class TestOpenDocumentNamespaceRouting: args, _ = mock_load.call_args assert args[1] == StorageConfig.ICEBERG_TABLE_STATE_NAMESPACE + def test_returns_the_document_and_the_schema_of_the_loaded_table( + self, mock_vfs, _icb, mock_load, mock_schema_cls, mock_doc + ): + mock_vfs.VFS_FILE_URI_SCHEME = "vfs" + mock_vfs.decode_uri.side_effect = _decode_returning(VFSResourceType.RESULT) + mock_load.return_value = self._stub_table() + + document, amber_schema = DocumentFactory.open_document(VFS_URI) + + # Bind the pair: with the result discarded, open_document could + # return (None, None) -- or transpose the namespace and storage key + # it hands the document -- with nothing in this file noticing. + assert mock_doc.call_args.args[:2] == ( + StorageConfig.ICEBERG_TABLE_RESULT_NAMESPACE, + VFS_URI_STORAGE_KEY, + ) + assert document is mock_doc.return_value + assert amber_schema is mock_schema_cls.return_value + def test_unsupported_resource_type_raises_value_error( self, mock_vfs, _icb, _load, _schema_cls, _doc ): @@ -134,6 +196,13 @@ class TestOpenDocumentNamespaceRouting: DocumentFactory.open_document(VFS_URI) +def test_open_document_rejects_non_vfs_scheme(): + # See test_create_document_rejects_non_vfs_scheme: the suffix is what + # identifies this raise site. + with pytest.raises(NotImplementedError, match="for opening the document"): + DocumentFactory.open_document("file:///tmp/x") + + @patch("core.storage.document_factory.IcebergCatalogInstance") @patch("core.storage.document_factory.VFSURIFactory") class TestDocumentExists: @@ -168,5 +237,7 @@ class TestDocumentExists: def test_document_exists_rejects_non_vfs_scheme(): - with pytest.raises(NotImplementedError, match="Unsupported URI scheme"): + # See test_create_document_rejects_non_vfs_scheme: the suffix is what + # identifies this raise site. + with pytest.raises(NotImplementedError, match="for checking document existence"): DocumentFactory.document_exists("file:///tmp/x") diff --git a/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py b/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py index c6558ca084..8754845cb1 100644 --- a/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py +++ b/amber/src/test/python/core/util/stoppable/test_stoppable_queue_blocking_thread.py @@ -157,3 +157,36 @@ class TestRun: runnable.run() assert runnable.events == [("pre_start",), ("post_stop",)] + + def test_the_base_receive_is_an_inert_no_op(self): + # Every other test subclasses the runnable and overrides receive, so + # the base implementation is never exercised. It must be a silent + # no-op: a subclass that only cares about pre_start/post_stop still + # has its entries consumed off the queue instead of the loop blowing + # up on the first one. + # + # Drive receive directly for the no-op claim. The drain in run() is + # performed by interruptible_get, not by receive, so an is_empty() + # assertion after run() is monotone in one direction only: it can see + # a receive that ADDS to the queue, but a receive that quietly + # swallows an extra entry -- dropping every other message off the + # wire -- leaves the queue just as empty. Pin both directions, plus + # the absence of any state written on the runnable itself. + queue = FakeQueue() + first, second = QueueElement(), QueueElement() + queue.put(first) + queue.put(second) + runnable = StoppableQueueBlockingRunnable(name="r", queue=queue) + state_before = dict(vars(runnable)) + + assert runnable.receive(QueueElement()) is None + + assert list(queue.items) == [first, second] + assert vars(runnable) == state_before + + # And through run(): the loop still drains to the stop sentinel. + runnable.stop() + + runnable.run() + + assert queue.is_empty() diff --git a/amber/src/test/python/pytexera/storage/test_large_binary_input_stream.py b/amber/src/test/python/pytexera/storage/test_large_binary_input_stream.py index 7e6a401aef..f0cf967d17 100644 --- a/amber/src/test/python/pytexera/storage/test_large_binary_input_stream.py +++ b/amber/src/test/python/pytexera/storage/test_large_binary_input_stream.py @@ -223,3 +223,20 @@ class TestLargeBinaryInputStream: with pytest.raises(ValueError, match="I/O operation on closed stream"): stream.read() # Stream is already closed, no need to close again + + def test_close_is_idempotent(self, large_binary): + """Test that a second close() does not re-close the S3 body. + + IOBase's finalizer calls close() again during garbage collection, so + the early-return guard is what keeps the botocore stream from being + closed twice. + """ + stream = LargeBinaryInputStream(large_binary) + underlying = MagicMock() + stream._underlying = underlying + + stream.close() + stream.close() + + assert underlying.close.call_count == 1 + assert stream.closed diff --git a/amber/src/test/python/pytexera/udf/examples/test_echo_operator.py b/amber/src/test/python/pytexera/udf/examples/test_echo_operator.py index 7741272594..7547e9fa31 100644 --- a/amber/src/test/python/pytexera/udf/examples/test_echo_operator.py +++ b/amber/src/test/python/pytexera/udf/examples/test_echo_operator.py @@ -36,3 +36,16 @@ class TestEchoOperator: assert output_tuple == tuple_ with pytest.raises(StopIteration): next(outputs) + + def test_on_finish_emits_a_single_none(self, echo_operator): + # The echo operator has nothing buffered, so end-of-port emits one + # placeholder and nothing more. + # + # Resolve the override out of EchoOperator.__dict__ rather than off + # the instance: UDFOperatorV2.on_finish has a byte-identical body, so + # a plain attribute lookup falls back to the base class and the + # assertion below would still hold with this operator's own override + # deleted -- i.e. it would cover line 28 without pinning it. + on_finish = EchoOperator.__dict__["on_finish"] + + assert list(on_finish(echo_operator, 0)) == [None]
