Coverage for tests / machinery / pytest.py: 100.000%

1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> 

2# 

3# SPDX-License-Identifier: Zlib 

4 

5"""`pytest` testing machinery for the `derivepassphrase` test suite. 

6 

7This is all the `pytest`-specific data and functionality used in the 

8`derivepassphrase` test suite; this includes `pytest` marks and 

9monkeypatched functions (or functions relying heavily internally on 

10monkeypatching). 

11 

12All code requiring *more* than plain `pytest` lives in its own sibling 

13module, e.g., the `hypothesis`-related stuff lives in the [`hypothesis` 

14sibling module][tests.machinery.hypothesis]. 

15 

16""" 

17 

18from __future__ import annotations (empty)

19 

20import base64 (empty)

21import contextlib (empty)

22import importlib.metadata (empty)

23import importlib.util (empty)

24import json (empty)

25import os (empty)

26import pathlib (empty)

27import sys (empty)

28import tempfile (empty)

29import types (empty)

30import zipfile (empty)

31from typing import TYPE_CHECKING (empty)

32 

33import pytest (empty)

34from typing_extensions import assert_never, overload (empty)

35 

36import tests.data (empty)

37import tests.machinery (empty)

38from derivepassphrase import _types, ssh_agent (empty)

39from derivepassphrase._internals import ( (empty)

40 cli_helpers, 

41 cli_machinery, 

42 cli_messages, 

43) 

44from derivepassphrase.ssh_agent import socketprovider (empty)

45 

46__all__ = () (empty)

47 

48if TYPE_CHECKING: (empty)

49 from collections.abc import Callable, Generator, Sequence 

50 from contextlib import AbstractContextManager 

51 

52 from typing_extensions import Any 

53 

54 

55# Marks 

56# ===== 

57 

58 

59skip_if_cryptography_support = pytest.mark.skipif( (empty)

60 importlib.util.find_spec("cryptography") is not None, 

61 reason='cryptography support available; cannot test "no support" scenario', 

62) 

63""" (empty)

64A cached pytest mark to skip this test if cryptography support is 

65available. Usually this means that the test targets 

66`derivepassphrase`'s fallback functionality, which is not available 

67whenever the primary functionality is. 

68""" 

69skip_if_no_cryptography_support = pytest.mark.skipif( (empty)

70 importlib.util.find_spec("cryptography") is None, 

71 reason='no "cryptography" support', 

72) 

73""" (empty)

74A cached pytest mark to skip this test if cryptography support is not 

75available. Usually this means that the test targets the 

76`derivepassphrase export vault` subcommand, whose functionality depends 

77on cryptography support being available. 

78""" 

79skip_if_on_the_annoying_os = pytest.mark.skipif( (empty)

80 sys.platform == "win32", 

81 reason="The Annoying OS behaves differently.", 

82) 

83""" (empty)

84A cached pytest mark to skip this test if running on The Annoying 

85Operating System, a.k.a. Microsoft Windows. Usually this is due to 

86unnecessary and stupid differences in the OS internals, and these 

87differences are deemed irreconcilable in the context of the decorated 

88test, so the test is to be skipped. 

89 

90See also: 

91 [`xfail_on_the_annoying_os`][] 

92 

93""" 

94skip_if_no_multiprocessing_support = pytest.mark.skipif( (empty)

95 importlib.util.find_spec("multiprocessing") is None, 

96 reason='no "multiprocessing" support', 

97) 

98""" (empty)

99A cached pytest mark to skip this test if multiprocessing support is not 

100available. Usually this means that the test targets the concurrency 

101features of `derivepassphrase`, which is generally only possible to test 

102in separate processes because the testing machinery operates on 

103process-global state. 

104""" 

105 

106 

107def xfail_on_the_annoying_os( (empty)

108 f: Callable | None = None, (empty)

109 /, 

110 *, 

111 reason: str = "", (empty)

112) -> pytest.MarkDecorator | Any: # pragma: no cover (empty)

