From 9c0f53e47c3e3a89c83159adc650c39322a6bcc1 Mon Sep 17 00:00:00 2001 From: Petr Date: Sat, 26 Sep 2026 07:13:45 +0200 Subject: [PATCH] test: make false-passing negative controls fail on regression (#790) Rewrite negative controls that passed even with the guarded behavior broken, add @enum.unique to ErrorCode, and move the offline check_error_codes.py self-tests out of the `integration` marker so CI actually runs them. --- src/keboola_agent_cli/errors.py | 3 +- tests/test_auto_update.py | 6 ++- tests/test_billing_cli.py | 16 ++++-- tests/test_errors.py | 15 ++++-- tests/test_file_locking.py | 14 ++++-- tests/test_integration.py | 86 ++++++++++++-------------------- tests/test_permissions_cli.py | 21 ++++---- tests/test_search_service.py | 9 ++-- tests/test_server_permissions.py | 8 ++- 9 files changed, 96 insertions(+), 82 deletions(-) diff --git a/src/keboola_agent_cli/errors.py b/src/keboola_agent_cli/errors.py index 2ceeadd17..8c4aa6fb2 100644 --- a/src/keboola_agent_cli/errors.py +++ b/src/keboola_agent_cli/errors.py @@ -1,8 +1,9 @@ """Error types and helpers for Keboola Agent CLI.""" -from enum import StrEnum +from enum import StrEnum, unique +@unique class ErrorCode(StrEnum): """Stable machine-readable error codes emitted by kbagent. diff --git a/tests/test_auto_update.py b/tests/test_auto_update.py index 57cbdedfa..92e742148 100644 --- a/tests/test_auto_update.py +++ b/tests/test_auto_update.py @@ -879,8 +879,12 @@ def test_sentinel_is_set_even_when_body_raises(self): with patch( "keboola_agent_cli.auto_update._should_skip_all", side_effect=RuntimeError("kaboom"), - ): + ) as mock_skip_all: maybe_auto_update() # blanket try/except swallows the RuntimeError + assert auto_update_module._AUTO_UPDATE_RAN is True + maybe_auto_update() # must short-circuit on the sentinel + # The crashing body ran once; the re-entry never reached it again. + assert mock_skip_all.call_count == 1 class TestSafeStartupUpdateOrder: diff --git a/tests/test_billing_cli.py b/tests/test_billing_cli.py index 7f3f07eb2..27a938bca 100644 --- a/tests/test_billing_cli.py +++ b/tests/test_billing_cli.py @@ -109,13 +109,15 @@ def test_no_project_flag_forwards_none(self, tmp_path: Path) -> None: mock_service.get_credits.assert_called_once_with(aliases=None) def test_per_project_errors_surface_as_warnings_exit_0(self, tmp_path: Path) -> None: - store = _setup_config(tmp_path / "cfg", {"a": {}, "b": {}}) + # Distinct multi-letter aliases: a single-letter alias would match + # almost any output and prove nothing. + store = _setup_config(tmp_path / "cfg", {"alpha": {}, "beta": {}}) mock_service = MagicMock() mock_service.get_credits.return_value = { - "credits": [_credit_row("a")], + "credits": [_credit_row("alpha")], "errors": [ { - "project_alias": "b", + "project_alias": "beta", "error_code": "PAYG_NOT_AVAILABLE", "message": "Project does not have the 'pay-as-you-go' feature enabled.", } @@ -123,8 +125,12 @@ def test_per_project_errors_surface_as_warnings_exit_0(self, tmp_path: Path) -> } result = _run(["billing", "credits"], store, mock_service) assert result.exit_code == 0, result.output - assert "b" in result.output - assert "PAYG_NOT_AVAILABLE" in result.output or "pay-as-you-go" in result.output + # Rich may wrap the warning line; compare on collapsed whitespace. + flat = " ".join(result.output.split()) + assert "Warning: Project 'beta': Project does not have the 'pay-as-you-go'" in flat + # The healthy project still renders despite the other one failing. + assert "alpha" in flat + assert "25.50" in flat def test_empty_result_prints_no_payg_projects_line(self, tmp_path: Path) -> None: store = _setup_config(tmp_path / "cfg", {"prod": {}}) diff --git a/tests/test_errors.py b/tests/test_errors.py index fe7ae5f87..6f9c0baf6 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -113,10 +113,17 @@ def test_json_serialisation(self) -> None: assert json.dumps(ErrorCode.UPLOAD_FAILED) == '"UPLOAD_FAILED"' - def test_no_duplicate_values(self) -> None: - """Every ErrorCode member has a unique string value.""" - values = [c.value for c in ErrorCode] - assert len(values) == len(set(values)), "Duplicate ErrorCode values detected" + def test_no_aliased_members(self) -> None: + """No ErrorCode name is an alias of another member. + + A duplicate value does not create a second member -- Enum turns the + later name into an alias, and iterating the enum skips aliases, so a + comparison over ``list(ErrorCode)`` can never see it. ``__members__`` + does include aliases. ``@enum.unique`` on the class rejects them at + import time; this pins that the decorator stays. + """ + aliases = [name for name, member in ErrorCode.__members__.items() if member.name != name] + assert aliases == [], f"ErrorCode aliases (duplicate values): {aliases}" def test_known_codes_present(self) -> None: """Spot-check that key codes defined in the spec are present.""" diff --git a/tests/test_file_locking.py b/tests/test_file_locking.py index 8f099bc42..582decc2e 100644 --- a/tests/test_file_locking.py +++ b/tests/test_file_locking.py @@ -44,10 +44,18 @@ def test_flock_suppresses_oserror(self, tmp_path: Path) -> None: _try_flock(42, fcntl.LOCK_EX) def test_flock_skipped_when_no_fcntl(self) -> None: - """_try_flock does nothing when _HAS_FCNTL is False.""" - with patch("keboola_agent_cli.config_store._HAS_FCNTL", False): - # Should not raise, should not call fcntl + """_try_flock never touches fcntl when _HAS_FCNTL is False. + + fcntl is replaced by a mock so the assertion is on the call itself: + a real flock on a bogus fd raises OSError, which _try_flock suppresses, + so "does not raise" alone would pass even without the guard. + """ + with ( + patch("keboola_agent_cli.config_store._HAS_FCNTL", False), + patch("keboola_agent_cli.config_store.fcntl", create=True) as mock_fcntl, + ): _try_flock(42, 0) + mock_fcntl.flock.assert_not_called() class TestFileLockingIntegration: diff --git a/tests/test_integration.py b/tests/test_integration.py index daf478e33..c43d37972 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -245,97 +245,77 @@ def test_context_command_works(self, integration_config_dir: Path) -> None: # =========================================================================== -# CI guard: check_error_codes.py catches planted raw strings +# CI guard: check_error_codes.py self-tests +# +# Pure offline checks of scripts/check_error_codes.py -- no network, no +# credentials -- so they are deliberately NOT marked `integration` and run in +# the normal CI suite (which deselects `-m integration`). # =========================================================================== +_REPO_ROOT = Path(__file__).resolve().parent.parent +_GUARD_SCRIPT = _REPO_ROOT / "scripts" / "check_error_codes.py" + + +def _load_guard_script(): + import importlib.util + + spec = importlib.util.spec_from_file_location("check_error_codes", _GUARD_SCRIPT) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + -@pytest.mark.integration class TestCheckErrorCodesGuard: """Verify the CI guard script rejects raw error_code string literals.""" def test_guard_passes_on_clean_source(self) -> None: """scripts/check_error_codes.py exits 0 on the current (clean) source.""" result = subprocess.run( - [sys.executable, "scripts/check_error_codes.py"], + [sys.executable, str(_GUARD_SCRIPT)], capture_output=True, check=False, text=True, + cwd=_REPO_ROOT, ) assert result.returncode == 0, ( f"Guard failed on clean source:\n{result.stdout}\n{result.stderr}" ) def test_guard_catches_planted_literal(self, tmp_path: Path) -> None: - """Guard exits 1 when a raw string literal is planted in a temp source file.""" - # Write a minimal Python file that uses a raw error_code string + """The guard's own scanner flags a raw error_code string literal.""" planted = tmp_path / "planted.py" planted.write_text( "from keboola_agent_cli.errors import KeboolaApiError\n" 'raise KeboolaApiError("oops", error_code="QUEUE_JOB_FAILED")\n', encoding="utf-8", ) - # Run the guard against only this file by patching SRC_ROOT via env isn't - # practical; instead verify the guard script's logic directly via import. - import ast - - source = planted.read_text(encoding="utf-8") - tree = ast.parse(source) - violations = [] - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - for kw in node.keywords: - if kw.arg == "error_code" and isinstance(kw.value, ast.Constant): - violations.append(kw.value.value) - - assert violations == ["QUEUE_JOB_FAILED"], ( - "Guard logic should detect the planted raw string literal" - ) + mod = _load_guard_script() + assert mod._collect_violations(planted) == [(2, "QUEUE_JOB_FAILED")] def test_guard_ignores_enum_usage(self, tmp_path: Path) -> None: - """Guard logic does NOT flag error_code=ErrorCode.X (non-Constant node).""" - import ast - - source = ( + """The guard's scanner does NOT flag error_code=ErrorCode.X.""" + clean = tmp_path / "clean.py" + clean.write_text( "from keboola_agent_cli.errors import ErrorCode, KeboolaApiError\n" - 'raise KeboolaApiError("oops", error_code=ErrorCode.QUEUE_JOB_FAILED)\n' + 'raise KeboolaApiError("oops", error_code=ErrorCode.QUEUE_JOB_FAILED)\n', + encoding="utf-8", ) - tree = ast.parse(source) - violations = [] - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - for kw in node.keywords: - if kw.arg == "error_code" and isinstance(kw.value, ast.Constant): - violations.append(kw.value.value) - - assert violations == [], "Enum usage should not be flagged as a violation" + mod = _load_guard_script() + assert mod._collect_violations(clean) == [] -@pytest.mark.integration class TestErrorCodesDocCompleteness: """Verify the enum-vs-docs/error-codes.md completeness guard.""" - @staticmethod - def _load_script(): - import importlib.util - - spec = importlib.util.spec_from_file_location( - "check_error_codes", Path("scripts") / "check_error_codes.py" - ) - assert spec is not None and spec.loader is not None - mod = importlib.util.module_from_spec(spec) - spec.loader.exec_module(mod) - return mod - def test_doc_matches_enum(self) -> None: """docs/error-codes.md documents exactly the ErrorCode members.""" - mod = self._load_script() + mod = _load_guard_script() assert mod._enum_members() == mod._documented_codes() def test_detects_missing_code(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Removing one documented code from the doc makes the check fail.""" - mod = self._load_script() + mod = _load_guard_script() doc_lines = mod.DOC_PATH.read_text(encoding="utf-8").splitlines(keepends=True) pruned = [line for line in doc_lines if not line.startswith("| `INVALID_TOKEN` |")] assert len(pruned) == len(doc_lines) - 1 @@ -346,7 +326,7 @@ def test_detects_missing_code(self, tmp_path: Path, monkeypatch: pytest.MonkeyPa def test_detects_stale_code(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """A doc row for a code that is not in the enum makes the check fail.""" - mod = self._load_script() + mod = _load_guard_script() doc = mod.DOC_PATH.read_text(encoding="utf-8") doc += "| `NO_SUCH_CODE_EVER` | Planted stale row |\n" stale_doc = tmp_path / "error-codes.md" diff --git a/tests/test_permissions_cli.py b/tests/test_permissions_cli.py index b0ae7b7d7..138c9b494 100644 --- a/tests/test_permissions_cli.py +++ b/tests/test_permissions_cli.py @@ -317,18 +317,21 @@ def test_set_allow_mode_with_deny(self, tmp_path: Path) -> None: assert config.permissions.deny == ["cli:write", "cli:destructive"] def test_set_rejected_without_confirmation(self, tmp_path: Path) -> None: - """set should fail when confirmation is not provided (non-interactive).""" + """set refuses without a TTY confirmation and persists nothing. + + The real confirmation helper runs (CliRunner has no TTY), so the exit + code comes from the production refusal path, not from a mock -- and + the store is checked so a write that happened before the refusal + would fail the test. + """ store = _make_store(tmp_path) - with ( - patch("keboola_agent_cli.cli.ConfigStore") as MockStore, - patch( - "keboola_agent_cli.commands.permissions.require_random_code_confirmation", - side_effect=typer.Exit(code=EXIT_PERMISSION_DENIED), - ), - ): + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: MockStore.return_value = store - result = runner.invoke(app, ["--json", "permissions", "set", "--mode", "allow"]) + result = runner.invoke( + app, ["--json", "permissions", "set", "--mode", "allow", "--deny", "cli:write"] + ) assert result.exit_code == EXIT_PERMISSION_DENIED + assert store.load().permissions is None def test_set_invalid_mode(self, tmp_path: Path) -> None: store = _make_store(tmp_path) diff --git a/tests/test_search_service.py b/tests/test_search_service.py index 023608208..c913ee400 100644 --- a/tests/test_search_service.py +++ b/tests/test_search_service.py @@ -93,10 +93,11 @@ def test_flow_maps_correctly(self) -> None: assert _resolve_api_types(["flow"]) == ["flow"] def test_multiple_types_deduped(self) -> None: - result = _resolve_api_types(["table", "bucket"]) - assert "table" in result - assert "bucket" in result - assert len(result) == 2 + # A repeated type, plus two spellings ("config" and the raw API + # "configuration") that resolve to the same API value. Each API type + # appears once, in first-seen order. + result = _resolve_api_types(["table", "config", "bucket", "table", "configuration"]) + assert result == ["table", "configuration", "bucket"] def test_unknown_type_passed_through(self) -> None: # Unknown types are passed directly to the API. diff --git a/tests/test_server_permissions.py b/tests/test_server_permissions.py index 82daa626c..4a549af46 100644 --- a/tests/test_server_permissions.py +++ b/tests/test_server_permissions.py @@ -34,6 +34,7 @@ from keboola_agent_cli.config_store import ConfigStore from keboola_agent_cli.constants import EXIT_PERMISSION_DENIED +from keboola_agent_cli.errors import PermissionDeniedError from keboola_agent_cli.models import PermissionPolicy from keboola_agent_cli.permissions import ( OPERATION_REGISTRY, @@ -190,8 +191,11 @@ def test_app_without_an_engine_fails_closed(self, tmp_path: Path) -> None: app = FastAPI() install_auth(app, AuthSettings(token=TOKEN)) _install_probes(app) - resp = TestClient(app, raise_server_exceptions=False).get(READ_PROBE, headers=AUTH) - assert resp.status_code != 200 + # The bare app has no PermissionDeniedError -> 403 handler, so any + # crash would also be "not 200". Let the exception surface and pin it + # to the deliberate refusal, not an accidental AttributeError. + with pytest.raises(PermissionDeniedError, match="not built by create_app"): + TestClient(app).get(READ_PROBE, headers=AUTH) class TestRegistryWiring: