From dec75f64eb0fe92cdc54f985e9ced4bfd9ca9d9f Mon Sep 17 00:00:00 2001 From: Kartheek Penagamuri Date: Tue, 6 Oct 2026 14:48:41 -0700 Subject: [PATCH] Add manual macOS terminal regression suite Import the pinned 51-case shared terminal suite, retain the Windows ConPTY backend, and add a standard-library POSIX PTY adapter with portable shell and locking fixtures. Document isolated manual Mac setup without changing production code or CI. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- dotnet-package-skills/CONTRIBUTING.md | 3 + .../tests/terminal/.gitignore | 3 + .../tests/terminal/README.md | 129 ++ .../tests/terminal/emulator.cjs | 49 + .../tests/terminal/mutex_holder.cs.txt | 22 + .../tests/terminal/package-lock.json | 19 + .../tests/terminal/package.json | 7 + .../tests/terminal/requirements.txt | 3 + .../tests/terminal/terminal_support.py | 230 +++ .../tests/terminal/test_driver.py | 123 ++ .../tests/terminal/test_terminal_support.py | 378 ++++ .../tests/terminal/verify_picker.py | 1595 +++++++++++++++++ 12 files changed, 2561 insertions(+) create mode 100644 dotnet-package-skills/tests/terminal/.gitignore create mode 100644 dotnet-package-skills/tests/terminal/README.md create mode 100644 dotnet-package-skills/tests/terminal/emulator.cjs create mode 100644 dotnet-package-skills/tests/terminal/mutex_holder.cs.txt create mode 100644 dotnet-package-skills/tests/terminal/package-lock.json create mode 100644 dotnet-package-skills/tests/terminal/package.json create mode 100644 dotnet-package-skills/tests/terminal/requirements.txt create mode 100644 dotnet-package-skills/tests/terminal/terminal_support.py create mode 100644 dotnet-package-skills/tests/terminal/test_driver.py create mode 100644 dotnet-package-skills/tests/terminal/test_terminal_support.py create mode 100644 dotnet-package-skills/tests/terminal/verify_picker.py diff --git a/dotnet-package-skills/CONTRIBUTING.md b/dotnet-package-skills/CONTRIBUTING.md index 8ebd8ed..b75ca83 100644 --- a/dotnet-package-skills/CONTRIBUTING.md +++ b/dotnet-package-skills/CONTRIBUTING.md @@ -26,6 +26,9 @@ eng\common\dotnet.cmd artifacts\bin\DotnetPackageSkills\Release\net10.0\dotnet-p The build creates an unsigned tool package under `artifacts\packages\Release\Shipping`. It does not replace a globally installed tool. Only the official pipeline signs the package. +For manual tests on a personal Mac, see the [full terminal regression suite](tests/terminal/README.md). +These tests are not wired into CI. + ## Layout ``` diff --git a/dotnet-package-skills/tests/terminal/.gitignore b/dotnet-package-skills/tests/terminal/.gitignore new file mode 100644 index 0000000..77ac754 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/.gitignore @@ -0,0 +1,3 @@ +.venv/ +__pycache__/ +*.pyc diff --git a/dotnet-package-skills/tests/terminal/README.md b/dotnet-package-skills/tests/terminal/README.md new file mode 100644 index 0000000..ba237a2 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/README.md @@ -0,0 +1,129 @@ +# Manual terminal regressions + +This is the full 51-case terminal suite, not a smoke test and not a CI job. It drives the +built `dotnet-package-skills` apphost through a real PTY, with `pyte` and +`@xterm/headless` checking rendering, colors, alternate buffers and resize reflow. +It also checks installs, removals, ownership, manifests, errors and cancellation against +isolated fake packages. **Actual macOS execution is still pending.** Linux PTY helper +results and Windows tool results are not a Mac pass. + +## Set up on a Mac + +Use Terminal, iTerm2 or another shell; no desktop automation is needed. Install: + +- The .NET SDK selected by the **repository root** `global.json` (currently 10.0.401). + Its .NET 10 runtime is required even when testing the net8.0 apphost. +- The .NET 8 runtime in that same installation if also testing net8.0. The apphost + permits major roll-forward, so require runtime 8 below to actually test .NET 8. +- Python 3.9 or newer, and a supported Node.js LTS release with npm. + +Use native arm64 SDK/runtime/Node/Python on Apple Silicon, or native x64 on Intel; +do not mix Rosetta x64 runtimes with an arm64 apphost. Run from this branch's checkout, +including when its path contains spaces: + +```bash +cd "/path/to/Client.Tools" +repo="$PWD" +terminal="$repo/dotnet-package-skills/tests/terminal" + +dotnet --version +dotnet --list-runtimes +export DOTNET_ROOT="$(python3 -c 'import os,shutil; print(os.path.dirname(os.path.realpath(shutil.which("dotnet"))))')" +export DOTNET_HOST_PATH="$DOTNET_ROOT/dotnet" + +python3 -m venv "$terminal/.venv" +"$terminal/.venv/bin/python" -m pip install -r "$terminal/requirements.txt" +npm --prefix "$terminal" ci +``` + +Python dependencies are pinned; `pywinpty` is installed only on Windows. The POSIX +backend uses Python's standard library, not a new terminal dependency. + +## Build and run the full suite + +Build from the repository root with the selected SDK and the existing Arcade imports. +This project-scoped build produces both native apphosts without running or changing CI: + +```bash +dotnet build "$repo/dotnet-package-skills/src/DotnetPackageSkills.csproj" \ + --configuration Release +``` + +Alternatively, use the native root build entry point: +`bash eng/common/build.sh --restore --build --configuration Release --projects dotnet-package-skills/src/DotnetPackageSkills.csproj`. +If Arcade selects a checkout-local `.dotnet`, put that installation on `PATH` and set +`DOTNET_ROOT`/`DOTNET_HOST_PATH` to it for the tests; runtime 8 must also be available +there for the optional run. Do not use `--prepareMachine`. + +The tool path is the executable **without `.exe`**, not the DLL, a Windows build copied +to the Mac, or a globally installed tool. Give each run a fresh artifact directory: + +```bash +run="$(mktemp -d "${TMPDIR:-/tmp}/package-skills-terminal.XXXXXX")" +printf 'Artifacts: %s\n' "$run" +set -o pipefail + +"$terminal/.venv/bin/python" "$terminal/verify_picker.py" \ + --tool "$repo/artifacts/bin/DotnetPackageSkills/Release/net10.0/dotnet-package-skills" \ + --artifacts "$run/net10.0" 2>&1 | tee "$run/net10.0.log" +``` + +Optional full run against .NET 8 (requires the SDK plus both runtimes listed above): + +```bash +DOTNET_ROLL_FORWARD=LatestPatch \ +"$terminal/.venv/bin/python" "$terminal/verify_picker.py" \ + --tool "$repo/artifacts/bin/DotnetPackageSkills/Release/net8.0/dotnet-package-skills" \ + --artifacts "$run/net8.0" 2>&1 | tee "$run/net8.0.log" +``` + +Exit 0 means all selected cases passed; a failure, timeout, missing dependency or invalid +apphost exits nonzero. `pipefail` keeps failures visible when using `tee`. A full run +selects 51 cases; do not report a filtered run as a full pass. +The console summary is saved in `net10.0.log` / `net8.0.log`; per-terminal `.ansi.txt`, +`.screen.txt` and (for reflow cases) `.buffers.json` files are under each run's `logs/`. +Keep the printed `$run` path and these files when reporting failures. + +The tests create Demo.Alpha/Demo.Beta packages and a local-only feed. Restore/discovery +cases need the SDK but no remote package feeds at runtime. Two lock cases build the +small `mutex_holder.cs.txt` fixture into the artifact directory, using .NET's actual +named mutex rather than an unrelated POSIX file lock. Its `.txt` extension prevents +the regular C# test project from compiling this manual-only fixture. +Package caches, SDK user profiles, HTTP caches, skill destinations and handwritten +sentinels are test-owned. SDK first-run certificate creation, global-tool PATH updates +and workload update checks are disabled for fixtures. Tests never install/replace a global tool or remove the user's +skills/cache directories. Only their freshly created temporary fixture directories +are cleaned; logs and the mutex fixture remain in `$run`. + +## Platform coverage and focused checks + +The driver selects all 51 tool cases on macOS; there are **no blanket Mac skips**. +PowerShell shell history is replaced by `/bin/sh` history, NTFS junctions by directory symlinks, and the +Win32 `Global\` mutex fixture by a .NET named mutex with POSIX case-sensitive canonical +paths. The Win32 `\\?\` spelling is exercised only on Windows; the Unix serialization +case instead uses an equivalent `..` destination spelling. These Unix equivalents do +not claim coverage of Win32/NTFS mechanisms on a Mac. All shared behavior assertions +are retained; failures are not converted to skips. + +Focused helper checks are separate from the full tool suite: + +```bash +"$terminal/.venv/bin/python" -m unittest discover \ + -s "$terminal" -p 'test_*.py' -v +``` + +They cover streaming UTF-8, terminal replies, isolation, named-mutex ownership, shell +quoting, controlling-terminal/session setup, real resize/SIGWINCH, Ctrl-C, exit/drain +behavior and owned-process/descriptor cleanup. POSIX PTY integration checks are +explicitly skipped **on Windows only**; this is not a skip of the real Mac picker suite. +The same PTY backend can be checked on Linux, but macOS and its .NET console behavior +must still be verified by running the full suite on a real Mac. + +## Provenance + +Adapted from [kartheekp-ms/dotnet-package-skills `tests/terminal`](https://github.com/kartheekp-ms/dotnet-package-skills/tree/59d3bc0d4fc80188d33bcc257d2a83c99a4cbfa3/tests/terminal) +at commit `59d3bc0d4fc80188d33bcc257d2a83c99a4cbfa3`. The five original files were +imported byte-for-byte before adapting the shared Python driver. The emulator and +pinned Python/Node manifests are unchanged. The upstream README declares MIT; +see this repository's [MIT license](../../../LICENSE). Dependencies retain their own +licenses (`@xterm/headless` is MIT). diff --git a/dotnet-package-skills/tests/terminal/emulator.cjs b/dotnet-package-skills/tests/terminal/emulator.cjs new file mode 100644 index 0000000..34d2f85 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/emulator.cjs @@ -0,0 +1,49 @@ +const { Terminal } = require('@xterm/headless'); +const readline = require('node:readline'); + +let terminal; + +function text(buffer) { + const rows = []; + for (let index = 0; index < buffer.length; index++) { + rows.push(buffer.getLine(index).translateToString(true)); + } + return rows.join('\n').trimEnd(); +} + +async function main() { + for await (const line of readline.createInterface({ input: process.stdin })) { + const command = JSON.parse(line); + switch (command.action) { + case 'open': + terminal = new Terminal({ + cols: command.columns, + rows: command.rows, + scrollback: 2000, + allowProposedApi: true + }); + break; + case 'write': + await new Promise(resolve => terminal.write(command.data, resolve)); + break; + case 'resize': + terminal.resize(command.columns, command.rows); + break; + case 'snapshot': + break; + default: + throw new Error(`Unknown emulator action: ${command.action}`); + } + console.log(JSON.stringify({ + alternate: terminal.buffer.active.type === 'alternate', + normal: text(terminal.buffer.normal), + active: text(terminal.buffer.active) + })); + } + terminal?.dispose(); +} + +main().catch(error => { + console.error(error); + process.exitCode = 1; +}); diff --git a/dotnet-package-skills/tests/terminal/mutex_holder.cs.txt b/dotnet-package-skills/tests/terminal/mutex_holder.cs.txt new file mode 100644 index 0000000..20c967d --- /dev/null +++ b/dotnet-package-skills/tests/terminal/mutex_holder.cs.txt @@ -0,0 +1,22 @@ +using System; +using System.Threading; + +using var mutex = new Mutex(initiallyOwned: false, args[0]); +if (!mutex.WaitOne(TimeSpan.FromSeconds(15))) +{ + throw new TimeoutException("The terminal fixture could not acquire the destination mutex."); +} + +try +{ + Console.WriteLine("locked"); + Console.Out.Flush(); + if (Console.ReadLine() is null) + { + throw new InvalidOperationException("The terminal fixture's release pipe closed unexpectedly."); + } +} +finally +{ + mutex.ReleaseMutex(); +} diff --git a/dotnet-package-skills/tests/terminal/package-lock.json b/dotnet-package-skills/tests/terminal/package-lock.json new file mode 100644 index 0000000..ff1ae5f --- /dev/null +++ b/dotnet-package-skills/tests/terminal/package-lock.json @@ -0,0 +1,19 @@ +{ + "name": "terminal", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "devDependencies": { + "@xterm/headless": "5.5.0" + } + }, + "node_modules/@xterm/headless": { + "version": "5.5.0", + "resolved": "https://registry.npmjs.org/@xterm/headless/-/headless-5.5.0.tgz", + "integrity": "sha512-5xXB7kdQlFBP82ViMJTwwEc3gKCLGKR/eoxQm4zge7GPBl86tCdI0IdPJjoKd8mUSFXz5V7i/25sfsEkP4j46g==", + "dev": true, + "license": "MIT" + } + } +} diff --git a/dotnet-package-skills/tests/terminal/package.json b/dotnet-package-skills/tests/terminal/package.json new file mode 100644 index 0000000..5f9516c --- /dev/null +++ b/dotnet-package-skills/tests/terminal/package.json @@ -0,0 +1,7 @@ +{ + "private": true, + "description": "Terminal-emulator regression support for the .NET skill picker", + "devDependencies": { + "@xterm/headless": "5.5.0" + } +} diff --git a/dotnet-package-skills/tests/terminal/requirements.txt b/dotnet-package-skills/tests/terminal/requirements.txt new file mode 100644 index 0000000..2ccfd29 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/requirements.txt @@ -0,0 +1,3 @@ +pyte==0.8.2 +pywinpty==3.0.5; sys_platform == "win32" +wcwidth==0.8.3 diff --git a/dotnet-package-skills/tests/terminal/terminal_support.py b/dotnet-package-skills/tests/terminal/terminal_support.py new file mode 100644 index 0000000..bd8b068 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/terminal_support.py @@ -0,0 +1,230 @@ +"""Small platform adapter for the manual terminal regression driver.""" + +import codecs +import errno +import hashlib +import os +from pathlib import Path +import struct +import subprocess +import sys +import threading +import time + +if sys.platform in ("darwin", "linux"): + import fcntl + import pty + import select + import signal + import termios + + +class OutputDecoder: + def __init__(self): + self.decoder = codecs.getincrementaldecoder("utf-8")(errors="strict") + + def decode(self, data, final=False): + return self.decoder.decode(data, final=final) + + +def winsize(rows, columns): + if any(type(value) is not int or not 1 <= value <= 65535 for value in (rows, columns)): + raise ValueError("Terminal rows and columns must be integers between 1 and 65535.") + return struct.pack("HHHH", rows, columns, 0, 0) + + +def spawn_pty(command, cwd, env, dimensions): + if sys.platform == "win32": + from winpty import Backend, PtyProcess + + return PtyProcess.spawn( + command, cwd=str(cwd), env=env, dimensions=dimensions, backend=Backend.ConPTY, + ) + return PosixPtyProcess.spawn(command, cwd=cwd, env=env, dimensions=dimensions) + + +class PosixPtyProcess: + # Exec in a fresh Python process instead of running Python preexec_fn after a + # multithreaded fork. The apphost then retains this PID, session and terminal. + CHILD = ( + "import fcntl,os,sys,termios\n" + "fcntl.ioctl(0, termios.TIOCSCTTY, 0)\n" + "os.execvpe(sys.argv[1], sys.argv[1:], os.environ)\n" + ) + + def __init__(self, process, master): + self.process = process + self.master = master + self.pid = process.pid + self.decoder = OutputDecoder() + self.closed = threading.Event() + self.fd_lock = threading.Lock() + self.write_lock = threading.Lock() + self.eof = False + + @classmethod + def spawn(cls, command, cwd, env, dimensions): + if sys.platform not in ("darwin", "linux"): + raise RuntimeError("POSIX terminal tests require macOS or Linux.") + if not command: + raise ValueError("A terminal command is required.") + size = winsize(*dimensions) + master, slave = pty.openpty() + try: + fcntl.ioctl(slave, termios.TIOCSWINSZ, size) + attributes = termios.tcgetattr(slave) + attributes[3] &= ~(termios.ECHO | termios.ICANON) + attributes[6][termios.VMIN] = 1 + attributes[6][termios.VTIME] = 0 + termios.tcsetattr(slave, termios.TCSANOW, attributes) + os.set_blocking(master, False) + process = subprocess.Popen( + [sys.executable, "-c", cls.CHILD, *command], + cwd=str(cwd), env=env, stdin=slave, stdout=slave, stderr=slave, + close_fds=True, start_new_session=True, + ) + except BaseException: + os.close(master) + raise + finally: + os.close(slave) + return cls(process, master) + + @property + def exitstatus(self): + return self.process.poll() + + def read(self, size): + if size <= 0: + raise ValueError("PTY read size must be positive.") + while not self.closed.is_set(): + if self.eof: + raise EOFError + try: + readable, _, _ = select.select([self.master], [], [], 0.1) + if not readable: + continue + with self.fd_lock: + if self.closed.is_set(): + raise EOFError + try: + data = os.read(self.master, size) + except BlockingIOError: + continue + except OSError as error: + if error.errno != errno.EIO: + raise + data = b"" + except (OSError, ValueError): + if self.closed.is_set(): + raise EOFError from None + raise + self.eof = not data + text = self.decoder.decode(data, final=self.eof) + if text: + return text + raise EOFError + + def write(self, text): + data = memoryview(text.encode("utf-8")) + deadline = time.monotonic() + 5 + with self.write_lock: + while data: + if self.closed.is_set(): + raise OSError(errno.EBADF, "The test PTY is closed.") + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Writing to the test PTY timed out.") + _, writable, _ = select.select([], [self.master], [], min(remaining, 0.1)) + if not writable: + continue + with self.fd_lock: + if self.closed.is_set(): + raise OSError(errno.EBADF, "The test PTY is closed.") + try: + written = os.write(self.master, data) + except BlockingIOError: + continue + data = data[written:] + + def setwinsize(self, rows, columns): + # The kernel sends SIGWINCH to the slave's foreground process group. + size = winsize(rows, columns) + with self.fd_lock: + if self.closed.is_set(): + raise OSError(errno.EBADF, "The test PTY is closed.") + fcntl.ioctl(self.master, termios.TIOCSWINSZ, size) + + def signal_group(self, number): + try: + os.killpg(self.pid, number) + except ProcessLookupError: + pass + + def close(self, force=False): + if self.closed.is_set(): + return + try: + if force: + self.signal_group(signal.SIGTERM) + try: + self.process.wait(timeout=2) + except subprocess.TimeoutExpired: + self.signal_group(signal.SIGKILL) + self.process.wait(timeout=3) + # A shell/tool child can outlive its session leader. + self.signal_group(signal.SIGKILL) + else: + try: + self.process.wait(timeout=3) + except subprocess.TimeoutExpired: + self.signal_group(signal.SIGKILL) + self.process.wait(timeout=3) + raise + finally: + with self.fd_lock: + if not self.closed.is_set(): + self.closed.set() + os.close(self.master) + + +def shell_history_command(command, windows=None): + if windows is None: + windows = sys.platform == "win32" + if windows: + invocation = "& " + " ".join("'" + argument.replace("'", "''") + "'" for argument in command) + return [ + "powershell.exe", "-NoLogo", "-NoProfile", "-Command", + '1..90 | ForEach-Object { Write-Output "earlier console output $_" }; ' + + invocation + "; exit $LASTEXITCODE", + ] + return [ + "/bin/sh", "-c", + 'i=1; while [ "$i" -le 90 ]; do printf "earlier console output %s\\n" "$i"; ' + 'i=$((i+1)); done; "$@"; exit "$?"', + "terminal-test-shell", *command, + ] + + +def destination_mutex_name(destination): + canonical = os.path.realpath(destination).rstrip("\\/") + if sys.platform == "win32": + canonical = canonical.upper() + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest().upper() + return ("Global\\" if sys.platform == "win32" else "") + "dotnet-package-skills-" + digest + + +def fixture_environment(directory): + directory = Path(directory) + return dict( + os.environ, + DOTNET_CLI_HOME=str(directory / "dotnet home"), + NUGET_PACKAGES=str(directory / "package cache"), + NUGET_HTTP_CACHE_PATH=str(directory / "http cache"), + DOTNET_CLI_TELEMETRY_OPTOUT="1", + DOTNET_SKIP_FIRST_TIME_EXPERIENCE="1", + DOTNET_NOLOGO="1", + DOTNET_GENERATE_ASPNET_CERTIFICATE="false", + DOTNET_ADD_GLOBAL_TOOLS_TO_PATH="false", + DOTNET_CLI_WORKLOAD_UPDATE_NOTIFY_DISABLE="true", + ) diff --git a/dotnet-package-skills/tests/terminal/test_driver.py b/dotnet-package-skills/tests/terminal/test_driver.py new file mode 100644 index 0000000..03e1453 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/test_driver.py @@ -0,0 +1,123 @@ +"""Focused shared-driver coverage without launching the tool.""" + +from pathlib import Path +import queue +import subprocess +import tempfile +import threading +from types import SimpleNamespace +import unittest + +import pyte + +from verify_picker import Terminal +import verify_picker as driver + + +class DriverTests(unittest.TestCase): + def test_fixture_environment_keeps_sdk_and_http_caches_inside_owned_scratch(self): + with tempfile.TemporaryDirectory(prefix="terminal environment ") as scratch: + previous = driver.OPTIONS + driver.OPTIONS = SimpleNamespace(artifacts=Path(scratch)) + fixture = driver.PickerRegression("test_empty_uninstall_does_not_prompt") + try: + fixture.setUp() + for key, folder in ( + ("DOTNET_CLI_HOME", "dotnet home"), + ("NUGET_HTTP_CACHE_PATH", "http cache"), + ): + self.assertEqual(str(fixture.root / folder), fixture.environment.get(key)) + self.assertEqual("1", fixture.environment.get("DOTNET_CLI_TELEMETRY_OPTOUT")) + finally: + fixture.doCleanups() + driver.OPTIONS = previous + + def terminal(self): + terminal = Terminal.__new__(Terminal) + self.replies = [] + terminal.process = SimpleNamespace(write=self.replies.append) + terminal.screen = pyte.Screen(80, 24) + terminal.stream = pyte.Stream(terminal.screen) + terminal.output = queue.Queue() + terminal.raw = [] + terminal.emulator = None + terminal.ended = False + return terminal + + def test_cursor_queries_split_between_reads_receive_the_current_position(self): + terminal = self.terminal() + for chunk in ("\x1b[3;7H\x1b[6", "n", "\x1b[5n"): + terminal.output.put(chunk) + terminal.pump() + self.assertEqual(["\x1b[3;7R", "\x1b[0n"], self.replies) + + def test_replaced_history_screen_still_answers_queries_after_resize(self): + terminal = self.terminal() + terminal.screen = pyte.HistoryScreen(40, 12, history=2000) + terminal.stream = pyte.Stream(terminal.screen) + terminal.output.put("\x1b[10;30H\x1b[6n") + terminal.pump() + self.assertEqual(["\x1b[10;30R"], self.replies) + + def test_eof_is_processed_after_all_final_control_output(self): + terminal = self.terminal() + for data in ("final output", "\x1b[?25h", None): + terminal.output.put(data) + terminal.pump() + self.assertTrue(terminal.ended) + self.assertEqual("final output", terminal.text) + self.assertFalse(terminal.screen.cursor.hidden) + self.assertEqual("final output\x1b[?25h", "".join(terminal.raw)) + + +class MutexFixtureTests(unittest.TestCase): + def test_dotnet_fixture_holds_the_named_mutex_until_released(self): + self.assertTrue( + callable(getattr(driver, "mutex_holder", None)), + "The portable named-mutex fixture has not been implemented.", + ) + with tempfile.TemporaryDirectory(prefix="terminal mutex ") as scratch: + previous = driver.OPTIONS + driver.OPTIONS = SimpleNamespace(artifacts=Path(scratch), mutex_holder=None) + try: + assembly = driver.mutex_holder() + name = "terminal-test-" + Path(scratch).name + first = self.holder(assembly, name) + self.assertEqual("locked\n", self.line(first)) + second = self.holder(assembly, name) + with self.assertRaises(subprocess.TimeoutExpired): + second.wait(timeout=0.3) + self.assertEqual(("", ""), first.communicate(input="\n", timeout=5)) + self.assertEqual(0, first.returncode) + self.assertEqual("locked\n", self.line(second)) + self.assertEqual(("", ""), second.communicate(input="\n", timeout=5)) + self.assertEqual(0, second.returncode) + finally: + driver.OPTIONS = previous + + def holder(self, assembly, name): + process = subprocess.Popen( + ["dotnet", str(assembly), name], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, encoding="utf-8", + ) + + def close(): + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + self.addCleanup(close) + return process + + def line(self, process): + output = queue.Queue() + reader = threading.Thread(target=lambda: output.put(process.stdout.readline()), daemon=True) + reader.start() + line = output.get(timeout=15) + reader.join(timeout=3) + return line + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/dotnet-package-skills/tests/terminal/test_terminal_support.py b/dotnet-package-skills/tests/terminal/test_terminal_support.py new file mode 100644 index 0000000..e438846 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/test_terminal_support.py @@ -0,0 +1,378 @@ +"""Focused adapter tests; POSIX cases launch real, owned Python processes.""" + +import importlib +import importlib.util +import json +import os +from pathlib import Path +import queue +import signal +import struct +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from unittest.mock import patch + + +class SupportTest(unittest.TestCase): + def setUp(self): + self.assertIsNotNone( + importlib.util.find_spec("terminal_support"), + "The portable terminal adapter has not been implemented.", + ) + self.support = importlib.import_module("terminal_support") + + +class PortableHelpers(SupportTest): + def test_utf8_decoder_retains_characters_split_at_every_byte(self): + text = "Caf\u00e9 \u6e2c\u8a66 \U0001f9ea e\u0301\x1b[6n" + decoder = self.support.OutputDecoder() + actual = "".join(decoder.decode(bytes([byte])) for byte in text.encode("utf-8")) + self.assertEqual(text, actual + decoder.decode(b"", final=True)) + + def test_utf8_decoder_rejects_invalid_and_incomplete_output(self): + with self.assertRaises(UnicodeDecodeError): + self.support.OutputDecoder().decode(b"\xff") + decoder = self.support.OutputDecoder() + self.assertEqual("", decoder.decode(b"\xf0\x9f")) + with self.assertRaises(UnicodeDecodeError): + decoder.decode(b"", final=True) + + def test_dimensions_use_rows_then_columns(self): + self.assertEqual( + (24, 100, 0, 0), + struct.unpack("HHHH", self.support.winsize(24, 100)), + ) + + def test_invalid_dimensions_are_rejected_before_opening_a_terminal(self): + for value in (0, -1, 65536, 1.5, True): + for rows, columns in ((value, 80), (24, value)): + with self.subTest(rows=rows, columns=columns): + with self.assertRaises(ValueError): + self.support.winsize(rows, columns) + + def test_shell_history_preserves_unix_arguments_without_interpolation(self): + command = ["/a path/tool", "install", "--destination", "a 'quoted' path", "; exit 99"] + wrapped = self.support.shell_history_command(command, windows=False) + self.assertEqual(["/bin/sh", "-c"], wrapped[:2]) + self.assertIn('"$@"', wrapped[2]) + self.assertEqual(command, wrapped[4:]) + + def test_shell_history_preserves_the_existing_powershell_invocation(self): + command = [r"C:\a path\tool.exe", "install", "a 'quoted' path"] + wrapped = self.support.shell_history_command(command, windows=True) + self.assertEqual( + ["powershell.exe", "-NoLogo", "-NoProfile", "-Command"], + wrapped[:4], + ) + self.assertIn("'a ''quoted'' path'", wrapped[4]) + self.assertTrue(wrapped[4].endswith("; exit $LASTEXITCODE")) + + def test_mutex_name_matches_the_existing_platform_canonicalization(self): + import hashlib + + with tempfile.TemporaryDirectory(prefix="terminal lock ") as scratch: + destination = Path(scratch) / "Skills" + canonical = os.path.realpath(destination).rstrip("\\/") + if sys.platform == "win32": + canonical = canonical.upper() + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest().upper() + prefix = "Global\\" if sys.platform == "win32" else "" + self.assertEqual( + prefix + "dotnet-package-skills-" + digest, + self.support.destination_mutex_name(destination), + ) + self.assertEqual( + self.support.destination_mutex_name(destination), + self.support.destination_mutex_name(destination / ".." / "Skills"), + ) + + def test_fixture_environment_disables_sdk_first_run_global_mutations_and_network_checks(self): + with tempfile.TemporaryDirectory(prefix="terminal sdk ") as scratch: + environment = self.support.fixture_environment(scratch) + self.assertEqual("false", environment.get("DOTNET_GENERATE_ASPNET_CERTIFICATE")) + self.assertEqual("false", environment.get("DOTNET_ADD_GLOBAL_TOOLS_TO_PATH")) + self.assertEqual("true", environment.get("DOTNET_CLI_WORKLOAD_UPDATE_NOTIFY_DISABLE")) + + +@unittest.skipUnless(sys.platform in ("darwin", "linux"), "Real POSIX PTYs require macOS or Linux.") +class PosixProcesses(SupportTest): + def setUp(self): + super().setUp() + self.scratch = tempfile.TemporaryDirectory(prefix="terminal adapter ") + self.addCleanup(self.scratch.cleanup) + + def spawn(self, source, *arguments, dimensions=(24, 80)): + environment = dict(os.environ, TERM="xterm-256color", TERMINAL_TEST="isolated") + process = self.support.PosixPtyProcess.spawn( + [sys.executable, "-c", source, *arguments], + cwd=self.scratch.name, env=environment, dimensions=dimensions, + ) + output = queue.Queue() + + def read(): + try: + while True: + output.put(process.read(65536)) + except EOFError: + output.put(None) + except Exception as error: + output.put(error) + + reader = threading.Thread(target=read, daemon=True) + reader.start() + + def close(): + process.close(force=True) + reader.join(timeout=3) + self.assertFalse(reader.is_alive(), "PTY reader did not stop after close.") + + self.addCleanup(close) + return process, output + + def read_until(self, output, expected=None): + text = "" + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + data = output.get(timeout=max(0.01, deadline - time.monotonic())) + if isinstance(data, Exception): + raise data + if data is None: + if expected is not None: + self.assertIn(expected, text) + return text + text += data + if expected is not None and expected in text: + return text + self.fail(f"PTY output timed out: {text!r}") + + def test_child_has_a_controlling_terminal_dimensions_environment_and_cwd(self): + process, output = self.spawn( + "import json,os; print(json.dumps({" + "'tty': [os.isatty(fd) for fd in (0,1,2)]," + "'foreground': os.tcgetpgrp(0) == os.getpgrp()," + "'session': os.getsid(0) == os.getpid()," + "'size': list(os.get_terminal_size(0))," + "'env': os.environ['TERMINAL_TEST'], 'cwd': os.getcwd()}), flush=True)", + dimensions=(29, 91), + ) + observed = json.loads(self.read_until(output).strip()) + self.assertEqual([True, True, True], observed["tty"]) + self.assertTrue(observed["foreground"]) + self.assertTrue(observed["session"]) + self.assertEqual([91, 29], observed["size"]) + self.assertEqual("isolated", observed["env"]) + self.assertEqual(os.path.realpath(self.scratch.name), observed["cwd"]) + process.close() + self.assertEqual(0, process.exitstatus) + + def test_read_decodes_real_utf8_chunks_and_drains_output_after_exit(self): + process, output = self.spawn( + "import os,time; data='Caf\\u00e9 \\U0001f9ea'.encode();" + "\nfor byte in data: os.write(1, bytes([byte])); time.sleep(.005)" + "\nos.write(1, b'x' * 131072 + b'FINAL'); os._exit(7)" + ) + self.assertEqual("Caf\u00e9 \U0001f9ea" + "x" * 131072 + "FINAL", self.read_until(output)) + process.close() + self.assertEqual(7, process.exitstatus) + + def test_live_resize_delivers_sigwinch_and_new_dimensions(self): + process, output = self.spawn( + "import json,os,signal;" + "\ndef resized(*_): print(json.dumps(list(os.get_terminal_size(0))), flush=True)" + "\nsignal.signal(signal.SIGWINCH, resized); print('READY', flush=True); signal.pause()" + ) + self.read_until(output, "READY") + process.setwinsize(17, 63) + self.assertIn("[63, 17]", self.read_until(output)) + process.close() + self.assertEqual(0, process.exitstatus) + + def test_keys_are_utf8_without_echo_or_canonical_line_buffering(self): + process, output = self.spawn( + "import os; print('READY', flush=True); data=b''" + "\nwhile len(data) < 3: data += os.read(0, 3-len(data))" + "\nprint('KEY:' + data.decode(), flush=True)" + ) + self.read_until(output, "READY") + process.write("q\u00e9") + self.assertEqual("KEY:q\u00e9", self.read_until(output).strip()) + process.close() + self.assertEqual(0, process.exitstatus) + + def test_control_c_is_input_when_the_application_disables_isig(self): + process, output = self.spawn( + "import os,termios; attrs=termios.tcgetattr(0); attrs[3] &= ~termios.ISIG;" + "termios.tcsetattr(0, termios.TCSANOW, attrs); print('READY', flush=True);" + "print('KEY:' + str(os.read(0,1)[0]), flush=True)" + ) + self.read_until(output, "READY") + process.write("\x03") + self.assertIn("KEY:3", self.read_until(output)) + process.close() + self.assertEqual(0, process.exitstatus) + + def test_control_c_reports_signal_exit_when_isig_is_enabled(self): + process, output = self.spawn( + "import signal,time; signal.signal(signal.SIGINT, signal.SIG_DFL);" + "print('READY', flush=True); time.sleep(30)" + ) + self.read_until(output, "READY") + process.write("\x03") + self.read_until(output) + process.close() + self.assertEqual(-signal.SIGINT, process.exitstatus) + + def test_force_close_kills_an_unresponsive_owned_group_and_is_idempotent(self): + process, output = self.spawn( + "import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN);" + "print('READY', flush=True); time.sleep(30)" + ) + self.read_until(output, "READY") + process.close(force=True) + process.close(force=True) + self.assertEqual(-signal.SIGKILL, process.exitstatus) + self.read_until(output) + + def test_force_close_does_not_signal_an_unrelated_process(self): + unrelated = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + try: + process, output = self.spawn("import time; print('READY', flush=True); time.sleep(30)") + self.read_until(output, "READY") + process.close(force=True) + self.assertIsNone(unrelated.poll()) + finally: + unrelated.terminate() + unrelated.wait(timeout=5) + + def test_force_close_cleans_owned_children_even_after_the_session_leader_exits(self): + worker = ( + "import signal,time; signal.signal(signal.SIGHUP, signal.SIG_IGN);" + "signal.signal(signal.SIGTERM, signal.SIG_IGN); print('READY', flush=True); time.sleep(30)" + ) + process, output = self.spawn( + "import os,subprocess,sys; child=subprocess.Popen(" + "[sys.executable, '-c', sys.argv[1]], stdin=subprocess.DEVNULL," + "stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True);" + "print('WORKER:' + str(child.pid) + ':' + child.stdout.readline().strip(), flush=True);" + "os._exit(0)", + worker, + ) + text = self.read_until(output, "READY") + pid = int(text.split("WORKER:")[1].split(":")[0]) + try: + process.process.wait(timeout=5) + self.read_until(output) + process.close(force=True) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + state = subprocess.run( + ["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True, timeout=3, + ).stdout.strip() + if not state or state.startswith("Z"): + break + time.sleep(0.05) + self.assertTrue(not state or state.startswith("Z"), f"Owned child {pid} is still running.") + finally: + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + + def test_failed_launch_reports_an_error_exit_instead_of_success(self): + process = self.support.PosixPtyProcess.spawn( + [str(Path(self.scratch.name) / "missing apphost")], + cwd=self.scratch.name, env=dict(os.environ), dimensions=(24, 80), + ) + self.addCleanup(process.close, force=True) + text = "" + while True: + try: + text += process.read(65536) + except EOFError: + break + process.close() + self.assertNotEqual(0, process.exitstatus) + self.assertIn("FileNotFoundError", text) + + def test_spawn_failure_closes_both_owned_descriptors(self): + before = set(os.listdir("/dev/fd")) + with self.assertRaises(FileNotFoundError): + self.support.PosixPtyProcess.spawn( + [sys.executable, "-c", "pass"], + cwd=str(Path(self.scratch.name) / "missing directory"), + env=dict(os.environ), dimensions=(24, 80), + ) + self.assertEqual(before, set(os.listdir("/dev/fd"))) + + def test_reader_cannot_consume_another_file_after_its_master_descriptor_is_closed(self): + process = self.support.PosixPtyProcess.spawn( + [sys.executable, "-c", "import time; time.sleep(30)"], + cwd=self.scratch.name, env=dict(os.environ), dimensions=(24, 80), + ) + self.addCleanup(process.close, force=True) + selected, resume = threading.Event(), threading.Event() + output = queue.Queue() + replacement = Path(self.scratch.name) / "not terminal output.txt" + replacement.write_bytes(b"not terminal output") + + def delayed_select(*_): + selected.set() + if not resume.wait(timeout=5): + raise TimeoutError("The test did not resume the PTY reader.") + return [process.master], [], [] + + def read(): + try: + output.put(process.read(65536)) + except EOFError: + output.put("EOF") + except Exception as error: + output.put(error) + + with patch.object(self.support.select, "select", side_effect=delayed_select): + reader = threading.Thread(target=read, daemon=True) + reader.start() + self.assertTrue(selected.wait(timeout=5)) + process.close(force=True) + descriptor = os.open(replacement, os.O_RDONLY) + try: + self.assertEqual(process.master, descriptor, "The regression requires descriptor reuse.") + resume.set() + self.assertEqual("EOF", output.get(timeout=5)) + finally: + resume.set() + reader.join(timeout=3) + os.close(descriptor) + self.assertFalse(reader.is_alive()) + + def test_symlink_and_dot_paths_share_a_case_sensitive_posix_mutex_name(self): + destination = Path(self.scratch.name) / "Skills" + destination.mkdir() + alias = Path(self.scratch.name) / "alias" + alias.symlink_to(destination, target_is_directory=True) + self.assertEqual( + self.support.destination_mutex_name(destination), + self.support.destination_mutex_name(alias / "." / "child" / ".."), + ) + self.assertNotEqual( + self.support.destination_mutex_name(destination), + self.support.destination_mutex_name(destination.with_name("skills")), + ) + + def test_shell_history_executes_the_exact_command_and_preserves_failure(self): + command = self.support.shell_history_command( + [sys.executable, "-c", "import sys; print(sys.argv[1]); sys.exit(7)", "a 'quoted' path"], + windows=False, + ) + result = subprocess.run(command, capture_output=True, text=True, timeout=10) + self.assertEqual(7, result.returncode, result.stderr) + self.assertEqual(90, result.stdout.count("earlier console output ")) + self.assertTrue(result.stdout.endswith("a 'quoted' path\n")) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/dotnet-package-skills/tests/terminal/verify_picker.py b/dotnet-package-skills/tests/terminal/verify_picker.py new file mode 100644 index 0000000..479ed59 --- /dev/null +++ b/dotnet-package-skills/tests/terminal/verify_picker.py @@ -0,0 +1,1595 @@ +"""Manual terminal regressions against a built native tool on Windows, macOS or Linux. + +python verify_picker.py --tool --artifacts +""" + +import argparse +import ctypes +from ctypes import wintypes +from contextlib import contextmanager +import hashlib +import json +import os +from pathlib import Path +import queue +import re +import shutil +import subprocess +import sys +import tempfile +import threading +import time +import unicodedata +import unittest +from xml.sax.saxutils import escape +from zipfile import ZipFile + +import pyte +from terminal_support import ( + destination_mutex_name, fixture_environment, shell_history_command, spawn_pty, +) + + +OPTIONS = None +ENTER, ESC, SPACE = "\r", "\x1b", " " +UP, DOWN, LEFT, RIGHT = "\x1b[A", "\x1b[B", "\x1b[D", "\x1b[C" +HOME, END = "\x1b[H", "\x1b[F" +PAGE_UP, PAGE_DOWN = "\x1b[5~", "\x1b[6~" +CTRL_UP, CTRL_DOWN, CTRL_C = "\x1b[1;5A", "\x1b[1;5B", "\x03" +ASPIRE_HINT = "(Press to select, to accept)" + + +class ReflowEmulator: + def __init__(self, rows, columns): + self.process = subprocess.Popen( + ["node", str(Path(__file__).with_name("emulator.cjs"))], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, encoding="utf-8", + ) + try: + self.state = self.send("open", rows=rows, columns=columns) + except Exception: + self.close() + raise + + def send(self, action, **parameters): + self.process.stdin.write(json.dumps({"action": action, **parameters}) + "\n") + self.process.stdin.flush() + line = self.process.stdout.readline() + if not line: + raise AssertionError("Terminal emulator failed: " + self.process.stderr.read()) + self.state = json.loads(line) + return self.state + + def close(self): + self.process.stdin.close() + try: + self.process.wait(timeout=5) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=5) + finally: + self.process.stdout.close() + self.process.stderr.close() + + +class Terminal: + def __init__(self, command, directory, environment, log, rows=24, columns=100): + self.process = spawn_pty( + command, cwd=directory, env=environment, dimensions=(rows, columns), + ) + self.screen = pyte.Screen(columns, rows) + self.stream = pyte.Stream(self.screen) + self.log = log + self.output = queue.Queue() + self.raw = [] + self.frames = [] + self.ended = False + self.closing = threading.Event() + self.emulator = None + self.reader = threading.Thread(target=self._read, daemon=True) + self.reader.start() + + def _read(self): + try: + while True: + data = self.process.read(65536) + if not data: + break + self.output.put(data) + except EOFError: + pass + except OSError as error: + if not self.closing.is_set(): + self.output.put(error) + except Exception as error: + self.output.put(error) + finally: + self.output.put(None) + + def pump(self, timeout=0.03): + try: + data = self.output.get(timeout=timeout) + except queue.Empty: + return False + if data is None: + self.ended = True + elif isinstance(data, Exception): + raise data + else: + self.raw.append(data) + self.screen.write_process_input = self.process.write + self.stream.feed(data) + if self.emulator is not None: + self.emulator.send("write", data=data) + return True + + @property + def text(self): + return "\n".join(line.rstrip() for line in self.screen.display).rstrip() + + @property + def compact(self): + return re.sub(r"\s+", " ", self.text) + + @property + def focused(self): + return next((line for line in self.screen.display if re.match(r"\s*>", line)), "") + + @property + def page_count(self): + match = re.search(r"page\s+\d+\s+of\s+(\d+)", self.compact, re.IGNORECASE) + return int(match[1]) if match else 1 + + def wait_for(self, predicate, label, timeout=15): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + self.pump() + if predicate(): + while self.pump(timeout=0.08): + pass + if predicate(): + self.frames.append(f"--- {label} ---\n{self.text}") + return + if self.ended: + break + self.frames.append(f"--- failed: {label} ---\n{self.text}") + raise AssertionError(f"Terminal did not reach {label}.\n{self.text}") + + def ready(self): + self.wait_for( + lambda: ASPIRE_HINT in self.compact and re.search(r"\[[ X]\]", self.text), + "initial picker", + ) + return self + + def press(self, keys, predicate=None, label="keypress"): + self.process.write(keys) + if predicate is not None: + self.wait_for(predicate, label) + + def focus(self, name, ordinal): + self.press(HOME) + self.press(DOWN * ordinal, lambda: name in self.focused, f"focus {name}") + + def colors(self, text): + for y, line in enumerate(self.screen.display): + start = line.find(text) + if start >= 0: + return [self.screen.buffer[y][x].fg for x in range(start, start + len(text))] + raise AssertionError(f"{text!r} is not visible.\n{self.text}") + + def resize(self, rows, columns): + if self.emulator is not None: + self.emulator.send("resize", rows=rows, columns=columns) + self.screen.resize(lines=rows, columns=columns) + self.process.setwinsize(rows, columns) + + def finish(self, keys=ENTER): + self.process.write(keys) + return self.exit_status() + + def exit_status(self): + self.wait_for(lambda: self.ended, "process exit") + # Close before querying status: PtyProcess.wait() marks the object closed without + # releasing its reader sockets in pywinpty 3.0.5. + self.process.close() + return self.process.exitstatus + + def close(self): + self.closing.set() + self.process.close(force=True) + self.reader.join(timeout=3) + while self.pump(timeout=0.01): + pass + self.log.parent.mkdir(parents=True, exist_ok=True) + self.log.with_suffix(".ansi.txt").write_text("".join(self.raw), encoding="utf-8") + self.log.with_suffix(".screen.txt").write_text("\n\n".join(self.frames), encoding="utf-8") + if self.emulator is not None: + self.log.with_suffix(".buffers.json").write_text( + json.dumps(self.emulator.state, indent=2), encoding="utf-8") + self.emulator.close() + + +def snapshot(directory): + return { + str(path.relative_to(directory)): hashlib.sha256(path.read_bytes()).hexdigest() + for path in directory.rglob("*") if path.is_file() + } + +def mutex_holder(): + if OPTIONS.mutex_holder is not None: + return OPTIONS.mutex_holder + directory = OPTIONS.artifacts / "mutex-holder" + directory.mkdir(parents=True, exist_ok=True) + shutil.copyfile(Path(__file__).with_name("mutex_holder.cs.txt"), directory / "Program.cs") + project = directory / "MutexHolder.csproj" + project.write_text( + '' + 'Exenet10.0' + 'false', + encoding="utf-8", + ) + config = directory / "NuGet.Config" + config.write_text( + '', + encoding="utf-8", + ) + result = subprocess.run( + ["dotnet", "build", str(project), "--configuration", "Release", "--nologo", + "--verbosity", "quiet", "--configfile", str(config), + "-p:ImportDirectoryBuildProps=false", "-p:ImportDirectoryBuildTargets=false"], + cwd=Path(__file__).resolve().parents[3], env=fixture_environment(directory), + capture_output=True, encoding="utf-8", timeout=90, + ) + if result.returncode: + raise AssertionError("Could not build the named-mutex fixture:\n" + result.stdout + result.stderr) + OPTIONS.mutex_holder = directory / "bin" / "Release" / "net10.0" / "MutexHolder.dll" + return OPTIONS.mutex_holder + + +@contextmanager +def destination_mutex(destination): + if sys.platform != "win32": + import select + + process = subprocess.Popen( + ["dotnet", str(mutex_holder()), destination_mutex_name(destination)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, encoding="utf-8", + ) + try: + readable, _, _ = select.select([process.stdout], [], [], 20) + if not readable: + raise AssertionError("The named-mutex fixture did not become ready within 20 seconds.") + if process.stdout.readline() != "locked\n": + raise AssertionError("The named-mutex fixture failed:\n" + process.stderr.read()) + yield + finally: + try: + stdout, stderr = process.communicate(input="\n", timeout=5) + except subprocess.TimeoutExpired: + process.kill() + stdout, stderr = process.communicate(timeout=5) + raise AssertionError("The named-mutex fixture did not release its lock:\n" + stdout + stderr) + if process.returncode: + raise AssertionError("The named-mutex fixture failed:\n" + stdout + stderr) + return + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.CreateMutexW.argtypes = [ctypes.c_void_p, wintypes.BOOL, wintypes.LPCWSTR] + kernel.CreateMutexW.restype = wintypes.HANDLE + kernel.ReleaseMutex.argtypes = [wintypes.HANDLE] + kernel.ReleaseMutex.restype = wintypes.BOOL + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + handle = kernel.CreateMutexW(None, True, destination_mutex_name(destination)) + if not handle: + raise ctypes.WinError(ctypes.get_last_error()) + try: + yield + finally: + released = kernel.ReleaseMutex(handle) + error = ctypes.get_last_error() + kernel.CloseHandle(handle) + if not released: + raise ctypes.WinError(error) + + +class PickerRegression(unittest.TestCase): + def setUp(self): + self.scratch = tempfile.TemporaryDirectory(prefix="picker-", dir=OPTIONS.artifacts) + self.addCleanup(self.scratch.cleanup) + self.root = Path(self.scratch.name) + self.cache = self.root / "package cache" + self.destination = self.root / "installed skills" + self.environment = fixture_environment(self.root) + self.environment.pop("NO_COLOR", None) + self.environment["TERM"] = "xterm-256color" + self.environment["NUGET_PACKAGES"] = str(self.cache) + self.names = [] + for package, version, prefix, count in ( + ("Demo.Alpha", "1.0.0", "alpha", 15), + ("Demo.Beta", "2.0.0", "beta", 10), + ): + for number in range(1, count + 1): + name = f"{prefix}-{number:02}" + self.names.append(name) + directory = self.cache / package.lower() / version / "skills" / name + directory.mkdir(parents=True) + description = f"Guidance for {name}: configure services and verify expected behavior." + if name == "alpha-01": + description = "ALPHA-FIRST: configure test doubles and check their calls." + if name == "alpha-02": + description += " Keep changes predictable." + if name == "alpha-05": + description = ( + "LONG-DESCRIPTION-START " + + " ".join( + f"Step {step:02}: every line of this description must remain readable." + for step in range(36) + ) + + "LONG-DESCRIPTION-END" + ) + contents = f"---\nname: {name}\ndescription: {json.dumps(description)}\n---\n# Body\n" + if name == "alpha-03": + contents = "# No frontmatter\n" + elif name == "alpha-04": + contents = "---\ndescription: [broken\n---\n" + (directory / "SKILL.md").write_text(contents, encoding="utf-8") + handwritten = self.destination / "team-owned" + handwritten.mkdir(parents=True) + (handwritten / "SKILL.md").write_text("Do not change our own skill.\n", encoding="utf-8") + self.manifest = self.destination / ".dotnet-package-skills.json" + self.target = None + self.terminal_count = 0 + + def command(self, verb, *extra, packages=None): + args = [str(OPTIONS.tool), verb, "--destination", str(self.destination)] + if verb != "uninstall": + args += ["--global-packages", str(self.cache)] + if self.target is not None: + args += ["--target", str(self.target)] + else: + for package in packages or ["Demo.Alpha@1.0.0", "Demo.Beta@2.0.0"]: + args += ["--package", package] + elif "--stale" in extra and self.target is not None: + args += ["--target", str(self.target)] + return args + list(extra) + + def cli(self, verb, *extra, packages=None, expected=0): + result = subprocess.run( + self.command(verb, *extra, packages=packages), + cwd=self.root, env=self.environment, capture_output=True, + encoding="utf-8", errors="replace", timeout=30, + ) + self.assertEqual(expected, result.returncode, result.stdout + result.stderr) + return result + + def suggested(self, arguments, with_target=False): + """The uninstall command the tool suggests, with this fixture's target and destination.""" + command = f"dotnet-package-skills uninstall {arguments}" + if with_target and self.target is not None: + command += f' --target "{self.target}"' + return command + f' --destination "{self.destination}"' + + def terminal(self, verb="install", *extra, packages=None, rows=24, columns=100, no_color=False, + shell_output=False): + environment = dict(self.environment) + if no_color: + environment["NO_COLOR"] = "1" + self.terminal_count += 1 + command = self.command(verb, "-i", *extra, packages=packages) + if shell_output: + command = shell_history_command(command) + terminal = Terminal( + command, + self.root, environment, + OPTIONS.artifacts / "logs" / f"{self._testMethodName}-{self.terminal_count}", + rows, columns, + ) + self.addCleanup(terminal.close) + return terminal + + def manifest_packages(self): + if not self.manifest.exists(): + return {} + return json.loads(self.manifest.read_text(encoding="utf-8-sig"))["packages"] + + def installed(self): + return {name for package in self.manifest_packages().values() for name in package["skills"]} + + def assert_shown(self, expected, terminal): + """Checks text the console may have wrapped mid-word by ignoring all whitespace.""" + self.assertIn(re.sub(r"\s+", "", expected), re.sub(r"\s+", "", terminal.text), terminal.text) + + def assert_frame(self, terminal): + self.assertIn("Which skills should", terminal.compact) + self.assertIn(ASPIRE_HINT, terminal.compact) + for hint in re.findall(r"\([^()]*<[^()]*\)", terminal.compact): + self.assertTrue(hint.startswith("(Press <"), hint) + self.assertNotIn("will install", terminal.text.lower()) + self.assertNotIn("will remove", terminal.text.lower()) + self.assertNotIn("(Demo.Alpha 1.0.0)", terminal.text) + self.assertNotIn("(Demo.Beta 2.0.0)", terminal.text) + self.assertIn("selected", terminal.text) + + def test_plain_commands_do_not_gain_descriptions_and_write_an_lf_manifest(self): + before = snapshot(self.destination) + listed = self.cli("list").stdout + for name in self.names: + self.assertIn(name, listed) + self.assertNotIn("ALPHA-FIRST", listed) + self.assertNotIn("Guidance for", listed) + self.assertEqual(before, snapshot(self.destination)) + self.cli("install", "--dry-run") + self.assertEqual(before, snapshot(self.destination)) + result = self.cli("install") + self.assertIn("Copied 25 skills", result.stdout) + self.assertNotIn("ALPHA-FIRST", result.stdout) + self.assertTrue(self.manifest.is_file()) + self.assertFalse(self.manifest.is_symlink()) + self.assertEqual(set(self.names), self.installed()) + raw = self.manifest.read_bytes() + self.assertNotIn(b"\r", raw) + self.assertFalse(raw.startswith(b"\xef\xbb\xbf")) + self.assertTrue(raw.endswith(b"}\n")) + self.assertEqual( + { + "version": 1, + "packages": { + "demo.alpha": {"version": "1.0.0", "skills": [f"alpha-{n:02}" for n in range(1, 16)]}, + "demo.beta": {"version": "2.0.0", "skills": [f"beta-{n:02}" for n in range(1, 11)]}, + }, + }, + json.loads(raw), + ) + installed = snapshot(self.destination) + self.cli("install") + self.assertEqual(installed, snapshot(self.destination)) + self.cli("uninstall", "--dry-run") + self.assertEqual(installed, snapshot(self.destination)) + self.assertIn("Removed 25 skills", self.cli("uninstall").stdout) + self.assertEqual(before, snapshot(self.destination)) + + def prepare_project(self): + feed = self.root / "local feed" + feed.mkdir() + for package in ("Demo.Alpha", "Demo.Beta"): + for source in sorted((self.cache / package.lower()).iterdir()): + version = source.name + with ZipFile(feed / f"{package}.{version}.nupkg", "w") as archive: + archive.writestr( + f"{package}.nuspec", + f'{package}' + f"{version}Regression fixture" + "Local terminal test fixture.", + ) + for path in source.rglob("*"): + if path.is_file(): + archive.write(path, path.relative_to(source).as_posix()) + project = self.root / "My App" + project.mkdir() + project_file = project / "App.csproj" + beta_reference = '' + project_xml = ( + '' + "net10.0false" + '' + f"{beta_reference}" + ) + project_file.write_text(project_xml, encoding="utf-8") + self.target = project / "App.slnx" + self.target.write_text('', encoding="utf-8") + config = project / "NuGet.Config" + config.write_text( + '' + f'', + encoding="utf-8", + ) + self.cache = self.root / "restored cache" + self.environment["NUGET_PACKAGES"] = str(self.cache) + + def restore(): + result = subprocess.run( + ["dotnet", "restore", str(self.target), "--configfile", str(config), + "--nologo", "--verbosity", "quiet"], + cwd=project, env=self.environment, capture_output=True, encoding="utf-8", timeout=60, + ) + self.assertEqual(0, result.returncode, result.stdout + result.stderr) + + restore() + return project_file, project_xml, beta_reference, restore + + def test_restored_solution_keeps_unreferenced_skills_until_uninstall_stale(self): + project_file, project_xml, beta_reference, restore = self.prepare_project() + listed = self.cli("list").stdout + self.assertIn(f"Target: {self.target}", listed) + for name in self.names: + self.assertIn(name, listed) + terminal = self.terminal().ready() + self.assertIn("App.slnx", terminal.text) + self.assertIn("ALPHA-FIRST", terminal.text) + terminal.press(SPACE) + self.assertEqual(0, terminal.finish()) + self.assertEqual({"alpha-01"}, self.installed()) + + terminal = self.terminal().ready() + self.assertIn("Installed skills aren't listed.", terminal.compact) + self.assertIn("0 of 24 selected", terminal.compact) + self.assertIn("alpha-02", terminal.focused) + self.assertNotIn("ALPHA-FIRST", terminal.text) + self.assertEqual(0, terminal.finish(ESC)) + + self.cli("install") + self.assertEqual(set(self.names), self.installed()) + project_file.write_text(project_xml.replace(beta_reference, ""), encoding="utf-8") + restore() + before = snapshot(self.destination) + for flags in (["--dry-run"], []): + with self.subTest(flags=flags): + result = self.cli("install", *flags) + self.assertIn( + "10 installed skills belong to a package that the target no longer references:", + result.stdout) + self.assertIn(" beta-10 (demo.beta 2.0.0)", result.stdout) + self.assertIn(f"Run '{self.suggested('--stale', with_target=True)}' to remove them.", result.stdout) + self.assertNotIn("Removed", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + preview = self.cli("uninstall", "--stale", "--dry-run").stdout + self.assertIn(f"Target: {self.target}", preview) + self.assertIn("Would remove 10 skills:", preview) + self.assertEqual(before, snapshot(self.destination)) + self.assertIn("Removed 10 skills:", self.cli("uninstall", "--stale").stdout) + self.assertEqual({f"alpha-{number:02}" for number in range(1, 16)}, self.installed()) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").exists()) + self.assertIn("No stale skills were found.", self.cli("uninstall", "--stale").stdout) + self.assertNotIn("uninstall --stale", self.cli("install").stdout) + + def test_interactive_install_stops_until_stale_skills_are_uninstalled(self): + project_file, project_xml, beta_reference, restore = self.prepare_project() + self.cli("install") + project_file.write_text(project_xml.replace(beta_reference, ""), encoding="utf-8") + restore() + before = snapshot(self.destination) + + terminal = self.terminal() + self.assertEqual(1, terminal.exit_status()) + self.assertNotIn("Which skills should", terminal.text) + self.assert_shown("10 installed skills don't match the target", terminal) + self.assert_shown(f"Run '{self.suggested('--stale', with_target=True)}' first", terminal) + self.assertEqual(before, snapshot(self.destination)) + + terminal = self.terminal("uninstall", "--stale").ready() + self.assertIn("Only skills that don't match the target are listed.", terminal.compact) + self.assertIn("0 of 10 selected", terminal.compact) + self.assertNotIn("alpha-", terminal.text) + terminal.press(END, lambda: "beta-10" in terminal.focused, "last stale skill") + terminal.press(SPACE, lambda: "1 to remove" in terminal.compact, "one stale removal") + self.assertEqual(0, terminal.finish()) + self.assertIn("Removed 1 skill", terminal.text) + self.assertEqual(set(self.names) - {"beta-10"}, self.installed()) + + terminal = self.terminal() + self.assertEqual(1, terminal.exit_status()) + self.assert_shown("9 installed skills don't match the target", terminal) + + self.cli("uninstall", "--stale") + before = snapshot(self.destination) + terminal = self.terminal() + self.assertEqual(0, terminal.exit_status()) + self.assertNotIn("Which skills should", terminal.text) + self.assert_shown("Nothing new to install. Every skill that these packages ship is already installed.", terminal) + self.assertEqual(before, snapshot(self.destination)) + + def test_empty_target_never_removes_skills_and_points_to_uninstall_stale(self): + project_file, project_xml, beta_reference, restore = self.prepare_project() + self.cli("install") + before = snapshot(self.destination) + project_file.write_text( + project_xml.replace(beta_reference, "").replace( + '', ""), + encoding="utf-8", + ) + restore() + + result = self.cli("install") + self.assertIn("No bundled skills found.", result.stdout) + self.assertIn( + "25 installed skills belong to packages that the target no longer references:", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + terminal = self.terminal() + self.assertEqual(1, terminal.exit_status()) + self.assert_shown("25 installed skills don't match the target", terminal) + self.assertEqual(before, snapshot(self.destination)) + + self.assertIn("Removed 25 skills:", self.cli("uninstall", "--stale").stdout) + self.assertEqual(set(), self.installed()) + self.assertFalse(self.manifest.exists()) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").exists()) + + def test_uninstall_stale_needs_a_solution_or_project(self): + self.cli("install") + before = snapshot(self.destination) + for flags in ([], ["--dry-run"]): + with self.subTest(flags=flags): + result = self.cli("uninstall", "--stale", *flags, expected=1) + self.assertIn("No solution or project found", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + + def test_incomplete_target_discovery_fails_before_writes_in_all_install_modes(self): + self.prepare_project() + self.cli("install") + before = snapshot(self.destination) + # dotnet list package restores into NUGET_PACKAGES, which stays complete. The tool reads + # the folder that --global-packages names, which lacks a package, as a fallback folder can. + partial = self.root / "partial cache" + shutil.copytree(self.cache, partial) + self.cache = partial + (self.cache / "demo.beta").rename(self.cache / "demo.beta-moved-aside") + available = self.cache / "demo.alpha" / "1.0.0" / "skills" / "alpha-01" / "SKILL.md" + available.write_text("---\ndescription: Must not be copied during incomplete discovery.\n---\n", encoding="utf-8") + + for flags in ([], ["--dry-run"], ["-i"], ["-i", "--dry-run"]): + with self.subTest(flags=flags): + result = self.cli("install", *flags, expected=1) + self.assertIn("resolved packages are missing", result.stderr) + self.assertIn("Demo.Beta 2.0.0", result.stderr) + self.assertEqual("", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + listed = self.cli("list").stdout + self.assertIn("alpha-01", listed) + self.assertNotIn("beta-01", listed) + + (self.cache / "demo.alpha").rename(self.cache / "demo.alpha-moved-aside") + result = self.cli("install", "-i", expected=1) + self.assertIn("resolved packages are missing", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + + def test_a_target_that_cannot_be_restored_is_reported_and_nothing_changes(self): + project_file, project_xml, _, _ = self.prepare_project() + self.cli("install") + before = snapshot(self.destination) + # The local feed has no 9.9.9, so the restore that dotnet list package runs fails. The + # tool reports what the SDK said and leaves restoring to the customer. + project_file.write_text(project_xml.replace('Version="2.0.0"', 'Version="9.9.9"'), encoding="utf-8") + for verb, flags in ( + ("install", []), ("install", ["--dry-run"]), ("install", ["-i"]), + ("list", []), ("uninstall", ["--stale"]), + ): + with self.subTest(verb=verb, flags=flags): + result = self.cli(verb, *flags, expected=1) + self.assertIn("package' failed with exit code", result.stderr) + self.assertIn("Restore failed", result.stderr) + self.assertIn("and then run this command again", result.stderr) + self.assertNotIn('"problems"', result.stderr) + self.assertEqual("", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + def add_shared_skills(self): + for package, version in (("demo.alpha", "1.0.0"), ("demo.beta", "2.0.0")): + directory = self.cache / package / version / "skills" / "shared-skill" + directory.mkdir() + (directory / "SKILL.md").write_text( + f"---\ndescription: {package} owns this guidance.\n---\n", encoding="utf-8") + + def shared_owner(self): + return next( + package for package, entry in self.manifest_packages().items() + if "shared-skill" in entry["skills"] + ) + + def test_package_filtered_picker_cannot_offer_or_remove_another_packages_shared_name(self): + self.add_shared_skills() + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + before = snapshot(self.destination) + terminal = self.terminal(packages=["Demo.Beta@2.0.0"]).ready() + self.assertIn("0 of 10 selected", terminal.compact) + terminal.press(END, lambda: "beta-10" in terminal.focused, "last eligible Beta skill") + self.assertNotIn("shared-skill", terminal.text) + terminal.press("c") + self.assertEqual(0, terminal.finish()) + self.assertEqual(before, snapshot(self.destination)) + self.assertEqual("demo.alpha", self.shared_owner()) + self.assertIn("Warning: skipped 1 colliding skill", terminal.text) + + def test_target_collision_preserves_ownership_until_explicit_uninstall(self): + self.add_shared_skills() + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + shared = (self.destination / "shared-skill" / "SKILL.md").read_bytes() + project_file, project_xml, _, restore = self.prepare_project() + project_file.write_text( + project_xml.replace('', ""), + encoding="utf-8", + ) + restore() + + result = self.cli("install").stdout + self.assertIn("Warning: skipped 1 colliding skill:", result) + self.assertIn(" shared-skill (Demo.Beta 2.0.0)", result) + self.assertIn("managed for demo.alpha 1.0.0", result) + self.assertIn( + "16 installed skills belong to a package that the target no longer references:", result) + self.assertNotIn("Removed", result) + self.assertEqual("demo.alpha", self.shared_owner()) + self.assertEqual(shared, (self.destination / "shared-skill" / "SKILL.md").read_bytes()) + + self.cli("uninstall", "--package", "Demo.Alpha") + self.cli("install") + self.assertEqual("demo.beta", self.shared_owner()) + + def test_blank_and_missing_uninstall_filters_cannot_broaden_removal(self): + self.cli("install") + before = snapshot(self.destination) + for value in ("", " ", "\t"): + for flags in ([], ["--dry-run"], ["-i"]): + with self.subTest(value=value, flags=flags): + result = self.cli("uninstall", "--package", value, *flags, expected=1) + self.assertIn("non-empty package ID", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + for flags in ([], ["--dry-run"], ["-i"]): + result = self.cli("uninstall", "--package", *flags, expected=1) + self.assertTrue(result.stderr.strip()) + self.assertEqual(before, snapshot(self.destination)) + for repeated in ("--package", "-p", "--package=", "--package=Demo.Alpha"): + result = self.cli( + "uninstall", "--package", "Demo.Alpha", repeated, expected=1) + self.assertTrue(result.stderr.strip()) + self.assertEqual(before, snapshot(self.destination)) + + def test_existing_ownership_data_cannot_be_missing_duplicated_or_ambiguous(self): + self.cli("install") + original = self.manifest.read_text(encoding="utf-8") + duplicate = json.loads(original) + duplicate["packages"]["other.owner"] = {"version": "1.0.0", "skills": ["ALPHA-01"]} + for damaged in ( + "{}", + '{"version":1,"packages":{},"Packages":{}}', + json.dumps(duplicate), + original.replace('"demo.beta"', '"Demo.Alpha"'), + '{"installed":[{"package":"Demo.Alpha","version":"1.0.0","skills":["alpha-01"]}]}', + ): + self.manifest.write_text(damaged, encoding="utf-8") + before = snapshot(self.destination) + for verb, flags in ( + ("install", []), ("install", ["-i"]), ("install", ["--dry-run"]), + ("uninstall", ["--package", "Demo.Alpha"]), ("uninstall", ["-i"]), + ): + with self.subTest(damaged=damaged, verb=verb, flags=flags): + result = self.cli(verb, *flags, expected=1) + self.assertIn("Could not read the install manifest", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + self.cli("list") + + def test_unsafe_manifest_names_fail_before_prompting_or_writing(self): + self.prepare_project() + self.cli("install") + original = self.manifest.read_text(encoding="utf-8") + for unsafe_name in ("...", ".. ", "team-owned.", "team-owned "): + damaged = json.loads(original) + damaged["packages"]["unsafe.owner"] = {"version": "1.0.0", "skills": [unsafe_name]} + self.manifest.write_text(json.dumps(damaged), encoding="utf-8") + before = snapshot(self.destination) + for verb in ("install", "uninstall"): + for flags in ([], ["--dry-run"], ["-i"], ["-i", "--dry-run"]): + with self.subTest(unsafe_name=unsafe_name, verb=verb, flags=flags): + result = self.cli(verb, *flags, expected=1) + self.assertEqual("", result.stdout) + self.assertIn("not a safe skill folder name", result.stderr) + self.assertIn("No skills were changed", result.stderr) + self.assertIn("preserved", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + self.cli("list") + self.assertEqual(before, snapshot(self.destination)) + + def test_current_owner_refresh_is_not_blocked_by_an_earlier_named_collision(self): + self.add_shared_skills() + self.cli("install", packages=["Demo.Beta@2.0.0"]) + owned_source = self.cache / "demo.beta" / "2.0.0" / "skills" / "shared-skill" / "SKILL.md" + owned_source.write_text("---\ndescription: Updated owner guidance.\n---\n", encoding="utf-8") + self.prepare_project() + + terminal = self.terminal().ready() + self.assertIn("0 of 15 selected", terminal.compact) + terminal.press(END, lambda: "alpha-15" in terminal.focused, "last skill that isn't installed") + self.assertNotIn("shared-skill", terminal.text) + self.assertEqual(0, terminal.finish(ESC)) + self.assertNotIn("Updated owner guidance.", (self.destination / "shared-skill" / "SKILL.md").read_text()) + + result = self.cli("install").stdout + self.assertIn(" shared-skill (Demo.Beta 2.0.0)", result) + self.assertIn(" shared-skill (Demo.Alpha 1.0.0)", result.split("Warning: skipped")[1]) + self.assertIn("which belongs to the current owner", result) + self.assertEqual("demo.beta", self.shared_owner()) + self.assertIn("Updated owner guidance.", (self.destination / "shared-skill" / "SKILL.md").read_text()) + + def test_repeated_equivalent_coordinates_do_not_self_collide(self): + result = self.cli( + "install", packages=["Demo.Alpha@1.0", "demo.alpha@1.0.0", "Demo.Alpha@1.0.0.0"]).stdout + self.assertIn("Scanned 1 package (named explicitly).", result) + self.assertIn("Copied 15 skills:", result) + self.assertNotIn("Warning", result) + self.assertEqual(["demo.alpha"], list(self.manifest_packages())) + self.assertEqual("1.0.0", self.manifest_packages()["demo.alpha"]["version"]) + + def test_version_change_removes_dropped_skills_even_when_the_folder_is_already_gone(self): + self.cli("install", packages=["Demo.Beta@2.0.0"]) + for number in range(1, 10): + directory = self.cache / "demo.beta" / "3.0.0" / "skills" / f"beta-{number:02}" + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text( + f"---\ndescription: beta-{number:02} for 3.0.\n---\n", encoding="utf-8") + target = self.destination / "beta-10" + for file in target.iterdir(): + file.unlink() + target.rmdir() + + result = self.cli("install", packages=["Demo.Beta@3.0.0"]).stdout + self.assertIn("Copied 9 skills:", result) + self.assertIn("Removed 1 skill:\n beta-10 (demo.beta 2.0.0)", result) + self.assertEqual( + {"demo.beta": {"version": "3.0.0", "skills": [f"beta-{n:02}" for n in range(1, 10)]}}, + self.manifest_packages(), + ) + self.assertIn("for 3.0.", (self.destination / "beta-01" / "SKILL.md").read_text(encoding="utf-8")) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").exists()) + + def test_interactive_install_cannot_change_an_installed_package_version(self): + self.cli("install", packages=["Demo.Beta@2.0.0"]) + newer = self.cache / "demo.beta" / "3.0.0" / "skills" / "beta-11" + newer.mkdir(parents=True) + (newer / "SKILL.md").write_text("---\ndescription: Only in 3.0.\n---\n", encoding="utf-8") + before = snapshot(self.destination) + + terminal = self.terminal(packages=["Demo.Beta@3.0.0"]) + self.assertEqual(1, terminal.exit_status()) + self.assertNotIn("Which skills should", terminal.text) + self.assert_shown("Demo.Beta 2.0.0 is already installed", terminal) + self.assert_shown(f"Run '{self.suggested('--package Demo.Beta')}' first", terminal) + self.assertEqual(before, snapshot(self.destination)) + + self.cli("uninstall", "--package", "Demo.Beta") + terminal = self.terminal(packages=["Demo.Beta@3.0.0"]).ready() + self.assertIn("beta-11", terminal.focused) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "the only 3.0 skill") + self.assertEqual(0, terminal.finish()) + self.assertEqual({"demo.beta": {"version": "3.0.0", "skills": ["beta-11"]}}, self.manifest_packages()) + + def test_a_version_change_that_would_hand_a_skill_to_another_package_stops(self): + self.add_shared_skills() + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + newer = self.cache / "demo.alpha" / "1.1.0" / "skills" / "alpha-01" + newer.mkdir(parents=True) + (newer / "SKILL.md").write_text("---\ndescription: Alpha 1.1 drops shared-skill.\n---\n", encoding="utf-8") + before = snapshot(self.destination) + packages = ["Demo.Alpha@1.1.0", "Demo.Beta@2.0.0"] + for flags in ([], ["--dry-run"]): + with self.subTest(flags=flags): + result = self.cli("install", *flags, packages=packages, expected=1) + self.assertIn( + "Demo.Alpha 1.1.0 no longer ships the installed skill 'shared-skill', " + "and Demo.Beta 2.0.0 ships a skill with that name", result.stderr) + self.assertIn(f"Run '{self.suggested('--package Demo.Alpha')}' first", result.stderr) + self.assertEqual("", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + self.cli("uninstall", "--package", "Demo.Alpha") + self.cli("install", packages=packages) + self.assertEqual("demo.beta", self.shared_owner()) + self.assertEqual({"version": "1.1.0", "skills": ["alpha-01"]}, self.manifest_packages()["demo.alpha"]) + + def test_a_package_id_with_letters_outside_ascii_survives_later_commands(self): + source = self.cache / "contoso.überlib" / "1.0.0" / "skills" / "uber-usage" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("---\ndescription: Unicode package id.\n---\n", encoding="utf-8") + self.cli("install", packages=["Contoso.Überlib@1.0.0"]) + self.assertEqual({"contoso.überlib": {"version": "1.0.0", "skills": ["uber-usage"]}}, self.manifest_packages()) + self.cli("install", packages=["Contoso.Überlib@1.0.0"]) + self.cli("uninstall", "--package", "Contoso.Überlib", "--dry-run") + self.cli("uninstall", "--package", "Contoso.Überlib") + self.assertFalse(self.manifest.exists()) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").exists()) + + def test_install_stops_when_packages_resolve_to_more_than_one_version(self): + newer = self.cache / "demo.alpha" / "1.1.0" / "skills" / "alpha-01" + newer.mkdir(parents=True) + (newer / "SKILL.md").write_text("---\ndescription: Alpha 1.1.\n---\n", encoding="utf-8") + before = snapshot(self.destination) + for flags in ([], ["--dry-run"], ["-i"]): + with self.subTest(mode="package", flags=flags): + result = self.cli( + "install", *flags, packages=["Demo.Alpha@1.0.0", "Demo.Alpha@1.1.0"], expected=1) + self.assertIn("--package names more than one version", result.stderr) + self.assertIn("Demo.Alpha (1.0.0, 1.1.0)", result.stderr) + self.assertEqual("", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + project_file, project_xml, beta_reference, restore = self.prepare_project() + self.cli("install") + installed = snapshot(self.destination) + # In its own folder, so the two restores don't share obj\project.assets.json. + other = project_file.parent / "Other" / "Other.csproj" + other.parent.mkdir() + other.write_text( + project_xml.replace(beta_reference, "").replace('Version="1.0.0"', 'Version="1.1.0"'), + encoding="utf-8", + ) + self.target.write_text( + '', + encoding="utf-8", + ) + restore() + for flags in ([], ["--dry-run"], ["-i"]): + with self.subTest(mode="target", flags=flags): + result = self.cli("install", *flags, expected=1) + self.assertIn("resolve to more than one version", result.stderr) + self.assertIn("Demo.Alpha (1.0.0, 1.1.0)", result.stderr) + self.assertIn("Central Package Management", result.stderr) + self.assertEqual("", result.stdout) + self.assertEqual(installed, snapshot(self.destination)) + self.assertIn(" alpha-01 (Demo.Alpha 1.1.0)", self.cli("list").stdout) + self.assertIn("No stale skills were found.", self.cli("uninstall", "--stale").stdout) + self.assertEqual(installed, snapshot(self.destination)) + + def test_valid_underscore_ids_work_for_coordinate_and_uninstall_filters(self): + for package in ("_Acme", "Acme_"): + source = self.cache / package.lower() / "1.0.0" / "skills" / "underscore-example" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("---\ndescription: Underscore package.\n---\n", encoding="utf-8") + self.cli("install", packages=[f"{package}@1.0.0"]) + terminal = self.terminal("uninstall", "--package", package).ready() + terminal.press(SPACE) + self.assertEqual(0, terminal.finish()) + self.assertEqual(set(), self.installed()) + + def test_interactive_uninstall_matches_the_same_normalized_version_as_noninteractive(self): + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + before = snapshot(self.destination) + preview = self.cli("uninstall", "--package", "demo.alpha@1.0", "--dry-run").stdout + self.assertIn("Would remove 15 skills:", preview) + + terminal = self.terminal("uninstall", "--package", "demo.alpha@1.0", "--dry-run").ready() + self.assertIn("0 of 15 selected", terminal.compact) + terminal.press("a", lambda: "15 to remove" in terminal.compact, "matching normalized version") + self.assertEqual(0, terminal.finish()) + self.assertEqual(before, snapshot(self.destination)) + + terminal = self.terminal("uninstall", "--package", "Demo.Alpha@1.0.0.0").ready() + terminal.press("a") + self.assertEqual(0, terminal.finish()) + self.assertEqual(set(), self.installed()) + + def test_ownership_changes_while_a_picker_is_open_are_rejected(self): + self.add_shared_skills() + for verb in ("install", "uninstall"): + with self.subTest(verb=verb): + self.cli("uninstall") + if verb == "uninstall": + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + terminal = self.terminal(verb, packages=["Demo.Alpha@1.0.0"]).ready() + terminal.press(END, lambda: "shared-skill" in terminal.focused, "reviewed owner") + terminal.press(SPACE) + self.cli("uninstall") + self.cli("install", packages=["Demo.Beta@2.0.0"]) + before = snapshot(self.destination) + self.assertEqual(1, terminal.finish()) + self.assertIn("ownership changed", terminal.compact) + self.assertEqual(before, snapshot(self.destination)) + + def test_destination_writes_are_serialized_across_processes(self): + self.cli("install") + for verb in ("install", "uninstall"): + with self.subTest(verb=verb): + before = snapshot(self.destination) + process = None + try: + command = self.command(verb) + if verb == "uninstall": + command[command.index("--destination") + 1] = ( + "\\\\?\\" + str(self.destination) if sys.platform == "win32" + else str(self.destination / ".." / self.destination.name) + ) + with destination_mutex(self.destination): + process = subprocess.Popen( + command, cwd=self.root, env=self.environment, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", + ) + with self.assertRaises(subprocess.TimeoutExpired): + process.wait(timeout=1) + self.assertEqual(before, snapshot(self.destination)) + stdout, stderr = process.communicate(timeout=30) + self.assertEqual(0, process.returncode, stdout + stderr) + self.assertIn("Copied 25 skills:" if verb == "install" else "Removed 25 skills:", stdout) + finally: + if process is not None and process.poll() is None: + process.kill() + process.communicate(timeout=10) + self.assertEqual(set(), self.installed()) + + def test_a_destination_alias_removed_while_waiting_invalidates_the_operation(self): + self.cli("install") + before = snapshot(self.destination) + alias = self.root / "destination alias" + if sys.platform == "win32": + result = subprocess.run( + ["powershell.exe", "-NoLogo", "-NoProfile", "-Command", + f"New-Item -ItemType Junction -Path '{alias}' -Target '{self.destination}' | Out-Null"], + capture_output=True, encoding="utf-8", timeout=15, + ) + self.assertEqual(0, result.returncode, result.stderr) + else: + alias.symlink_to(self.destination, target_is_directory=True) + process = None + try: + command = self.command("install") + command[command.index("--destination") + 1] = str(alias) + with destination_mutex(self.destination): + process = subprocess.Popen( + command, cwd=self.root, env=self.environment, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8", + ) + with self.assertRaises(subprocess.TimeoutExpired): + process.wait(timeout=1) + if sys.platform == "win32": + alias.rmdir() + else: + alias.unlink() + stdout, stderr = process.communicate(timeout=30) + self.assertEqual(1, process.returncode, stdout + stderr) + self.assertIn("changed while waiting", stderr) + self.assertFalse(alias.exists()) + self.assertEqual(before, snapshot(self.destination)) + finally: + if process is not None and process.poll() is None: + process.kill() + process.communicate(timeout=10) + if alias.exists() or alias.is_symlink(): + if sys.platform == "win32": + alias.rmdir() + else: + alias.unlink() + + def test_long_destination_paths_remain_manageable_after_the_first_install(self): + destination = self.root + for _ in range(5): + destination /= "a" * 60 + self.destination = destination / "skills" + self.manifest = self.destination / ".dotnet-package-skills.json" + self.assertGreater(len(str(self.destination)), 260) + + self.cli("install") + self.assertEqual(set(self.names), self.installed()) + self.cli("install") + before = snapshot(self.destination) + self.cli("uninstall", "--dry-run") + self.assertEqual(before, snapshot(self.destination)) + self.cli("uninstall") + self.assertFalse(self.manifest.exists()) + + def test_aspire_hint_descriptions_pagination_and_action_colors(self): + terminal = self.terminal().ready() + self.assert_frame(terminal) + self.assertIn("ALPHA-FIRST", terminal.text) + rows = terminal.screen.display + first_line = next(index for index, line in enumerate(rows) if "Guidance for alpha-02" in line) + description_column = rows[first_line].index("Guidance") + self.assertEqual( + 6, + len(rows[first_line + 1]) - len(rows[first_line + 1].lstrip()), + ) + self.assertEqual( + "Guidance for alpha-02: configure services and verify expected behavior. Keep changes predictable.", + rows[first_line][description_column:].strip() + + " " + rows[first_line + 1][6:].strip(), + ) + pages = re.search(r"page\s+1\s+of\s+(\d+)", terminal.compact, re.IGNORECASE) + self.assertIsNotNone(pages, terminal.text) + self.assertGreaterEqual(int(pages[1]), 3) + self.assertEqual({"brightblue"}, set(terminal.colors(">"))) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "pending install") + self.assertEqual({"brightblue"}, set(terminal.colors("alpha-01"))) + self.assertEqual({"brightblue"}, set(terminal.colors("[X]"))) + self.assertEqual({"brightblue"}, set(terminal.colors(">"))) + self.assertEqual({"brightblue"}, set(terminal.colors("ALPHA-FIRST"))) + terminal.press(DOWN, lambda: "alpha-02" in terminal.focused, "move focus off the checked skill") + self.assertNotIn("brightblue", terminal.colors("alpha-01")) + self.assertEqual("brightblue", terminal.colors("[X]")[1]) + self.assertNotIn("brightblue", terminal.colors("ALPHA-FIRST")) + terminal.press(UP, lambda: "alpha-01" in terminal.focused, "return to checked skill") + terminal.press(SPACE, lambda: "[ ]" in terminal.focused, "undo install") + self.assertNotIn("brightgreen", terminal.colors("alpha-01")) + self.assertEqual(0, terminal.finish(ESC)) + self.assertFalse(self.manifest.exists()) + + def test_focus_colors_all_wrapped_lines_and_checked_mark_stays_blue_off_focus(self): + terminal = self.terminal().ready() + terminal.press(DOWN, lambda: "alpha-02" in terminal.focused, "focus multiline skill") + first = next(i for i, line in enumerate(terminal.screen.display) if "alpha-02 - " in line) + self.assertTrue(terminal.screen.display[first + 1].strip()) + for row in (first, first + 1): + self.assertEqual( + {"brightblue"}, + {cell.fg for cell in terminal.screen.buffer[row].values() if cell.data.strip()}, + ) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "check multiline skill") + terminal.press(DOWN, lambda: "alpha-03" in terminal.focused, "move off multiline skill") + self.assertNotIn("brightblue", terminal.colors("alpha-02")) + self.assertNotIn("brightblue", terminal.colors("Guidance for alpha-02")) + self.assertEqual("brightblue", terminal.colors("[X]")[1]) + self.assertEqual(0, terminal.finish(ESC)) + + def test_each_name_is_followed_directly_by_the_description_without_package_metadata(self): + self.cache = self.root / "compact cache" + self.environment["NUGET_PACKAGES"] = str(self.cache) + names = ["demo.alpha-longer-name", "demo.alpha-short"] + for name in names: + directory = self.cache / "demo.alpha" / "1.0.0" / "skills" / name + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text( + f"---\ndescription: Guidance for {name}.\n---\n", encoding="utf-8") + + for verb in ("install", "uninstall"): + if verb == "uninstall": + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + before = snapshot(self.destination) + for no_color in (False, True): + with self.subTest(verb=verb, no_color=no_color): + terminal = self.terminal( + verb, packages=["Demo.Alpha@1.0.0"], no_color=no_color, + ).ready() + for name in names: + self.assertIn(f"{name} - Guidance for {name}.", terminal.text) + self.assertNotIn("(Demo.Alpha 1.0.0)", terminal.text) + self.assertIn(ASPIRE_HINT, terminal.compact) + self.assertEqual(0, terminal.finish(ESC)) + self.assertEqual(before, snapshot(self.destination)) + + def test_select_and_clear_apply_across_all_pages(self): + terminal = self.terminal().ready() + terminal.press("a", lambda: "25 of 25" in terminal.compact, "select all pages") + terminal.press(END, lambda: "beta-10" in terminal.focused, "last skill") + self.assertIn("[X]", terminal.focused) + terminal.press("c", lambda: "0 of 25" in terminal.compact, "clear all pages") + terminal.press(HOME, lambda: "alpha-01" in terminal.focused, "first skill") + self.assertIn("[ ]", terminal.focused) + self.assertEqual(0, terminal.finish()) + self.assertEqual(set(), self.installed()) + + def test_page_keys_home_end_and_wraparound_keep_a_complete_frame(self): + terminal = self.terminal().ready() + terminal.press(END, lambda: "beta-10" in terminal.focused, "last page") + self.assert_frame(terminal) + self.assertNotIn("alpha-01", terminal.text) + terminal.press(DOWN, lambda: "alpha-01" in terminal.focused, "wrap to first") + terminal.press(UP, lambda: "beta-10" in terminal.focused, "wrap to last") + terminal.press(HOME, lambda: "alpha-01" in terminal.focused, "home") + for forward, backward in ((RIGHT, LEFT), (PAGE_DOWN, PAGE_UP)): + terminal.press(forward, lambda: "page 2 of" in terminal.compact.lower(), "next page") + self.assert_frame(terminal) + terminal.press(backward, lambda: "page 1 of" in terminal.compact.lower(), "previous page") + self.assert_frame(terminal) + self.assertEqual(0, terminal.finish(ESC)) + + def test_confirmation_only_installs_checked_skills(self): + terminal = self.terminal().ready() + terminal.press(SPACE) + terminal.press(DOWN, lambda: "alpha-02" in terminal.focused, "second skill") + terminal.press(SPACE) + self.assertEqual(0, terminal.finish()) + self.assertEqual({"alpha-01", "alpha-02"}, self.installed()) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").exists()) + + def test_reopening_lists_only_skills_that_are_not_installed(self): + self.cli("install", packages=["Demo.Alpha@1.0.0"]) + terminal = self.terminal().ready() + self.assertIn("Installed skills aren't listed.", terminal.compact) + self.assertIn("0 of 10 selected", terminal.compact) + self.assertIn("beta-01", terminal.focused) + self.assertIn("[ ]", terminal.focused) + self.assertNotIn("alpha-", terminal.text) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "pending installation") + terminal.press(SPACE, lambda: "[ ]" in terminal.focused, "undo") + self.assertNotIn("brightred", terminal.colors("[ ]")) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "install beta-01") + self.assertNotIn("brightred", terminal.colors("[X]")) + self.assertEqual(0, terminal.finish()) + self.assertEqual({f"alpha-{n:02}" for n in range(1, 16)} | {"beta-01"}, self.installed()) + + self.cli("install") + before = snapshot(self.destination) + terminal = self.terminal() + self.assertEqual(0, terminal.exit_status()) + self.assertNotIn("Which skills should", terminal.text) + self.assert_shown("Nothing new to install. Every skill that these packages ship is already installed.", terminal) + self.assertEqual(before, snapshot(self.destination)) + + def test_package_install_never_removes_skills_of_packages_it_does_not_name(self): + self.cli("install", packages=["Demo.Beta@2.0.0"]) + beta = {f"beta-{number:02}" for number in range(1, 11)} + terminal = self.terminal(packages=["Demo.Alpha@1.0.0"]).ready() + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "select one Alpha skill") + self.assertEqual(0, terminal.finish()) + self.assertEqual(beta | {"alpha-01"}, self.installed()) + result = self.cli("install", packages=["Demo.Alpha@1.0.0"]).stdout + self.assertEqual(set(self.names), self.installed()) + self.assertNotIn("Removed", result) + self.assertNotIn("uninstall --stale", result) + + def test_uninstall_reads_installed_descriptions_without_the_package_cache(self): + self.cli("install", packages=["Demo.Beta@2.0.0"]) + (self.destination / "beta-01" / "SKILL.md").write_text( + "---\ndescription: INSTALLED-COPY-ONLY\n---\n", encoding="utf-8") + self.cache.rename(self.root / "cache moved aside") + terminal = self.terminal("uninstall").ready() + self.assertIn("INSTALLED-COPY-ONLY", terminal.text) + self.assertNotIn("team-owned", terminal.text) + self.assertNotIn("alpha-", terminal.text) + self.assertIn("0 of 10", terminal.compact) + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "select removal") + self.assertEqual({"brightblue"}, set(terminal.colors("[X]"))) + self.assertEqual({"brightblue"}, set(terminal.colors("beta-01"))) + self.assertIn("Blue X: selected", terminal.text) + terminal.press(DOWN, lambda: "beta-02" in terminal.focused, "leave the removal checked") + self.assertNotIn("brightblue", terminal.colors("beta-01")) + self.assertEqual(["default", "brightblue", "default"], terminal.colors("[X]")) + self.assertNotIn( + "brightred", + {cell.fg for line in terminal.screen.buffer.values() for cell in line.values()}) + self.assertEqual(0, terminal.finish()) + self.assertEqual({f"beta-{number:02}" for number in range(2, 11)}, self.installed()) + + def test_missing_and_invalid_descriptions_are_visible_and_selectable(self): + terminal = self.terminal().ready() + terminal.focus("alpha-03", 2) + self.assertIn("No description provided.", terminal.compact) + terminal.press(SPACE) + terminal.focus("alpha-04", 3) + self.assertIn("Description unavailable", terminal.compact) + terminal.press(SPACE) + self.assertEqual(0, terminal.finish()) + self.assertEqual({"alpha-03", "alpha-04"}, self.installed()) + + def test_unicode_descriptions_are_readable_and_cannot_send_terminal_controls(self): + description = "Café, 測試, 🧪, Cafe\u0301. \x1b[2JUNICODE-END" + path = self.cache / "demo.alpha" / "1.0.0" / "skills" / "alpha-01" / "SKILL.md" + path.write_text( + f"---\ndescription: {json.dumps(description)}\n---\n", + encoding="utf-8", + ) + before = snapshot(self.destination) + terminal = self.terminal(columns=120).ready() + self.assert_frame(terminal) + self.assertIn("Café", unicodedata.normalize("NFC", terminal.text)) + self.assertIn("測試", terminal.text) + self.assertIn("🧪", terminal.text) + self.assertIn("UNICODE-END", terminal.text) + self.assertEqual(0, terminal.finish(ESC)) + self.assertEqual(before, snapshot(self.destination)) + + def test_oversized_description_can_be_scrolled_to_its_end(self): + terminal = self.terminal(rows=18, columns=80).ready() + terminal.focus("alpha-05", 4) + self.assertIn("LONG-DESCRIPTION-START", terminal.compact) + self.assertIn("scroll", terminal.compact.lower()) + for scroll in range(250): + if "LONG-DESCRIPTION-END" in terminal.compact: + break + previous = terminal.text + terminal.press( + CTRL_DOWN, + lambda: terminal.text != previous, + f"description scroll {scroll}", + ) + self.assertIn("LONG-DESCRIPTION-END", terminal.compact) + self.assertIn("0 of 25", terminal.compact) + self.assertEqual(0, terminal.finish(ESC)) + + def test_resize_preserves_focus_selection_and_keyboard_hints(self): + terminal = self.terminal(rows=32, columns=120).ready() + terminal.press(SPACE) + terminal.focus("alpha-02", 1) + initial_pages = terminal.page_count + terminal.resize(18, 46) + terminal.wait_for( + lambda: terminal.page_count > initial_pages + and "alpha-02" in terminal.focused and ASPIRE_HINT in terminal.compact, + "narrow resize without keyboard input", + ) + self.assert_frame(terminal) + terminal.press(DOWN, lambda: "alpha-03" in terminal.focused, "narrow resized frame") + self.assert_frame(terminal) + self.assertIn("1 of 25", terminal.compact) + narrow_pages = terminal.page_count + terminal.resize(40, 140) + terminal.wait_for( + lambda: terminal.page_count < narrow_pages + and "alpha-03" in terminal.focused and ASPIRE_HINT in terminal.compact, + "wide resize without keyboard input", + ) + self.assert_frame(terminal) + terminal.press(HOME, lambda: "alpha-01" in terminal.focused, "wide resized frame") + self.assert_frame(terminal) + self.assertIn("[X]", terminal.focused) + self.assertEqual(0, terminal.finish()) + self.assertEqual({"alpha-01"}, self.installed()) + + def test_rapid_live_resizes_do_not_leave_old_picker_frames_in_scrollback(self): + before = snapshot(self.destination) + for shell_output in (False, True): + with self.subTest(shell_output=shell_output): + terminal = self.terminal(rows=70, columns=210, shell_output=shell_output) + terminal.screen = pyte.HistoryScreen(210, 70, history=2000) + terminal.stream = pyte.Stream(terminal.screen) + terminal.ready() + terminal.press(SPACE, lambda: "1 of 25" in terminal.compact, "selected before resizing") + for cycle in range(3): + for columns, rows in ((170, 60), (130, 45), (80, 24), (110, 40), (210, 70)): + terminal.resize(rows, columns) + time.sleep(0.006) + terminal.pump(timeout=0.01) + terminal.resize(30, 100) + terminal.wait_for( + lambda: terminal.page_count > 1 and "alpha-01" in terminal.focused + and "1 of 25" in terminal.compact and ASPIRE_HINT in terminal.compact, + f"settled rapid resize {cycle}", + ) + self.assert_frame(terminal) + history = "\n".join( + "".join(cell.data for _, cell in sorted(row.items())) + for row in terminal.screen.history.top + ) + self.assertNotIn("Which skills should be installed?", history) + if shell_output: + self.assertIn("earlier console output", history) + self.assertEqual(0, terminal.finish(ESC)) + self.assertEqual(before, snapshot(self.destination)) + + def test_first_picker_frame_starts_at_the_top_after_prior_shell_output(self): + coordinate = "Demo.Compact@1.0.0" + for name in ("compact-01", "compact-02", "compact-03"): + directory = self.cache / "demo.compact" / "1.0.0" / "skills" / name + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text( + "---\ndescription: Short guidance for a compact checklist.\n---\n", + encoding="utf-8", + ) + for verb in ("install", "uninstall"): + if verb == "uninstall": + self.cli("install", packages=[coordinate]) + for no_color in (False, True): + with self.subTest(verb=verb, no_color=no_color): + before = snapshot(self.destination) + terminal = self.terminal( + verb, "--dry-run", packages=[coordinate], rows=50, columns=140, + no_color=no_color, shell_output=True, + ) + terminal.emulator = ReflowEmulator(50, 140) + terminal.ready() + state = terminal.emulator.state + self.assertTrue(state["alternate"]) + self.assertTrue(state["active"].startswith("Which skills should"), state["active"]) + self.assertNotIn("Which skills should", state["normal"]) + self.assertIn("earlier console output", state["normal"]) + self.assertEqual(0, terminal.finish(ESC)) + self.assertFalse(terminal.emulator.state["alternate"]) + self.assertIn("earlier console output", terminal.emulator.state["normal"]) + self.assertIn("Cancelled.", terminal.emulator.state["normal"]) + self.assertEqual(before, snapshot(self.destination)) + + def test_resize_reflow_does_not_leave_duplicate_picker_frames_in_normal_scrollback(self): + before = snapshot(self.destination) + terminal = self.terminal(rows=70, columns=210, shell_output=True) + terminal.emulator = ReflowEmulator(70, 210) + terminal.ready() + terminal.press(SPACE, lambda: "1 of 25" in terminal.compact, "selected before host reflow") + terminal.resize(24, 80) + terminal.wait_for( + lambda: ASPIRE_HINT in terminal.compact and "1 of 25" in terminal.compact + and terminal.page_count > 1, + "host-reflowed narrow picker", + ) + + state = terminal.emulator.state + self.assertNotIn("Which skills should be installed?", state["normal"]) + self.assertTrue(state["alternate"]) + self.assertEqual(1, state["active"].count("Which skills should be installed?")) + terminal.resize(50, 160) + terminal.wait_for(lambda: ASPIRE_HINT in terminal.compact and "[X]" in terminal.focused, "grown picker") + self.assertNotIn("Which skills should be installed?", terminal.emulator.state["normal"]) + self.assertEqual(0, terminal.finish(ESC)) + state = terminal.emulator.state + self.assertFalse(state["alternate"]) + self.assertNotIn("Which skills should be installed?", state["normal"]) + self.assertIn("earlier console output", state["normal"]) + self.assertIn("Cancelled.", state["normal"]) + self.assertEqual(before, snapshot(self.destination)) + + def test_interactive_screen_restores_shell_output_on_accept_cancel_and_interrupt(self): + for verb in ("install", "uninstall"): + if verb == "uninstall": + self.cli("install") + for key in (ENTER, ESC, CTRL_C): + with self.subTest(verb=verb, key=repr(key)): + before = snapshot(self.destination) + terminal = self.terminal(verb, "--dry-run", rows=32, columns=100, shell_output=True) + terminal.emulator = ReflowEmulator(32, 100) + terminal.ready() + self.assertTrue(terminal.emulator.state["alternate"]) + terminal.press("a") + self.assertEqual(0, terminal.finish(key)) + state = terminal.emulator.state + self.assertFalse(state["alternate"]) + self.assertIn("earlier console output", state["normal"]) + self.assertNotIn("Which skills should", state["normal"]) + self.assertIn("Cancelled." if key != ENTER else "Would", state["normal"]) + self.assertEqual(before, snapshot(self.destination)) + + def test_no_color_mode_shows_ticks_without_markers_or_color(self): + terminal = self.terminal(no_color=True).ready() + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "no-color installation") + self.assertTrue(terminal.focused.startswith("> [X] alpha-01 - "), terminal.focused) + self.assertEqual({"default"}, set(terminal.colors("alpha-01"))) + self.assertNotIn("brightblue", terminal.colors(">")) + self.assertNotIn("Blue X", terminal.text) + self.assertEqual(0, terminal.finish()) + terminal = self.terminal("uninstall", no_color=True).ready() + terminal.press(SPACE, lambda: "[X]" in terminal.focused, "no-color removal") + self.assertTrue(terminal.focused.startswith("> [X] alpha-01 - "), terminal.focused) + self.assertNotIn("Blue X", terminal.text) + self.assertEqual(0, terminal.finish()) + self.assertEqual(set(), self.installed()) + + def test_cancel_keys_preserve_files_and_restore_the_cursor(self): + for verb, pending in (("install", "25 of 25"), ("uninstall", "25 to remove")): + if verb == "uninstall": + self.cli("install") + before = snapshot(self.destination) + for key in (ESC, "q", CTRL_C): + with self.subTest(verb=verb, key=repr(key)): + terminal = self.terminal(verb).ready() + terminal.press("a", lambda: pending in terminal.compact, f"pending {verb} of all") + self.assertEqual(0, terminal.finish(key)) + self.assertEqual(before, snapshot(self.destination)) + self.assertFalse(terminal.screen.cursor.hidden) + self.assertNotIn(terminal.screen.cursor.attrs.fg, {"brightred", "brightgreen", "brightblue"}) + + def test_interactive_dry_runs_change_no_files(self): + before = snapshot(self.destination) + terminal = self.terminal("install", "--dry-run").ready() + terminal.press("a") + self.assertEqual(0, terminal.finish()) + self.assertEqual(before, snapshot(self.destination)) + self.cli("install") + before = snapshot(self.destination) + terminal = self.terminal("uninstall", "--dry-run").ready() + terminal.press("a") + self.assertEqual(0, terminal.finish()) + self.assertEqual(before, snapshot(self.destination)) + + def test_corrupt_manifest_still_fails_before_prompting_and_preserves_bytes(self): + self.cli("install") + self.manifest.write_text("<<<<<<< HEAD\n{}\n=======\n{}\n>>>>>>> branch", encoding="utf-8") + before = snapshot(self.destination) + for verb in ("install", "uninstall"): + for flags in ([], ["--dry-run"], ["-i"]): + result = self.cli(verb, *flags, expected=1) + self.assertEqual("", result.stdout) + self.assertIn("Could not read the install manifest", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + self.cli("list") + self.assertEqual(before, snapshot(self.destination)) + + def test_manifest_metadata_cannot_inject_terminal_controls_into_human_reports(self): + self.add_shared_skills() + self.cli("install") + original = self.manifest.read_text(encoding="utf-8") + controls = ( + "\x1b]52;c;ZWNobyBleGFtcGxl\x07", + "\x1b]52;c;ZWNobyBleGFtcGxl\x1b\\", + "\x9d52;c;ZWNobyBleGFtcGxl\x9c", + ) + for control in controls: + damaged = json.loads(original) + damaged["packages"]["demo.alpha"]["version"] += control + self.manifest.write_text(json.dumps(damaged), encoding="utf-8") + before = snapshot(self.destination) + for verb, packages, expected in ( + ("uninstall", None, "alpha-01 (demo.alpha 1.0.0"), + ("install", ["Demo.Beta@2.0.0"], "managed for demo.alpha 1.0.0"), + ): + with self.subTest(verb=verb, control=ascii(control)): + result = self.cli(verb, "--dry-run", packages=packages) + self.assert_plain_output(result.stdout) + self.assertIn(expected, result.stdout) + self.assertEqual(before, snapshot(self.destination)) + result = self.cli("uninstall") + self.assert_plain_output(result.stdout) + self.assertIn("Removed 26 skills", result.stdout) + self.assertTrue((self.destination / "team-owned" / "SKILL.md").is_file()) + + def test_operational_errors_cannot_emit_terminal_controls(self): + before = snapshot(self.destination) + result = self.cli( + "install", packages=["Example\x1b]52;c;ZWNobyBleGFtcGxl\x07"], expected=1) + + self.assertEqual("", result.stdout) + self.assert_plain_output(result.stderr) + self.assertIn("error:", result.stderr) + self.assertIn("Id@Version", result.stderr) + self.assertEqual(before, snapshot(self.destination)) + + def test_parser_diagnostics_cannot_emit_controls_on_either_output_stream(self): + before = snapshot(self.destination) + controls = ( + "\x1b]52;c;ZWNobyBleGFtcGxl\x07", + "\x1b]52;c;ZWNobyBleGFtcGxl\x1b\\", + "\x9d52;c;ZWNobyBleGFtcGxl\x9c", + ) + for control in controls: + for arguments in ( + ["--package", "Ex" + control + "ample"], + ["--package", "Demo.Alpha@1.*" + control], + ["--unknown" + control], + ["--dry-run=false" + control], + ): + with self.subTest(arguments=ascii(arguments)): + result = self.cli("uninstall", *arguments, expected=1) + self.assert_plain_output(result.stdout) + self.assert_plain_output(result.stderr) + self.assertNotIn("ZWNobyBleGFtcGxl", result.stdout + result.stderr) + self.assertEqual(before, snapshot(self.destination)) + typo = subprocess.run( + [str(OPTIONS.tool), "uninstal\x1b"], cwd=self.root, env=self.environment, + capture_output=True, encoding="utf-8", timeout=30, + ) + self.assertEqual(1, typo.returncode) + self.assertIn("Did you mean", typo.stdout) + self.assert_plain_output(typo.stdout) + self.assert_plain_output(typo.stderr) + self.assertEqual(before, snapshot(self.destination)) + + def assert_plain_output(self, text): + self.assertFalse( + any(unicodedata.category(character) == "Cc" and character not in "\r\n" + or character in "\u202e\u2066" for character in text), + "Captured output contains unsafe terminal controls.", + ) + + def test_redirected_interactive_runs_and_json_output_are_rejected(self): + for verb in ("install", "uninstall"): + if verb == "uninstall": + self.cli("install") + result = self.cli(verb, "-i", expected=1) + self.assertIn("needs a terminal", result.stderr) + before = snapshot(self.destination) + for verb, flags in ( + ("install", ["--json"]), ("install", ["-i", "--json"]), ("list", ["--json"]), + ("uninstall", ["--json"]), ("uninstall", ["-i", "--json"]), + ): + with self.subTest(verb=verb, flags=flags): + result = self.cli(verb, *flags, expected=1) + self.assertIn("Unrecognized command or argument '--json'", result.stdout + result.stderr) + self.assertEqual(before, snapshot(self.destination)) + + def test_empty_uninstall_does_not_prompt(self): + before = snapshot(self.destination) + result = self.cli("uninstall", "-i") + self.assertIn("Nothing to remove", result.stdout) + self.assertEqual(before, snapshot(self.destination)) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--tool", type=Path, required=True) + parser.add_argument("--artifacts", type=Path, required=True) + OPTIONS, unittest_args = parser.parse_known_args() + if sys.platform not in ("win32", "darwin", "linux"): + parser.error("These real terminal tests support Windows ConPTY, macOS and Linux PTYs.") + OPTIONS.tool = OPTIONS.tool.resolve(strict=True) + if sys.platform != "win32" and not os.access(OPTIONS.tool, os.X_OK): + parser.error("--tool must name an executable native apphost, not a DLL.") + OPTIONS.artifacts = OPTIONS.artifacts.resolve() + OPTIONS.artifacts.mkdir(parents=True, exist_ok=True) + OPTIONS.mutex_holder = None + unittest.main(argv=[sys.argv[0], *unittest_args], verbosity=2)