113 """Annotate a test which fails on The Annoying OS. 

114 

115 Annotate a test to indicate that it fails on The Annoying Operating 

116 System, a.k.a. Microsoft Windows. Usually this is due to 

117 differences in the design of OS internals, and usually, these 

118 differences are both unnecessary and stupid. 

119 

120 Args: 

121 f: 

122 A callable to decorate. If not given, return the pytest 

123 mark directly. 

124 reason: 

125 An optional, more detailed reason stating why this test 

126 fails on The Annoying OS. 

127 

128 Returns: 

129 The callable, marked as an expected failure on the Annoying OS, 

130 or alternatively a suitable pytest mark if no callable was 

131 passed. The reason will begin with the phrase "The Annoying OS 

132 behaves differently.", and the optional detailed reason, if not 

133 empty, will follow. 

134 

135 """ 

136 import hypothesis.errors # noqa: PLC0415 

137 

138 base_reason = "The Annoying OS behaves differently." 

139 full_reason = base_reason if not reason else f"{base_reason} {reason}" 

140 mark = pytest.mark.xfail( 

141 sys.platform == "win32", 

142 reason=full_reason, 

143 raises=(AssertionError, hypothesis.errors.FailedHealthCheck), 

144 strict=True, 

145 ) 

146 return mark if f is None else mark(f) 

147 

148 

149heavy_duty = pytest.mark.heavy_duty (empty)

150""" (empty)

151Test functions/classes/modules with this `pytest` mark are slow, heavy 

152duty tests. Users who are impatient (or otherwise cannot afford to wait 

153for these tests to complete) may wish to exclude these tests; this mark 

154helps in achieving that. 

155""" 

156 

157correctness_focused = pytest.mark.correctness_focused (empty)

158""" (empty)

159Test functions/classes/modules with this `pytest` mark are designed 

160primarily to verify correctness and to find novel failure cases, even at 

161the cost of high CPU or wall clock time. 

162 

163Contrast this with tests designed primarily to verify that the existing 

164system still behaves as before in all the important ways ("is healthy") 

165even after introducing a small change, but before committing it. (See 

166[Two kinds of testing][TWO_KINDS] for more elaboration.) 

167 

168[TWO_KINDS]: https://blog.nelhage.com/post/two-kinds-of-testing/ 

169""" 

170 

171non_isolated_agent_use = pytest.mark.xdist_group("non_isolated_agent_use") (empty)

172""" (empty)

173Tests with this `pytest` mark (for use with xdist's "loadgroup" 

174scheduling), comprise the group of tests that use a potentially 

175non-isolated agent, and thus need to be run serially, in the same worker 

176(because of the setup and teardown). 

177 

178This is a pessimistic approach: all agent use is marked with this mark 

179*unless* we know for sure that this constellation is guaranteed safe. 

180To date, this is only the case for the fake agents from this very test 

181suite. 

182 

183The use of this mark is, if not outright a hack, at least an abuse of 

184xdist features. The "correct" way to handle this situation is to employ 

185a lock, but we have had bad experiences with trying to do this in 

186a portable way without adding oodles of extra code (that also needs to be 

187tested), or effectively forcing *everything* into serial execution. It 

188also only works because we have precisely one group, and because we have 

189no work stealing (at least, not with the only scheduler that honors 

190`xdist_group`). In particular, it bars us from adding more groups or 

191grouping along different criteria. 

192""" 

193 

194 

195# Parameter sets 

196# ============== 

197 

198 

199class Parametrize(types.SimpleNamespace): (empty)

200 VAULT_CONFIG_FORMATS_DATA = pytest.mark.parametrize( (empty)

201 ["config", "format", "config_data"], 

202 [ 

203 pytest.param( 

204 tests.data.VAULT_V02_CONFIG, 

205 "v0.2", 

206 tests.data.VAULT_V02_CONFIG_DATA, 

207 id="0.2", 

208 ), 

209 pytest.param( 

210 tests.data.VAULT_V03_CONFIG, 

211 "v0.3", 

212 tests.data.VAULT_V03_CONFIG_DATA, 

213 id="0.3", 

214 ), 

215 pytest.param( 

216 tests.data.VAULT_STOREROOM_CONFIG_ZIPPED, 

217 "storeroom", 

218 tests.data.VAULT_STOREROOM_CONFIG_DATA, 

219 id="storeroom", 

220 ), 

221 ], 

222 ) 

