Coverage for tests / test_derivepassphrase_cli / test_heavy_duty.py: 100.000%
387 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: heavy-duty tests."""
7from __future__ import annotations
9import contextlib
10import copy
11import errno
12import json
13import pathlib
14import queue
15import shutil
16from typing import TYPE_CHECKING, cast
18import hypothesis
19import pytest
20from hypothesis import stateful, strategies
21from typing_extensions import Any, NamedTuple, TypeAlias
23from derivepassphrase import _types, cli
24from derivepassphrase._internals import cli_helpers
25from tests import data, machinery
26from tests.machinery import hypothesis as hypothesis_machinery
27from tests.machinery import pytest as pytest_machinery
29if TYPE_CHECKING:
30 import multiprocessing
31 from collections.abc import Iterable, Sequence
32 from collections.abc import Set as AbstractSet
34 from typing_extensions import Literal
36# All tests in this module are heavy-duty tests.
37pytestmark = [pytest_machinery.heavy_duty]
39KNOWN_SERVICES = (data.DUMMY_SERVICE, "email", "bank", "work")
40"""Known service names. Used for the [`ConfigManagementStateMachine`][]."""
41VALID_PROPERTIES = (
42 "length",
43 "repeat",
44 "upper",
45 "lower",
46 "number",
47 "space",
48 "dash",
49 "symbol",
50)
51"""Known vault properties. Used for the [`ConfigManagementStateMachine`][]."""
54class Strategies:
55 """Common hypothesis data generation strategies."""
57 @staticmethod
58 def assemble_config(
59 global_data: _types.VaultConfigGlobalSettings,
60 service_data: Sequence[tuple[str, _types.VaultConfigServicesSettings]],
61 ) -> _types.VaultConfig:
62 """Return a vault config using the global and service data."""
63 services_dict = dict(service_data) 1a
64 return ( 1a
65 {"global": global_data, "services": services_dict}
66 if global_data
67 else {"services": services_dict}
68 )
70 @staticmethod
71 def build_reduced_vault_config_settings(
72 config: _types.VaultConfigServicesSettings,
73 keys_to_prune: AbstractSet[str],
74 ) -> _types.VaultConfigServicesSettings:
75 """Return a service settings object with certain keys pruned.
77 Args:
78 config:
79 The original service settings object.
80 keys_to_prune:
81 The keys to prune from the settings object.
83 """
84 config2 = copy.deepcopy(config) 1a
85 for key in keys_to_prune: 1a
86 config2.pop(key, None) # type: ignore[misc] 1a
87 return config2 1a
89 # Prefer this explicit composite strategy over `strategies.builds`
90 # because the type checker can better introspect this.
91 @strategies.composite
92 @staticmethod
93 def services_strategy(
94 draw: strategies.DrawFn,
95 ) -> _types.VaultConfigServicesSettings:
96 """Return a strategy to build incomplete service configurations.
98 Args:
99 draw:
100 The `draw` function, as provided for by hypothesis.
102 Returns:
103 A strategy that generates `vault` service configurations,
104 with some settings left unspecified.
106 """
107 config = draw( 1a
108 hypothesis_machinery.vault_full_service_config(), label="config"
109 )
110 keys_to_prune = draw( 1a
111 strategies.sets(
112 strategies.sampled_from(VALID_PROPERTIES), max_size=7
113 ),
114 label="keys_to_prune",
115 )
116 # The `hypothesis.strategies.composite` decorator cannot handle
117 # bound class or instance methods, so we must code this as
118 # a static method. As such, we cannot access sibling helper
119 # methods via `self` or `cls`, but must go via the class name.
120 return Strategies.build_reduced_vault_config_settings( 1a
121 config, keys_to_prune
122 )
124 @strategies.composite
125 @staticmethod
126 def service_name_and_data_strategy(
127 draw: strategies.DrawFn,
128 num_entries: int,
129 ) -> tuple[tuple[str, _types.VaultConfigServicesSettings], ...]:
130 """Return a strategy for tuples of service names and settings.
132 Will draw service names from [`KNOWN_SERVICES`][] and service
133 settings via [`services_strategy`][].
135 Args:
136 draw:
137 The `draw` function, as provided for by hypothesis.
138 num_entries:
139 The number of services to draw.
141 Returns:
142 A strategy that generates a sequence of pairs of service
143 names and service settings.
145 """
146 possible_services = list(KNOWN_SERVICES) 1a
147 selected_services: list[str] = [] 1a
148 for _ in range(num_entries): 1a
149 selected_services.append( 1a
150 draw(strategies.sampled_from(possible_services))
151 )
152 possible_services.remove(selected_services[-1]) 1a
153 return tuple( 1a
154 (service, draw(Strategies.services_strategy()))
155 for service in selected_services
156 )
158 @strategies.composite
159 @staticmethod
160 def vault_full_config(draw: strategies.DrawFn) -> _types.VaultConfig:
161 """Return a strategy to build full vault configurations."""
162 services = draw(Strategies.services_strategy(), label="services") 1a
163 num_entries = draw( 1a
164 strategies.integers(min_value=2, max_value=4), label="num_entries"
165 )
166 service_name_and_data = draw( 1a
167 Strategies.service_name_and_data_strategy(num_entries),
168 label="service_name_and_data",
169 )
170 return Strategies.assemble_config(services, service_name_and_data) 1a
172 @staticmethod
173 def maybe_unset_strategy() -> strategies.SearchStrategy[AbstractSet[str]]:
174 """Return a strategy to build sets of names to maybe unset."""
175 return strategies.sets(
176 strategies.sampled_from(VALID_PROPERTIES), max_size=3
177 )
179 @staticmethod
180 def service_name_strategy() -> strategies.SearchStrategy[str]:
181 """Return a strategy for service names."""
182 return strategies.sampled_from(KNOWN_SERVICES)
184 @staticmethod
185 def setting_strategy(
186 setting_bundle: stateful.Bundle[_types.VaultConfigServicesSettings],
187 ) -> strategies.SearchStrategy[_types.VaultConfigServicesSettings]:
188 """Return a strategy for setting objects, given a setting bundle."""
189 return setting_bundle.filter(bool)
191 @strategies.composite
192 @staticmethod
193 def config_and_service_strategy(
194 draw: strategies.DrawFn,
195 configuration: stateful.Bundle[_types.VaultConfig],
196 ) -> tuple[_types.VaultConfig, str]:
197 """Return a strategy for a vault config and a service name tuple."""
198 config_strategy = configuration.filter(lambda c: bool(c["services"])) 1a
199 config = draw(config_strategy, label="config") 1a
200 keys = tuple(config["services"].keys()) 1a
201 key = draw(strategies.sampled_from(keys), label="key") 1a
202 return (config, key) 1a
205class ConfigManagementStateMachine(stateful.RuleBasedStateMachine):
206 """A state machine recording changes in the vault configuration.
208 Record possible configuration states in bundles, then in each rule,
209 take a configuration and manipulate it somehow.
211 Attributes:
212 setting:
213 A bundle for single-service settings.
214 configuration:
215 A bundle for full vault configurations.
217 """
219 def __init__(self) -> None: 1ba
220 """Initialize self, set up context managers and enter them."""
221 super().__init__() 1a
222 self.runner = machinery.CliRunner(mix_stderr=False) 1a
223 self.exit_stack = contextlib.ExitStack().__enter__() 1a
224 self.monkeypatch = self.exit_stack.enter_context( 1a
225 pytest.MonkeyPatch().context()
226 )
227 self.isolated_config = self.exit_stack.enter_context( 1a
228 pytest_machinery.isolated_vault_config(
229 monkeypatch=self.monkeypatch,
230 runner=self.runner,
231 vault_config={"services": {}},
232 )
233 )
235 setting: stateful.Bundle[_types.VaultConfigServicesSettings] = (
236 stateful.Bundle("setting")
237 )
238 """"""
239 configuration: stateful.Bundle[_types.VaultConfig] = stateful.Bundle(
240 "configuration"
241 )
242 """"""
244 @stateful.initialize(
245 target=configuration,
246 configs=strategies.lists(
247 Strategies.vault_full_config(),
248 min_size=8,
249 max_size=8,
250 ),
251 )
252 def declare_initial_configs(
253 self,
254 configs: Iterable[_types.VaultConfig],
255 ) -> stateful.MultipleResults[_types.VaultConfig]:
256 """Initialize the configuration bundle with eight configurations."""
257 return stateful.multiple(*configs) 1a
259 @stateful.initialize(
260 target=setting,
261 configs=strategies.lists(
262 Strategies.vault_full_config(),
263 min_size=4,
264 max_size=4,
265 ),
266 )
267 def extract_initial_settings(
268 self,
269 configs: list[_types.VaultConfig],
270 ) -> stateful.MultipleResults[_types.VaultConfigServicesSettings]:
271 """Initialize the settings bundle with four service settings."""
272 settings: list[_types.VaultConfigServicesSettings] = [] 1a
273 for c in configs: 1a
274 settings.extend(c["services"].values()) 1a
275 return stateful.multiple(*map(copy.deepcopy, settings)) 1a
277 @staticmethod
278 def fold_configs(
279 c1: _types.VaultConfig, c2: _types.VaultConfig
280 ) -> _types.VaultConfig:
281 """Fold `c1` into `c2`, overriding the latter."""
282 new_global_dict = c1.get("global", c2.get("global")) 1a
283 if new_global_dict is not None: 1a
284 return { 1a
285 "global": new_global_dict,
286 "services": {**c2["services"], **c1["services"]},
287 }
288 return { 1a
289 "services": {**c2["services"], **c1["services"]},
290 }
292 def call_cli(
293 self,
294 command_line: list[str],
295 expected_config: _types.VaultConfig,
296 /,
297 *,
298 input: str | bytes | None = None,
299 overwrite: bool = False,
300 ) -> _types.VaultConfig:
301 """Call the command-line interface for config manipulation.
303 Args:
304 command_line:
305 The command-line to execute via
306 a [`machinery.CliRunner`][].
308 The `--overwrite-existing`/`--merge-existing` options
309 should be managed via the `overwrite` option instead of
310 being explicitly specified.
311 expected_config:
312 The expected configuration after calling this
313 command-line.
315 Raises:
316 AssertionError:
317 The command exited with an error status.
319 *Or*, the actual resulting configuration does not match
320 the expected configuration.
322 """
323 overwriting = ( 1a
324 ["--overwrite-existing"] if overwrite else ["--merge-existing"]
325 )
326 result = self.runner.invoke( 1a
327 cli.derivepassphrase_vault,
328 [*overwriting, *command_line],
329 input=input,
330 catch_exceptions=False,
331 )
332 assert result.clean_exit(empty_stderr=False) 1a
333 assert cli_helpers.load_config() == expected_config 1a
334 return expected_config 1a
336 def set_globals_expected_result(
337 self,
338 config: _types.VaultConfig,
339 setting: _types.VaultConfigGlobalSettings,
340 maybe_unset: AbstractSet[str],
341 overwrite: bool,
342 ) -> tuple[_types.VaultConfig, AbstractSet[str]]:
343 """Set the global settings of a configuration.
345 This is the "calculate the correct result" section of the
346 `set_globals` rule.
348 Args:
349 config:
350 The configuration to edit.
351 setting:
352 The new global settings.
353 maybe_unset:
354 Settings keys to additionally unset, if not already
355 present in the new settings. Corresponds to the
356 `--unset` command-line argument.
357 overwrite:
358 Overwrite the settings object if true, or merge if
359 false. Corresponds to the `--overwrite-existing` and
360 `--merge-existing` command-line arguments.
362 Returns:
363 A 2-tuple containing the amended configuration, then the
364 settings keys that were actually unset.
366 """
367 config = copy.deepcopy(config) 1a
368 setting = copy.deepcopy(setting) 1a
369 config_global = config.get( 1a
370 "global", cast("_types.VaultConfigGlobalSettings", {})
371 )
372 maybe_unset = set(maybe_unset) - setting.keys() 1a
373 if overwrite: 1a
374 config["global"] = config_global = cast( 1a
375 "_types.VaultConfigGlobalSettings", {}
376 )
377 elif maybe_unset: 1a
378 for key in maybe_unset: 1a
379 config_global.pop(key, None) # type: ignore[misc] 1a
380 config.setdefault("global", {}).update(setting) 1a
381 assert _types.is_vault_config(config) 1a
382 return config, maybe_unset 1a
384 @stateful.rule(
385 target=configuration,
386 config=configuration,
387 setting=Strategies.setting_strategy(setting),
388 maybe_unset=Strategies.maybe_unset_strategy(),
389 overwrite=strategies.booleans(),
390 )
391 def set_globals(
392 self,
393 config: _types.VaultConfig,
394 setting: _types.VaultConfigGlobalSettings,
395 maybe_unset: AbstractSet[str],
396 overwrite: bool,
397 ) -> _types.VaultConfig:
398 """Set the global settings of a configuration.
400 Args:
401 config:
402 The configuration to edit.
403 setting:
404 The new global settings.
405 maybe_unset:
406 Settings keys to additionally unset, if not already
407 present in the new settings. Corresponds to the
408 `--unset` command-line argument.
409 overwrite:
410 Overwrite the settings object if true, or merge if
411 false. Corresponds to the `--overwrite-existing` and
412 `--merge-existing` command-line arguments.
414 Returns:
415 The amended configuration.
417 """
418 cli_helpers.save_config(config) 1a
419 expected_config, maybe_unset = self.set_globals_expected_result( 1a
420 config=config,
421 setting=setting,
422 maybe_unset=maybe_unset,
423 overwrite=overwrite,
424 )
425 # NOTE: This relies on `settings` containing only the keys
426 # "length", "repeat", "upper", "lower", "number", "space",
427 # "dash" and "symbol".
428 unset_commands = [f"--unset={key}" for key in maybe_unset] 1a
429 set_commands = [ 1a
430 f"--{key}={value}"
431 for key, value in setting.items()
432 if key in VALID_PROPERTIES
433 ]
434 return self.call_cli( 1a
435 ["--config", *unset_commands, *set_commands],
436 expected_config,
437 overwrite=overwrite,
438 )
440 def set_service_expected_result(
441 self,
442 service: str,
443 config: _types.VaultConfig,
444 setting: _types.VaultConfigServicesSettings,
445 maybe_unset: AbstractSet[str],
446 overwrite: bool,
447 ) -> tuple[_types.VaultConfig, AbstractSet[str]]:
448 """Set the named service settings for a configuration.
450 This is the "calculate the correct result" section of the
451 `set_service` rule.
453 Args:
454 config:
455 The configuration to edit.
456 service:
457 The name of the service to set.
458 setting:
459 The new service settings.
460 maybe_unset:
461 Settings keys to additionally unset, if not already
462 present in the new settings. Corresponds to the
463 `--unset` command-line argument.
464 overwrite:
465 Overwrite the settings object if true, or merge if
466 false. Corresponds to the `--overwrite-existing` and
467 `--merge-existing` command-line arguments.
469 Returns:
470 A 2-tuple containing the amended configuration, then the
471 settings keys that were actually unset.
473 """
474 config = copy.deepcopy(config) 1a
475 config_service = config["services"].get(service, {}) 1a
476 maybe_unset = set(maybe_unset) - setting.keys() 1a
477 if overwrite: 1a
478 config["services"][service] = config_service = cast( 1a
479 "_types.VaultConfigServicesSettings", {}
480 )
481 elif maybe_unset: 1a
482 for key in maybe_unset: 1a
483 config_service.pop(key, None) # type: ignore[misc] 1a
484 config["services"].setdefault(service, {}).update(setting) 1a
485 assert _types.is_vault_config(config) 1a
486 return config, maybe_unset 1a
488 @stateful.rule(
489 target=configuration,
490 config=configuration,
491 service=Strategies.service_name_strategy(),
492 setting=Strategies.setting_strategy(setting),
493 maybe_unset=Strategies.maybe_unset_strategy(),
494 overwrite=strategies.booleans(),
495 )
496 def set_service(
497 self,
498 config: _types.VaultConfig,
499 service: str,
500 setting: _types.VaultConfigServicesSettings,
501 maybe_unset: AbstractSet[str],
502 overwrite: bool,
503 ) -> _types.VaultConfig:
504 """Set the named service settings for a configuration.
506 Args:
507 config:
508 The configuration to edit.
509 service:
510 The name of the service to set.
511 setting:
512 The new service settings.
513 maybe_unset:
514 Settings keys to additionally unset, if not already
515 present in the new settings. Corresponds to the
516 `--unset` command-line argument.
517 overwrite:
518 Overwrite the settings object if true, or merge if
519 false. Corresponds to the `--overwrite-existing` and
520 `--merge-existing` command-line arguments.
522 Returns:
523 The amended configuration.
525 """
526 cli_helpers.save_config(config) 1a
527 expected_config, maybe_unset = self.set_service_expected_result( 1a
528 service,
529 config=config,
530 setting=setting,
531 maybe_unset=maybe_unset,
532 overwrite=overwrite,
533 )
534 # NOTE: This relies on `settings` containing only the keys
535 # "length", "repeat", "upper", "lower", "number", "space",
536 # "dash" and "symbol".
537 unset_commands = [f"--unset={key}" for key in maybe_unset] 1a
538 set_commands = [ 1a
539 f"--{key}={value}"
540 for key, value in setting.items()
541 if key in VALID_PROPERTIES
542 ]
543 service_arg = ["--", service] 1a
544 return self.call_cli( 1a
545 ["--config", *unset_commands, *set_commands, *service_arg],
546 expected_config,
547 overwrite=overwrite,
548 )
550 def purge_global_expected_result(
551 self,
552 config: _types.VaultConfig,
553 ) -> _types.VaultConfig:
554 """Purge the globals of a configuration.
556 This is the "calculate the correct result" section of the
557 `purge_global` rule.
559 Args:
560 config:
561 The configuration to edit.
563 Returns:
564 The pruned configuration.
566 """
567 config = copy.deepcopy(config) 1a
568 config.pop("global", None) 1a
569 return config 1a
571 @stateful.rule(
572 target=configuration,
573 config=configuration,
574 )
575 def purge_global(
576 self,
577 config: _types.VaultConfig,
578 ) -> _types.VaultConfig:
579 """Purge the globals of a configuration.
581 Args:
582 config:
583 The configuration to edit.
585 Returns:
586 The pruned configuration.
588 """
589 cli_helpers.save_config(config) 1a
590 expected_config = self.purge_global_expected_result(config) 1a
591 return self.call_cli(["--delete-globals"], expected_config) 1a
593 def purge_service_expected_result(
594 self,
595 config: _types.VaultConfig,
596 service: str,
597 ) -> _types.VaultConfig:
598 """Purge the settings of a named service in a configuration.
600 This is the "calculate the correct result" section of the
601 `purge_global` rule.
603 Args:
604 config:
605 The configuration to edit.
606 service:
607 The service name to purge.
609 Returns:
610 The pruned configuration.
612 """
613 config = copy.deepcopy(config) 1a
614 config["services"].pop(service, None) 1a
615 return config 1a
617 @stateful.rule(
618 target=configuration,
619 config_and_service=Strategies.config_and_service_strategy(
620 configuration
621 ),
622 )
623 def purge_service(
624 self,
625 config_and_service: tuple[_types.VaultConfig, str],
626 ) -> _types.VaultConfig:
627 """Purge the settings of a named service in a configuration.
629 Args:
630 config_and_service:
631 A 2-tuple containing the configuration to edit, and the
632 service name to purge.
634 Returns:
635 The pruned configuration.
637 """
638 config, service = config_and_service 1a
639 cli_helpers.save_config(config) 1a
640 expected_config = self.purge_service_expected_result(config, service) 1a
641 return self.call_cli(["--delete", "--", service], expected_config) 1a
643 def purge_all_expected_result(self) -> _types.VaultConfig:
644 """Purge the entire configuration.
646 This is the "calculate the correct result" section of the
647 `purge_all` rule.
649 Returns:
650 The empty configuration.
652 """
653 return {"services": {}} 1a
655 @stateful.rule(
656 target=configuration,
657 config=configuration,
658 )
659 def purge_all(
660 self,
661 config: _types.VaultConfig,
662 ) -> _types.VaultConfig:
663 """Purge the entire configuration.
665 Args:
666 config:
667 The configuration to edit.
669 Returns:
670 The empty configuration.
672 """
673 cli_helpers.save_config(config) 1a
674 expected_config = self.purge_all_expected_result() 1a
675 return self.call_cli(["--clear"], expected_config) 1a
677 def import_configuration_expected_result(
678 self,
679 base_config: _types.VaultConfig,
680 config_to_import: _types.VaultConfig,
681 overwrite: bool,
682 ) -> _types.VaultConfig:
683 """Import the given configuration into a base configuration.
685 This is the "calculate the correct result" section of the
686 `import_configuration` rule.
688 Args:
689 base_config:
690 The configuration to import into.
691 config_to_import:
692 The configuration to import.
693 overwrite:
694 Overwrite the base configuration if true, or merge if
695 false. Corresponds to the `--overwrite-existing` and
696 `--merge-existing` command-line arguments.
698 Returns:
699 The imported or merged configuration.
701 """
702 result = ( 1a
703 copy.deepcopy(config_to_import)
704 if overwrite
705 else self.fold_configs(config_to_import, base_config)
706 )
707 assert _types.is_vault_config(result) 1a
708 return result 1a
710 @stateful.rule(
711 target=configuration,
712 base_config=configuration,
713 config_to_import=configuration,
714 overwrite=strategies.booleans(),
715 )
716 def import_configuration(
717 self,
718 base_config: _types.VaultConfig,
719 config_to_import: _types.VaultConfig,
720 overwrite: bool,
721 ) -> _types.VaultConfig:
722 """Import the given configuration into a base configuration.
724 Args:
725 base_config:
726 The configuration to import into.
727 config_to_import:
728 The configuration to import.
729 overwrite:
730 Overwrite the base configuration if true, or merge if
731 false. Corresponds to the `--overwrite-existing` and
732 `--merge-existing` command-line arguments.
734 Returns:
735 The imported or merged configuration.
737 """
738 cli_helpers.save_config(base_config) 1a
739 expected_config = self.import_configuration_expected_result( 1a
740 base_config=base_config,
741 config_to_import=config_to_import,
742 overwrite=overwrite,
743 )
744 return self.call_cli( 1a
745 ["--import", "-"],
746 expected_config,
747 input=json.dumps(config_to_import),
748 overwrite=overwrite,
749 )
751 def teardown(self) -> None:
752 """Upon teardown, exit all contexts entered in `__init__`."""
753 self.exit_stack.close() 1a
756TestConfigManagement = ConfigManagementStateMachine.TestCase
757"""The [`unittest.TestCase`][] class that will actually be run."""
760class FakeConfigurationMutexAction(NamedTuple):
761 """An action/a step in the [`FakeConfigurationMutexStateMachine`][].
763 Attributes:
764 command_line:
765 The command-line for `derivepassphrase vault` to execute.
766 input:
767 The input to this command.
769 """
771 command_line: list[str]
772 """"""
773 input: str | bytes | None = None
774 """"""
777def run_actions_handler(
778 id_num: int,
779 action: FakeConfigurationMutexAction,
780 *,
781 input_queue: queue.Queue,
782 output_queue: queue.Queue,
783 timeout: int,
784) -> None:
785 """Prepare the faked mutex, then run `action`.
787 This is a top-level handler function -- to be used in a new
788 [`multiprocessing.Process`][] -- to run a single action from the
789 [`FakeConfigurationMutexStateMachine`][]. Output from this function
790 must be sent down the output queue instead of relying on the call
791 stack.
793 Args:
794 id_num:
795 The internal ID of this subprocess.
796 action:
797 The action to execute.
798 input_queue:
799 The queue for data passed from the manager/parent process to
800 this subprocess.
801 output_queue:
802 The queue for data passed from this subprocess to the
803 manager/parent process.
804 timeout:
805 The maximum amount of time to wait for a data transfer along
806 the input or the output queue. If exceeded, we exit
807 immediately.
809 """
810 with pytest.MonkeyPatch.context() as monkeypatch:
811 monkeypatch.setattr(
812 cli_helpers,
813 "configuration_mutex",
814 lambda: FakeConfigurationMutexStateMachine.ConfigurationMutexStub(
815 my_id=id_num,
816 input_queue=input_queue,
817 output_queue=output_queue,
818 timeout=timeout,
819 ),
820 )
821 runner = machinery.CliRunner(mix_stderr=False)
822 try:
823 result = runner.invoke(
824 cli.derivepassphrase_vault,
825 args=action.command_line,
826 input=action.input,
827 catch_exceptions=True,
828 )
829 output_queue.put(
830 FakeConfigurationMutexStateMachine.IPCMessage(
831 id_num,
832 "result",
833 (
834 result.clean_exit(empty_stderr=False),
835 copy.copy(result.stdout),
836 copy.copy(result.stderr),
837 ),
838 ),
839 block=True,
840 timeout=timeout,
841 )
842 except Exception as exc: # pragma: no cover # noqa: BLE001
843 output_queue.put(
844 FakeConfigurationMutexStateMachine.IPCMessage(
845 id_num, "exception", exc
846 ),
847 block=False,
848 )
851@hypothesis.settings(
852 stateful_step_count=hypothesis_machinery.get_concurrency_step_count(),
853 max_examples=hypothesis_machinery.get_process_spawning_state_machine_examples_count(),
854 deadline=None,
855)
856class FakeConfigurationMutexStateMachine(stateful.RuleBasedStateMachine):
857 """A state machine simulating the (faked) configuration mutex.
859 Generate an ordered set of concurrent writers to the
860 derivepassphrase configuration, then test that the writers' accesses
861 are serialized correctly, i.e., test that the writers correctly use
862 the mutex to avoid concurrent accesses, under the assumption that
863 the mutex itself is correctly implemented.
865 We use a custom mutex implementation to both ensure that all writers
866 attempt to lock the configuration at the same time and that the lock
867 is granted in our desired order. This test is therefore independent
868 of the actual (operating system-specific) mutex implementation in
869 `derivepassphrase`.
871 Attributes:
872 setting:
873 A bundle for single-service settings.
874 configuration:
875 A bundle for full vault configurations.
876 actions:
877 A list of [actions][FakeConfigurationMutexAction] to
878 take. These will be executed in [`run_actions`][].
879 step_count:
880 The currently valid `hypothesis` step count for state
881 machine testing (if we can successfully determine it;
882 else a default value).
885 """
887 class IPCMessage(NamedTuple):
888 """A message for inter-process communication.
890 Used by the configuration mutex stub class to affect/signal the
891 control flow amongst the linked mutex clients.
893 Attributes:
894 child_id:
895 The ID of the sending or receiving child process.
896 message:
897 One of "ready", "go", "config", "result" or "exception".
898 payload:
899 The (optional) message payload.
901 """
903 child_id: int
904 """"""
905 message: Literal["ready", "go", "config", "result", "exception"]
906 """"""
907 payload: object | None
908 """"""
910 class ConfigurationMutexStub(cli_helpers.ConfigurationMutex):
911 """Configuration mutex subclass that enforces a locking order.
913 Each configuration mutex stub object ("mutex client") has an
914 associated ID, and one read-only and one write-only pipe
915 (actually: [`multiprocessing.Queue`][] objects) to the "manager"
916 instance coordinating these stub objects. First, the mutex
917 client signals readiness, then the manager signals when the
918 mutex shall be considered "acquired", then finally the mutex
919 client sends the result back (simultaneously releasing the mutex
920 again). The manager may optionally send an abort signal if the
921 operations take too long.
923 This subclass also copies the effective vault configuration
924 to `intermediate_configs` upon releasing the lock.
926 """
928 def __init__(
929 self,
930 *,
931 my_id: int,
932 timeout: int,
933 input_queue: queue.Queue[
934 FakeConfigurationMutexStateMachine.IPCMessage
935 ],
936 output_queue: queue.Queue[
937 FakeConfigurationMutexStateMachine.IPCMessage
938 ],
939 ) -> None:
940 """Initialize this mutex client.
942 Args:
943 my_id:
944 The ID of this client.
945 timeout:
946 The timeout for each get and put operation on the
947 queues.
948 input_queue:
949 The message queue for IPC messages from the manager
950 instance to this mutex client.
951 output_queue:
952 The message queue for IPC messages from this mutex
953 client to the manager instance.
955 """
956 super().__init__()
958 def lock() -> None:
959 """Simulate locking of the mutex.
961 Issue a "ready" message, wait for a "go", then return.
962 If an exception occurs, issue an "exception" message,
963 then raise the exception.
965 """
966 IPCMessage: TypeAlias = (
967 FakeConfigurationMutexStateMachine.IPCMessage
968 )
969 try:
970 output_queue.put(
971 IPCMessage(my_id, "ready", None),
972 block=True,
973 timeout=timeout,
974 )
975 ok = input_queue.get(block=True, timeout=timeout)
976 if ok != IPCMessage(my_id, "go", None): # pragma: no cover
977 output_queue.put(
978 IPCMessage(my_id, "exception", ok), block=False
979 )
980 raise (
981 ok[2]
982 if isinstance(ok[2], BaseException)
983 else RuntimeError(ok[2])
984 )
985 except (queue.Empty, queue.Full) as exc: # pragma: no cover
986 output_queue.put(
987 IPCMessage(my_id, "exception", exc), block=False
988 )
989 return
991 def unlock() -> None:
992 """Simulate unlocking of the mutex.
994 Issue a "config" message, then return. If an exception
995 occurs, issue an "exception" message, then raise the
996 exception.
998 """
999 IPCMessage: TypeAlias = (
1000 FakeConfigurationMutexStateMachine.IPCMessage
1001 )
1002 try:
1003 output_queue.put(
1004 IPCMessage(
1005 my_id,
1006 "config",
1007 copy.copy(cli_helpers.load_config()),
1008 ),
1009 block=True,
1010 timeout=timeout,
1011 )
1012 except (queue.Empty, queue.Full) as exc: # pragma: no cover
1013 output_queue.put(
1014 IPCMessage(my_id, "exception", exc), block=False
1015 )
1016 raise
1018 self.lock = lock
1019 self.unlock = unlock
1021 setting: stateful.Bundle[_types.VaultConfigServicesSettings] = (
1022 stateful.Bundle("setting")
1023 )
1024 """"""
1025 configuration: stateful.Bundle[_types.VaultConfig] = stateful.Bundle(
1026 "configuration"
1027 )
1028 """"""
1030 def __init__(self, *args: Any, **kwargs: Any) -> None: 1ba
1031 """Initialize the state machine."""
1032 super().__init__(*args, **kwargs) 1a
1033 self.actions: list[FakeConfigurationMutexAction] = [] 1a
1034 """""" 1a
1035 # Determine the step count by poking around in the hypothesis
1036 # internals. As this isn't guaranteed to be stable, turn off
1037 # coverage.
1038 try: # pragma: no cover 1a
1039 settings: hypothesis.settings | None
1040 settings = FakeConfigurationMutexStateMachine.TestCase.settings 1a
1041 except AttributeError: # pragma: no cover
1042 settings = None
1043 self.step_count: int = hypothesis_machinery.get_concurrency_step_count( 1a
1044 settings
1045 )
1046 """""" 1a
1048 @staticmethod
1049 def overwrite_or_merge_commands(*, overwrite: bool = False) -> list[str]:
1050 """Return a partial command-line for overwriting or merging.
1052 Args:
1053 overwrite:
1054 If true, overwrite the configuration, else merge in the
1055 relevant parts.
1057 Returns:
1058 A list of command-line options for selecting config
1059 overwriting or config merging.
1061 """
1062 return ["--overwrite-existing"] if overwrite else ["--merge-existing"] 1a
1064 @stateful.initialize(
1065 target=configuration,
1066 configs=strategies.lists(
1067 Strategies.vault_full_config(),
1068 min_size=8,
1069 max_size=8,
1070 ),
1071 )
1072 def declare_initial_configs(
1073 self,
1074 configs: list[_types.VaultConfig],
1075 ) -> stateful.MultipleResults[_types.VaultConfig]:
1076 """Initialize the configuration bundle with eight configurations."""
1077 return stateful.multiple(*configs) 1a
1079 @stateful.initialize(
1080 target=setting,
1081 configs=strategies.lists(
1082 Strategies.vault_full_config(),
1083 min_size=4,
1084 max_size=4,
1085 ),
1086 )
1087 def extract_initial_settings(
1088 self,
1089 configs: list[_types.VaultConfig],
1090 ) -> stateful.MultipleResults[_types.VaultConfigServicesSettings]:
1091 """Initialize the settings bundle with four service settings."""
1092 settings: list[_types.VaultConfigServicesSettings] = [] 1a
1093 for c in configs: 1a
1094 settings.extend(c["services"].values()) 1a
1095 return stateful.multiple(*map(copy.deepcopy, settings)) 1a
1097 @stateful.initialize(
1098 config=Strategies.vault_full_config(),
1099 )
1100 def declare_initial_action(
1101 self,
1102 config: _types.VaultConfig,
1103 ) -> None:
1104 """Initialize the actions bundle from the configuration bundle.
1106 This is roughly comparable to the
1107 [`add_import_configuration_action`][] general rule, but adding
1108 it as a separate initialize rule avoids having to guard every
1109 other action-amending rule against empty action sequences, which
1110 would discard huge portions of the rule selection search space
1111 and thus trigger loads of hypothesis health check warnings.
1113 """
1114 command_line = ["--import", "-", "--overwrite-existing"] 1a
1115 input = json.dumps(config) # noqa: A001 1a
1116 hypothesis.note(f"# {command_line = }, {input = }") 1a
1117 action = FakeConfigurationMutexAction( 1a
1118 command_line=command_line, input=input
1119 )
1120 self.actions.append(action) 1a
1122 @stateful.rule(
1123 setting=Strategies.setting_strategy(setting),
1124 maybe_unset=Strategies.maybe_unset_strategy(),
1125 overwrite=strategies.booleans(),
1126 )
1127 def add_set_globals_action(
1128 self,
1129 setting: _types.VaultConfigGlobalSettings,
1130 maybe_unset: set[str],
1131 overwrite: bool,
1132 ) -> None:
1133 """Set the global settings of a configuration.
1135 Args:
1136 setting:
1137 The new global settings.
1138 maybe_unset:
1139 Settings keys to additionally unset, if not already
1140 present in the new settings. Corresponds to the
1141 `--unset` command-line argument.
1142 overwrite:
1143 Overwrite the settings object if true, or merge if
1144 false. Corresponds to the `--overwrite-existing` and
1145 `--merge-existing` command-line arguments.
1147 """
1148 maybe_unset = set(maybe_unset) - setting.keys() 1a
1149 # NOTE: This relies on `settings` containing only the keys
1150 # "length", "repeat", "upper", "lower", "number", "space",
1151 # "dash" and "symbol".
1152 unset_commands = [f"--unset={key}" for key in maybe_unset] 1a
1153 set_commands = [ 1a
1154 f"--{key}={value}"
1155 for key, value in setting.items()
1156 if key in VALID_PROPERTIES
1157 ]
1158 command_line = [ 1a
1159 "--config",
1160 *self.overwrite_or_merge_commands(overwrite=overwrite),
1161 *unset_commands,
1162 *set_commands,
1163 ]
1164 input = None # noqa: A001 1a
1165 hypothesis.note(f"# {command_line = }, {input = }") 1a
1166 action = FakeConfigurationMutexAction( 1a
1167 command_line=command_line, input=input
1168 )
1169 self.actions.append(action) 1a
1171 @stateful.rule(
1172 service=Strategies.service_name_strategy(),
1173 setting=Strategies.setting_strategy(setting),
1174 maybe_unset=Strategies.maybe_unset_strategy(),
1175 overwrite=strategies.booleans(),
1176 )
1177 def add_set_service_action(
1178 self,
1179 service: str,
1180 setting: _types.VaultConfigServicesSettings,
1181 maybe_unset: set[str],
1182 overwrite: bool,
1183 ) -> None:
1184 """Set the named service settings for a configuration.
1186 Args:
1187 service:
1188 The name of the service to set.
1189 setting:
1190 The new service settings.
1191 maybe_unset:
1192 Settings keys to additionally unset, if not already
1193 present in the new settings. Corresponds to the
1194 `--unset` command-line argument.
1195 overwrite:
1196 Overwrite the settings object if true, or merge if
1197 false. Corresponds to the `--overwrite-existing` and
1198 `--merge-existing` command-line arguments.
1200 """
1201 maybe_unset = set(maybe_unset) - setting.keys() 1a
1202 # NOTE: This relies on `settings` containing only the keys
1203 # "length", "repeat", "upper", "lower", "number", "space",
1204 # "dash" and "symbol".
1205 unset_commands = [f"--unset={key}" for key in maybe_unset] 1a
1206 set_commands = [ 1a
1207 f"--{key}={value}"
1208 for key, value in setting.items()
1209 if key in VALID_PROPERTIES
1210 ]
1211 command_line = [ 1a
1212 "--config",
1213 *self.overwrite_or_merge_commands(overwrite=overwrite),
1214 *unset_commands,
1215 *set_commands,
1216 "--",
1217 service,
1218 ]
1219 input = None # noqa: A001 1a
1220 hypothesis.note(f"# {command_line = }, {input = }") 1a
1221 action = FakeConfigurationMutexAction( 1a
1222 command_line=command_line, input=input
1223 )
1224 self.actions.append(action) 1a
1226 @stateful.rule()
1227 def add_purge_global_action(
1228 self,
1229 ) -> None:
1230 """Purge the globals of a configuration."""
1231 command_line = ["--delete-globals"] 1a
1232 input = None # 'y' # noqa: A001 1a
1233 hypothesis.note(f"# {command_line = }, {input = }") 1a
1234 action = FakeConfigurationMutexAction( 1a
1235 command_line=command_line, input=input
1236 )
1237 self.actions.append(action) 1a
1239 @stateful.rule(
1240 service=Strategies.service_name_strategy(),
1241 )
1242 def add_purge_service_action(
1243 self,
1244 service: str,
1245 ) -> None:
1246 """Purge the settings of a named service in a configuration.
1248 Args:
1249 service:
1250 The service name to purge.
1252 """
1253 command_line = ["--delete", "--", service] 1a
1254 input = None # 'y' # noqa: A001 1a
1255 hypothesis.note(f"# {command_line = }, {input = }") 1a
1256 action = FakeConfigurationMutexAction( 1a
1257 command_line=command_line, input=input
1258 )
1259 self.actions.append(action) 1a
1261 @stateful.rule()
1262 def add_purge_all_action(
1263 self,
1264 ) -> None:
1265 """Purge the entire configuration."""
1266 command_line = ["--clear"] 1a
1267 input = None # 'y' # noqa: A001 1a
1268 hypothesis.note(f"# {command_line = }, {input = }") 1a
1269 action = FakeConfigurationMutexAction( 1a
1270 command_line=command_line, input=input
1271 )
1272 self.actions.append(action) 1a
1274 @stateful.rule(
1275 config_to_import=configuration,
1276 overwrite=strategies.booleans(),
1277 )
1278 def add_import_configuration_action(
1279 self,
1280 config_to_import: _types.VaultConfig,
1281 overwrite: bool,
1282 ) -> None:
1283 """Import the given configuration.
1285 Args:
1286 config_to_import:
1287 The configuration to import.
1288 overwrite:
1289 Overwrite the base configuration if true, or merge if
1290 false. Corresponds to the `--overwrite-existing` and
1291 `--merge-existing` command-line arguments.
1293 """
1294 command_line = [ 1a
1295 "--import",
1296 "-",
1297 *self.overwrite_or_merge_commands(overwrite=overwrite),
1298 ]
1299 input = json.dumps(config_to_import) # noqa: A001 1a
1300 hypothesis.note(f"# {command_line = }, {input = }") 1a
1301 action = FakeConfigurationMutexAction( 1a
1302 command_line=command_line, input=input
1303 )
1304 self.actions.append(action) 1a
1306 @stateful.precondition(lambda self: len(self.actions) > 0) 1ba
1307 @stateful.invariant()
1308 def run_actions( # noqa: C901
1309 self,
1310 ) -> None:
1311 """Run the actions, serially and concurrently.
1313 Run the actions once serially, then once more concurrently with
1314 the faked configuration mutex, and assert that both runs yield
1315 identical intermediate and final results.
1317 We must run the concurrent version in processes, not threads or
1318 Python async functions, because the `click` testing machinery
1319 manipulates global properties (e.g. the standard I/O streams,
1320 the current directory, and the environment), and we require this
1321 manipulation to happen in a time-overlapped manner.
1323 However, running multiple processes increases the risk of the
1324 operating system imposing process count or memory limits on us.
1325 We therefore skip the test as a whole if we fail to start a new
1326 process due to lack of necessary resources (memory, processes,
1327 or open file descriptors).
1329 """
1330 if not TYPE_CHECKING: # pragma: no branch 1a
1331 multiprocessing = pytest.importorskip("multiprocessing") 1a
1332 IPCMessage: TypeAlias = FakeConfigurationMutexStateMachine.IPCMessage 1a
1333 intermediate_configs: dict[int, _types.VaultConfig] = {} 1a
1334 intermediate_results: dict[ 1a
1335 int, tuple[bool, str | None, str | None]
1336 ] = {}
1337 true_configs: dict[int, _types.VaultConfig] = {} 1a
1338 true_results: dict[int, tuple[bool, str | None, str | None]] = {} 1a
1339 timeout = 30 # Hopefully slow enough to accomodate The Annoying OS. 1a
1340 actions = self.actions 1a
1341 mp = multiprocessing.get_context() 1a
1342 # Coverage tracking writes coverage data to the current working
1343 # directory, but because the subprocesses are spawned within the
1344 # `pytest_machinery.isolated_vault_config` context manager,
1345 # their starting working directory is the isolated one, not the
1346 # original one.
1347 orig_cwd = pathlib.Path.cwd() 1a
1349 fatal_process_creation_errnos = { 1a
1350 # Specified by POSIX for fork(3).
1351 errno.ENOMEM,
1352 # Specified by POSIX for fork(3).
1353 errno.EAGAIN,
1354 # Specified by Linux/glibc for fork(3)
1355 getattr(errno, "ENOSYS", errno.ENOMEM),
1356 # Specified by POSIX for posix_spawn(3).
1357 errno.EINVAL,
1358 }
1360 hypothesis.note(f"# {actions = }") 1a
1362 stack = contextlib.ExitStack() 1a
1363 with stack: 1a
1364 runner = machinery.CliRunner(mix_stderr=False) 1a
1365 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1a
1366 stack.enter_context( 1a
1367 pytest_machinery.isolated_vault_config(
1368 monkeypatch=monkeypatch,
1369 runner=runner,
1370 vault_config={"services": {}},
1371 )
1372 )
1373 for i, action in enumerate(actions): 1a
1374 result = runner.invoke( 1a
1375 cli.derivepassphrase_vault,
1376 args=action.command_line,
1377 input=action.input,
1378 catch_exceptions=True,
1379 )
1380 true_configs[i] = copy.copy(cli_helpers.load_config()) 1a
1381 true_results[i] = ( 1a
1382 result.clean_exit(empty_stderr=False),
1383 result.stdout,
1384 result.stderr,
1385 )
1387 with stack: # noqa: PLR1702 1a
1388 runner = machinery.CliRunner(mix_stderr=False) 1a
1389 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1a
1390 stack.enter_context( 1a
1391 pytest_machinery.isolated_vault_config(
1392 monkeypatch=monkeypatch,
1393 runner=runner,
1394 vault_config={"services": {}},
1395 )
1396 )
1398 child_output_queue: multiprocessing.Queue[IPCMessage] = mp.Queue() 1a
1399 child_input_queues: list[ 1a
1400 multiprocessing.Queue[IPCMessage] | None
1401 ] = []
1402 processes: list[multiprocessing.process.BaseProcess] = [] 1a
1403 processes_pending: set[multiprocessing.process.BaseProcess] = set() 1a
1404 ready_wait: set[int] = set() 1a
1406 try: 1a
1407 for i, action in enumerate(actions): 1a
1408 q: multiprocessing.Queue[IPCMessage] | None = mp.Queue() 1a
1409 try: 1a
1410 p: multiprocessing.process.BaseProcess = mp.Process( 1a
1411 name=f"fake-mutex-action-{i:02d}",
1412 target=run_actions_handler,
1413 kwargs={
1414 "id_num": i,
1415 "timeout": timeout,
1416 "action": action,
1417 "input_queue": q,
1418 "output_queue": child_output_queue,
1419 },
1420 daemon=False,
1421 )
1422 p.start() 1a
1423 except OSError as exc: # pragma: no cover
1424 if exc.errno in fatal_process_creation_errnos:
1425 pytest.skip(
1426 "cannot test mutex functionality due to "
1427 "lack of system resources for "
1428 "creating enough subprocesses"
1429 )
1430 raise
1431 else:
1432 processes.append(p) 1a
1433 processes_pending.add(p) 1a
1434 child_input_queues.append(q) 1a
1435 ready_wait.add(i) 1a
1437 while processes_pending: 1a
1438 try: 1a
1439 self.mainloop( 1a
1440 timeout=timeout,
1441 child_output_queue=child_output_queue,
1442 child_input_queues=child_input_queues,
1443 ready_wait=ready_wait,
1444 intermediate_configs=intermediate_configs,
1445 intermediate_results=intermediate_results,
1446 processes=processes,
1447 processes_pending=processes_pending,
1448 block=True,
1449 )
1450 except Exception as exc: # pragma: no cover
1451 for i, q in enumerate(child_input_queues):
1452 if q:
1453 q.put(IPCMessage(i, "exception", exc))
1454 for p in processes_pending:
1455 p.join(timeout=timeout)
1456 raise
1457 finally:
1458 try: 1a
1459 while True: 1a
1460 try: 1a
1461 self.mainloop( 1a
1462 timeout=timeout,
1463 child_output_queue=child_output_queue,
1464 child_input_queues=child_input_queues,
1465 ready_wait=ready_wait,
1466 intermediate_configs=intermediate_configs,
1467 intermediate_results=intermediate_results,
1468 processes=processes,
1469 processes_pending=processes_pending,
1470 block=False,
1471 )
1472 except queue.Empty: 1a
1473 break 1a
1474 finally:
1475 # The subprocesses have this
1476 # `pytest_machinery.isolated_vault_config` directory
1477 # as their startup and working directory, so systems
1478 # like coverage tracking write their data files to
1479 # this directory. We need to manually move them
1480 # back to the starting working directory if they are
1481 # to survive this test.
1482 for coverage_file in pathlib.Path.cwd().glob( 1a
1483 ".coverage.*"
1484 ):
1485 shutil.move(coverage_file, orig_cwd) 1a
1486 hypothesis.note( 1a
1487 f"# {true_results = }, {intermediate_results = }, "
1488 f"identical = {true_results == intermediate_results}"
1489 )
1490 hypothesis.note( 1a
1491 f"# {true_configs = }, {intermediate_configs = }, "
1492 f"identical = {true_configs == intermediate_configs}"
1493 )
1494 assert intermediate_results == true_results 1a
1495 assert intermediate_configs == true_configs 1a
1497 @staticmethod
1498 def mainloop(
1499 *,
1500 timeout: int,
1501 child_output_queue: multiprocessing.Queue[
1502 FakeConfigurationMutexStateMachine.IPCMessage
1503 ],
1504 child_input_queues: list[
1505 multiprocessing.Queue[
1506 FakeConfigurationMutexStateMachine.IPCMessage
1507 ]
1508 | None
1509 ],
1510 ready_wait: set[int],
1511 intermediate_configs: dict[int, _types.VaultConfig],
1512 intermediate_results: dict[int, tuple[bool, str | None, str | None]],
1513 processes: list[multiprocessing.process.BaseProcess],
1514 processes_pending: set[multiprocessing.process.BaseProcess],
1515 block: bool = True,
1516 ) -> None:
1517 """Run a single step of the IPC main loop operation.
1519 We check for a message (in a blocking manner, or not, depending
1520 on the options) on the common child output queue and react to
1521 it, accordingly. This may entail sending other child processes
1522 a message via their input queue, or otherwise updating the
1523 bookkeeping data structures.
1525 Specifically, we take the following actions:
1527 1. We re-raise any exceptions that occurred in a child process.
1529 2. We issue work clearance for child number 0 once all child
1530 processes have signalled readiness.
1532 3. We store any results that a child process sends via the
1533 output queue.
1535 4. We react to any child process exiting by checking its return
1536 status and by issuing work clearance to the next child
1537 process, if any.
1539 We take a number of arguments, many of which are data structures
1540 that, once they are set up, will be managed by us exclusively.
1541 These are marked as "managed by us" below. **Do not
1542 manipulate** these arguments once they have been passed in to us
1543 **unless you are absolutely 100% sure you know what you're
1544 doing**.
1546 Args:
1547 timeout:
1548 The maximum time to wait for sending or receiving an IPC
1549 message on a queue.
1550 block:
1551 If true, block until a message has been sent or
1552 received, else return immediately.
1553 intermediate_configs:
1554 A `dict` in which to store the `vault` configurations
1555 returned from the child processes. Each configuration
1556 is keyed by the step number (which is equal to the child
1557 process ID).
1558 intermediate_results:
1559 A `dict` in which to store the result tuples of running
1560 the chain of actions at the given step number, based on
1561 the results of the previous step number. Each result
1562 tuple contains the exit status (`True` if successful,
1563 else `False`), the contents of standard output, and the
1564 contents of standard error.
1566 Other args:
1567 child_output_queue:
1568 The common output queue of all child processes. Managed
1569 by us.
1570 child_input_queues:
1571 A list of input queues for the child processes. Managed
1572 by us.
1573 ready_wait:
1574 The set of child process IDs who have yet to signal
1575 readiness. Managed by us.
1576 processes:
1577 The child process objects sitting at the other end of
1578 the respective input queue. Managed by us.
1579 processes_pending:
1580 The set of child processes currently still alive.
1581 Managed by us.
1583 """
1584 IPCMessage: TypeAlias = FakeConfigurationMutexStateMachine.IPCMessage 1a
1585 msg = child_output_queue.get(block=block, timeout=timeout) 1a
1586 # TODO(the-13th-letter): Rewrite using structural pattern
1587 # matching.
1588 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
1589 if ( # pragma: no cover 1a
1590 isinstance(msg, IPCMessage)
1591 and msg[1] == "exception"
1592 and isinstance(msg[2], Exception)
1593 ):
1594 e = msg[2]
1595 raise e
1596 if isinstance(msg, IPCMessage) and msg[1] == "ready": 1a
1597 n = msg[0] 1a
1598 ready_wait.remove(n) 1a
1599 if not ready_wait: 1a
1600 assert child_input_queues 1a
1601 assert child_input_queues[0] 1a
1602 child_input_queues[0].put( 1a
1603 IPCMessage(0, "go", None),
1604 block=True,
1605 timeout=timeout,
1606 )
1607 elif isinstance(msg, IPCMessage) and msg[1] == "config": 1a
1608 n = msg[0] 1a
1609 config = msg[2] 1a
1610 intermediate_configs[n] = cast("_types.VaultConfig", config) 1a
1611 elif isinstance(msg, IPCMessage) and msg[1] == "result": 1a
1612 n = msg[0] 1a
1613 result_ = msg[2] 1a
1614 result_tuple: tuple[bool, str | None, str | None] = cast( 1a
1615 "tuple[bool, str | None, str | None]", result_
1616 )
1617 intermediate_results[n] = result_tuple 1a
1618 child_input_queues[n] = None 1a
1619 p = processes[n] 1a
1620 p.join(timeout=timeout) 1a
1621 assert not p.is_alive() 1a
1622 processes_pending.remove(p) 1a
1623 assert result_tuple[0], ( 1a
1624 f"action #{n} exited with an error: {result_tuple!r}"
1625 )
1626 if n + 1 < len(processes): 1a
1627 next_child_input_queue = child_input_queues[n + 1] 1a
1628 assert next_child_input_queue 1a
1629 next_child_input_queue.put( 1a
1630 IPCMessage(n + 1, "go", None),
1631 block=True,
1632 timeout=timeout,
1633 )
1634 else:
1635 raise AssertionError()
1638TestFakedConfigurationMutex = (
1639 pytest_machinery.skip_if_no_multiprocessing_support(
1640 FakeConfigurationMutexStateMachine.TestCase
1641 )
1642)
1643"""The [`unittest.TestCase`][] class that will actually be run."""