Marco Ricci commited on 2026-07-04 21:08:08
Zeige 2 geänderte Dateien mit 244 Einfügungen und 237 Löschungen.
Move the ListKeysAction, SignAction, SocketAddressAction and SystemSupportAction classes into tests.data. We intend to reuse the SystemSupportAction class for a new purpose next.
| ... | ... |
@@ -21,7 +21,11 @@ from __future__ import annotations |
| 21 | 21 |
|
| 22 | 22 |
import base64 |
| 23 | 23 |
import copy |
| 24 |
+import ctypes |
|
| 24 | 25 |
import enum |
| 26 |
+import errno |
|
| 27 |
+import os |
|
| 28 |
+import socket |
|
| 25 | 29 |
from typing import TYPE_CHECKING, cast |
| 26 | 30 |
|
| 27 | 31 |
from typing_extensions import NamedTuple, assert_never |
| ... | ... |
@@ -33,10 +37,226 @@ __all__ = () |
| 33 | 37 |
|
| 34 | 38 |
if TYPE_CHECKING: |
| 35 | 39 |
from collections.abc import Mapping |
| 40 |
+ from typing import NoReturn |
|
| 36 | 41 |
|
| 42 |
+ import pytest |
|
| 37 | 43 |
from typing_extensions import Any |
| 38 | 44 |
|
| 39 | 45 |
|
| 46 |
+# Actions |
|
| 47 |
+# ======= |
|
| 48 |
+ |
|
| 49 |
+ |
|
| 50 |
+class ListKeysAction(str, enum.Enum): |
|
| 51 |
+ """Test fixture settings for [`ssh_agent.SSHAgentClient.list_keys`][]. |
|
| 52 |
+ |
|
| 53 |
+ Attributes: |
|
| 54 |
+ EMPTY: Return an empty key list. |
|
| 55 |
+ FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][]. |
|
| 56 |
+ FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][]. |
|
| 57 |
+ |
|
| 58 |
+ """ |
|
| 59 |
+ |
|
| 60 |
+ EMPTY = enum.auto() |
|
| 61 |
+ """""" |
|
| 62 |
+ FAIL = enum.auto() |
|
| 63 |
+ """""" |
|
| 64 |
+ FAIL_RUNTIME = enum.auto() |
|
| 65 |
+ """""" |
|
| 66 |
+ |
|
| 67 |
+ def __call__(self, *_args: Any, **_kwargs: Any) -> Any: |
|
| 68 |
+ """Execute the respective action.""" |
|
| 69 |
+ # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 70 |
+ # matching. |
|
| 71 |
+ # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 72 |
+ if self == self.EMPTY: |
|
| 73 |
+ return [] |
|
| 74 |
+ if self == self.FAIL: |
|
| 75 |
+ raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"") |
|
| 76 |
+ if self == self.FAIL_RUNTIME: |
|
| 77 |
+ raise ssh_agent.TrailingDataError() |
|
| 78 |
+ raise AssertionError() |
|
| 79 |
+ |
|
| 80 |
+ |
|
| 81 |
+class SignAction(str, enum.Enum): |
|
| 82 |
+ """Test fixture settings for [`ssh_agent.SSHAgentClient.sign`][]. |
|
| 83 |
+ |
|
| 84 |
+ Attributes: |
|
| 85 |
+ FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][]. |
|
| 86 |
+ FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][]. |
|
| 87 |
+ |
|
| 88 |
+ """ |
|
| 89 |
+ |
|
| 90 |
+ FAIL = enum.auto() |
|
| 91 |
+ """""" |
|
| 92 |
+ FAIL_RUNTIME = enum.auto() |
|
| 93 |
+ """""" |
|
| 94 |
+ |
|
| 95 |
+ def __call__(self, *_args: Any, **_kwargs: Any) -> Any: |
|
| 96 |
+ """Execute the respective action.""" |
|
| 97 |
+ # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 98 |
+ # matching. |
|
| 99 |
+ # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 100 |
+ if self == self.FAIL: |
|
| 101 |
+ raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"") |
|
| 102 |
+ if self == self.FAIL_RUNTIME: |
|
| 103 |
+ raise ssh_agent.TrailingDataError() |
|
| 104 |
+ raise AssertionError() |
|
| 105 |
+ |
|
| 106 |
+ |
|
| 107 |
+class SocketAddressAction(str, enum.Enum): |
|
| 108 |
+ """Test fixture settings for the SSH agent socket address. |
|
| 109 |
+ |
|
| 110 |
+ Attributes: |
|
| 111 |
+ MANGLE_ADDRESS: |
|
| 112 |
+ Mangle the socket address. |
|
| 113 |
+ |
|
| 114 |
+ For UNIX domain sockets, mangle the `SSH_AUTH_SOCK` |
|
| 115 |
+ environment variable. |
|
| 116 |
+ |
|
| 117 |
+ For Windows named pipes, skip the test. (Addresses are |
|
| 118 |
+ fixed, so cannot be mangled.) |
|
| 119 |
+ UNSET_ADDRESS: |
|
| 120 |
+ Unset the socket address. |
|
| 121 |
+ |
|
| 122 |
+ For UNIX domain sockets, unset the `SSH_AUTH_SOCK` |
|
| 123 |
+ environment variable. |
|
| 124 |
+ |
|
| 125 |
+ For Windows named pipes, skip the test. (Adresses are |
|
| 126 |
+ fixed, so cannot be unset.) |
|
| 127 |
+ |
|
| 128 |
+ """ |
|
| 129 |
+ |
|
| 130 |
+ MANGLE_ADDRESS = enum.auto() |
|
| 131 |
+ """""" |
|
| 132 |
+ UNSET_ADDRESS = enum.auto() |
|
| 133 |
+ """""" |
|
| 134 |
+ |
|
| 135 |
+ def __call__( |
|
| 136 |
+ self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any |
|
| 137 |
+ ) -> None: |
|
| 138 |
+ """Execute the respective action.""" |
|
| 139 |
+ # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 140 |
+ # matching. |
|
| 141 |
+ # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 142 |
+ if self == self.MANGLE_ADDRESS: |
|
| 143 |
+ |
|
| 144 |
+ def mangled_address( |
|
| 145 |
+ *_args: Any, **_kwargs: Any |
|
| 146 |
+ ) -> NoReturn: # pragma: no cover [failsafe] |
|
| 147 |
+ raise OSError(errno.ENOENT, os.strerror(errno.ENOENT)) |
|
| 148 |
+ |
|
| 149 |
+ monkeypatch.setattr( |
|
| 150 |
+ socketprovider.WindowsNamedPipeHandle, |
|
| 151 |
+ "for_pageant", |
|
| 152 |
+ mangled_address, |
|
| 153 |
+ ) |
|
| 154 |
+ monkeypatch.setattr( |
|
| 155 |
+ socketprovider.WindowsNamedPipeHandle, |
|
| 156 |
+ "for_openssh", |
|
| 157 |
+ mangled_address, |
|
| 158 |
+ ) |
|
| 159 |
+ monkeypatch.setenv( |
|
| 160 |
+ "SSH_AUTH_SOCK", |
|
| 161 |
+ os.environ["SSH_AUTH_SOCK"] + "~" |
|
| 162 |
+ if "SSH_AUTH_SOCK" in os.environ |
|
| 163 |
+ else socketprovider.PIPE_PREFIX |
|
| 164 |
+ if os.name == "nt" |
|
| 165 |
+ else "/", |
|
| 166 |
+ ) |
|
| 167 |
+ elif self == self.UNSET_ADDRESS: |
|
| 168 |
+ |
|
| 169 |
+ def no_address( |
|
| 170 |
+ *_args: Any, **_kwargs: Any |
|
| 171 |
+ ) -> NoReturn: # pragma: no cover [failsafe] |
|
| 172 |
+ raise KeyError("SSH_AUTH_SOCK") # noqa: EM101
|
|
| 173 |
+ |
|
| 174 |
+ monkeypatch.setattr( |
|
| 175 |
+ socketprovider.WindowsNamedPipeHandle, |
|
| 176 |
+ "for_pageant", |
|
| 177 |
+ no_address, |
|
| 178 |
+ ) |
|
| 179 |
+ monkeypatch.setattr( |
|
| 180 |
+ socketprovider.WindowsNamedPipeHandle, |
|
| 181 |
+ "for_openssh", |
|
| 182 |
+ no_address, |
|
| 183 |
+ ) |
|
| 184 |
+ monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
|
|
| 185 |
+ else: |
|
| 186 |
+ raise AssertionError() |
|
| 187 |
+ |
|
| 188 |
+ |
|
| 189 |
+class SystemSupportAction(str, enum.Enum): |
|
| 190 |
+ """Test fixture settings for [`ssh_agent.SSHAgentClient`][] system support. |
|
| 191 |
+ |
|
| 192 |
+ Attributes: |
|
| 193 |
+ UNSET_AF_UNIX_AND_ENSURE_USE: |
|
| 194 |
+ Ensure lack of support for UNIX domain sockets, and that the |
|
| 195 |
+ agent will use this socket provider. |
|
| 196 |
+ UNSET_NATIVE_AND_ENSURE_USE: |
|
| 197 |
+ Ensure both `UNSET_AF_UNIX` and `UNSET_WINDLL`, and that the |
|
| 198 |
+ agent will use the native socket provider. |
|
| 199 |
+ UNSET_PROVIDER_LIST: |
|
| 200 |
+ Ensure an empty list of SSH agent socket providers. |
|
| 201 |
+ UNSET_WINDLL_AND_ENSURE_USE: |
|
| 202 |
+ Ensure lack of support for Windows named pipes, and that the |
|
| 203 |
+ agent will use this socket provider. |
|
| 204 |
+ |
|
| 205 |
+ """ |
|
| 206 |
+ |
|
| 207 |
+ UNSET_AF_UNIX_AND_ENSURE_USE = enum.auto() |
|
| 208 |
+ """""" |
|
| 209 |
+ UNSET_NATIVE_AND_ENSURE_USE = enum.auto() |
|
| 210 |
+ """""" |
|
| 211 |
+ UNSET_PROVIDER_LIST = enum.auto() |
|
| 212 |
+ """""" |
|
| 213 |
+ UNSET_WINDLL_AND_ENSURE_USE = enum.auto() |
|
| 214 |
+ """""" |
|
| 215 |
+ |
|
| 216 |
+ def __call__( |
|
| 217 |
+ self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any |
|
| 218 |
+ ) -> None: |
|
| 219 |
+ """Execute the respective action. |
|
| 220 |
+ |
|
| 221 |
+ Args: |
|
| 222 |
+ monkeypatch: The current monkeypatch context. |
|
| 223 |
+ |
|
| 224 |
+ """ |
|
| 225 |
+ # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 226 |
+ # matching. |
|
| 227 |
+ # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 228 |
+ if self == self.UNSET_PROVIDER_LIST: |
|
| 229 |
+ monkeypatch.setattr( |
|
| 230 |
+ ssh_agent.SSHAgentClient, "SOCKET_PROVIDERS", [] |
|
| 231 |
+ ) |
|
| 232 |
+ elif self == self.UNSET_NATIVE_AND_ENSURE_USE: |
|
| 233 |
+ monkeypatch.setattr( |
|
| 234 |
+ ssh_agent.SSHAgentClient, |
|
| 235 |
+ "SOCKET_PROVIDERS", |
|
| 236 |
+ [_types.BuiltinSSHAgentSocketProvider.NATIVE], |
|
| 237 |
+ ) |
|
| 238 |
+ monkeypatch.delattr(socket, "AF_UNIX", raising=False) |
|
| 239 |
+ monkeypatch.delattr(ctypes, "WinDLL", raising=False) |
|
| 240 |
+ monkeypatch.delattr(ctypes, "windll", raising=False) |
|
| 241 |
+ elif self == self.UNSET_AF_UNIX_AND_ENSURE_USE: |
|
| 242 |
+ monkeypatch.setattr( |
|
| 243 |
+ ssh_agent.SSHAgentClient, |
|
| 244 |
+ "SOCKET_PROVIDERS", |
|
| 245 |
+ [_types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX], |
|
| 246 |
+ ) |
|
| 247 |
+ monkeypatch.delattr(socket, "AF_UNIX", raising=False) |
|
| 248 |
+ elif self == self.UNSET_WINDLL_AND_ENSURE_USE: |
|
| 249 |
+ monkeypatch.setattr( |
|
| 250 |
+ ssh_agent.SSHAgentClient, |
|
| 251 |
+ "SOCKET_PROVIDERS", |
|
| 252 |
+ [_types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE], |
|
| 253 |
+ ) |
|
| 254 |
+ monkeypatch.delattr(ctypes, "WinDLL", raising=False) |
|
| 255 |
+ monkeypatch.delattr(ctypes, "windll", raising=False) |
|
| 256 |
+ else: |
|
| 257 |
+ raise AssertionError() |
|
| 258 |
+ |
|
| 259 |
+ |
|
| 40 | 260 |
# Types |
| 41 | 261 |
# ===== |
| 42 | 262 |
|
| ... | ... |
@@ -8,8 +8,6 @@ from __future__ import annotations |
| 8 | 8 |
|
| 9 | 9 |
import base64 |
| 10 | 10 |
import contextlib |
| 11 |
-import ctypes |
|
| 12 |
-import enum |
|
| 13 | 11 |
import errno |
| 14 | 12 |
import io |
| 15 | 13 |
import json |
| ... | ... |
@@ -35,7 +33,6 @@ from derivepassphrase._internals import ( |
| 35 | 33 |
cli_helpers, |
| 36 | 34 |
cli_machinery, |
| 37 | 35 |
) |
| 38 |
-from derivepassphrase.ssh_agent import socketprovider |
|
| 39 | 36 |
from tests import data, machinery |
| 40 | 37 |
from tests.data import callables |
| 41 | 38 |
from tests.machinery import hypothesis as hypothesis_machinery |
| ... | ... |
@@ -150,216 +147,6 @@ def vault_config_exporter_shell_interpreter( # noqa: C901 |
| 150 | 147 |
) |
| 151 | 148 |
|
| 152 | 149 |
|
| 153 |
-class ListKeysAction(str, enum.Enum): |
|
| 154 |
- """Test fixture settings for [`ssh_agent.SSHAgentClient.list_keys`][]. |
|
| 155 |
- |
|
| 156 |
- Attributes: |
|
| 157 |
- EMPTY: Return an empty key list. |
|
| 158 |
- FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][]. |
|
| 159 |
- FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][]. |
|
| 160 |
- |
|
| 161 |
- """ |
|
| 162 |
- |
|
| 163 |
- EMPTY = enum.auto() |
|
| 164 |
- """""" |
|
| 165 |
- FAIL = enum.auto() |
|
| 166 |
- """""" |
|
| 167 |
- FAIL_RUNTIME = enum.auto() |
|
| 168 |
- """""" |
|
| 169 |
- |
|
| 170 |
- def __call__(self, *_args: Any, **_kwargs: Any) -> Any: |
|
| 171 |
- """Execute the respective action.""" |
|
| 172 |
- # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 173 |
- # matching. |
|
| 174 |
- # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 175 |
- if self == self.EMPTY: |
|
| 176 |
- return [] |
|
| 177 |
- if self == self.FAIL: |
|
| 178 |
- raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"") |
|
| 179 |
- if self == self.FAIL_RUNTIME: |
|
| 180 |
- raise ssh_agent.TrailingDataError() |
|
| 181 |
- raise AssertionError() |
|
| 182 |
- |
|
| 183 |
- |
|
| 184 |
-class SignAction(str, enum.Enum): |
|
| 185 |
- """Test fixture settings for [`ssh_agent.SSHAgentClient.sign`][]. |
|
| 186 |
- |
|
| 187 |
- Attributes: |
|
| 188 |
- FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][]. |
|
| 189 |
- FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][]. |
|
| 190 |
- |
|
| 191 |
- """ |
|
| 192 |
- |
|
| 193 |
- FAIL = enum.auto() |
|
| 194 |
- """""" |
|
| 195 |
- FAIL_RUNTIME = enum.auto() |
|
| 196 |
- """""" |
|
| 197 |
- |
|
| 198 |
- def __call__(self, *_args: Any, **_kwargs: Any) -> Any: |
|
| 199 |
- """Execute the respective action.""" |
|
| 200 |
- # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 201 |
- # matching. |
|
| 202 |
- # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 203 |
- if self == self.FAIL: |
|
| 204 |
- raise ssh_agent.SSHAgentFailedError(_types.SSH_AGENT.FAILURE, b"") |
|
| 205 |
- if self == self.FAIL_RUNTIME: |
|
| 206 |
- raise ssh_agent.TrailingDataError() |
|
| 207 |
- raise AssertionError() |
|
| 208 |
- |
|
| 209 |
- |
|
| 210 |
-class SocketAddressAction(str, enum.Enum): |
|
| 211 |
- """Test fixture settings for the SSH agent socket address. |
|
| 212 |
- |
|
| 213 |
- Attributes: |
|
| 214 |
- MANGLE_ADDRESS: |
|
| 215 |
- Mangle the socket address. |
|
| 216 |
- |
|
| 217 |
- For UNIX domain sockets, mangle the `SSH_AUTH_SOCK` |
|
| 218 |
- environment variable. |
|
| 219 |
- |
|
| 220 |
- For Windows named pipes, skip the test. (Addresses are |
|
| 221 |
- fixed, so cannot be mangled.) |
|
| 222 |
- UNSET_ADDRESS: |
|
| 223 |
- Unset the socket address. |
|
| 224 |
- |
|
| 225 |
- For UNIX domain sockets, unset the `SSH_AUTH_SOCK` |
|
| 226 |
- environment variable. |
|
| 227 |
- |
|
| 228 |
- For Windows named pipes, skip the test. (Adresses are |
|
| 229 |
- fixed, so cannot be unset.) |
|
| 230 |
- |
|
| 231 |
- """ |
|
| 232 |
- |
|
| 233 |
- MANGLE_ADDRESS = enum.auto() |
|
| 234 |
- """""" |
|
| 235 |
- UNSET_ADDRESS = enum.auto() |
|
| 236 |
- """""" |
|
| 237 |
- |
|
| 238 |
- def __call__( |
|
| 239 |
- self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any |
|
| 240 |
- ) -> None: |
|
| 241 |
- """Execute the respective action.""" |
|
| 242 |
- # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 243 |
- # matching. |
|
| 244 |
- # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 245 |
- if self == self.MANGLE_ADDRESS: |
|
| 246 |
- |
|
| 247 |
- def mangled_address( |
|
| 248 |
- *_args: Any, **_kwargs: Any |
|
| 249 |
- ) -> NoReturn: # pragma: no cover [failsafe] |
|
| 250 |
- raise OSError(errno.ENOENT, os.strerror(errno.ENOENT)) |
|
| 251 |
- |
|
| 252 |
- monkeypatch.setattr( |
|
| 253 |
- socketprovider.WindowsNamedPipeHandle, |
|
| 254 |
- "for_pageant", |
|
| 255 |
- mangled_address, |
|
| 256 |
- ) |
|
| 257 |
- monkeypatch.setattr( |
|
| 258 |
- socketprovider.WindowsNamedPipeHandle, |
|
| 259 |
- "for_openssh", |
|
| 260 |
- mangled_address, |
|
| 261 |
- ) |
|
| 262 |
- monkeypatch.setenv( |
|
| 263 |
- "SSH_AUTH_SOCK", |
|
| 264 |
- os.environ["SSH_AUTH_SOCK"] + "~" |
|
| 265 |
- if "SSH_AUTH_SOCK" in os.environ |
|
| 266 |
- else socketprovider.PIPE_PREFIX |
|
| 267 |
- if os.name == "nt" |
|
| 268 |
- else "/", |
|
| 269 |
- ) |
|
| 270 |
- elif self == self.UNSET_ADDRESS: |
|
| 271 |
- |
|
| 272 |
- def no_address( |
|
| 273 |
- *_args: Any, **_kwargs: Any |
|
| 274 |
- ) -> NoReturn: # pragma: no cover [failsafe] |
|
| 275 |
- raise KeyError("SSH_AUTH_SOCK") # noqa: EM101
|
|
| 276 |
- |
|
| 277 |
- monkeypatch.setattr( |
|
| 278 |
- socketprovider.WindowsNamedPipeHandle, |
|
| 279 |
- "for_pageant", |
|
| 280 |
- no_address, |
|
| 281 |
- ) |
|
| 282 |
- monkeypatch.setattr( |
|
| 283 |
- socketprovider.WindowsNamedPipeHandle, |
|
| 284 |
- "for_openssh", |
|
| 285 |
- no_address, |
|
| 286 |
- ) |
|
| 287 |
- monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
|
|
| 288 |
- else: |
|
| 289 |
- raise AssertionError() |
|
| 290 |
- |
|
| 291 |
- |
|
| 292 |
-class SystemSupportAction(str, enum.Enum): |
|
| 293 |
- """Test fixture settings for [`ssh_agent.SSHAgentClient`][] system support. |
|
| 294 |
- |
|
| 295 |
- Attributes: |
|
| 296 |
- UNSET_AF_UNIX_AND_ENSURE_USE: |
|
| 297 |
- Ensure lack of support for UNIX domain sockets, and that the |
|
| 298 |
- agent will use this socket provider. |
|
| 299 |
- UNSET_NATIVE_AND_ENSURE_USE: |
|
| 300 |
- Ensure both `UNSET_AF_UNIX` and `UNSET_WINDLL`, and that the |
|
| 301 |
- agent will use the native socket provider. |
|
| 302 |
- UNSET_PROVIDER_LIST: |
|
| 303 |
- Ensure an empty list of SSH agent socket providers. |
|
| 304 |
- UNSET_WINDLL_AND_ENSURE_USE: |
|
| 305 |
- Ensure lack of support for Windows named pipes, and that the |
|
| 306 |
- agent will use this socket provider. |
|
| 307 |
- |
|
| 308 |
- """ |
|
| 309 |
- |
|
| 310 |
- UNSET_AF_UNIX_AND_ENSURE_USE = enum.auto() |
|
| 311 |
- """""" |
|
| 312 |
- UNSET_NATIVE_AND_ENSURE_USE = enum.auto() |
|
| 313 |
- """""" |
|
| 314 |
- UNSET_PROVIDER_LIST = enum.auto() |
|
| 315 |
- """""" |
|
| 316 |
- UNSET_WINDLL_AND_ENSURE_USE = enum.auto() |
|
| 317 |
- """""" |
|
| 318 |
- |
|
| 319 |
- def __call__( |
|
| 320 |
- self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any |
|
| 321 |
- ) -> None: |
|
| 322 |
- """Execute the respective action. |
|
| 323 |
- |
|
| 324 |
- Args: |
|
| 325 |
- monkeypatch: The current monkeypatch context. |
|
| 326 |
- |
|
| 327 |
- """ |
|
| 328 |
- # TODO(the-13th-letter): Rewrite using structural pattern |
|
| 329 |
- # matching. |
|
| 330 |
- # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 |
|
| 331 |
- if self == self.UNSET_PROVIDER_LIST: |
|
| 332 |
- monkeypatch.setattr( |
|
| 333 |
- ssh_agent.SSHAgentClient, "SOCKET_PROVIDERS", [] |
|
| 334 |
- ) |
|
| 335 |
- elif self == self.UNSET_NATIVE_AND_ENSURE_USE: |
|
| 336 |
- monkeypatch.setattr( |
|
| 337 |
- ssh_agent.SSHAgentClient, |
|
| 338 |
- "SOCKET_PROVIDERS", |
|
| 339 |
- [_types.BuiltinSSHAgentSocketProvider.NATIVE], |
|
| 340 |
- ) |
|
| 341 |
- monkeypatch.delattr(socket, "AF_UNIX", raising=False) |
|
| 342 |
- monkeypatch.delattr(ctypes, "WinDLL", raising=False) |
|
| 343 |
- monkeypatch.delattr(ctypes, "windll", raising=False) |
|
| 344 |
- elif self == self.UNSET_AF_UNIX_AND_ENSURE_USE: |
|
| 345 |
- monkeypatch.setattr( |
|
| 346 |
- ssh_agent.SSHAgentClient, |
|
| 347 |
- "SOCKET_PROVIDERS", |
|
| 348 |
- [_types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX], |
|
| 349 |
- ) |
|
| 350 |
- monkeypatch.delattr(socket, "AF_UNIX", raising=False) |
|
| 351 |
- elif self == self.UNSET_WINDLL_AND_ENSURE_USE: |
|
| 352 |
- monkeypatch.setattr( |
|
| 353 |
- ssh_agent.SSHAgentClient, |
|
| 354 |
- "SOCKET_PROVIDERS", |
|
| 355 |
- [_types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE], |
|
| 356 |
- ) |
|
| 357 |
- monkeypatch.delattr(ctypes, "WinDLL", raising=False) |
|
| 358 |
- monkeypatch.delattr(ctypes, "windll", raising=False) |
|
| 359 |
- else: |
|
| 360 |
- raise AssertionError() |
|
| 361 |
- |
|
| 362 |
- |
|
| 363 | 150 |
class Parametrize(types.SimpleNamespace): |
| 364 | 151 |
"""Common test parametrizations.""" |
| 365 | 152 |
|
| ... | ... |
@@ -425,46 +212,46 @@ class Parametrize(types.SimpleNamespace): |
| 425 | 212 |
], |
| 426 | 213 |
[ |
| 427 | 214 |
pytest.param( |
| 428 |
- ListKeysAction.EMPTY, |
|
| 215 |
+ data.ListKeysAction.EMPTY, |
|
| 429 | 216 |
None, |
| 430 | 217 |
None, |
| 431 |
- SignAction.FAIL, |
|
| 218 |
+ data.SignAction.FAIL, |
|
| 432 | 219 |
"not loaded into the agent", |
| 433 | 220 |
[], |
| 434 | 221 |
id="key-not-loaded", |
| 435 | 222 |
), |
| 436 | 223 |
pytest.param( |
| 437 |
- ListKeysAction.FAIL, |
|
| 224 |
+ data.ListKeysAction.FAIL, |
|
| 438 | 225 |
None, |
| 439 | 226 |
None, |
| 440 |
- SignAction.FAIL, |
|
| 227 |
+ data.SignAction.FAIL, |
|
| 441 | 228 |
"SSH agent failed to or refused to", |
| 442 | 229 |
[], |
| 443 | 230 |
id="list-keys-refused", |
| 444 | 231 |
), |
| 445 | 232 |
pytest.param( |
| 446 |
- ListKeysAction.FAIL_RUNTIME, |
|
| 233 |
+ data.ListKeysAction.FAIL_RUNTIME, |
|
| 447 | 234 |
None, |
| 448 | 235 |
None, |
| 449 |
- SignAction.FAIL, |
|
| 236 |
+ data.SignAction.FAIL, |
|
| 450 | 237 |
"SSH agent failed to or refused to", |
| 451 | 238 |
[], |
| 452 | 239 |
id="list-keys-protocol-error", |
| 453 | 240 |
), |
| 454 | 241 |
pytest.param( |
| 455 | 242 |
None, |
| 456 |
- SocketAddressAction.MANGLE_ADDRESS, |
|
| 243 |
+ data.SocketAddressAction.MANGLE_ADDRESS, |
|
| 457 | 244 |
None, |
| 458 |
- SignAction.FAIL, |
|
| 245 |
+ data.SignAction.FAIL, |
|
| 459 | 246 |
"Cannot connect to the SSH agent", |
| 460 | 247 |
[], |
| 461 | 248 |
id="agent-address-mangled", |
| 462 | 249 |
), |
| 463 | 250 |
pytest.param( |
| 464 | 251 |
None, |
| 465 |
- SocketAddressAction.UNSET_ADDRESS, |
|
| 252 |
+ data.SocketAddressAction.UNSET_ADDRESS, |
|
| 466 | 253 |
None, |
| 467 |
- SignAction.FAIL, |
|
| 254 |
+ data.SignAction.FAIL, |
|
| 468 | 255 |
"Cannot find any running SSH agent", |
| 469 | 256 |
[], |
| 470 | 257 |
id="agent-address-missing", |
| ... | ... |
@@ -472,8 +259,8 @@ class Parametrize(types.SimpleNamespace): |
| 472 | 259 |
pytest.param( |
| 473 | 260 |
None, |
| 474 | 261 |
None, |
| 475 |
- SystemSupportAction.UNSET_NATIVE_AND_ENSURE_USE, |
|
| 476 |
- SignAction.FAIL, |
|
| 262 |
+ data.SystemSupportAction.UNSET_NATIVE_AND_ENSURE_USE, |
|
| 263 |
+ data.SignAction.FAIL, |
|
| 477 | 264 |
"does not support communicating with it", |
| 478 | 265 |
[], |
| 479 | 266 |
id="no-native-agent-available", |
| ... | ... |
@@ -481,8 +268,8 @@ class Parametrize(types.SimpleNamespace): |
| 481 | 268 |
pytest.param( |
| 482 | 269 |
None, |
| 483 | 270 |
None, |
| 484 |
- SystemSupportAction.UNSET_PROVIDER_LIST, |
|
| 485 |
- SignAction.FAIL, |
|
| 271 |
+ data.SystemSupportAction.UNSET_PROVIDER_LIST, |
|
| 272 |
+ data.SignAction.FAIL, |
|
| 486 | 273 |
"does not support communicating with it", |
| 487 | 274 |
[], |
| 488 | 275 |
id="no-agents-in-agent-provider-list", |
| ... | ... |
@@ -490,8 +277,8 @@ class Parametrize(types.SimpleNamespace): |
| 490 | 277 |
pytest.param( |
| 491 | 278 |
None, |
| 492 | 279 |
None, |
| 493 |
- SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE, |
|
| 494 |
- SignAction.FAIL, |
|
| 280 |
+ data.SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE, |
|
| 281 |
+ data.SignAction.FAIL, |
|
| 495 | 282 |
"does not support communicating with it", |
| 496 | 283 |
["Cannot connect to an SSH agent via UNIX domain sockets"], |
| 497 | 284 |
id="no-unix-domain-sockets", |
| ... | ... |
@@ -499,8 +286,8 @@ class Parametrize(types.SimpleNamespace): |
| 499 | 286 |
pytest.param( |
| 500 | 287 |
None, |
| 501 | 288 |
None, |
| 502 |
- SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE, |
|
| 503 |
- SignAction.FAIL, |
|
| 289 |
+ data.SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE, |
|
| 290 |
+ data.SignAction.FAIL, |
|
| 504 | 291 |
"does not support communicating with it", |
| 505 | 292 |
["Cannot connect to an SSH agent via Windows named pipes"], |
| 506 | 293 |
id="no-windows-named-pipes", |
| ... | ... |
@@ -509,7 +296,7 @@ class Parametrize(types.SimpleNamespace): |
| 509 | 296 |
None, |
| 510 | 297 |
None, |
| 511 | 298 |
None, |
| 512 |
- SignAction.FAIL_RUNTIME, |
|
| 299 |
+ data.SignAction.FAIL_RUNTIME, |
|
| 513 | 300 |
"violates the communication protocol", |
| 514 | 301 |
[], |
| 515 | 302 |
id="sign-violates-protocol", |
| ... | ... |
@@ -1411,10 +1198,10 @@ class TestMisc: |
| 1411 | 1198 |
def test_key_to_phrase( |
| 1412 | 1199 |
self, |
| 1413 | 1200 |
use_stub_agent_with_address: None, |
| 1414 |
- list_keys_action: ListKeysAction | None, |
|
| 1415 |
- system_support_action: SystemSupportAction | None, |
|
| 1416 |
- address_action: SocketAddressAction | None, |
|
| 1417 |
- sign_action: SignAction, |
|
| 1201 |
+ list_keys_action: data.ListKeysAction | None, |
|
| 1202 |
+ system_support_action: data.SystemSupportAction | None, |
|
| 1203 |
+ address_action: data.SocketAddressAction | None, |
|
| 1204 |
+ sign_action: data.SignAction, |
|
| 1418 | 1205 |
pattern: str, |
| 1419 | 1206 |
warnings_patterns: list[str], |
| 1420 | 1207 |
) -> None: |
| ... | ... |
@@ -1458,7 +1245,7 @@ class TestMisc: |
| 1458 | 1245 |
assert any([pat in string for string in captured_warnings]), ( |
| 1459 | 1246 |
f"expected some warning message to match {pat}"
|
| 1460 | 1247 |
) |
| 1461 |
- if list_keys_action == ListKeysAction.FAIL_RUNTIME: |
|
| 1248 |
+ if list_keys_action == data.ListKeysAction.FAIL_RUNTIME: |
|
| 1462 | 1249 |
assert excinfo.value.kwargs |
| 1463 | 1250 |
assert isinstance( |
| 1464 | 1251 |
excinfo.value.kwargs["exc_info"], |
| 1465 | 1252 |