223 MISSING_AGENT_SUPPORT = pytest.mark.parametrize( (empty)

224 ["provider", "sys_action", "exception", "message"], 

225 [ 

226 pytest.param( 

227 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX, 

228 tests.data.SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE, 

229 socketprovider.UnixDomainSocketsNotAvailableError, 

230 str( 

231 cli_messages.TranslatedString( 

232 cli_messages.WarnMsgTemplate.NO_UNIX_DOMAIN_SOCKETS 

233 ) 

234 ), 

235 id="unix_domain_sockets", 

236 ), 

237 pytest.param( 

238 _types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE, 

239 tests.data.SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE, 

240 socketprovider.WindowsNamedPipesNotAvailableError, 

241 str( 

242 cli_messages.TranslatedString( 

243 cli_messages.WarnMsgTemplate.NO_WINDOWS_NAMED_PIPES 

244 ) 

245 ), 

246 id="windows_named_pipes", 

247 ), 

248 ], 

249 ) 

250 

251 

252# Monkeypatchings 

253# =============== 

254 

255 

256@contextlib.contextmanager (empty)

257def faked_entry_point_list( # noqa: C901 (empty)

258 additional_entry_points: Sequence[importlib.metadata.EntryPoint], 

259 remove_conflicting_entries: bool = False, 

260) -> Generator[Sequence[str], None, None]: 

261 """Yield a context where additional entry points are visible. 

262 

263 Args: 

264 additional_entry_points: 

265 A sequence of entry point objects that should additionally 

266 be visible. 

267 remove_conflicting_entries: 

268 If true, remove all names provided by the additional entry 

269 points, otherwise leave them untouched. 

270 

271 Yields: 

272 A sequence of registry names that are newly available within the 

273 context. 

274 

275 """ 

276 true_entry_points = importlib.metadata.entry_points() 2 ctx1bc

277 additional_entry_points = list(additional_entry_points) 2 ctx1bc

278 

279 if sys.version_info >= (3, 12): 2 ctx1bc

280 new_entry_points = importlib.metadata.EntryPoints( 2 ctx1bc

281 list(true_entry_points) + additional_entry_points 

282 ) 

283 

284 @overload 2 ctx1bc

285 def mangled_entry_points( 2 ctx1bc

286 *, group: None = None 2 ctx1bc

287 ) -> importlib.metadata.EntryPoints: ... 2 ctx1bc

288 

289 @overload 2 ctx1bc

290 def mangled_entry_points( 2 ctx1bc

291 *, group: str 2 ctx1bc

292 ) -> importlib.metadata.EntryPoints: ... 2 ctx1bc

293 

294 def mangled_entry_points( 2 ctx1bc

295 **params: Any, 

296 ) -> importlib.metadata.EntryPoints: 

297 return new_entry_points.select(**params) 2 ctx1bc

298 

299 elif sys.version_info >= (3, 10): 2 ctx1bc

300 # Compatibility concerns within importlib.metadata: depending on 

301 # whether the .select() API is used, the result is either the dict 

302 # of groups of points (as in < 3.10), or the EntryPoints iterable 

303 # (as in >= 3.12). So our wrapper needs to duplicate that 

304 # interface. FUN. 

305 new_entry_points_dict = { 2 ctx1bc

306 k: list(v) for k, v in true_entry_points.items() 

307 } 

308 for ep in additional_entry_points: 2 ctx1bc

309 new_entry_points_dict.setdefault(ep.group, []).append(ep) 2 ctx1bc

310 new_entry_points = importlib.metadata.EntryPoints([ 2 ctx1bc

311 ep for group in new_entry_points_dict.values() for ep in group 

312 ]) 

313 

314 @overload 2 ctx1bc

