diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2f18f66c..010b960d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,6 +29,19 @@ jobs: - run: uvx deptry src/ - run: uv run pytest --cov --cov-fail-under=80 + windows-test: + runs-on: windows-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + with: + persist-credentials: false + - uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 + with: + python-version: "3.12" + enable-cache: true + - run: uv sync --locked --dev --all-extras + - run: uv run pytest -q + pre-commit: runs-on: korvid-runners steps: diff --git a/README.md b/README.md index eac6a0c8..27d078e3 100644 --- a/README.md +++ b/README.md @@ -185,3 +185,5 @@ uv sync --dev --all-extras # create .venv with locked deps + all extras uv run korvid # run against your current kubeconfig context make check # lint + mypy --strict + tach + tests ``` + +Contributor docs: [Windows contributor notes](docs/windows.md). diff --git a/docs/ops.md b/docs/ops.md index a033559c..66d5211d 100644 --- a/docs/ops.md +++ b/docs/ops.md @@ -78,7 +78,7 @@ protected contexts. ### Crash recovery If a fatal exception escapes the TUI, korvid restores the terminal, logs -the traceback, and asks `korvid crashed — restart? [Y/n]` instead of just +the traceback, and asks `korvid crashed -- restart? [Y/n]` instead of just dying. A restart rebuilds everything from scratch — a fresh event loop, Kubernetes client, agent provider, and MCP server; nothing from the crashed run is reused, and no approval state or pending proposal survives (the diff --git a/docs/windows.md b/docs/windows.md new file mode 100644 index 00000000..39708ce4 --- /dev/null +++ b/docs/windows.md @@ -0,0 +1,48 @@ +# Windows contributor notes + +Native Windows development is supported for dependency sync and the full test +suite: + +```sh +uv sync --dev --all-extras +uv run pytest -q +``` + +PRs that touch shared/runtime behavior must also keep the required +`windows-test` CI job green. The proving run for issue #173 was +`30936032385`: **3376 passed / 37 skipped / 0 failures**. + +## Expected skips on Windows + +- **21 opt-in contract-suite skips** when `KORVID_CONTRACT_RUN_ID` is unset. +- **16 capability skips**: + - **3 newly classified capability skips**: 2 `~user` POSIX account-lookup + cases (`tests/core/test_transfer.py`, `tests/ui/test_transfer_picker.py`) + and 1 POSIX directory-fsync failure case. + - **13 pre-existing platform skips** for POSIX-only permission semantics: + 7 local transfer permission-bit cases, 3 audit-log mode cases, + 2 transfer-stream late-permission-loss cases, and 1 unreadable CA bundle + permission case. + +## Current Windows limits + +- Symlink tests depend on Windows **Developer Mode** (or an elevated shell): + native `Path.symlink_to()` can fail before korvid logic runs if symlink + creation is not allowed. The shared helper turns that into a capability skip + only when Windows symlink privilege is absent; the final hosted runner had privilege, so the count remained 37. +- Interactive shell attach can hit Textual's `SuspendNotSupported` path on + Windows/non-suspending drivers. That refusal is expected; render-only tests + pin `legacy_windows=False` for deterministic Rich output. +- NTFS uses ACLs, not POSIX mode bits. Tests verify atomic create/replace, the requested `0o600` mode, and durability semantics, but they do not claim ACL confidentiality. + +## Windows-specific fixes covered by the green run + +- Log export opens files with `newline=""`, preserving exact LF bytes instead + of writing CRLF. +- Terminal/status text now stays ASCII-safe on cp1252-style consoles; the + literal replacements are `->` and `--`: + `korvid shell -> ...`, `korvid node shell -> ...`, and + `korvid crashed -- restart? [Y/n]`. +- The write gate now takes an `op_factory`, so blocked or cancelled writes + never create eager mutation coroutines; cancellation safety is + cross-platform. diff --git a/src/korvid/__main__.py b/src/korvid/__main__.py index 7a200035..efd096f9 100644 --- a/src/korvid/__main__.py +++ b/src/korvid/__main__.py @@ -1132,7 +1132,7 @@ def _run_with_recovery( if len(crash_times) >= RESTART_CAP: print( f"korvid crashed {len(crash_times)} times within" - f" {RESTART_WINDOW_SECONDS:.0f}s — not restarting.", + f" {RESTART_WINDOW_SECONDS:.0f}s -- not restarting.", file=sys.stderr, ) raise @@ -1145,7 +1145,7 @@ def _run_with_recovery( def _restart_prompt() -> str: # Interactivity keys off stdin/stderr; a redirected stdout must neither # swallow the question nor be contaminated by it. - print("korvid crashed — restart? [Y/n] ", end="", file=sys.stderr, flush=True) + print("korvid crashed -- restart? [Y/n] ", end="", file=sys.stderr, flush=True) return input() diff --git a/src/korvid/core/config.py b/src/korvid/core/config.py index ac66072d..b55a826a 100644 --- a/src/korvid/core/config.py +++ b/src/korvid/core/config.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os import re import unicodedata from collections.abc import Mapping @@ -348,10 +349,14 @@ def _atomic_write_text(path: Path, text: str) -> None: unrelated keys), a power loss cannot leave an empty file, and concurrent writers cannot race on a shared temp name.""" try: - # Preserve an existing restrictive mode; default new files to 0600. - mode = S_IMODE(path.stat().st_mode) + existing_mode = S_IMODE(path.stat().st_mode) except OSError: - mode = 0o600 + existing_mode = None + # On Windows, POSIX stat mode emulation returns 0o666 for readable+writable + # files regardless of actual ACLs; we cannot trust it as a "preserve" signal + # and always request the restrictive 0o600. On POSIX the real mode is + # meaningful, so we honour it when present. + mode = existing_mode if os.name != "nt" and existing_mode is not None else 0o600 fd, tmp_name = mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp") tmp = Path(tmp_name) try: diff --git a/src/korvid/core/logexport.py b/src/korvid/core/logexport.py index 2a3b485a..21a1cdaa 100644 --- a/src/korvid/core/logexport.py +++ b/src/korvid/core/logexport.py @@ -83,7 +83,7 @@ def export_log_lines( # (cluster logs may be sensitive), and explicit UTF-8 so a # non-UTF-8 locale can't fail on non-ASCII log text. fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) - with open(fd, "w", encoding="utf-8") as fh: + with open(fd, "w", encoding="utf-8", newline="") as fh: fh.write(content) except FileExistsError: continue diff --git a/src/korvid/core/transfer.py b/src/korvid/core/transfer.py index 06e4236a..0158ff19 100644 --- a/src/korvid/core/transfer.py +++ b/src/korvid/core/transfer.py @@ -311,9 +311,10 @@ def default_local_path(remote_path: str) -> str: ``validate_spec``'s parent checks. """ base = posixpath.basename(remote_path.rstrip("/")) or "download" - downloads = Path("~/Downloads").expanduser() + home = Path.home() + downloads = home / "Downloads" usable = downloads.is_dir() and os.access(downloads, os.W_OK | os.X_OK) - directory = downloads if usable else Path("~").expanduser() + directory = downloads if usable else home return str(directory / base) diff --git a/src/korvid/providers/net.py b/src/korvid/providers/net.py index a8ad716f..dc585c8c 100644 --- a/src/korvid/providers/net.py +++ b/src/korvid/providers/net.py @@ -56,7 +56,7 @@ async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response: if "CERTIFICATE_VERIFY_FAILED" in str(exc): raise httpx.ConnectError( f"TLS verification failed against network.ca_bundle" - f" {self._ca_bundle_path!r}: {exc}", + f" '{self._ca_bundle_path}': {exc}", request=request, ) from exc raise diff --git a/src/korvid/ui/app.py b/src/korvid/ui/app.py index 7a457e14..e40681c1 100644 --- a/src/korvid/ui/app.py +++ b/src/korvid/ui/app.py @@ -4243,7 +4243,7 @@ def _run_shell(self, namespace: str, name: str, container: str | None) -> None: argv = build_exec_argv(namespace, name, container, context=self.config.kube_context) target = f"{name}/{container}" if container else name with self.suspend(): - exit_code = self._run_interactive(argv, f"korvid shell → {target} (exit to return)") + exit_code = self._run_interactive(argv, f"korvid shell -> {target} (exit to return)") self.refresh() if exit_code == 0: return @@ -5182,7 +5182,7 @@ async def _run_write( meta: ResourceMeta, namespace: str | None, name: str, - op: Awaitable[None], + op_factory: Callable[[], Awaitable[None]], detail: str = "", ) -> str: """Execute an approved write with fail-closed auditing (AGENTS.md): @@ -5190,12 +5190,18 @@ async def _run_write( the write is blocked. Returns a short outcome string ('done' / 'blocked: ...' / 'failed: ...') for callers that report back. + Takes an operation *factory* so the mutation coroutine is never + created until intent is audited — a declined or blocked write + produces no unawaited coroutine to leak. + The whole span publishes an in-flight progress label (issue #143): between approval and the outcome toast there was previously no visible state at all.""" kind = meta.plural with self._progress(f"{action} {kind}/{name}"): - return await self._run_write_inner(action, meta, namespace, name, op, detail, kind) + return await self._run_write_inner( + action, meta, namespace, name, op_factory, detail, kind + ) async def _run_write_inner( self, @@ -5203,19 +5209,14 @@ async def _run_write_inner( meta: ResourceMeta, namespace: str | None, name: str, - op: Awaitable[None], + op_factory: Callable[[], Awaitable[None]], detail: str, kind: str, ) -> str: try: await self._audit_write(action, meta, namespace, name, detail, "intent") except Exception as exc: - close = getattr(op, "close", None) - if callable(close): - close() # avoid "coroutine was never awaited" for the blocked op - # The exception can embed the local audit path (home directory): - # log it here, but keep the notification and the tool result - - # which is sent to the LLM provider - free of filesystem details. + # Factory was never called — no coroutine to leak. logger.exception("audit intent record failed; write blocked: %s", exc) self.notify( f"{action} {kind}/{name} blocked: audit log unavailable", @@ -5223,7 +5224,7 @@ async def _run_write_inner( ) return "blocked: audit log unavailable" try: - await op + await op_factory() except ApiStatusError as exc: with contextlib.suppress(Exception): await self._audit_write(action, meta, namespace, name, detail, f"error: {exc}") @@ -5294,7 +5295,7 @@ async def _push_write_confirmation( def _done(confirmed: bool | None) -> None: if confirmed: self.run_worker( - self._run_write(action, meta, namespace, name, op_factory(), detail=detail) + self._run_write(action, meta, namespace, name, op_factory, detail=detail) ) await self.push_screen( @@ -6255,7 +6256,7 @@ async def _wait_and_attach_node_shell(self, node: str, namespace: str, pod_name: try: with self.suspend(): exit_code = self._run_interactive( - attach_argv, f"korvid node shell → {node} (exit to return)" + attach_argv, f"korvid node shell -> {node} (exit to return)" ) except SuspendNotSupported: # Non-suspending drivers (e.g. Windows, web): refuse gracefully — @@ -6666,7 +6667,7 @@ def _done(confirmed: bool | None) -> None: sub_meta, namespace, facts.package, - ops.create_object(sub_meta, namespace, manifest), + lambda: ops.create_object(sub_meta, namespace, manifest), detail=f"channel={channel} approval={approval} source={facts.catalog_source}", ) ) @@ -6942,7 +6943,7 @@ async def _operator_apply_uninstall( sub_meta, ns, name, - ops.delete_object(sub_meta, ns, name, uid=uid), + lambda: ops.delete_object(sub_meta, ns, name, uid=uid), detail=f"csv={csv_name or '-'}", ) if outcome != "done" or csv_meta is None or not csv_name: @@ -6952,7 +6953,7 @@ async def _operator_apply_uninstall( csv_meta, ns, csv_name, - ops.delete_object(csv_meta, ns, csv_name, uid=csv_uid), + lambda: ops.delete_object(csv_meta, ns, csv_name, uid=csv_uid), detail=f"subscription={name}", ) @@ -7978,17 +7979,22 @@ def _refresh_status(self) -> None: mcp_label = self._mcp.status() if self._mcp is not None else "" if mcp_label and self._mcp is not None and self._mcp.running and self._mcp_follow: mcp_label += " ·follow" - self._status_bar.update_status( - self.config.kube_context, - self.current_scope, - label, - breadcrumb=self._drill.breadcrumb(), - mcp_label=mcp_label, - filter_label=self._resource_filter.describe(), - progress_label=" · ".join(label for label in self._progress_labels.values() if label), - proposals_label=self._proposals_label(), - protected=self._protected_context is not None, - ) + try: + self._status_bar.update_status( + self.config.kube_context, + self.current_scope, + label, + breadcrumb=self._drill.breadcrumb(), + mcp_label=mcp_label, + filter_label=self._resource_filter.describe(), + progress_label=" · ".join( + label for label in self._progress_labels.values() if label + ), + proposals_label=self._proposals_label(), + protected=self._protected_context is not None, + ) + except NoMatches: + return # StatusBar unmounted during teardown def _proposals_label(self) -> str: """Status-bar text for pending external write proposals (issue #110): @@ -8998,7 +9004,9 @@ async def agent_request_write( return ( f"denied: the user declined the {action} request for {self._gvr_label(meta)}/{name}" ) - outcome = await self._run_write_shielded(action, meta, ns, name, op(uid), detail=detail) + outcome = await self._run_write_shielded( + action, meta, ns, name, lambda: op(uid), detail=detail + ) if outcome != "done": return f"ERROR: {action} {self._gvr_label(meta)}/{name} {outcome}" self._mark_agent_action(f"{action} → {self._gvr_label(meta)}/{name}") @@ -9010,7 +9018,7 @@ async def _run_write_shielded( meta: ResourceMeta, ns: str | None, name: str, - op: Awaitable[None], + op_factory: Callable[[], Awaitable[None]], *, detail: str, ) -> str: @@ -9019,7 +9027,9 @@ async def _run_write_shielded( mutation is worse than finishing what the user explicitly approved. Every wait stays shielded — repeated cancellations are absorbed until the write reaches a terminal state, then cancellation is re-raised.""" - write = asyncio.ensure_future(self._run_write(action, meta, ns, name, op, detail=detail)) + write = asyncio.ensure_future( + self._run_write(action, meta, ns, name, op_factory, detail=detail) + ) interrupted = False while not write.done(): try: @@ -9587,7 +9597,12 @@ async def _execute_proposal( detail = self._proposal_provenance(proposal) write = asyncio.ensure_future( self._run_write( - proposal.action, meta, ns, proposal.name, op(proposal.uid), detail=detail + proposal.action, + meta, + ns, + proposal.name, + lambda: op(proposal.uid), + detail=detail, ) ) try: diff --git a/src/korvid/ui/debug.py b/src/korvid/ui/debug.py index bde3f742..adcddf1b 100644 --- a/src/korvid/ui/debug.py +++ b/src/korvid/ui/debug.py @@ -106,7 +106,7 @@ async def run( with self._suspend(): exit_code, pull_failure = self.run_process( argv, - f"korvid debug → {target} (exit to return)", + f"korvid debug -> {target} (exit to return)", namespace, name, image, diff --git a/src/korvid/ui/widgets/path_picker.py b/src/korvid/ui/widgets/path_picker.py index 9d51e55c..4f03fc65 100644 --- a/src/korvid/ui/widgets/path_picker.py +++ b/src/korvid/ui/widgets/path_picker.py @@ -10,6 +10,7 @@ from __future__ import annotations +import os import posixpath import unicodedata from collections.abc import Awaitable, Callable @@ -121,7 +122,9 @@ def action_select_dir(self) -> None: path = self._start if node is None or node.data is None else Path(node.data.path) if not path.is_dir(): path = path.parent - self.dismiss(str(path).rstrip("/") + "/") + # rstrip(os.sep) avoids "//"; add trailing sep so the caller knows + # this is a directory (on root "/" or "C:\", rstrip + sep round-trips). + self.dismiss(str(path).rstrip(os.sep) + os.sep) def action_cancel(self) -> None: self.dismiss(None) diff --git a/src/korvid/ui/widgets/transfer_screen.py b/src/korvid/ui/widgets/transfer_screen.py index 0bafe1f8..9f02338e 100644 --- a/src/korvid/ui/widgets/transfer_screen.py +++ b/src/korvid/ui/widgets/transfer_screen.py @@ -162,7 +162,8 @@ def _apply_local_pick(self, result: str | None) -> None: # Basename taken verbatim — it may end in whitespace. remote = self.query_one("#transfer-remote", Input).value base = posixpath.basename(remote.rstrip("/")) if remote.strip() else "" - result += base + if base: + result = str(Path(result) / base) field.value = result field.focus() diff --git a/tests/core/test_audit.py b/tests/core/test_audit.py index 677beca5..39189fc4 100644 --- a/tests/core/test_audit.py +++ b/tests/core/test_audit.py @@ -8,6 +8,7 @@ import korvid.core.audit as audit_module from korvid.core.audit import AuditLog +from tests.platforms import POSIX, posix_only def test_append_writes_jsonl_entry(tmp_path: Path) -> None: @@ -138,8 +139,10 @@ def test_constructor_does_not_touch_filesystem(tmp_path: Path) -> None: bad = tmp_path / "not-a-dir" bad.write_text("file, not a directory") log = AuditLog(bad / "audit.jsonl") # must not raise - # macOS raises FileExistsError, Linux NotADirectoryError for the mkdir - with pytest.raises((FileExistsError, NotADirectoryError)): + # macOS raises FileExistsError, Linux NotADirectoryError, Windows + # FileNotFoundError — all are OSError subtypes. The constructor is lazy; + # only append() touches the filesystem and propagates the failure. + with pytest.raises(OSError, match=r"(exist|directory|not found|denied)"): log.append(action="delete", kind="pods", namespace="default", name="w") @@ -220,13 +223,26 @@ def recording_fsync(fd: int) -> None: monkeypatch.setattr(os, "fsync", recording_fsync) log = AuditLog(tmp_path / "audit.jsonl") log.append(action="delete", kind="pods", namespace="default", name="web-1") - # at least the log file and its parent directory were synced - assert len(synced) >= 2 + # POSIX: file fsync + parent directory fsync (at least 2 calls). + # Windows: file fsync only — NTFS metadata ops are journal-durable, so + # directory fsync is skipped (os.open(dir, O_RDONLY) is unsupported). + if POSIX: + assert len(synced) >= 2 + else: + assert len(synced) >= 1 def test_append_fails_closed_when_fsync_fails( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: + """File-level fsync failure propagates on every platform (the cross- + platform counterpart to the POSIX-only directory-sync test below). + + On Windows, where directory fsync is inapplicable (NTFS metadata is + journaled), this is the sole durability-failure assertion: a buffered- + only record does not satisfy the fail-closed invariant. + """ + def failing_fsync(fd: int) -> None: raise OSError("disk gone") @@ -236,6 +252,7 @@ def failing_fsync(fd: int) -> None: log.append(action="delete", kind="pods", namespace="default", name="web-1") +@posix_only("directory fsync requires POSIX os.open(dir, O_RDONLY)") def test_append_fails_closed_when_dir_sync_fails( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 5ab0817f..29a927bd 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -6,6 +6,7 @@ import yaml from korvid.core.config import KorvidConfig, load_config, save_agent_config +from tests.platforms import POSIX def _load_agent_options_config(tmp_path: Path, options: object) -> KorvidConfig: @@ -567,22 +568,53 @@ def failing_fsync(fd: int) -> None: def test_save_agent_config_preserves_restrictive_file_mode(tmp_path: Path) -> None: """Atomic replacement must not widen an existing 0600 config to the - umask-derived default, exposing preserved values.""" - import os + umask-derived default, exposing preserved values. + + On POSIX we verify the effective stat mode. On Windows/NTFS, Python's + POSIX-mode emulation does not enforce real file permissions; the code + calls os.chmod(tmp, mode) before os.replace — we verify via spy that + the code *requests* the restrictive mode. + """ import stat + import korvid.core.config as cfg_mod + p = tmp_path / "c.yaml" p.write_text("agent:\n provider: ollama\n model: llama3\n") os.chmod(p, 0o600) - save_agent_config( - p, - provider="ollama", - auth_method="none", - base_url=None, - model="llama3", - api_key_env=None, - ) - assert stat.S_IMODE(p.stat().st_mode) == 0o600 + + if POSIX: + save_agent_config( + p, + provider="ollama", + auth_method="none", + base_url=None, + model="llama3", + api_key_env=None, + ) + assert stat.S_IMODE(p.stat().st_mode) == 0o600 + else: + # Windows: stat mode doesn't reflect POSIX bits; spy on os.chmod + # to prove the code requests 0o600. + chmod_calls: list[tuple[object, int]] = [] + real_chmod = os.chmod + + def spy_chmod(path: Path, mode: int) -> None: + chmod_calls.append((path, mode)) + real_chmod(path, mode) + + from unittest.mock import patch + + with patch.object(cfg_mod, "os_chmod", spy_chmod): + save_agent_config( + p, + provider="ollama", + auth_method="none", + base_url=None, + model="llama3", + api_key_env=None, + ) + assert any(mode == 0o600 for _, mode in chmod_calls) def test_save_agent_config_preserves_auth_extension_keys(tmp_path: Path) -> None: @@ -622,7 +654,7 @@ def spy_fsync(fd: int) -> None: # Windows' fsync (_commit) requires a writable handle: syncing an # O_RDONLY fd raises there, so the implementation must sync the fd # it wrote through. fcntl itself is POSIX-only, so guard the check. - if os.name != "nt": + if POSIX: import fcntl assert fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_ACCMODE != os.O_RDONLY diff --git a/tests/core/test_logexport.py b/tests/core/test_logexport.py index ec02f826..7630883c 100644 --- a/tests/core/test_logexport.py +++ b/tests/core/test_logexport.py @@ -9,6 +9,7 @@ from korvid.core.logexport import default_log_export_dir, export_log_lines from korvid.k8s.logs import LogLine +from tests.platforms import POSIX def _line( @@ -57,10 +58,21 @@ def test_export_includes_timestamp_when_present(tmp_path: Path) -> None: def test_export_file_is_private(tmp_path: Path) -> None: - """Exported cluster logs must not be readable by group/other users.""" + """Exported cluster logs must not be readable by group/other users. + + On POSIX we can verify the effective mode bits. On Windows/NTFS, Python's + POSIX-mode emulation (stat.st_mode) does not reflect true ACLs; the code + passes 0o600 to os.open(O_CREAT|O_EXCL) which is the strongest portable + guarantee available. We verify the file was created and exists. + """ path = export_log_lines([_line("secretish")], tmp_path) - assert path.stat().st_mode & 0o077 == 0 + if POSIX: + assert path.stat().st_mode & 0o077 == 0 + else: + # Windows: file exists and was created atomically (O_EXCL); POSIX + # mode bits are not enforced by NTFS — assert creation succeeded. + assert path.is_file() def test_export_writes_utf8(tmp_path: Path) -> None: diff --git a/tests/core/test_transfer.py b/tests/core/test_transfer.py index 283c121e..afc2676b 100644 --- a/tests/core/test_transfer.py +++ b/tests/core/test_transfer.py @@ -19,6 +19,7 @@ upload_command, validate_spec, ) +from tests.platforms import posix_only, symlink_or_skip class TestTransferSpec: @@ -35,8 +36,8 @@ def test_fields(self) -> None: class TestValidateSpec: - def test_valid_download(self) -> None: - spec = TransferSpec("download", "/var/log/app.log", "/tmp/app.log") + def test_valid_download(self, tmp_path: Path) -> None: + spec = TransferSpec("download", "/var/log/app.log", str(tmp_path / "app.log")) assert validate_spec(spec) is None def test_valid_upload(self, tmp_path: Path) -> None: @@ -63,10 +64,10 @@ def test_remote_path_trailing_slash(self) -> None: assert error is not None assert "file" in error - def test_remote_path_with_trailing_space_is_validated_verbatim(self) -> None: + def test_remote_path_with_trailing_space_is_validated_verbatim(self, tmp_path: Path) -> None: # "/srv/ " names the file " " in /srv — valid; stripping before # validation turned it into the directory "/srv/" and rejected it. - spec = TransferSpec("download", "/srv/ ", "/tmp/x") + spec = TransferSpec("download", "/srv/ ", str(tmp_path / "x")) assert validate_spec(spec) is None def test_remote_path_with_leading_space_rejected_as_relative(self) -> None: @@ -126,9 +127,11 @@ def test_local_path_tilde_expanded( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) spec = TransferSpec("download", "/tmp/x", "~/x.log") assert validate_spec(spec) is None + @posix_only("requires POSIX ~user account expansion behavior") def test_unknown_user_tilde_is_a_validation_error(self) -> None: # Path.expanduser raises RuntimeError for an unknown user; that must # surface as a validation message, not escape the submit handler. @@ -163,6 +166,7 @@ def test_download_command_option_looking_basename_is_neutralised(self) -> None: class TestDefaultLocalPath: def test_uses_downloads_dir(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) (tmp_path / "Downloads").mkdir() assert default_local_path("/var/log/app.log") == str(tmp_path / "Downloads" / "app.log") @@ -172,6 +176,7 @@ def test_falls_back_to_home_without_downloads_dir( # Not every home has ~/Downloads; the default must still pass # validate_spec (parent exists), so fall back to the home directory. monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) assert default_local_path("/var/log/app.log") == str(tmp_path / "app.log") @@ -183,7 +188,7 @@ def test_pack_dereferences_symlink_source(self, tmp_path: Path) -> None: target = tmp_path / "real.txt" target.write_bytes(b"real bytes") link = tmp_path / "link.txt" - link.symlink_to(target) + symlink_or_skip(link, target) archive = tmp_path / "out.tar" size = pack_file(link, "f.txt", archive) assert size == len(b"real bytes") @@ -345,6 +350,7 @@ def test_skips_unwritable_downloads_dir( # The default must always survive validate_spec's new writability # check, so a read-only ~/Downloads falls back to the home directory. monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) downloads = tmp_path / "Downloads" downloads.mkdir(mode=0o500) assert default_local_path("/var/log/app.log") == str(tmp_path / "app.log") @@ -353,6 +359,7 @@ def test_skips_unsearchable_downloads_dir( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) downloads = tmp_path / "Downloads" downloads.mkdir(mode=0o600) try: diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py index 9dfcdb76..63acfffd 100644 --- a/tests/mcp/test_server.py +++ b/tests/mcp/test_server.py @@ -20,6 +20,7 @@ default_endpoint_path, ) from korvid.tools.executor import PROPOSAL_TOOLS, READ_TOOLS, UI_TOOLS, ToolExecutor +from tests.platforms import POSIX from tests.tools.test_executor import FakeBridge @@ -811,6 +812,12 @@ async def test_streamable_http_proposal_roundtrip(tmp_path: Path) -> None: async def test_endpoint_file_publishes_capability_with_owner_only_mode(tmp_path: Path) -> None: + """The endpoint file carries a capability token and must be created 0600. + + On POSIX we verify the effective stat mode bits. On Windows/NTFS, Python's + POSIX-mode emulation does not enforce real ACLs; we verify the file was + created and the server published the expected capability value. + """ endpoint_file = tmp_path / "mcp-endpoint.json" server = make_proposal_server(port=0, endpoint_path=endpoint_file) task = asyncio.create_task(server.run()) @@ -818,7 +825,12 @@ async def test_endpoint_file_publishes_capability_with_owner_only_mode(tmp_path: await asyncio.wait_for(server.wait_started(), timeout=10) entry = json.loads(endpoint_file.read_text())["servers"][str(os.getpid())] assert entry["capability"] == "cap-tok" - assert (endpoint_file.stat().st_mode & 0o777) == 0o600 + if POSIX: + assert (endpoint_file.stat().st_mode & 0o777) == 0o600 + else: + # Windows: POSIX mode bits are not meaningful on NTFS; assert the + # file was created atomically (exists) with the capability token. + assert endpoint_file.is_file() finally: server.request_shutdown() await asyncio.wait_for(task, timeout=10) @@ -842,12 +854,29 @@ def test_endpoint_file_is_created_owner_only_not_merely_chmodded( ) -> None: """The capability-bearing registry file must never be observable with group/other bits: the mode has to come from atomic 0600 creation, not - from a chmod racing the umask-default file.""" - monkeypatch.setattr(Path, "chmod", lambda self, mode: None) - target = tmp_path / "mcp-endpoint.json" - _replace_atomically(target, {"servers": {}}) - assert stat.S_IMODE(target.stat().st_mode) == 0o600 - assert json.loads(target.read_text()) == {"servers": {}} + from a chmod racing the umask-default file. + + On POSIX we disable chmod and prove os.open(O_CREAT|O_EXCL, 0o600) + alone achieves the correct mode. On Windows/NTFS, Python's stat does + not reflect POSIX permission bits (NTFS uses ACLs, not mode bits); + we can only verify the atomic-creation path produces valid content. + The code passes 0o600 to os.open which is the strongest portable + guarantee — no ACL confidentiality claim is made here. + """ + if POSIX: + monkeypatch.setattr(Path, "chmod", lambda self, mode: None) + target = tmp_path / "mcp-endpoint.json" + _replace_atomically(target, {"servers": {}}) + assert stat.S_IMODE(target.stat().st_mode) == 0o600 + assert json.loads(target.read_text()) == {"servers": {}} + else: + # Windows: POSIX mode bits are not enforced by NTFS. Verify the + # atomic creation path works and produces valid content. The code + # passes 0o600 to os.open which is the strongest portable guarantee. + target = tmp_path / "mcp-endpoint.json" + _replace_atomically(target, {"servers": {}}) + assert target.is_file() + assert json.loads(target.read_text()) == {"servers": {}} async def test_client_info_is_sanitized_before_crossing_the_boundary() -> None: diff --git a/tests/platforms.py b/tests/platforms.py new file mode 100644 index 00000000..7e47d9c5 --- /dev/null +++ b/tests/platforms.py @@ -0,0 +1,47 @@ +"""Shared test helpers for platform-specific expectations.""" + +import errno +import os +import re +from pathlib import Path + +import pytest + +WINDOWS = os.name == "nt" +POSIX = os.name == "posix" + + +def posix_only(reason: str) -> pytest.MarkDecorator: + """Skip a test when POSIX-specific behavior is unavailable.""" + return pytest.mark.skipif(not POSIX, reason=reason) + + +def read_text_utf8(path: Path) -> str: + """Read a repository text file with an explicit UTF-8 encoding.""" + return path.read_text(encoding="utf-8") + + +def symlink_or_skip(link: Path, target: Path) -> None: + """Create a symlink, or skip only when Windows symlink privilege is absent.""" + try: + link.symlink_to(target) + except OSError as exc: + missing_privilege = WINDOWS and ( + getattr(exc, "winerror", None) == 1314 or exc.errno == errno.EPERM + ) + if missing_privilege: + pytest.skip( + "Windows symlink creation requires Developer Mode or an elevated" + " administrator shell" + ) + raise + + +def assert_pinned_action_ref(workflow_text: str, action: str) -> str: + """Require an action use-site to be pinned to a lowercase full commit SHA.""" + pattern = re.compile( + rf"(?m)^\s*-\s+uses:\s+{re.escape(action)}@(?P[0-9a-f]{{40}})\s*(?:#.*)?$" + ) + match = pattern.search(workflow_text) + assert match is not None, f"expected {action}@<40 lowercase hex characters>" + return match.group("ref") diff --git a/tests/providers/test_token_store.py b/tests/providers/test_token_store.py index d415fbbb..b9669499 100644 --- a/tests/providers/test_token_store.py +++ b/tests/providers/test_token_store.py @@ -7,6 +7,7 @@ import pytest from korvid.providers.token_store import TokenStore +from tests.platforms import POSIX def _no_keyring(monkeypatch: pytest.MonkeyPatch) -> None: @@ -23,10 +24,25 @@ def test_file_fallback_roundtrip(tmp_path: Path, monkeypatch: pytest.MonkeyPatch def test_file_mode_0600(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Credential file must be created with restrictive permissions. + + On POSIX we verify the effective stat mode. On Windows/NTFS, Python's + POSIX-mode emulation does not enforce real ACLs; mkstemp creates the + temp file with 0600 (the strongest portable guarantee) but stat does + not reflect it. We verify the file was created via atomic replace. + """ _no_keyring(monkeypatch) p = tmp_path / "creds.json" TokenStore(fallback_path=p).save("k", "v") - assert stat.S_IMODE(p.stat().st_mode) == 0o600 + if POSIX: + assert stat.S_IMODE(p.stat().st_mode) == 0o600 + else: + # Windows: POSIX mode bits are not meaningful on NTFS. Verify the + # file was created and contains valid JSON (atomic write succeeded). + assert p.is_file() + import json + + assert json.loads(p.read_text()) == {"k": "v"} def test_keyring_preferred(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -169,7 +185,7 @@ def spy_fsync(fd: int) -> None: # Windows' fsync (_commit) requires a writable handle: syncing an # O_RDONLY fd raises there, so the implementation must sync the fd # it wrote through. fcntl itself is POSIX-only, so guard the check. - if os.name != "nt": + if POSIX: import fcntl assert fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_ACCMODE != os.O_RDONLY diff --git a/tests/test_main_recovery.py b/tests/test_main_recovery.py index 8a4e41b1..3f5897b0 100644 --- a/tests/test_main_recovery.py +++ b/tests/test_main_recovery.py @@ -77,6 +77,20 @@ def test_crash_cap_stops_a_deterministic_crash_loop( assert "not restarting" in capsys.readouterr().err +def test_crash_cap_message_is_ascii_encodable( + capsys: pytest.CaptureFixture[str], +) -> None: + """All stderr output from the crash-cap path must be encodable in ASCII + (and therefore cp1252/any single-byte codepage) — em dashes or other + non-ASCII would raise UnicodeEncodeError on Windows terminals.""" + runner = FlakyRunner(failures=100) + with pytest.raises(RuntimeError, match="boom"): + _run_with_recovery(runner, allow_restart=True, prompt=lambda: "y", clock=lambda: 0.0) + captured = capsys.readouterr().err + # This encodes the entire cap diagnostic; would raise on non-ASCII. + captured.encode("ascii") + + def test_old_crashes_age_out_of_the_window() -> None: runner = FlakyRunner(failures=RESTART_CAP + 1) times = iter( diff --git a/tests/test_platforms.py b/tests/test_platforms.py new file mode 100644 index 00000000..e9842441 --- /dev/null +++ b/tests/test_platforms.py @@ -0,0 +1,266 @@ +"""Tests for shared platform helpers and CI workflow invariants.""" + +import ast +import errno +import os +from pathlib import Path +from typing import Any + +import pytest +import yaml + +import tests.platforms as platform_helpers +from tests.platforms import ( + POSIX, + WINDOWS, + assert_pinned_action_ref, + posix_only, + read_text_utf8, +) + + +def test_windows_and_posix_flags_follow_os_name() -> None: + assert WINDOWS is (os.name == "nt") + assert POSIX is (os.name == "posix") + + +def test_posix_only_returns_a_skipif_mark_with_the_given_reason() -> None: + mark = posix_only("POSIX permissions required") + + assert isinstance(mark, pytest.MarkDecorator) + assert mark.mark.name == "skipif" + assert mark.mark.args == (not POSIX,) + assert mark.mark.kwargs == {"reason": "POSIX permissions required"} + + +def _find_test_function(module: ast.Module, name: str) -> ast.FunctionDef | ast.AsyncFunctionDef: + for node in ast.walk(module): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name: + return node + raise AssertionError(f"expected to find test function {name}") + + +def _decorates_with_posix_only(node: ast.FunctionDef | ast.AsyncFunctionDef, reason: str) -> bool: + for decorator in node.decorator_list: + if not isinstance(decorator, ast.Call): + continue + if not isinstance(decorator.func, ast.Name) or decorator.func.id != "posix_only": + continue + if len(decorator.args) != 1 or decorator.keywords: + continue + if isinstance(decorator.args[0], ast.Constant) and decorator.args[0].value == reason: + return True + return False + + +@pytest.mark.parametrize( + ("relative_path", "test_name"), + [ + ("tests/core/test_transfer.py", "test_unknown_user_tilde_is_a_validation_error"), + ("tests/ui/test_transfer_picker.py", "test_unexpandable_tilde_falls_back_to_home"), + ], +) +def test_posix_user_tilde_cases_use_the_shared_marker(relative_path: str, test_name: str) -> None: + module = ast.parse(read_text_utf8(Path(__file__).parents[1] / relative_path)) + + assert _decorates_with_posix_only( + _find_test_function(module, test_name), + "requires POSIX ~user account expansion behavior", + ) + + +def _ci_workflow() -> str: + return read_text_utf8(Path(__file__).parents[1] / ".github" / "workflows" / "ci.yml") + + +def _workflow_job(workflow: str, name: str) -> dict[str, Any]: + parsed = yaml.safe_load(workflow) + assert isinstance(parsed, dict), "expected workflow YAML to parse to a mapping" + jobs = parsed.get("jobs") + assert isinstance(jobs, dict), "expected workflow to define a jobs mapping" + job = jobs.get(name) + assert isinstance(job, dict), f"expected workflow to define jobs[{name!r}]" + return job + + +def _readme() -> str: + return read_text_utf8(Path(__file__).parents[1] / "README.md") + + +def _windows_doc() -> str: + return read_text_utf8(Path(__file__).parents[1] / "docs" / "windows.md") + + +def test_assert_pinned_action_ref_accepts_full_lowercase_commit_shas() -> None: + workflow = "- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1" + + assert ( + assert_pinned_action_ref(workflow, "actions/checkout") + == "34e114876b0b11c390a56381ad16ebd13914f8d5" + ) + + +@pytest.mark.parametrize("bad_ref", ["v4", "v4.3.1", "34E114876B0B11C390A56381AD16EBD13914F8D5"]) +def test_assert_pinned_action_ref_rejects_tags_and_non_lowercase_refs(bad_ref: str) -> None: + workflow = f"- uses: astral-sh/setup-uv@{bad_ref}" + + with pytest.raises( + AssertionError, match="expected astral-sh/setup-uv@<40 lowercase hex characters>" + ): + assert_pinned_action_ref(workflow, "astral-sh/setup-uv") + + +def test_read_text_utf8_uses_utf8_encoding(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + path = tmp_path / "ci.yml" + captured: dict[str, object] = {} + + def fake_read_text(self: Path, encoding: str | None = None, errors: str | None = None) -> str: + captured["path"] = self + captured["encoding"] = encoding + captured["errors"] = errors + return "name: CI" + + monkeypatch.setattr(Path, "read_text", fake_read_text) + + assert read_text_utf8(path) == "name: CI" + assert captured == {"path": path, "encoding": "utf-8", "errors": None} + + +def test_ci_workflow_defines_the_required_windows_test_job() -> None: + windows_job = _workflow_job(_ci_workflow(), "windows-test") + segment = yaml.safe_dump(windows_job, sort_keys=False) + steps = windows_job["steps"] + assert isinstance(steps, list) + runs = [step["run"] for step in steps if isinstance(step, dict) and "run" in step] + setup_uv = next( + step + for step in steps + if isinstance(step, dict) + and step.get("uses") == "astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9" + ) + assert windows_job["runs-on"] == "windows-latest" + assert_pinned_action_ref(segment, "actions/checkout") + assert_pinned_action_ref(segment, "astral-sh/setup-uv") + assert isinstance(setup_uv.get("with"), dict) + assert setup_uv["with"]["python-version"] == "3.12" + assert "uv sync --locked --dev --all-extras" in runs + assert "uv run pytest -q" in runs + + +def test_workflow_job_lookup_is_order_independent() -> None: + workflow = """ +name: CI +jobs: + pre-commit: + runs-on: ubuntu-latest + windows-test: + runs-on: windows-latest + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + - uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0 + - run: uv sync --locked --dev --all-extras + - run: uv run pytest -q + test: + runs-on: ubuntu-latest +""" + + job = _workflow_job(workflow, "windows-test") + + assert job["runs-on"] == "windows-latest" + assert "uv run pytest -q" in yaml.safe_dump(job, sort_keys=False) + + +def test_symlink_or_skip_calls_path_symlink( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + target = tmp_path / "target.txt" + link = tmp_path / "link.txt" + called: dict[str, Path] = {} + + def fake(self: Path, dest: Path, target_is_directory: bool = False) -> None: + assert target_is_directory is False + called["link"] = self + called["target"] = dest + + monkeypatch.setattr(Path, "symlink_to", fake) + + platform_helpers.symlink_or_skip(link, target) + + assert called == {"link": link, "target": target} + + +@pytest.mark.parametrize("winerror", [1314, None]) +def test_symlink_or_skip_skips_windows_privilege_errors( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, winerror: int | None +) -> None: + link = tmp_path / "link.txt" + target = tmp_path / "target.txt" + + class FakePrivilegeError(PermissionError): + def __init__(self, value: int | None) -> None: + super().__init__(errno.EPERM, "privilege unavailable") + self.winerror = value + + def fail(self: Path, dest: Path, target_is_directory: bool = False) -> None: + raise FakePrivilegeError(winerror) + + monkeypatch.setattr(platform_helpers, "WINDOWS", True) + monkeypatch.setattr(Path, "symlink_to", fail) + + with pytest.raises(pytest.skip.Exception, match=r"Developer Mode|administrator"): + platform_helpers.symlink_or_skip(link, target) + + +def test_symlink_or_skip_propagates_unrelated_errors( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + link = tmp_path / "link.txt" + target = tmp_path / "target.txt" + + def fail(self: Path, dest: Path, target_is_directory: bool = False) -> None: + raise FileNotFoundError(errno.ENOENT, "missing target") + + monkeypatch.setattr(platform_helpers, "WINDOWS", True) + monkeypatch.setattr(Path, "symlink_to", fail) + + with pytest.raises(FileNotFoundError, match="missing target"): + platform_helpers.symlink_or_skip(link, target) + + +def test_readme_links_windows_contributor_notes() -> None: + readme = _readme() + + assert "[Windows contributor notes](docs/windows.md)" in readme + + +def test_windows_doc_records_verified_support_contract() -> None: + doc = _windows_doc() + + for snippet in ( + "uv sync --dev --all-extras", + "uv run pytest -q", + "windows-test", + "30936032385", + "3376 passed / 37 skipped / 0 failures", + "21 opt-in contract-suite skips", + "16 capability skips", + "3 newly classified capability skips", + "13 pre-existing platform skips", + "`~user`", + "Developer Mode", + "symlink", + "capability skip", + "hosted runner had privilege", + "count remained 37", + "SuspendNotSupported", + "legacy_windows=False", + 'newline=""', + "`->`", + "`--`", + "0o600", + "ACL", + "do not claim ACL confidentiality", + "op_factory", + "cancelled writes", + ): + assert snippet in doc diff --git a/tests/ui/test_busy_bird.py b/tests/ui/test_busy_bird.py index b7f9cb1d..4d2c009d 100644 --- a/tests/ui/test_busy_bird.py +++ b/tests/ui/test_busy_bird.py @@ -169,9 +169,7 @@ async def slow_op() -> None: await gate.wait() async with app.run_test() as pilot: - task = asyncio.create_task( - app._run_write("delete", PODS_META, "default", "web-1", slow_op()) - ) + task = asyncio.create_task(app._run_write("delete", PODS_META, "default", "web-1", slow_op)) await until( pilot, lambda: any("delete pods/web-1" in v for v in app._progress_labels.values()), diff --git a/tests/ui/test_node_ops.py b/tests/ui/test_node_ops.py index f66ea047..2b5ebc2f 100644 --- a/tests/ui/test_node_ops.py +++ b/tests/ui/test_node_ops.py @@ -183,6 +183,14 @@ async def test_c_cordons_node_after_approval(tmp_path: Path) -> None: await pilot.press("y") await until(pilot, lambda: rec.calls, label="cordon executed") assert rec.calls == [("cordon", "worker-1", True, "node-uid-1")] + + def _success_audited() -> bool: + if not audit_path.exists(): + return False + lines = audit_path.read_text().splitlines() + return any('"success"' in ln for ln in lines) + + await until(pilot, _success_audited, label="success audit record") entries = [json.loads(ln) for ln in audit_path.read_text().splitlines()] assert entries[0]["action"] == "cordon" assert entries[0]["outcome"] == "intent" diff --git a/tests/ui/test_proposals_ui.py b/tests/ui/test_proposals_ui.py index bc9a07ee..ab1a49d8 100644 --- a/tests/ui/test_proposals_ui.py +++ b/tests/ui/test_proposals_ui.py @@ -232,11 +232,14 @@ async def test_review_approve_executes_with_the_bound_uid(tmp_path: Path) -> Non await until(pilot, lambda: isinstance(app.screen, ConfirmScreen)) await pilot.press("y") await until(pilot, lambda: rec.calls != []) + + def _executed() -> bool: + found = store.get(pid) + return found is not None and found[1] == "executed" + + await until(pilot, _executed, label="proposal executed") assert rec.calls == [("delete", "deployments", "default", "web")] assert rec.uids == ["uid-1"] - found = store.get(pid) - assert found is not None - assert found[1] == "executed" entries = [json.loads(line) for line in audit_path.read_text().splitlines()] details = " ".join(e.get("detail", "") for e in entries) assert "external_mcp" in details @@ -1223,3 +1226,58 @@ def outcome_context() -> object: await until(pilot, lambda: outcome_context() != "missing") assert outcome_context() == "ctx-a" + + +async def test_proposals_changed_during_teardown_does_not_raise(tmp_path: Path) -> None: + """ExternalProposalsChanged arriving after StatusBar is unmounted must + not raise NoMatches — _refresh_status handles it internally. + + Removes the StatusBar while the app message pump is active, then calls + the handler directly; a spy on _refresh_status proves the code path + executed (not silently discarded) and that _refresh_status caught + the NoMatches from the missing StatusBar. + """ + from unittest.mock import patch + + from korvid.ui.messages import ExternalProposalsChanged + from korvid.ui.widgets.status_bar import StatusBar + + store = ProposalStore() + app = make_app(Recorder(), tmp_path / "a.jsonl", store) + async with app.run_test(): + await _submit(app) + # Remove the StatusBar to reproduce teardown widget state while + # the app message pump is still active. + bar = app.query_one(StatusBar) + await bar.remove() + # Spy on _refresh_status to prove the handler entered it (the + # suppress catches the NoMatches raised by query_one inside). + refresh_calls: list[bool] = [] + real_refresh = app._refresh_status + + def spy_refresh() -> None: + refresh_calls.append(True) + real_refresh() + + with patch.object(app, "_refresh_status", spy_refresh): + app.on_external_proposals_changed(ExternalProposalsChanged()) + assert refresh_calls, "_refresh_status must be called (NoMatches suppressed, not skipped)" + + +async def test_refresh_status_tolerates_missing_status_bar(tmp_path: Path) -> None: + """_refresh_status is called from 20+ sites (MCP switch worker, + navigation, proposals handler, etc.); all must survive when StatusBar + is unmounted during teardown. Removing the widget while the app is + live and calling _refresh_status directly exercises the guard.""" + from korvid.ui.widgets.status_bar import StatusBar + + app = make_app(Recorder(), tmp_path / "a.jsonl", None) + async with app.run_test(): + # Confirm normal refresh works with the bar present. + app._refresh_status() # must not raise + # Remove StatusBar to simulate teardown. + bar = app.query_one(StatusBar) + await bar.remove() + # Must not raise NoMatches — any caller (MCP _switch, proposals + # handler, navigation, etc.) hitting this after teardown is safe. + app._refresh_status() diff --git a/tests/ui/test_transfer.py b/tests/ui/test_transfer.py index 4d814745..6f08f571 100644 --- a/tests/ui/test_transfer.py +++ b/tests/ui/test_transfer.py @@ -222,6 +222,11 @@ async def test_download_writes_file_and_audits(tmp_path: Path) -> None: lambda: any("downloaded" in str(n.message).lower() for n in app._notifications), label="success toast", ) + await until( + pilot, + lambda: any(e.get("outcome") == "success" for e in audit_entries(audit_path)), + label="success audit", + ) assert dest.read_bytes() == payload assert opener.calls == [ { @@ -366,6 +371,13 @@ async def test_download_failure_notifies_and_audits_error(tmp_path: Path) -> Non lambda: any("not found" in str(n.message) for n in app._notifications), label="error toast", ) + await until( + pilot, + lambda: any( + e.get("outcome", "").startswith("error") for e in audit_entries(audit_path) + ), + label="error audit", + ) entries = audit_entries(audit_path) assert [e["outcome"] for e in entries[:1]] == ["intent"] assert entries[-1]["outcome"].startswith("error") @@ -397,6 +409,11 @@ async def test_upload_requires_approval_then_transfers(tmp_path: Path) -> None: lambda: any("uploaded" in str(n.message).lower() for n in app._notifications), label="success toast", ) + await until( + pilot, + lambda: any(e.get("outcome") == "success" for e in audit_entries(audit_path)), + label="success audit", + ) assert opener.calls[0]["command"] == ["tar", "xf", "-", "-C", "/opt"] assert opener.calls[0]["stdin"] is True assert opener.ws is not None @@ -562,6 +579,7 @@ async def test_download_default_local_path_from_remote_basename( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) (tmp_path / "Downloads").mkdir() payload = b"data" opener = FakeExecOpener([b"\x01" + tar_bytes("app.log", payload), b"\x03" + SUCCESS]) @@ -611,6 +629,11 @@ async def test_progress_screen_escape_cancels_transfer(tmp_path: Path) -> None: lambda: not isinstance(app.screen, TransferProgressScreen), label="progress closed", ) + await until( + pilot, + lambda: any(e.get("outcome") == "cancelled" for e in audit_entries(audit_path)), + label="cancelled audit", + ) entries = audit_entries(audit_path) assert [e["outcome"] for e in entries] == ["intent", "cancelled"] # Partial transfers stay auditable: the outcome records what was moved. diff --git a/tests/ui/test_transfer_picker.py b/tests/ui/test_transfer_picker.py index deb1f5c9..b347d864 100644 --- a/tests/ui/test_transfer_picker.py +++ b/tests/ui/test_transfer_picker.py @@ -20,6 +20,7 @@ from korvid.ui.widgets.path_picker import LocalPathPickerScreen, RemotePathPickerScreen from korvid.ui.widgets.resource_table import ResourceTable from korvid.ui.widgets.transfer_screen import TransferScreen +from tests.platforms import posix_only from tests.ui.test_app import make_app from tests.ui.test_transfer import SUCCESS, FakeExecOpener, _dialog, _pod from tests.ui.waits import until @@ -266,6 +267,7 @@ async def test_ctrl_o_outside_path_fields_does_nothing(self) -> None: await pilot.pause() assert app.screen is dialog + @posix_only("requires POSIX ~user account expansion behavior") async def test_unexpandable_tilde_falls_back_to_home(self) -> None: # Path.expanduser raises RuntimeError for "~no_such_user/f": browsing # must fall back to home, not escape the dialog handler. @@ -521,7 +523,7 @@ async def test_local_dir_pick_appends_remote_basename_verbatim(self, tmp_path: P app.screen.query_one(DirectoryTree).focus() await pilot.press("s") await until(pilot, lambda: app.screen is dialog, label="picker closed") - assert local.value == str(tmp_path) + "/report " + assert local.value == str(tmp_path / "report ") async def test_remote_dir_pick_appends_local_basename_verbatim(self, tmp_path: Path) -> None: opener = FakeExecOpener(_listing("config/")) diff --git a/tests/ui/test_write_confirm_characterization.py b/tests/ui/test_write_confirm_characterization.py index cbdf8d1a..66a63366 100644 --- a/tests/ui/test_write_confirm_characterization.py +++ b/tests/ui/test_write_confirm_characterization.py @@ -10,10 +10,11 @@ Group (a) — standard `_run_write` launch (9): delete, rollout restart, edit, scale, resize, cordon/uncordon, installplan approve, helm install/upgrade, helm rollback. Shape: `if confirmed: -self.run_worker(self._run_write(...))`. The mutation coroutine is -constructed *inside* the confirmed branch, so a declined dialog never -creates it (no unawaited-coroutine leak — pinned here by the decline -tests under warnings-as-errors). +self.run_worker(self._run_write(...))`. The confirmed branch passes +the operation factory to `_run_write`, which constructs the mutation +coroutine only after the intent audit succeeds (no unawaited-coroutine +leak on decline or audit failure — pinned here by the decline tests +under warnings-as-errors). Group (b) — launch + UID recheck (1): operator install re-checks the catalog incarnation inside `_done` before launching, because create has diff --git a/tests/ui/test_write_ops.py b/tests/ui/test_write_ops.py index a1e15607..bafea2eb 100644 --- a/tests/ui/test_write_ops.py +++ b/tests/ui/test_write_ops.py @@ -6,6 +6,7 @@ """ import asyncio +import contextlib import copy import json import logging @@ -1384,3 +1385,100 @@ async def get_manifest(kind: str, ns: str | None, name: str) -> dict[str, Any]: } } assert await app._managed_note_from(pod, "default") is None + + +# -- _run_write factory-based design regression tests ------------------------- + + +async def test_blocked_audit_never_invokes_op_factory(tmp_path: Path) -> None: + """When the intent audit fails, the operation factory must never be called. + + The factory design guarantees no mutation coroutine is created before + intent is persisted — a blocked write cannot leak an unawaited coroutine. + """ + factory_calls: list[str] = [] + + async def spy_op() -> None: + factory_calls.append("invoked") + + def factory() -> Awaitable[None]: + factory_calls.append("created") + return spy_op() + + audit_path = tmp_path / "audit.jsonl" + audit_path.mkdir() # directory makes appends fail → intent blocked + app = make_app(Recorder(), audit_path) + async with app.run_test(): + result = await app._run_write("delete", _PODS_META, "default", "web-1", factory) + assert "blocked" in result + assert factory_calls == [], "factory must not be called when audit intent fails" + + +async def test_cancelled_before_factory_leaks_no_coroutine(tmp_path: Path) -> None: + """Cancellation while the audit intent is in-flight must never invoke the + factory — no unawaited coroutine, no mutation before intent. + + Uses a gated audit (threading.Event inside to_thread) to deterministically + cancel at the right instant. + """ + import threading + + factory_calls: list[str] = [] + + async def spy_op() -> None: + factory_calls.append("invoked") + + def factory() -> Awaitable[None]: + factory_calls.append("created") + return spy_op() + + audit_gate = threading.Event() + real_audit_path = tmp_path / "audit.jsonl" + entered = threading.Event() + + class GatedAudit(AuditLog): + def append( + self, + *, + action: str, + kind: str, + namespace: str | None, + name: str, + group: str = "", + version: str = "", + detail: str = "", + outcome: str = "success", + context: object = None, + ) -> None: + entered.set() + audit_gate.wait() + super().append( + action=action, + kind=kind, + namespace=namespace, + name=name, + group=group, + version=version, + detail=detail, + outcome=outcome, + ) + + audit = GatedAudit(real_audit_path, context="test") + rec = Recorder() + app = make_app(rec, real_audit_path) + app._audit = audit + + async with app.run_test() as pilot: + task = asyncio.create_task( + app._run_write("delete", _PODS_META, "default", "web-1", factory) + ) + try: + await until(pilot, entered.is_set, label="audit entered") + task.cancel() + finally: + # Always release so the executor thread cannot hang at exit. + audit_gate.set() + with contextlib.suppress(asyncio.CancelledError): + await task + assert factory_calls == [], "factory must not be called when cancelled during audit" + assert rec.calls == [], "no mutation must reach the recorder"