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-8050-eddec267a2c6974d0ea0f29b63e9c7506e12f9b4 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 8b3aefec9dadd2933fa63b53c05e6a152f435175 Author: Xinyuan Lin <[email protected]> AuthorDate: Fri Aug 28 02:41:42 2026 +0000 test(pyamber): cover the python worker entry point (#8050) ### What changes were proposed in this PR? `test_run_python_worker.py` goes from 27 tests to 42, covering the worker entry point's configuration-failure paths. | Metric | Before | After | |---|---|---| | Codecov fully-covered lines **today** | 32/39 | **39/39** | | Codecov fully-covered lines **after #8040** | 31/39 | **39/39** | | Branch arms | 9/12 | **12/12** | **`texera_run_python_worker.py` reaches 100% on both lines and branches.** +7 fully-covered lines today, +8 and +3 arms after #8040 — line 155 is a partial that the current branch-less report already counts covered, because it executes at import. The uncovered set was the argument-parsing failures (26–28), the loguru level configuration (35), and the missing-dependency and R_HOME paths (48, 51, 155–156). These are what run when the worker starts with a malformed or incomplete configuration, so a regression there surfaces as a worker that fails to start with an unhelpful message. One correction to the assessment worth recording: **line 143, the R_HOME guard, was already at 2/2** — it was not part of the gap. ### Verification Two reviewers returned nine findings and reported **thirteen surviving mutants** against the first draft. All thirteen were independently reproduced on a pristine tree before being fixed, and all thirteen now die. **39 mutations in the final table: 38 killed, 1 equivalent, 0 real survivors.** The equivalent one is `frozenset` to `set` for `EXPECTED_CONFIG_KEYS` — reported rather than omitted. The repair also confirmed something worth stating about the measurement itself: `pytest`'s `importlib` import mode re-executes this module per test, and there is **exactly one** `<class filename="texera_run_python_worker.py">` entry in every report, on both sides and in the full-suite run. So the fresh execs attribute to the same source path rather than splitting the coverage entry — checked rather than assumed. Measured from `amber/` with an identical single-spec test-path scope on both sides, both with and without `--cov-branch` (the no-branch figure measured directly, not derived), per-line `hits` and `condition-coverage` read out of `coverage.xml` rather than from the summary percentage. Full pyamber suite: `5 failed, 1174 passed, 7 errors` against main's `1159 passed`. Failure sets sorted and diffed — **identical by identity**, 12 entries. `ruff check` and `ruff format --check` pass on CI's exact scope. No production file is touched; `git diff -- 'amber/src/main/*'` is empty. ### Any related issues, documentation, discussions? Closes #8046 ### How was this PR tested? ``` cd amber && python -m pytest -m "not integration" -q src/test/python/test_run_python_worker.py ``` ``` 42 passed ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- amber/src/test/python/test_run_python_worker.py | 309 +++++++++++++++++++++++- 1 file changed, 307 insertions(+), 2 deletions(-) diff --git a/amber/src/test/python/test_run_python_worker.py b/amber/src/test/python/test_run_python_worker.py index 7deec97d48..2c4280ac59 100644 --- a/amber/src/test/python/test_run_python_worker.py +++ b/amber/src/test/python/test_run_python_worker.py @@ -16,10 +16,18 @@ # under the License. import base64 +import builtins +import importlib.util +import io import json +import os +import sys +import types +from pathlib import Path from unittest import mock import pytest +from loguru import logger import texera_run_python_worker as entry @@ -155,10 +163,15 @@ def test_parse_rejects_a_missing_key(missing_key): def test_parse_rejects_an_unexpected_key(): - """An extra key (e.g. the JVM side added a field) fails instead of being ignored.""" + """An extra key (e.g. the JVM side added a field) fails instead of being ignored, + and is reported on the `unexpected` side — that message is the whole diagnostic + an operator gets when the two sides drift, so naming the wrong side sends the + reader to the wrong end of the contract.""" config = _full_config() config["someNewField"] = "value" - with pytest.raises(ValueError, match="key mismatch"): + with pytest.raises( + ValueError, match=r"missing=\[\], unexpected=\['someNewField'\]" + ): entry.parse_startup_config(_encode(config)) @@ -179,3 +192,295 @@ def test_parse_round_trips_a_base64_encoded_config(): """The config is passed as Base64-encoded JSON; parsing decodes it back.""" config = _full_config() assert entry.parse_startup_config(_encode(config)) == config + + +ENTRY_MODULE_PATH = Path(entry.__file__) + + +def _exec_entry_module(module_name: str): + """Execute the entry script afresh from its file under the given module name. + + `exec_module` deliberately does not register the result in `sys.modules`, so a + run that aborts half-way through its imports — or one executed as `__main__` — + cannot disturb the already-imported `texera_run_python_worker` that the tests + above use. Loading by file path keeps the executed code objects anchored to the + real source file, so coverage still attributes the lines to it. + """ + spec = importlib.util.spec_from_file_location(module_name, ENTRY_MODULE_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _exec_entry_module_with_failing_collaborator_imports(error): + """Re-run the entry script with its `core.*` imports raising `error`. + + Only the `core` package is failed: the stdlib and loguru imports above the + try/except must still succeed, otherwise the failure would never reach the + handler under test. + """ + real_import = builtins.__import__ + + def failing_import(name, globals=None, locals=None, fromlist=(), level=0): + if name.split(".")[0] == "core": + raise error + return real_import(name, globals, locals, fromlist, level) + + with mock.patch.object(builtins, "__import__", failing_import): + return _exec_entry_module("texera_run_python_worker_reexec") + + [email protected] +def restored_loguru_sink(): + """Undo `init_loguru_logger`'s process-global handler surgery. + + `logger.remove()` stops every loguru handler for the remainder of the pytest + process, so a test that lets it run must hand a working stderr sink back to the + suites that follow. The stopped handler objects cannot be re-registered, so a + fresh sink is added instead. `LOGURU_LEVEL` is safe to reuse verbatim: loguru + validates it while configuring its own default handler at import time, so an + unusable value would already have failed long before this fixture runs. + """ + yield + logger.remove() + logger.add( + sys.__stderr__ or sys.stderr, level=os.environ.get("LOGURU_LEVEL", "DEBUG") + ) + + +def test_missing_proto_package_exits_with_generation_guidance(): + """The generated proto bindings are not checked in, so their absence is the + single most likely first-run failure; it must exit with the regeneration + recipe rather than a bare ImportError traceback.""" + with pytest.raises(SystemExit) as excinfo: + _exec_entry_module_with_failing_collaborator_imports( + ModuleNotFoundError("No module named 'proto'", name="proto") + ) + + message = str(excinfo.value) + assert "amber/src/main/python/proto/" in message + assert "bin/python-proto-gen.sh" in message + + +def test_missing_proto_submodule_exits_with_generation_guidance(): + """A partially generated tree reports a submodule rather than `proto` itself, + which has to be recognised as the same failure.""" + with pytest.raises(SystemExit) as excinfo: + _exec_entry_module_with_failing_collaborator_imports( + ModuleNotFoundError( + "No module named 'proto.edu'", name="proto.edu.uci.ics.amber" + ) + ) + + assert "bin/python-proto-gen.sh" in str(excinfo.value) + + [email protected]( + "missing_name", + [ + "numpy", + # A dependency whose name merely *begins with* "proto" — protobuf sits right + # next to the generated `proto/` package in this project — must not match. + "protobuf", + # `core` is the very package the try block imports: a broken PYTHONPATH or a + # worker spawned from the wrong cwd makes it unimportable, and blaming the + # proto bindings for that sends the operator off to regenerate bindings that + # are already present. + "core", + "core.models", + ], +) +def test_unrelated_missing_dependency_propagates(missing_name): + """A genuinely missing third-party dependency must not be misreported as + missing proto bindings: the original error has to reach the caller. The guard + has to be specific as well as complete — matching a name that merely starts + with or contains "proto" is as wrong as failing to match "proto." itself.""" + with pytest.raises(ModuleNotFoundError) as excinfo: + _exec_entry_module_with_failing_collaborator_imports( + ModuleNotFoundError(f"No module named {missing_name!r}", name=missing_name) + ) + + assert excinfo.value.name == missing_name + + +def test_missing_dependency_without_a_name_propagates(): + """`ModuleNotFoundError.name` is optional, so the proto check must tolerate a + `None` name instead of failing with an AttributeError of its own.""" + with pytest.raises(ModuleNotFoundError) as excinfo: + _exec_entry_module_with_failing_collaborator_imports( + ModuleNotFoundError("the import machinery reported no module name") + ) + + assert excinfo.value.name is None + + [email protected]( + ("requested_level", "info_expected"), + # Two rows, and a probe *between* DEBUG(10) and WARNING(30), are what make the + # level parameter observable. A single row bounded only by debug and warning is + # satisfied by every hardcoded constant in {INFO, SUCCESS, WARNING}, because + # loguru puts INFO(20) and SUCCESS(25) inside that window. + [("WARNING", False), ("INFO", True)], +) +def test_init_loguru_logger_replaces_handlers_at_the_requested_level( + monkeypatch, restored_loguru_sink, requested_level, info_expected +): + """Configuring the logger means *replacing* loguru's handlers: any sink that + was already registered (loguru installs one by default) must be gone, and the + new stderr sink must honour the level it was *asked* for rather than a level of + its own choosing.""" + pre_existing_records = [] + logger.add(pre_existing_records.append, level="DEBUG") + stderr_sink = io.StringIO() + monkeypatch.setattr(sys, "stderr", stderr_sink) + + entry.init_loguru_logger(requested_level) + + logger.debug("below-every-requested-level") + logger.info("between-the-requested-levels") + logger.warning("at-or-above-every-requested-level") + + emitted = stderr_sink.getvalue() + assert "at-or-above-every-requested-level" in emitted + assert "below-every-requested-level" not in emitted + assert ("between-the-requested-levels" in emitted) is info_expected + assert pre_existing_records == [] + + +def test_script_entry_point_starts_the_worker_from_argv(restored_loguru_sink): + """Run as a script — the way PythonWorkflowWorker on the JVM side spawns it — + the module must start a worker from the config in argv[1].""" + python_worker = mock.MagicMock() + worker_module = types.ModuleType("core.python_worker") + worker_module.PythonWorker = python_worker + storage_module = types.ModuleType("core.storage.storage_config") + storage_module.StorageConfig = mock.MagicMock() + # The third element exists only so that index 1 and index -1 differ, making the + # argv *index* observable. The JVM side passes exactly one argument, so nothing + # is asserted about trailing arguments beyond their not being what gets parsed. + argv = [ + "src/main/python/texera_run_python_worker.py", + _encode(_full_config()), + "trailing-argument-the-entry-point-must-ignore", + ] + + with ( + mock.patch.dict( + sys.modules, + { + "core.python_worker": worker_module, + "core.storage.storage_config": storage_module, + }, + ), + mock.patch.object(sys, "argv", argv), + ): + _exec_entry_module("__main__") + + python_worker.assert_called_once_with( + worker_id="worker-1", host="localhost", output_port=5005 + ) + python_worker.return_value.run.assert_called_once() + + +def test_parse_names_the_missing_key_on_the_missing_side(): + """The companion of test_parse_rejects_an_unexpected_key: a key the JVM stopped + sending has to be reported as *missing*. The 19-way parametrized test above can + only afford a loose match, so the side of the diagnostic is fenced here.""" + config = _full_config() + del config["s3Region"] + with pytest.raises(ValueError, match=r"missing=\['s3Region'\], unexpected=\[\]"): + entry.parse_startup_config(_encode(config)) + + +def _encode_raw_utf8(config) -> str: + """Encode a config the way Jackson actually does: with raw, unescaped UTF-8. + + `_encode` cannot model the real producer here. `json.dumps` defaults to + `ensure_ascii=True`, which escapes every non-ASCII character, so any payload it + builds is pure ASCII. PythonWorkflowWorker on the JVM side does + `objectMapper.writeValueAsString(...)` then + `Base64.getEncoder.encodeToString(json.getBytes(StandardCharsets.UTF_8))`, and + Jackson emits non-ASCII characters literally. `ensure_ascii=False` is therefore + the entire point of this helper - do not "simplify" it back to `_encode`. + """ + return base64.b64encode( + json.dumps(config, ensure_ascii=False).encode("utf-8") + ).decode("ascii") + + +def test_parse_decodes_non_ascii_values_as_utf8(): + """Config values arrive as raw UTF-8 bytes, so they must be decoded as UTF-8. + + A non-ASCII value is entirely reachable in production - a Windows profile + directory under a non-ASCII user name is the obvious case - and decoding the + payload as ASCII would kill every worker on such a host at startup, before a + single log line. + """ + non_ascii_path = "/tmp/\u0444\u0430\u0439\u043b\u044b-\u6570\u636e" + config = _full_config() + config["icebergFileStorageDirectoryPath"] = non_ascii_path + + parsed = entry.parse_startup_config(_encode_raw_utf8(config)) + + assert parsed["icebergFileStorageDirectoryPath"] == non_ascii_path + + +def test_main_forwards_the_configured_logger_level(): + """`loggerLevel` carries the JVM's UdfConfig.pythonLogStreamHandlerLevel, and it + is the one named field no other test observes reaching its destination: the three + main() tests replace init_loguru_logger with a mock and never look at the call. + The level used here differs from the sample config's on purpose, so no hardcoded + constant at the call site can satisfy the assertion.""" + assert _full_config()["loggerLevel"] != "WARNING" + config = _full_config() + config["loggerLevel"] = "WARNING" + storage_patch, worker_patch, _unused_logger_patch = _patched_collaborators() + with ( + storage_patch, + worker_patch, + mock.patch.object(entry, "init_loguru_logger") as init_logger, + ): + entry.main(_encode(config)) + + init_logger.assert_called_once_with("WARNING") + + +def test_main_keeps_the_rest_catalog_uri_and_warehouse_name_in_order(): + """Fence the one adjacent pair of StorageConfig.initialize arguments the sample + config cannot tell apart: `icebergRestCatalogUri` and + `icebergRestCatalogWarehouseName` are both "" there, so a swap of the two would + satisfy assert_called_once_with in either order. On a REST catalog both fields + are non-empty, and swapping them points the catalog URI at a warehouse name. + + The values are overridden locally rather than in `_full_config()` so the shared + fixture, and the two full-mapping assertions built on it, stay untouched. + """ + config = _full_config() + config["icebergRestCatalogUri"] = "http://rest-catalog:8181"; + config["icebergRestCatalogWarehouseName"] = "warehouse-name" + storage_patch, worker_patch, logger_patch = _patched_collaborators() + with storage_patch as storage_config, worker_patch, logger_patch: + entry.main(_encode(config)) + + positional = storage_config.initialize.call_args.args + assert positional[4] == "http://rest-catalog:8181"; + assert positional[5] == "warehouse-name" + + +def test_main_leaves_r_home_unset_when_r_path_is_blank(monkeypatch): + """The false arm of the R_HOME guard: a non-R worker must not be handed an empty + R_HOME. An empty value is worse than an absent one for rpy2's R discovery, which + is exactly what the guard exists to prevent. Every other main() test runs the + false arm too, but none of them looks at the environment.""" + # Load-bearing, not decorative: test_main_sets_r_home_when_r_path_present lets + # production write R_HOME directly, and monkeypatch.delenv records nothing to + # restore when the variable was absent to begin with - so R_HOME leaks out of + # that test and this one would otherwise depend on collection order. + monkeypatch.delenv("R_HOME", raising=False) + config = _full_config() + assert config["rPath"] == "" + storage_patch, worker_patch, logger_patch = _patched_collaborators() + with storage_patch, worker_patch, logger_patch: + entry.main(_encode(config)) + + assert "R_HOME" not in os.environ