315 def mangled_entry_points( 2 ctx1bc

316 *, group: None = None 2 ctx1bc

317 ) -> dict[ 2 ctx1bc

318 str, 

319 list[importlib.metadata.EntryPoint] 

320 | tuple[importlib.metadata.EntryPoint, ...], 

321 ]: ... 

322 

323 @overload 2 ctx1bc

324 def mangled_entry_points( 2 ctx1bc

325 *, group: str 2 ctx1bc

326 ) -> importlib.metadata.EntryPoints: ... 2 ctx1bc

327 

328 def mangled_entry_points( 2 ctx1bc

329 **params: Any, 

330 ) -> ( 

331 importlib.metadata.EntryPoints 

332 | dict[ 

333 str, 

334 list[importlib.metadata.EntryPoint] 

335 | tuple[importlib.metadata.EntryPoint, ...], 

336 ] 

337 ): 

338 return ( 2 ctx1bc

339 new_entry_points.select(**params) 

340 if params 

341 else new_entry_points_dict 

342 ) 

343 

344 else: 

345 new_entry_points: dict[ 2 ctx1bc

346 str, 

347 list[importlib.metadata.EntryPoint] 

348 | tuple[importlib.metadata.EntryPoint, ...], 

349 ] = { 

350 group_name: list(group) 

351 for group_name, group in true_entry_points.items() 

352 } 

353 for ep in additional_entry_points: 2 ctx1bc

354 new_entry_points.setdefault(ep.group, []) 2 ctx1bc

355 new_entry_points[ep.group].append(ep) 2 ctx1bc

356 new_entry_points = { 2 ctx1bc

357 group_name: tuple(group) 

358 for group_name, group in new_entry_points.items() 

359 } 

360 

361 @overload 2 ctx1bc

362 def mangled_entry_points( 2 ctx1bc

363 *, group: None = None 2 ctx1bc

364 ) -> dict[str, tuple[importlib.metadata.EntryPoint, ...]]: ... 2 ctx1bc

365 

366 @overload 2 ctx1bc

367 def mangled_entry_points( 2 ctx1bc

368 *, group: str 2 ctx1bc

369 ) -> tuple[importlib.metadata.EntryPoint, ...]: ... 2 ctx1bc

370 

371 def mangled_entry_points( 2 ctx1bc

372 *, group: str | None = None 

373 ) -> ( 

374 dict[str, tuple[importlib.metadata.EntryPoint, ...]] 

375 | tuple[importlib.metadata.EntryPoint, ...] 

376 ): 

377 return ( 

378 new_entry_points.get(group, ()) 

379 if group is not None 

380 else new_entry_points 

381 ) 

382 

383 registry = socketprovider.SocketProvider.registry 2 ctx1bc

384 new_registry = registry.copy() 2 ctx1bc

385 keys = [ep.load().key for ep in additional_entry_points] 2 ctx1bc

386 aliases = [a for ep in additional_entry_points for a in ep.load().aliases] 2 ctx1bc

387 if remove_conflicting_entries: # pragma: no cover [unused] 2 ctx1bc

388 for name in [*keys, *aliases]: 

389 new_registry.pop(name, None) 

390 

391 with pytest.MonkeyPatch.context() as monkeypatch: 2 ctx1bc

392 monkeypatch.setattr( 2 ctx1bc

393 socketprovider.SocketProvider, "registry", new_registry 

394 ) 

395 monkeypatch.setattr( 2 ctx1bc

396 importlib.metadata, "entry_points", mangled_entry_points 

397 ) 

398 yield (*keys, *aliases) 2 ctx1bc

399 

400 

401@contextlib.contextmanager (empty)

402def isolated_config( (empty)

403 monkeypatch: pytest.MonkeyPatch, 

404 runner: tests.machinery.CliRunner, 

405 main_config_str: str | None = None, 

406) -> Generator[None, None, None]: 

407 """Provide an isolated configuration setup, as a context. 

408 

409 This context manager sets up (and changes into) a temporary 

410 directory, which holds the user configuration specified in 

411 `main_config_str`, if any. The manager also ensures that the 

412 environment variables `HOME` and `USERPROFILE` are set, and that 

413 `DERIVEPASSPHRASE_PATH` is unset. Upon exiting the context, the 

414 changes are undone and the temporary directory is removed. 

415 

416 Args: 

417 monkeypatch: 

418 A monkeypatch fixture object. 

419 runner: 

420 A `click` CLI runner harness. 

421 main_config_str: 

422 Optional TOML file contents, to be used as the user 

423 configuration. 

424 

425 Returns: 

426 A context manager, without a return value. 

427 

428 """ 

429 prog_name = cli_helpers.PROG_NAME 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

430 env_name = prog_name.replace(" ", "_").upper() + "_PATH" 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

431 # TODO(the-13th-letter): Rewrite using parenthesized with-statements. 

432 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

433 with contextlib.ExitStack() as stack: 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

434 stack.enter_context(runner.isolated_filesystem()) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

435 stack.enter_context( 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

436 cli_machinery.StandardCLILogging.ensure_standard_logging() 

437 ) 

438 stack.enter_context( 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

439 cli_machinery.StandardCLILogging.ensure_standard_warnings_logging() 

440 ) 

441 cwd = str(pathlib.Path.cwd().resolve()) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

442 monkeypatch.setenv("HOME", cwd) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

443 monkeypatch.setenv("APPDATA", cwd) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

444 monkeypatch.setenv("LOCALAPPDATA", cwd) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

445 monkeypatch.delenv(env_name, raising=False) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

446 config_dir = cli_helpers.config_filename(subsystem=None) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

447 config_dir.mkdir(parents=True, exist_ok=True) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

448 if isinstance(main_config_str, str): 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

449 cli_helpers.config_filename("user configuration").write_text( 6 ctx1xyzABC

450 main_config_str, encoding="UTF-8" 

451 ) 

452 try: 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

453 yield 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

454 finally: 

455 cli_helpers.config_filename("write lock").unlink(missing_ok=True) 87 ctx2D x y E z A } ~ abbbcbdbebF G H I J fbgbhbibjbkbK lbmbnbL M N O P Q R S T obpbqbU V W X Y Z 0 1 B C 2 3 4 5 6 7 8 9 ! rb# sbtbub$ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { |

456 

457 

458@contextlib.contextmanager (empty)

459def isolated_vault_config( (empty)

460 monkeypatch: pytest.MonkeyPatch, 

461 runner: tests.machinery.CliRunner, 

462 vault_config: Any, 

463 main_config_str: str | None = None, 

464) -> Generator[None, None, None]: 

465 """Provide an isolated vault configuration setup, as a context. 

466 

467 Uses [`isolated_config`][] internally. Beyond those actions, this 

468 manager also loads the specified vault configuration into the 

469 context. 

470 

471 Args: 

472 monkeypatch: 

473 A monkeypatch fixture object. 

474 runner: 

475 A `click` CLI runner harness. 

476 vault_config: 

477 A valid vault configuration, to be integrated into the 

478 context. 

479 main_config_str: 

480 Optional TOML file contents, to be used as the user 

481 configuration. 

482 

483 Returns: 

484 A context manager, without a return value. 

485 

486 """ 

487 with isolated_config( 64 ctx1DxyEzAFGHIJKLMNOPQRSTUVWXYZ01BC23456789!#$%'()*+,-./:;=?@[]^_`{|

488 monkeypatch=monkeypatch, runner=runner, main_config_str=main_config_str 

489 ): 

490 config_filename = cli_helpers.config_filename(subsystem="vault") 64 ctx1DxyEzAFGHIJKLMNOPQRSTUVWXYZ01BC23456789!#$%'()*+,-./:;=?@[]^_`{|

491 with config_filename.open("w", encoding="UTF-8") as outfile: 64 ctx1DxyEzAFGHIJKLMNOPQRSTUVWXYZ01BC23456789!#$%'()*+,-./:;=?@[]^_`{|

492 json.dump(vault_config, outfile) 64 ctx1DxyEzAFGHIJKLMNOPQRSTUVWXYZ01BC23456789!#$%'()*+,-./:;=?@[]^_`{|

493 yield 64 ctx1DxyEzAFGHIJKLMNOPQRSTUVWXYZ01BC23456789!#$%'()*+,-./:;=?@[]^_`{|

494 

495 

496@contextlib.contextmanager (empty)

497def isolated_vault_exporter_config( (empty)

498 monkeypatch: pytest.MonkeyPatch, 

499 runner: tests.machinery.CliRunner, 

500 vault_config: str | bytes | None = None, 

501 vault_key: str | None = None, 

502) -> Generator[None, None, None]: 

503 """Provide an isolated vault configuration setup, as a context. 

504 

505 Works similarly to [`isolated_config`][], except that no user 

506 configuration is accepted or integrated into the context. This 

507 manager also accepts a serialized vault-native configuration and 

508 a vault encryption key to integrate into the context. 

509 

510 Args: 

511 monkeypatch: 

512 A monkeypatch fixture object. 

513 runner: 

514 A `click` CLI runner harness. 

515 vault_config: 

516 An optional serialized vault-native configuration, to be 

517 integrated into the context. If a text string, then the 

518 contents are written to the file `.vault`. If a byte 

519 string, then it is treated as base64-encoded zip file 

520 contents, which---once inside the `.vault` directory---will 

521 be extracted into the current directory. 

522 vault_key: 

523 An optional encryption key presumably for the stored 

524 vault-native configuration. If given, then the environment 

525 variable `VAULT_KEY` will be populated with this key while 

526 the context is active. 

527 

528 Returns: 

529 A context manager, without a return value. 

530 

531 """ 

532 # TODO(the-13th-letter): Remove the fallback implementation. 

533 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.10 

534 if TYPE_CHECKING: 20 ctx1lmnopqrdsfghituvjwek

535 chdir: Callable[..., AbstractContextManager] 

536 else: 

537 try: 20 ctx1lmnopqrdsfghituvjwek

538 chdir = contextlib.chdir # type: ignore[attr] 20 ctx1lmnopqrdsfghituvjwek

539 except AttributeError: 20 ctx1lmnopqrdsfghituvjwek

540 

541 @contextlib.contextmanager 20 ctx1lmnopqrdsfghituvjwek

542 def chdir( 20 ctx1lmnopqrdsfghituvjwek

543 newpath: str | bytes | os.PathLike, 

544 ) -> Generator[None, None, None]: # pragma: no branch 

545 oldpath = pathlib.Path.cwd().resolve() 6 ctx1dfghie

546 os.chdir(newpath) 6 ctx1dfghie

547 try: 6 ctx1dfghie

548 yield 6 ctx1dfghie

549 finally: 

550 os.chdir(oldpath) 6 ctx1dfghie

551 

552 with runner.isolated_filesystem(): 20 ctx1lmnopqrdsfghituvjwek

553 cwd = str(pathlib.Path.cwd().resolve()) 20 ctx1lmnopqrdsfghituvjwek

554 monkeypatch.setenv("HOME", cwd) 20 ctx1lmnopqrdsfghituvjwek

555 monkeypatch.setenv("USERPROFILE", cwd) 20 ctx1lmnopqrdsfghituvjwek

556 monkeypatch.delenv( 20 ctx1lmnopqrdsfghituvjwek

557 cli_helpers.PROG_NAME.replace(" ", "_").upper() + "_PATH", 

558 raising=False, 

559 ) 

560 monkeypatch.delenv("VAULT_PATH", raising=False) 20 ctx1lmnopqrdsfghituvjwek

561 monkeypatch.delenv("VAULT_KEY", raising=False) 20 ctx1lmnopqrdsfghituvjwek

562 monkeypatch.delenv("LOGNAME", raising=False) 20 ctx1lmnopqrdsfghituvjwek

563 monkeypatch.delenv("USER", raising=False) 20 ctx1lmnopqrdsfghituvjwek

564 monkeypatch.delenv("USERNAME", raising=False) 20 ctx1lmnopqrdsfghituvjwek

565 if vault_key is not None: 20 ctx1lmnopqrdsfghituvjwek

566 monkeypatch.setenv("VAULT_KEY", vault_key) 18 ctx1lmnopqrdsfghituvwe

567 vault_config_path = pathlib.Path(".vault").resolve() 20 ctx1lmnopqrdsfghituvjwek

568 # TODO(the-13th-letter): Rewrite using structural pattern matching. 

569 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

570 if isinstance(vault_config, str): 20 ctx1lmnopqrdsfghituvjwek

571 vault_config_path.write_text(f"{vault_config}\n", encoding="UTF-8") 14 ctx1lmnopqrdstuvwe

572 elif isinstance(vault_config, bytes): 8 ctx1dfghijek

573 vault_config_path.mkdir(parents=True, mode=0o700, exist_ok=True) 6 ctx1dfghie

574 # TODO(the-13th-letter): Rewrite using parenthesized 

575 # with-statements. 

576 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

577 with contextlib.ExitStack() as stack: 6 ctx1dfghie

578 stack.enter_context(chdir(vault_config_path)) 6 ctx1dfghie

579 tmpzipfile = stack.enter_context( 6 ctx1dfghie

580 tempfile.NamedTemporaryFile(suffix=".zip") 

581 ) 

582 for line in vault_config.splitlines(): 6 ctx1dfghie

583 tmpzipfile.write(base64.standard_b64decode(line)) 6 ctx1dfghie

584 tmpzipfile.flush() 6 ctx1dfghie

585 tmpzipfile.seek(0, 0) 6 ctx1dfghie

586 with zipfile.ZipFile(tmpzipfile.file) as zipfileobj: 6 ctx1dfghie

587 zipfileobj.extractall() 6 ctx1dfghie

588 elif vault_config is None: 2 ctx1jk

589 pass 2 ctx1jk

590 else: # pragma: no cover 

591 assert_never(vault_config) 

592 try: 20 ctx1lmnopqrdsfghituvjwek

593 yield 20 ctx1lmnopqrdsfghituvjwek

594 finally: 

595 cli_helpers.config_filename("write lock").unlink(missing_ok=True) 20 ctx1lmnopqrdsfghituvjwek

596 

597 

598def ensure_singleton_client_if_non_reentrant_agent( (empty)

599 *, 

600 request: pytest.FixtureRequest, 

601 monkeypatch: pytest.MonkeyPatch, 

602 client: ssh_agent.SSHAgentClient, 

603) -> None: 

604 """Mangle SSH agent construction if the spawned agent is non-reentrant. 

605 

606 Monkeypatch a suitable environment to ensure that constructing a new 

607 SSH agent client -- usually via the 

608 [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][] context 

609 manager -- doesn't really construct a new client, in case the agent 

610 is non-reentrant. 

611 

612 Agent reentrancy is determined by the presence of the 

613 `non_reentrant_agent` mark. We change nothing if the agent is 

614 actually reentrant. 

615 

616 Args: 

617 request: 

618 The `pytest` request fixture. Needed to get the active 

619 markers on this test function execution, which detail 

620 whether the agent is potentially non-reentrant. 

621 monkeypatch: 

622 A monkeypatching context to enrich. 

623 client: 

624 The singleton SSH agent client that is to be used. 

625 

626 """ 

627 if ( 

628 request.node.get_closest_marker("non_reentrant_agent") is not None 

629 ): # pragma: no cover [external] 

630 real_ensure_agent_subcontext = ( 

631 ssh_agent.SSHAgentClient.ensure_agent_subcontext 

632 ) 

633 

634 def ensure_agent_subcontext( 

635 conn: ssh_agent.SSHAgentClient 

636 | _types.SSHAgentSocket 

637 | Sequence[str] 

638 | None = None, 

639 ) -> contextlib.AbstractContextManager[ssh_agent.SSHAgentClient]: 

640 if isinstance(conn, ssh_agent.SSHAgentClient): 

641 return real_ensure_agent_subcontext(conn) 

642 return real_ensure_agent_subcontext(client) 

643 

644 monkeypatch.setattr( 

645 ssh_agent.SSHAgentClient, 

646 "ensure_agent_subcontext", 

647 ensure_agent_subcontext, 

648 )