test: make false-passing negative controls fail on regression (#790)

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.
 

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:

tests/test_billing_cli.py

@@ -109,22 +109,28 @@ 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.",
                 }
             ],
         }
         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": {}})

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."""

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:

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"

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)

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.

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: