Coverage for tests / test_derivepassphrase_cli / test_utils.py: 100.000%
335 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-06 21:34 +0200
1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info>
2#
3# SPDX-License-Identifier: Zlib
5"""Tests for the `derivepassphrase` command-line interface: internal utility functions."""
7from __future__ import annotations
9import base64
10import contextlib
11import errno
12import io
13import json
14import logging
15import operator
16import os
17import pathlib
18import shlex
19import shutil
20import socket
21import tempfile
22import types
23import warnings
24from typing import TYPE_CHECKING
26import click
27import hypothesis
28import pytest
29from hypothesis import strategies
31from derivepassphrase import _types, cli, ssh_agent, vault
32from derivepassphrase._internals import (
33 cli_helpers,
34 cli_machinery,
35)
36from tests import data, machinery
37from tests.data import callables
38from tests.machinery import hypothesis as hypothesis_machinery
39from tests.machinery import pytest as pytest_machinery
41if TYPE_CHECKING:
42 from collections.abc import Callable, Generator, Iterable
43 from typing import NoReturn
45 from typing_extensions import Any
48DUMMY_SERVICE = data.DUMMY_SERVICE
49DUMMY_PASSPHRASE = data.DUMMY_PASSPHRASE
50DUMMY_CONFIG_SETTINGS = data.DUMMY_CONFIG_SETTINGS
51DUMMY_CONFIG_SETTINGS_AS_CONSTRUCTOR_ARGS = (
52 data.DUMMY_CONFIG_SETTINGS_AS_CONSTRUCTOR_ARGS
53)
54DUMMY_RESULT_PASSPHRASE = data.DUMMY_RESULT_PASSPHRASE
55DUMMY_RESULT_KEY1 = data.DUMMY_RESULT_KEY1
56DUMMY_PHRASE_FROM_KEY1_RAW = data.DUMMY_PHRASE_FROM_KEY1_RAW
57DUMMY_PHRASE_FROM_KEY1 = data.DUMMY_PHRASE_FROM_KEY1
59DUMMY_KEY1 = data.DUMMY_KEY1
60DUMMY_KEY1_B64 = data.DUMMY_KEY1_B64
61DUMMY_KEY2 = data.DUMMY_KEY2
62DUMMY_KEY2_B64 = data.DUMMY_KEY2_B64
63DUMMY_KEY3 = data.DUMMY_KEY3
64DUMMY_KEY3_B64 = data.DUMMY_KEY3_B64
66TEST_CONFIGS = data.TEST_CONFIGS
69def vault_config_exporter_shell_interpreter( # noqa: C901
70 script: str | Iterable[str],
71 /,
72 *,
73 prog_name_list: list[str] | None = None,
74 # The click.BaseCommand abstract base class, previously the base
75 # class of click.Command and click.Group, was removed in click
76 # 8.2.0 without a transition period, and click.Command instated
77 # as the common base class instead. To keep some degree of
78 # compatibility with both old click and new click, we explicitly
79 # list the (somewhat) concrete base classes we actually care
80 # about here.
81 command: click.Command | click.Group | None = None,
82 runner: machinery.CliRunner | None = None,
83) -> Generator[machinery.ReadableResult, None, None]:
84 """A rudimentary sh(1) interpreter for `--export-as=sh` output.
86 Assumes a script as emitted by `derivepassphrase vault
87 --export-as=sh --export -` and interprets the calls to
88 `derivepassphrase vault` within. (One call per line, skips all
89 other lines.) Also has rudimentary support for (quoted)
90 here-documents using `HERE` as the marker.
92 """
93 if isinstance(script, str): # pragma: no cover 1cbde
94 script = script.splitlines(False) 1cbde
95 if prog_name_list is None: # pragma: no cover 1cbde
96 prog_name_list = ["derivepassphrase", "vault"] 1cbde
97 if command is None: # pragma: no cover 1cbde
98 command = cli.derivepassphrase_vault 1cbde
99 if runner is None: # pragma: no cover 1cbde
100 runner = machinery.CliRunner(mix_stderr=False) 1cbde
101 n = len(prog_name_list) 1cbde
102 it = iter(script) 1cbde
103 while True: 1cbde
104 try: 1cbde
105 raw_line = next(it) 1cbde
106 except StopIteration: 1cbde
107 break 1cbde
108 else:
109 line = shlex.split(raw_line) 1cbde
110 input_buffer: list[str] = [] 1cbde
111 if line[:n] != prog_name_list: 1cbde
112 continue 1cbde
113 line[:n] = [] 1cbde
114 if line and line[-1] == "<<HERE": 1cbde
115 # naive HERE document support
116 while True: 1cbde
117 try: 1cbde
118 raw_line = next(it) 1cbde
119 except StopIteration as exc: # pragma: no cover
120 msg = "incomplete here document"
121 raise EOFError(msg) from exc
122 else:
123 if raw_line == "HERE": 1cbde
124 break 1cbde
125 input_buffer.append(raw_line) 1cbde
126 line.pop() 1cbde
127 yield runner.invoke( 1cbde
128 command,
129 line,
130 catch_exceptions=False,
131 input=("".join(x + "\n" for x in input_buffer) or None),
132 )
135class Parametrize(types.SimpleNamespace):
136 """Common test parametrizations."""
138 DELETE_CONFIG_INPUT = pytest.mark.parametrize(
139 ["command_line", "config", "result_config"],
140 [
141 pytest.param(
142 ["--delete-globals"],
143 {"global": {"phrase": "abc"}, "services": {}},
144 {"services": {}},
145 id="globals",
146 ),
147 pytest.param(
148 ["--delete", "--", DUMMY_SERVICE],
149 {
150 "global": {"phrase": "abc"},
151 "services": {DUMMY_SERVICE: {"notes": "..."}},
152 },
153 {"global": {"phrase": "abc"}, "services": {}},
154 id="service",
155 ),
156 pytest.param(
157 ["--clear"],
158 {
159 "global": {"phrase": "abc"},
160 "services": {DUMMY_SERVICE: {"notes": "..."}},
161 },
162 {"services": {}},
163 id="all",
164 ),
165 ],
166 )
167 BASE_CONFIG_VARIATIONS = pytest.mark.parametrize(
168 "config",
169 [
170 {"global": {"phrase": "my passphrase"}, "services": {}},
171 {"global": {"key": DUMMY_KEY1_B64}, "services": {}},
172 {
173 "global": {"phrase": "abc"},
174 "services": {"sv": {"phrase": "my passphrase"}},
175 },
176 {
177 "global": {"phrase": "abc"},
178 "services": {"sv": {"key": DUMMY_KEY1_B64}},
179 },
180 {
181 "global": {"phrase": "abc"},
182 "services": {"sv": {"key": DUMMY_KEY1_B64, "length": 15}},
183 },
184 ],
185 )
186 CONNECTION_HINTS = pytest.mark.parametrize(
187 "conn_hint", ["none", "socket", "client"]
188 )
189 KEY_TO_PHRASE_SETTINGS = pytest.mark.parametrize(
190 [
191 "list_keys_action",
192 "address_action",
193 "system_support_action",
194 "sign_action",
195 "pattern",
196 "warnings_patterns",
197 ],
198 [
199 pytest.param(
200 data.ListKeysAction.EMPTY,
201 None,
202 None,
203 data.SignAction.FAIL,
204 "not loaded into the agent",
205 [],
206 id="key-not-loaded",
207 ),
208 pytest.param(
209 data.ListKeysAction.FAIL,
210 None,
211 None,
212 data.SignAction.FAIL,
213 "SSH agent failed to or refused to",
214 [],
215 id="list-keys-refused",
216 ),
217 pytest.param(
218 data.ListKeysAction.FAIL_RUNTIME,
219 None,
220 None,
221 data.SignAction.FAIL,
222 "SSH agent failed to or refused to",
223 [],
224 id="list-keys-protocol-error",
225 ),
226 pytest.param(
227 None,
228 data.SocketAddressAction.MANGLE_ADDRESS,
229 None,
230 data.SignAction.FAIL,
231 "Cannot connect to the SSH agent",
232 [],
233 id="agent-address-mangled",
234 ),
235 pytest.param(
236 None,
237 data.SocketAddressAction.UNSET_ADDRESS,
238 None,
239 data.SignAction.FAIL,
240 "Cannot find any running SSH agent",
241 [],
242 id="agent-address-missing",
243 ),
244 pytest.param(
245 None,
246 None,
247 data.SystemSupportAction.UNSET_NATIVE_AND_ENSURE_USE,
248 data.SignAction.FAIL,
249 "does not support communicating with it",
250 [],
251 id="no-native-agent-available",
252 ),
253 pytest.param(
254 None,
255 None,
256 data.SystemSupportAction.UNSET_PROVIDER_LIST,
257 data.SignAction.FAIL,
258 "does not support communicating with it",
259 [],
260 id="no-agents-in-agent-provider-list",
261 ),
262 pytest.param(
263 None,
264 None,
265 data.SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE,
266 data.SignAction.FAIL,
267 "does not support communicating with it",
268 ["Cannot connect to an SSH agent via UNIX domain sockets"],
269 id="no-unix-domain-sockets",
270 ),
271 pytest.param(
272 None,
273 None,
274 data.SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE,
275 data.SignAction.FAIL,
276 "does not support communicating with it",
277 ["Cannot connect to an SSH agent via Windows named pipes"],
278 id="no-windows-named-pipes",
279 ),
280 pytest.param(
281 None,
282 None,
283 None,
284 data.SignAction.FAIL_RUNTIME,
285 "violates the communication protocol",
286 [],
287 id="sign-violates-protocol",
288 ),
289 ],
290 )
291 VALIDATION_FUNCTION_INPUT = pytest.mark.parametrize(
292 ["vfunc", "input"],
293 [
294 (cli_machinery.validate_occurrence_constraint, 20),
295 (cli_machinery.validate_length, 20),
296 ],
297 )
300class TestConfigLoadingSaving:
301 """Tests for the config loading and saving utility functions."""
303 @Parametrize.BASE_CONFIG_VARIATIONS
304 def test_load(
305 self,
306 config: Any,
307 ) -> None:
308 """[`cli_helpers.load_config`][] works for valid configurations."""
309 runner = machinery.CliRunner(mix_stderr=False) 1p
310 # TODO(the-13th-letter): Rewrite using parenthesized
311 # with-statements.
312 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
313 with contextlib.ExitStack() as stack: 1p
314 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1p
315 stack.enter_context( 1p
316 pytest_machinery.isolated_vault_config(
317 monkeypatch=monkeypatch,
318 runner=runner,
319 vault_config=config,
320 )
321 )
322 config_filename = cli_helpers.config_filename(subsystem="vault") 1p
323 with config_filename.open(encoding="UTF-8") as fileobj: 1p
324 assert json.load(fileobj) == config 1p
325 assert cli_helpers.load_config() == config 1p
327 def test_save_bad_config(
328 self,
329 ) -> None:
330 """[`cli_helpers.save_config`][] fails for bad configurations."""
331 runner = machinery.CliRunner(mix_stderr=False) 1q
332 # TODO(the-13th-letter): Rewrite using parenthesized
333 # with-statements.
334 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
335 with contextlib.ExitStack() as stack: 1q
336 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1q
337 stack.enter_context( 1q
338 pytest_machinery.isolated_vault_config(
339 monkeypatch=monkeypatch,
340 runner=runner,
341 vault_config={},
342 )
343 )
344 stack.enter_context( 1q
345 pytest.raises(ValueError, match="Invalid vault config")
346 )
347 cli_helpers.save_config(None) # type: ignore[arg-type] 1q
350class TestPrompting:
351 """Tests for the prompting utility functions."""
353 def test_selection_multiple(self) -> None:
354 """[`cli_helpers.prompt_for_selection`][] works in the "multiple" case."""
356 @click.command() 1j
357 @click.option("--heading", default="Our menu:") 1j
358 @click.argument("items", nargs=-1) 1j
359 def driver(heading: str, items: list[str]) -> None: 1j
360 # from https://montypython.fandom.com/wiki/Spam#The_menu
361 items = items or [ 1j
362 "Egg and bacon",
363 "Egg, sausage and bacon",
364 "Egg and spam",
365 "Egg, bacon and spam",
366 "Egg, bacon, sausage and spam",
367 "Spam, bacon, sausage and spam",
368 "Spam, egg, spam, spam, bacon and spam",
369 "Spam, spam, spam, egg and spam",
370 (
371 "Spam, spam, spam, spam, spam, spam, baked beans, "
372 "spam, spam, spam and spam"
373 ),
374 (
375 "Lobster thermidor aux crevettes with a mornay sauce "
376 "garnished with truffle paté, brandy "
377 "and a fried egg on top and spam"
378 ),
379 ]
380 index = cli_helpers.prompt_for_selection(items, heading=heading) 1j
381 click.echo("A fine choice: ", nl=False) 1j
382 click.echo(items[index]) 1j
383 click.echo("(Note: Vikings strictly optional.)") 1j
385 runner = machinery.CliRunner(mix_stderr=True) 1j
386 result = runner.invoke(driver, [], input="9") 1j
387 assert result.clean_exit( 1j
388 output="""\
389Our menu:
390[1] Egg and bacon
391[2] Egg, sausage and bacon
392[3] Egg and spam
393[4] Egg, bacon and spam
394[5] Egg, bacon, sausage and spam
395[6] Spam, bacon, sausage and spam
396[7] Spam, egg, spam, spam, bacon and spam
397[8] Spam, spam, spam, egg and spam
398[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
399[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
400Your selection? (1-10, leave empty to abort): 9
401A fine choice: Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
402(Note: Vikings strictly optional.)
403"""
404 ), "expected clean exit"
405 result = runner.invoke( 1j
406 driver, ["--heading="], input="\n", catch_exceptions=True
407 )
408 assert result.error_exit(error=IndexError), ( 1j
409 "expected error exit and known error type"
410 )
411 assert ( 1j
412 result.stdout
413 == """\
414[1] Egg and bacon
415[2] Egg, sausage and bacon
416[3] Egg and spam
417[4] Egg, bacon and spam
418[5] Egg, bacon, sausage and spam
419[6] Spam, bacon, sausage and spam
420[7] Spam, egg, spam, spam, bacon and spam
421[8] Spam, spam, spam, egg and spam
422[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
423[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
424Your selection? (1-10, leave empty to abort):\x20
425"""
426 ), "expected known output"
427 # click.testing.CliRunner on click < 8.2.1 incorrectly mocks the
428 # click prompting machinery, meaning that the mixed output will
429 # incorrectly contain a line break, contrary to what the
430 # documentation for click.prompt prescribes.
431 result = runner.invoke( 1j
432 driver, ["--heading="], input="", catch_exceptions=True
433 )
434 assert result.error_exit(error=IndexError), ( 1j
435 "expected error exit and known error type"
436 )
437 assert result.stdout in { 1j
438 """\
439[1] Egg and bacon
440[2] Egg, sausage and bacon
441[3] Egg and spam
442[4] Egg, bacon and spam
443[5] Egg, bacon, sausage and spam
444[6] Spam, bacon, sausage and spam
445[7] Spam, egg, spam, spam, bacon and spam
446[8] Spam, spam, spam, egg and spam
447[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
448[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
449Your selection? (1-10, leave empty to abort):\x20
450""",
451 """\
452[1] Egg and bacon
453[2] Egg, sausage and bacon
454[3] Egg and spam
455[4] Egg, bacon and spam
456[5] Egg, bacon, sausage and spam
457[6] Spam, bacon, sausage and spam
458[7] Spam, egg, spam, spam, bacon and spam
459[8] Spam, spam, spam, egg and spam
460[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
461[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
462Your selection? (1-10, leave empty to abort): """,
463 }, "expected known output"
465 def test_selection_single(self) -> None:
466 """[`cli_helpers.prompt_for_selection`][] works in the "single" case."""
468 @click.command() 1i
469 @click.option("--item", default="baked beans") 1i
470 @click.argument("prompt") 1i
471 def driver(item: str, prompt: str) -> None: 1i
472 try: 1i
473 cli_helpers.prompt_for_selection( 1i
474 [item], heading="", single_choice_prompt=prompt
475 )
476 except IndexError: 1i
477 click.echo("Boo.") 1i
478 raise 1i
479 else:
480 click.echo("Great!") 1i
482 runner = machinery.CliRunner(mix_stderr=True) 1i
483 result = runner.invoke( 1i
484 driver, ["Will replace with spam. Confirm, y/n?"], input="y"
485 )
486 assert result.clean_exit( 1i
487 output="""\
488[1] baked beans
489Will replace with spam. Confirm, y/n? y
490Great!
491"""
492 ), "expected clean exit"
493 result = runner.invoke( 1i
494 driver,
495 ['Will replace with spam, okay? (Please say "y" or "n".)'],
496 input="\n",
497 )
498 assert result.error_exit(error=IndexError), ( 1i
499 "expected error exit and known error type"
500 )
501 assert ( 1i
502 result.stdout
503 == """\
504[1] baked beans
505Will replace with spam, okay? (Please say "y" or "n".):\x20
506Boo.
507"""
508 ), "expected known output"
509 # click.testing.CliRunner on click < 8.2.1 incorrectly mocks the
510 # click prompting machinery, meaning that the mixed output will
511 # incorrectly contain a line break, contrary to what the
512 # documentation for click.prompt prescribes.
513 result = runner.invoke( 1i
514 driver,
515 ['Will replace with spam, okay? (Please say "y" or "n".)'],
516 input="",
517 )
518 assert result.error_exit(error=IndexError), ( 1i
519 "expected error exit and known error type"
520 )
521 assert result.stdout in { 1i
522 """\
523[1] baked beans
524Will replace with spam, okay? (Please say "y" or "n".):\x20
525Boo.
526""",
527 """\
528[1] baked beans
529Will replace with spam, okay? (Please say "y" or "n".): Boo.
530""",
531 # Some versions of click concatenate stdout and stderr
532 # (in the test runner) instead of interleaving them. This
533 # looks okay on-screen, but rather crazy in the test suite.
534 """\
535[1] baked beans
536Boo.
537Will replace with spam, okay? (Please say "y" or "n".): """,
538 }, "expected known output"
540 def test_passphrase(
541 self,
542 ) -> None:
543 """[`cli_helpers.prompt_for_passphrase`][] works."""
544 with pytest.MonkeyPatch.context() as monkeypatch: 1n
545 monkeypatch.setattr( 1n
546 click,
547 "prompt",
548 lambda *a, **kw: json.dumps({"args": a, "kwargs": kw}),
549 )
550 res = json.loads(cli_helpers.prompt_for_passphrase()) 1n
551 err_msg = "missing arguments to passphrase prompt" 1n
552 assert "args" in res, err_msg 1n
553 assert "kwargs" in res, err_msg 1n
554 assert res["args"][:1] == ["Passphrase"], err_msg 1n
555 assert res["kwargs"].get("default") == "", err_msg 1n
556 assert not res["kwargs"].get("show_default", True), err_msg 1n
557 assert res["kwargs"].get("err"), err_msg 1n
558 assert res["kwargs"].get("hide_input"), err_msg 1n
561class TestLoggingMachinery:
562 """Tests for the logging utility functions."""
564 def test_standard_logging_context_manager(
565 self,
566 caplog: pytest.LogCaptureFixture,
567 capsys: pytest.CaptureFixture[str],
568 ) -> None:
569 """The standard logging context manager works.
571 It registers its handlers, once, and emits formatted calls to
572 standard error prefixed with the program name.
574 """
575 prog_name = cli_machinery.StandardCLILogging.prog_name 1l
576 package_name = cli_machinery.StandardCLILogging.package_name 1l
577 logger = logging.getLogger(package_name) 1l
578 deprecation_logger = logging.getLogger(f"{package_name}.deprecation") 1l
579 logging_cm = cli_machinery.StandardCLILogging.ensure_standard_logging() 1l
580 with logging_cm: 1l
581 assert ( 1l
582 sum(
583 1
584 for h in logger.handlers
585 if h is cli_machinery.StandardCLILogging.cli_handler
586 )
587 == 1
588 )
589 logger.warning("message 1") 1l
590 with logging_cm: 1l
591 deprecation_logger.warning("message 2") 1l
592 assert ( 1l
593 sum(
594 1
595 for h in logger.handlers
596 if h is cli_machinery.StandardCLILogging.cli_handler
597 )
598 == 1
599 )
600 assert capsys.readouterr() == ( 1l
601 "",
602 (
603 f"{prog_name}: Warning: message 1\n"
604 f"{prog_name}: Deprecation warning: message 2\n"
605 ),
606 )
607 logger.warning("message 3") 1l
608 assert ( 1l
609 sum(
610 1
611 for h in logger.handlers
612 if h is cli_machinery.StandardCLILogging.cli_handler
613 )
614 == 1
615 )
616 assert capsys.readouterr() == ( 1l
617 "",
618 f"{prog_name}: Warning: message 3\n",
619 )
620 assert caplog.record_tuples == [ 1l
621 (package_name, logging.WARNING, "message 1"),
622 (f"{package_name}.deprecation", logging.WARNING, "message 2"),
623 (package_name, logging.WARNING, "message 3"),
624 ]
626 def test_warnings_context_manager(
627 self,
628 caplog: pytest.LogCaptureFixture,
629 capsys: pytest.CaptureFixture[str],
630 ) -> None:
631 """The standard warnings logging context manager works.
633 It registers its handlers, once, and emits formatted calls to
634 standard error prefixed with the program name. It also adheres
635 to the global warnings filter concerning which messages it
636 actually emits to standard error.
638 """
639 warnings_cm = ( 1h
640 cli_machinery.StandardCLILogging.ensure_standard_warnings_logging()
641 )
642 THE_FUTURE = "the future will be here sooner than you think" # noqa: N806 1h
643 JUST_TESTING = "just testing whether warnings work" # noqa: N806 1h
644 with warnings_cm: 1h
645 assert ( 1h
646 sum(
647 1
648 for h in logging.getLogger("py.warnings").handlers
649 if h is cli_machinery.StandardCLILogging.warnings_handler
650 )
651 == 1
652 )
653 warnings.warn(UserWarning(JUST_TESTING), stacklevel=1) 1h
654 with warnings_cm: 1h
655 warnings.warn(FutureWarning(THE_FUTURE), stacklevel=1) 1h
656 _out, err = capsys.readouterr() 1h
657 err_lines = err.splitlines(True) 1h
658 assert any( 1h
659 f"UserWarning: {JUST_TESTING}" in line
660 for line in err_lines
661 )
662 assert any( 1h
663 f"FutureWarning: {THE_FUTURE}" in line
664 for line in err_lines
665 )
666 warnings.warn(UserWarning(JUST_TESTING), stacklevel=1) 1h
667 _out, err = capsys.readouterr() 1h
668 err_lines = err.splitlines(True) 1h
669 assert any( 1h
670 f"UserWarning: {JUST_TESTING}" in line for line in err_lines
671 )
672 assert not any( 1h
673 f"FutureWarning: {THE_FUTURE}" in line for line in err_lines
674 )
675 record_tuples = caplog.record_tuples 1h
676 assert [tup[:2] for tup in record_tuples] == [ 1h
677 ("py.warnings", logging.WARNING),
678 ("py.warnings", logging.WARNING),
679 ("py.warnings", logging.WARNING),
680 ]
681 assert f"UserWarning: {JUST_TESTING}" in record_tuples[0][2] 1h
682 assert f"FutureWarning: {THE_FUTURE}" in record_tuples[1][2] 1h
683 assert f"UserWarning: {JUST_TESTING}" in record_tuples[2][2] 1h
686class TestExportConfigAsShellScript:
687 """Tests the utility functions for exporting configs in `sh` format."""
689 def export_as_sh_helper(
690 self,
691 config: Any,
692 ) -> None:
693 """Emits a config in sh(1) format, then reads it back to verify it.
695 This function exports the configuration, sets up a new
696 enviroment, then calls
697 [`vault_config_exporter_shell_interpreter`][] on the export
698 script, verifying that each command ran successfully and that
699 the final configuration matches the initial one.
701 Args:
702 config:
703 The configuration to emit and read back.
705 """
706 prog_name_list = ("derivepassphrase", "vault") 1cbde
707 with io.StringIO() as outfile: 1cbde
708 cli_helpers.print_config_as_sh_script( 1cbde
709 config, outfile=outfile, prog_name_list=prog_name_list
710 )
711 script = outfile.getvalue() 1cbde
712 runner = machinery.CliRunner(mix_stderr=False) 1cbde
713 # TODO(the-13th-letter): Rewrite using parenthesized
714 # with-statements.
715 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
716 with contextlib.ExitStack() as stack: 1cbde
717 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1cbde
718 stack.enter_context( 1cbde
719 pytest_machinery.isolated_vault_config(
720 monkeypatch=monkeypatch,
721 runner=runner,
722 vault_config={"services": {}},
723 )
724 )
725 for result in vault_config_exporter_shell_interpreter(script): 1cbde
726 assert result.clean_exit() 1cbde
727 assert cli_helpers.load_config() == config 1cbde
729 @hypothesis.given(
730 global_config_settable=hypothesis_machinery.vault_full_service_config(),
731 global_config_importable=strategies.fixed_dictionaries(
732 {},
733 optional={
734 "key": strategies.text(
735 alphabet=strategies.characters(
736 min_codepoint=32,
737 max_codepoint=126,
738 ),
739 max_size=128,
740 ),
741 "phrase": strategies.text(
742 alphabet=strategies.characters(
743 min_codepoint=32,
744 max_codepoint=126,
745 ),
746 max_size=64,
747 ),
748 },
749 ),
750 )
751 def test_export_as_sh_global(
752 self,
753 global_config_settable: _types.VaultConfigServicesSettings,
754 global_config_importable: _types.VaultConfigServicesSettings,
755 ) -> None:
756 """Exporting configurations as sh(1) script works.
758 Here, we check global-only configurations which use both
759 settings settable via `--config` and settings requiring
760 `--import`.
762 The actual verification is done by [`export_as_sh_helper`][].
764 """
765 config: _types.VaultConfig = { 1c
766 "global": global_config_settable | global_config_importable,
767 "services": {},
768 }
769 assert _types.clean_up_falsy_vault_config_values(config) is not None 1c
770 assert _types.is_vault_config(config) 1c
771 return self.export_as_sh_helper(config) 1c
773 @hypothesis.given(
774 global_config_importable=strategies.fixed_dictionaries(
775 {},
776 optional={
777 "key": strategies.text(
778 alphabet=strategies.characters(
779 min_codepoint=32,
780 max_codepoint=126,
781 ),
782 max_size=128,
783 ),
784 "phrase": strategies.text(
785 alphabet=strategies.characters(
786 min_codepoint=32,
787 max_codepoint=126,
788 ),
789 max_size=64,
790 ),
791 },
792 ),
793 )
794 def test_export_as_sh_global_only_imports(
795 self,
796 global_config_importable: _types.VaultConfigServicesSettings,
797 ) -> None:
798 """Exporting configurations as sh(1) script works.
800 Here, we check global-only configurations which only use
801 settings requiring `--import`.
803 The actual verification is done by [`export_as_sh_helper`][].
805 """
806 config: _types.VaultConfig = { 1b
807 "global": global_config_importable,
808 "services": {},
809 }
810 assert _types.clean_up_falsy_vault_config_values(config) is not None 1b
811 assert _types.is_vault_config(config) 1b
812 if not config["global"]: 1b
813 config.pop("global") 1b
814 return self.export_as_sh_helper(config) 1b
816 @hypothesis.given(
817 service_name=strategies.text(
818 alphabet=strategies.characters(
819 min_codepoint=32,
820 max_codepoint=126,
821 ),
822 min_size=4,
823 max_size=64,
824 ),
825 service_config_settable=hypothesis_machinery.vault_full_service_config(),
826 service_config_importable=strategies.fixed_dictionaries(
827 {},
828 optional={
829 "key": strategies.text(
830 alphabet=strategies.characters(
831 min_codepoint=32,
832 max_codepoint=126,
833 ),
834 max_size=128,
835 ),
836 "phrase": strategies.text(
837 alphabet=strategies.characters(
838 min_codepoint=32,
839 max_codepoint=126,
840 ),
841 max_size=64,
842 ),
843 "notes": strategies.text(
844 alphabet=strategies.characters(
845 min_codepoint=32,
846 max_codepoint=126,
847 include_characters=("\n", "\f", "\t"),
848 ),
849 max_size=256,
850 ),
851 },
852 ),
853 )
854 def test_export_as_sh_service(
855 self,
856 service_name: str,
857 service_config_settable: _types.VaultConfigServicesSettings,
858 service_config_importable: _types.VaultConfigServicesSettings,
859 ) -> None:
860 """Exporting configurations as sh(1) script works.
862 Here, we check service-only configurations which use both
863 settings settable via `--config` and settings requiring
864 `--import`.
866 The actual verification is done by [`export_as_sh_helper`][].
868 """
869 config: _types.VaultConfig = { 1d
870 "services": {
871 service_name: (
872 service_config_settable | service_config_importable
873 ),
874 },
875 }
876 assert _types.clean_up_falsy_vault_config_values(config) is not None 1d
877 assert _types.is_vault_config(config) 1d
878 return self.export_as_sh_helper(config) 1d
880 @hypothesis.given(
881 service_name=strategies.text(
882 alphabet=strategies.characters(
883 min_codepoint=32,
884 max_codepoint=126,
885 ),
886 min_size=4,
887 max_size=64,
888 ),
889 service_config_importable=strategies.fixed_dictionaries(
890 {},
891 optional={
892 "key": strategies.text(
893 alphabet=strategies.characters(
894 min_codepoint=32,
895 max_codepoint=126,
896 ),
897 max_size=128,
898 ),
899 "phrase": strategies.text(
900 alphabet=strategies.characters(
901 min_codepoint=32,
902 max_codepoint=126,
903 ),
904 max_size=64,
905 ),
906 "notes": strategies.text(
907 alphabet=strategies.characters(
908 min_codepoint=32,
909 max_codepoint=126,
910 include_characters=("\n", "\f", "\t"),
911 ),
912 max_size=256,
913 ),
914 },
915 ),
916 )
917 def test_export_as_sh_service_only_imports(
918 self,
919 service_name: str,
920 service_config_importable: _types.VaultConfigServicesSettings,
921 ) -> None:
922 """Exporting configurations as sh(1) script works.
924 Here, we check service-only configurations which only use
925 settings requiring `--import`.
927 The actual verification is done by [`export_as_sh_helper`][].
929 """
930 config: _types.VaultConfig = { 1e
931 "services": {
932 service_name: service_config_importable,
933 },
934 }
935 assert _types.clean_up_falsy_vault_config_values(config) is not None 1e
936 assert _types.is_vault_config(config) 1e
937 return self.export_as_sh_helper(config) 1e
940class TestTempdir:
941 """Tests for the temporary directory handling utility functions."""
943 # The Annoying OS appears to silently truncate spaces at the end of
944 # filenames.
945 @hypothesis.given(
946 env_var=strategies.sampled_from(["TMPDIR", "TEMP", "TMP"]),
947 subdir=strategies.builds(
948 operator.add,
949 strategies.text(
950 tuple(" 0123456789abcdefghijklmnopqrstuvwxyz"),
951 min_size=11,
952 max_size=11,
953 ),
954 strategies.text(
955 tuple("0123456789abcdefghijklmnopqrstuvwxyz"),
956 min_size=1,
957 max_size=1,
958 ),
959 ),
960 )
961 @hypothesis.example(env_var="", subdir=".").via(
962 "static example; branch coverage (test)"
963 )
964 def test_get_tempdir(
965 self,
966 env_var: str,
967 subdir: str,
968 ) -> None:
969 """[`cli_helpers.get_tempdir`][] returns a temporary directory.
971 If it is not the same as the temporary directory determined by
972 [`tempfile.gettempdir`][], then assert that
973 `tempfile.gettempdir` returned the current directory and
974 `cli_helpers.get_tempdir` returned the configuration directory.
976 """
978 @contextlib.contextmanager 1g
979 def make_temporary_directory( 1g
980 path: pathlib.Path,
981 ) -> Generator[pathlib.Path, None, None]:
982 try: 1g
983 path.mkdir() 1g
984 yield path 1g
985 finally:
986 shutil.rmtree(path) 1g
988 runner = machinery.CliRunner(mix_stderr=False) 1g
989 # TODO(the-13th-letter): Rewrite using parenthesized
990 # with-statements.
991 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
992 with contextlib.ExitStack() as stack: 1g
993 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1g
994 stack.enter_context( 1g
995 pytest_machinery.isolated_vault_config(
996 monkeypatch=monkeypatch,
997 runner=runner,
998 vault_config={"services": {}},
999 )
1000 )
1001 old_tempdir = os.fsdecode(tempfile.gettempdir()) 1g
1002 monkeypatch.delenv("TMPDIR", raising=False) 1g
1003 monkeypatch.delenv("TEMP", raising=False) 1g
1004 monkeypatch.delenv("TMP", raising=False) 1g
1005 monkeypatch.setattr(tempfile, "tempdir", None) 1g
1006 temp_path = pathlib.Path.cwd() / subdir 1g
1007 if env_var: 1g
1008 monkeypatch.setenv(env_var, os.fsdecode(temp_path)) 1g
1009 stack.enter_context(make_temporary_directory(temp_path)) 1g
1010 new_tempdir = os.fsdecode(tempfile.gettempdir()) 1g
1011 hypothesis.assume( 1g
1012 temp_path.resolve() == pathlib.Path.cwd().resolve()
1013 or old_tempdir != new_tempdir
1014 )
1015 system_tempdir = os.fsdecode(tempfile.gettempdir()) 1g
1016 our_tempdir = cli_helpers.get_tempdir() 1g
1017 assert system_tempdir == os.fsdecode(our_tempdir) or ( 1g
1018 # TODO(the-13th-letter): `pytest_machinery.isolated_config`
1019 # guarantees that `Path.cwd() == config_filename(None)`.
1020 # So this sub-branch ought to never trigger in our
1021 # tests.
1022 system_tempdir == os.getcwd() # noqa: PTH109
1023 and our_tempdir == cli_helpers.config_filename(subsystem=None)
1024 )
1025 assert not temp_path.exists(), f"temp path {temp_path} not cleaned up!" 1g
1027 def test_get_tempdir_force_default(self) -> None:
1028 """[`cli_helpers.get_tempdir`][] returns a temporary directory.
1030 If all candidates are mocked to fail for the standard temporary
1031 directory choices, then we return the `derivepassphrase`
1032 configuration directory.
1034 """
1035 runner = machinery.CliRunner(mix_stderr=False) 1k
1036 # TODO(the-13th-letter): Rewrite using parenthesized
1037 # with-statements.
1038 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
1039 with contextlib.ExitStack() as stack: 1k
1040 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1k
1041 stack.enter_context( 1k
1042 pytest_machinery.isolated_vault_config(
1043 monkeypatch=monkeypatch,
1044 runner=runner,
1045 vault_config={"services": {}},
1046 )
1047 )
1048 monkeypatch.delenv("TMPDIR", raising=False) 1k
1049 monkeypatch.delenv("TEMP", raising=False) 1k
1050 monkeypatch.delenv("TMP", raising=False) 1k
1051 config_dir = cli_helpers.config_filename(subsystem=None) 1k
1053 def is_dir_false( 1k
1054 self: pathlib.Path,
1055 /,
1056 *,
1057 follow_symlinks: bool = False,
1058 ) -> bool:
1059 del self, follow_symlinks 1k
1060 return False 1k
1062 def is_dir_error( 1k
1063 self: pathlib.Path,
1064 /,
1065 *,
1066 follow_symlinks: bool = False,
1067 ) -> bool:
1068 del follow_symlinks 1k
1069 raise OSError( 1k
1070 errno.EACCES,
1071 os.strerror(errno.EACCES),
1072 str(self),
1073 )
1075 monkeypatch.setattr(pathlib.Path, "is_dir", is_dir_false) 1k
1076 assert cli_helpers.get_tempdir() == config_dir 1k
1078 monkeypatch.setattr(pathlib.Path, "is_dir", is_dir_error) 1k
1079 assert cli_helpers.get_tempdir() == config_dir 1k
1082# TODO(the-13th-letter): Use a better class name, or consider keeping them
1083# as top-level functions.
1084class TestMisc:
1085 """Miscellaneous tests for the command-line utility functions."""
1087 @Parametrize.DELETE_CONFIG_INPUT
1088 def test_repeated_config_deletion(
1089 self,
1090 command_line: list[str],
1091 config: _types.VaultConfig,
1092 result_config: _types.VaultConfig,
1093 ) -> None:
1094 """Repeatedly removing the same parts of a configuration works."""
1095 for start_config in [config, result_config]: 1o
1096 runner = machinery.CliRunner(mix_stderr=False) 1o
1097 # TODO(the-13th-letter): Rewrite using parenthesized
1098 # with-statements.
1099 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
1100 with contextlib.ExitStack() as stack: 1o
1101 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1o
1102 stack.enter_context( 1o
1103 pytest_machinery.isolated_vault_config(
1104 monkeypatch=monkeypatch,
1105 runner=runner,
1106 vault_config=start_config,
1107 )
1108 )
1109 result = runner.invoke( 1o
1110 cli.derivepassphrase_vault,
1111 command_line,
1112 catch_exceptions=False,
1113 )
1114 assert result.clean_exit(empty_stderr=True), ( 1o
1115 "expected clean exit"
1116 )
1117 with cli_helpers.config_filename(subsystem="vault").open( 1o
1118 encoding="UTF-8"
1119 ) as infile:
1120 config_readback = json.load(infile) 1o
1121 assert config_readback == result_config 1o
1123 def test_phrase_from_key_manually(self) -> None:
1124 """The dummy service, key and config settings are consistent."""
1125 assert ( 1s
1126 vault.Vault(
1127 phrase=DUMMY_PHRASE_FROM_KEY1,
1128 **DUMMY_CONFIG_SETTINGS_AS_CONSTRUCTOR_ARGS,
1129 ).generate(DUMMY_SERVICE)
1130 == DUMMY_RESULT_KEY1
1131 )
1133 @Parametrize.VALIDATION_FUNCTION_INPUT
1134 def test_validate_constraints_manually(
1135 self,
1136 vfunc: Callable[[click.Context, click.Parameter, Any], int | None],
1137 input: int,
1138 ) -> None:
1139 """Command-line argument constraint validation works."""
1140 ctx = cli.derivepassphrase_vault.make_context(cli.PROG_NAME, []) 1r
1141 param = cli.derivepassphrase_vault.params[0] 1r
1142 assert vfunc(ctx, param, input) == input 1r
1144 @Parametrize.CONNECTION_HINTS
1145 def test_get_suitable_ssh_keys(
1146 self,
1147 running_ssh_agent: data.RunningSSHAgentInfo,
1148 conn_hint: str,
1149 ) -> None:
1150 """[`cli_helpers.get_suitable_ssh_keys`][] works."""
1151 with pytest.MonkeyPatch.context() as monkeypatch: 1m
1152 monkeypatch.setattr( 1m
1153 ssh_agent.SSHAgentClient,
1154 "list_keys",
1155 callables.list_keys,
1156 )
1157 hint: ssh_agent.SSHAgentClient | _types.SSHAgentSocket | None
1158 # TODO(the-13th-letter): Rewrite using structural pattern
1159 # matching.
1160 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
1161 if conn_hint == "client": 1m
1162 hint = ssh_agent.SSHAgentClient() 1m
1163 elif conn_hint == "socket": 1m
1164 if isinstance( 1m
1165 running_ssh_agent.socket, str
1166 ): # pragma: no cover
1167 if not hasattr(socket, "AF_UNIX"): 1m
1168 pytest.skip("socket module does not support AF_UNIX") 1m
1169 # socket.AF_UNIX is not defined everywhere.
1170 hint = socket.socket(family=socket.AF_UNIX) # type: ignore[attr-defined] 1m
1171 hint.connect(running_ssh_agent.socket) 1m
1172 else: # pragma: no cover
1173 hint = running_ssh_agent.socket()
1174 else:
1175 assert conn_hint == "none" 1m
1176 hint = None 1m
1177 exception: Exception | None = None 1m
1178 try: 1m
1179 list(cli_helpers.get_suitable_ssh_keys(hint)) 1m
1180 except RuntimeError: # pragma: no cover
1181 pass
1182 except Exception as e: # noqa: BLE001 # pragma: no cover
1183 exception = e
1184 finally:
1185 assert exception is None, ( 1m
1186 "exception querying suitable SSH keys"
1187 )
1189 @Parametrize.KEY_TO_PHRASE_SETTINGS
1190 def test_key_to_phrase(
1191 self,
1192 use_stub_agent_with_address: None,
1193 list_keys_action: data.ListKeysAction | None,
1194 system_support_action: data.SystemSupportAction | None,
1195 address_action: data.SocketAddressAction | None,
1196 sign_action: data.SignAction,
1197 pattern: str,
1198 warnings_patterns: list[str],
1199 ) -> None:
1200 """All errors in [`cli_helpers.key_to_phrase`][] are handled."""
1201 del use_stub_agent_with_address 1f
1202 captured_warnings: list[str] = [] 1f
1204 class ErrCallback(BaseException): 1f
1205 def __init__(self, *args: Any, **kwargs: Any) -> None: 1f
1206 super().__init__(*args[:1]) 1f
1207 self.args = args 1f
1208 self.kwargs = kwargs 1f
1210 def err(*args: Any, **kwargs: Any) -> NoReturn: 1f
1211 raise ErrCallback(*args, **kwargs) 1f
1213 def warn(*args: Any) -> None: 1f
1214 if args: # pragma: no branch 1f
1215 captured_warnings.append(str(args[0])) 1f
1217 with ssh_agent.SSHAgentClient.ensure_agent_subcontext() as client: 1f
1218 loaded_keys = list(client.list_keys()) 1f
1219 loaded_key = base64.standard_b64encode(loaded_keys[0][0]) 1f
1220 with pytest.MonkeyPatch.context() as monkeypatch: 1f
1221 monkeypatch.setattr(ssh_agent.SSHAgentClient, "sign", sign_action) 1f
1222 if list_keys_action: 1f
1223 monkeypatch.setattr( 1f
1224 ssh_agent.SSHAgentClient, "list_keys", list_keys_action
1225 )
1226 if address_action: 1f
1227 address_action(monkeypatch) 1f
1228 if system_support_action: 1f
1229 system_support_action(monkeypatch) 1f
1231 with pytest.raises(ErrCallback, match=pattern) as excinfo: 1f
1232 cli_helpers.key_to_phrase( 1f
1233 loaded_key, error_callback=err, warning_callback=warn
1234 )
1236 for pat in warnings_patterns: 1f
1237 assert any([pat in string for string in captured_warnings]), ( 1f
1238 f"expected some warning message to match {pat}"
1239 )
1240 if list_keys_action == data.ListKeysAction.FAIL_RUNTIME: 1f
1241 assert excinfo.value.kwargs 1f
1242 assert isinstance( 1f
1243 excinfo.value.kwargs["exc_info"],
1244 ssh_agent.SSHAgentFailedError,
1245 )
1246 assert excinfo.value.kwargs["exc_info"].__context__ is not None 1f
1247 assert isinstance( 1f
1248 excinfo.value.kwargs["exc_info"].__context__,
1249 ssh_agent.TrailingDataError,
1250 )