Add a stateful hypothesis test for config importing and merging (eb2e29c) - derivepassphrase.git

@@ -11,21 +11,19 @@ import json
 import os
 import shutil
 import socket
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 import click.testing
 import hypothesis
 import pytest
-from hypothesis import strategies
-from typing_extensions import NamedTuple
+from hypothesis import stateful, strategies
+from typing_extensions import Any, NamedTuple
 
 import tests
 from derivepassphrase import _types, cli, ssh_agent, vault
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
-
-    from typing_extensions import Any
+    from collections.abc import Callable, Iterable
 
 DUMMY_SERVICE = tests.DUMMY_SERVICE
 DUMMY_PASSPHRASE = tests.DUMMY_PASSPHRASE
@@ -486,9 +484,7 @@ class TestCLI:
             },
             {
                 'global': {'key': DUMMY_KEY1_B64},
-                'services': {
-                    DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS.copy()
-                },
+                'services': {DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS.copy()},
             },
         ],
     )
@@ -2100,3 +2096,283 @@ class TestCLITransition:
         assert (
             'Warning: Failed to migrate to ' in result.stderr
         ), 'expected known warning message in stderr'
+
+
+class ConfigMergingStateMachine(stateful.RuleBasedStateMachine):
+    def __init__(self) -> None:
+        super().__init__()
+        self.runner = click.testing.CliRunner(mix_stderr=False)
+        self.exit_stack = contextlib.ExitStack().__enter__()
+        self.monkeypatch = self.exit_stack.enter_context(
+            pytest.MonkeyPatch().context()
+        )
+        self.isolated_config = self.exit_stack.enter_context(
+            tests.isolated_vault_config(
+                monkeypatch=self.monkeypatch,
+                runner=self.runner,
+                config={'services': {}},
+            )
+        )
+        self.current_config = cli._load_config()
+
+    known_services = stateful.Bundle('known_services')
+    settings = stateful.Bundle('settings')
+    configurations = stateful.Bundle('configurations')
+
+    @stateful.initialize(target=configurations)
+    def init_empty_configuration(self) -> _types.VaultConfig:
+        return copy.deepcopy(self.current_config)
+
+    @stateful.initialize(target=configurations)
+    def init_standard_testing_configuration(self) -> _types.VaultConfig:
+        return cast(
+            _types.VaultConfig,
+            {
+                'services': {
+                    DUMMY_SERVICE: copy.deepcopy(DUMMY_CONFIG_SETTINGS)
+                }
+            },
+        )
+
+    @stateful.initialize(
+        target=known_services,
+        service_names=strategies.lists(
+            strategies.text(
+                strategies.characters(min_codepoint=32, max_codepoint=126),
+                min_size=1,
+                max_size=50,
+            ),
+            min_size=10,
+            max_size=10,
+            unique=True,
+        ),
+    )
+    def init_random_service_names(
+        self, service_names: list[str]
+    ) -> Iterable[str]:
+        return stateful.multiple(*service_names)
+
+    # Don't include key or phrase settings here.  While easy to
+    # implement when manipulating the stored config directly, the
+    # command-line interface for changing the passphrase and key values
+    # is not straight-forward, and key values require a running SSH
+    # agent and the key to be loaded.
+    @stateful.initialize(
+        target=settings,
+        settings_list=strategies.lists(
+            tests.vault_full_service_config(),
+            min_size=10,
+            max_size=10,
+            unique_by=lambda obj: json.dumps(obj, sort_keys=True),
+        ),
+    )
+    def init_random_settings(
+        self, settings_list: list[_types.VaultConfigGlobalSettings]
+    ) -> Iterable[_types.VaultConfigGlobalSettings]:
+        return stateful.multiple(*settings_list)
+
+    @stateful.invariant()
+    def check_consistency_of_configs(self) -> None:
+        _types.clean_up_falsy_vault_config_values(self.current_config)
+        assert self.current_config == cli._load_config()
+
+    @stateful.rule(
+        target=settings,
+        settings_obj=tests.vault_full_service_config(),
+    )
+    def prepare_settings(
+        self, settings_obj: dict[str, int]
+    ) -> _types.VaultConfigGlobalSettings:
+        return cast(_types.VaultConfigGlobalSettings, settings_obj.copy())
+
+    @stateful.rule(
+        target=known_services,
+        name=strategies.text(
+            strategies.characters(min_codepoint=32, max_codepoint=126),
+            min_size=1,
+            max_size=50,
+        ),
+    )
+    def prepare_service_name(self, name: str) -> str:
+        return name
+
+    @stateful.rule(
+        target=configurations,
+        settings_obj=stateful.consumes(settings),
+    )
+    def prepare_global_config(
+        self,
+        settings_obj: dict[str, int],
+    ) -> _types.VaultConfig:
+        return {
+            'global': cast(_types.VaultConfigGlobalSettings, settings_obj),
+            'services': {},
+        }
+
+    @stateful.rule(
+        target=configurations,
+        service=known_services,
+        settings_obj=stateful.consumes(settings),
+    )
+    def prepare_service_config(
+        self,
+        service: str,
+        settings_obj: dict[str, int],
+    ) -> _types.VaultConfig:
+        return {
+            'services': {
+                service: cast(
+                    _types.VaultConfigServicesSettings, settings_obj
+                ),
+            },
+        }
+
+    @staticmethod
+    def fold_configs(
+        c1: _types.VaultConfig, c2: _types.VaultConfig
+    ) -> _types.VaultConfig:
+        new_global_dict = c1.get('global', c2.get('global'))
+        if new_global_dict is not None:
+            return {
+                'global': new_global_dict,
+                'services': {**c2['services'], **c1['services']},
+            }
+        return {
+            'services': {**c2['services'], **c1['services']},
+        }
+
+    @stateful.rule(
+        target=configurations,
+        config_base=stateful.consumes(configurations),
+        config_folded=stateful.consumes(configurations),
+    )
+    def fold_configuration_into(
+        self,
+        config_base: _types.VaultConfig,
+        config_folded: _types.VaultConfig,
+    ) -> _types.VaultConfig:
+        return self.fold_configs(config_folded, config_base)
+
+    @stateful.rule(
+        settings_obj=stateful.consumes(settings),
+    )
+    def set_globals(
+        self,
+        settings_obj: _types.VaultConfigGlobalSettings,
+    ) -> None:
+        self.current_config['global'] = settings_obj
+        assert _types.is_vault_config(self.current_config)
+        # NOTE: This relies on settings_obj containing only the keys
+        # "length", "repeat", "upper", "lower", "number", "space",
+        # "dash" and "symbol".
+        _result = self.runner.invoke(
+            cli.derivepassphrase_vault,
+            ['--config']
+            + [f'--{key}={value}' for key, value in settings_obj.items()],
+            catch_exceptions=False,
+        )
+        result = tests.ReadableResult.parse(_result)
+        assert result.clean_exit(empty_stderr=False)
+
+    # No check for whether the service settings currently exist.  This
+    # may therefore actually "create" the settings, not merely "modify"
+    # them.
+    #
+    # (There is no check because this appears to be hard or impossible
+    # to express as a hypothesis strategy: it would depend on the
+    # current state of the state machine instance.  This could be
+    # circumvented with `hypothesis.assume`, but that may likely trigger
+    # health check errors.)
+    @stateful.rule(
+        service=known_services,
+        settings_obj=stateful.consumes(settings),
+    )
+    def set_service(
+        self,
+        service: str,
+        settings_obj: _types.VaultConfigServicesSettings,
+    ) -> None:
+        self.current_config['services'][service] = settings_obj
+        assert _types.is_vault_config(self.current_config)
+        # NOTE: This relies on settings_obj containing only the keys
+        # "length", "repeat", "upper", "lower", "number", "space",
+        # "dash" and "symbol".
+        _result = self.runner.invoke(
+            cli.derivepassphrase_vault,
+            ['--config']
+            + [f'--{key}={value}' for key, value in settings_obj.items()]
+            + ['--', service],
+            catch_exceptions=False,
+        )
+        result = tests.ReadableResult.parse(_result)
+        assert result.clean_exit(empty_stderr=False)
+
+    @stateful.precondition(lambda self: 'global' in self.current_config)
+    @stateful.rule()
+    def purge_global(self) -> None:
+        self.current_config.pop('global')
+        _result = self.runner.invoke(
+            cli.derivepassphrase_vault,
+            ['--delete-globals'],
+            input='y',
+            catch_exceptions=False,
+        )
+        result = tests.ReadableResult.parse(_result)
+        assert result.clean_exit(empty_stderr=False)
+
+    # No check for whether the service settings currently exist.  This
+    # may therefore actually be almost a no-op, purging settings that
+    # aren't set in the first place.
+    #
+    # (There is no check because this appears to be hard or impossible
+    # to express as a hypothesis strategy: it would depend on the
+    # current state of the state machine instance.  This could be
+    # circumvented with `hypothesis.assume`, but that may likely trigger
+    # health check errors.)
+    @stateful.precondition(lambda self: bool(self.current_config['services']))
+    @stateful.rule(service=stateful.consumes(known_services))
+    def purge_service(self, service: str) -> None:
+        ret = self.current_config['services'].pop(service, None)
+        if ret is not None:
+            _result = self.runner.invoke(
+                cli.derivepassphrase_vault,
+                ['--delete', service],
+                input='y',
+                catch_exceptions=False,
+            )
+            result = tests.ReadableResult.parse(_result)
+            assert result.clean_exit(empty_stderr=False)
+
+    @stateful.rule()
+    def purge_all(self) -> None:
+        self.current_config = {'services': {}}
+        _result = self.runner.invoke(
+            cli.derivepassphrase_vault,
+            ['--clear'],
+            input='y',
+            catch_exceptions=False,
+        )
+        result = tests.ReadableResult.parse(_result)
+        assert result.clean_exit(empty_stderr=False)
+
+    @stateful.rule(
+        config=stateful.consumes(configurations),
+    )
+    def import_configuraton(self, config: _types.VaultConfig) -> None:
+        self.current_config = self.fold_configs(config, self.current_config)
+        assert _types.is_vault_config(self.current_config)
+        _result = self.runner.invoke(
+            cli.derivepassphrase_vault,
+            ['--import', '-'],
+            input=json.dumps(config),
+            catch_exceptions=False,
+        )
+        assert tests.ReadableResult.parse(_result).clean_exit(
+            empty_stderr=False
+        )
+
+    def teardown(self) -> None:
+        self.exit_stack.close()
+
+
+TestConfigMerging = ConfigMergingStateMachine.TestCase


...	...	@@ -11,21 +11,19 @@ import json
11	11	import os
12	12	import shutil
13	13	import socket
14		-from typing import TYPE_CHECKING
	14	+from typing import TYPE_CHECKING, cast
15	15
16	16	import click.testing
17	17	import hypothesis
18	18	import pytest
19		-from hypothesis import strategies
20		-from typing_extensions import NamedTuple
	19	+from hypothesis import stateful, strategies
	20	+from typing_extensions import Any, NamedTuple
21	21
22	22	import tests
23	23	from derivepassphrase import _types, cli, ssh_agent, vault
24	24
25	25	if TYPE_CHECKING:
26		- from collections.abc import Callable
27		-
28		- from typing_extensions import Any
	26	+ from collections.abc import Callable, Iterable
29	27
30	28	DUMMY_SERVICE = tests.DUMMY_SERVICE
31	29	DUMMY_PASSPHRASE = tests.DUMMY_PASSPHRASE
...	...	@@ -486,9 +484,7 @@ class TestCLI:
486	484	},
487	485	{
488	486	'global': {'key': DUMMY_KEY1_B64},
489		- 'services': {
490		- DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS.copy()
491		- },
	487	+ 'services': {DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS.copy()},
492	488	},
493	489	],
494	490	)
...	...	@@ -2100,3 +2096,283 @@ class TestCLITransition:
2100	2096	assert (
2101	2097	'Warning: Failed to migrate to ' in result.stderr
2102	2098	), 'expected known warning message in stderr'
	2099	+
	2100	+
	2101	+class ConfigMergingStateMachine(stateful.RuleBasedStateMachine):
	2102	+ def __init__(self) -> None:
	2103	+ super().__init__()
	2104	+ self.runner = click.testing.CliRunner(mix_stderr=False)
	2105	+ self.exit_stack = contextlib.ExitStack().__enter__()
	2106	+ self.monkeypatch = self.exit_stack.enter_context(
	2107	+ pytest.MonkeyPatch().context()
	2108	+ )
	2109	+ self.isolated_config = self.exit_stack.enter_context(
	2110	+ tests.isolated_vault_config(
	2111	+ monkeypatch=self.monkeypatch,
	2112	+ runner=self.runner,
	2113	+ config={'services': {}},
	2114	+ )
	2115	+ )
	2116	+ self.current_config = cli._load_config()
	2117	+
	2118	+ known_services = stateful.Bundle('known_services')
	2119	+ settings = stateful.Bundle('settings')
	2120	+ configurations = stateful.Bundle('configurations')
	2121	+
	2122	+ @stateful.initialize(target=configurations)
	2123	+ def init_empty_configuration(self) -> _types.VaultConfig:
	2124	+ return copy.deepcopy(self.current_config)
	2125	+
	2126	+ @stateful.initialize(target=configurations)
	2127	+ def init_standard_testing_configuration(self) -> _types.VaultConfig:
	2128	+ return cast(
	2129	+ _types.VaultConfig,
	2130	+ {
	2131	+ 'services': {
	2132	+ DUMMY_SERVICE: copy.deepcopy(DUMMY_CONFIG_SETTINGS)
	2133	+ }
	2134	+ },
	2135	+ )
	2136	+
	2137	+ @stateful.initialize(
	2138	+ target=known_services,
	2139	+ service_names=strategies.lists(
	2140	+ strategies.text(
	2141	+ strategies.characters(min_codepoint=32, max_codepoint=126),
	2142	+ min_size=1,
	2143	+ max_size=50,
	2144	+ ),
	2145	+ min_size=10,
	2146	+ max_size=10,
	2147	+ unique=True,
	2148	+ ),
	2149	+ )
	2150	+ def init_random_service_names(
	2151	+ self, service_names: list[str]
	2152	+ ) -> Iterable[str]:
	2153	+ return stateful.multiple(*service_names)
	2154	+
	2155	+ # Don't include key or phrase settings here. While easy to
	2156	+ # implement when manipulating the stored config directly, the
	2157	+ # command-line interface for changing the passphrase and key values
	2158	+ # is not straight-forward, and key values require a running SSH
	2159	+ # agent and the key to be loaded.
	2160	+ @stateful.initialize(
	2161	+ target=settings,
	2162	+ settings_list=strategies.lists(
	2163	+ tests.vault_full_service_config(),
	2164	+ min_size=10,
	2165	+ max_size=10,
	2166	+ unique_by=lambda obj: json.dumps(obj, sort_keys=True),
	2167	+ ),
	2168	+ )
	2169	+ def init_random_settings(
	2170	+ self, settings_list: list[_types.VaultConfigGlobalSettings]
	2171	+ ) -> Iterable[_types.VaultConfigGlobalSettings]:
	2172	+ return stateful.multiple(*settings_list)
	2173	+
	2174	+ @stateful.invariant()
	2175	+ def check_consistency_of_configs(self) -> None:
	2176	+ _types.clean_up_falsy_vault_config_values(self.current_config)
	2177	+ assert self.current_config == cli._load_config()
	2178	+
	2179	+ @stateful.rule(
	2180	+ target=settings,
	2181	+ settings_obj=tests.vault_full_service_config(),
	2182	+ )
	2183	+ def prepare_settings(
	2184	+ self, settings_obj: dict[str, int]
	2185	+ ) -> _types.VaultConfigGlobalSettings:
	2186	+ return cast(_types.VaultConfigGlobalSettings, settings_obj.copy())
	2187	+
	2188	+ @stateful.rule(
	2189	+ target=known_services,
	2190	+ name=strategies.text(
	2191	+ strategies.characters(min_codepoint=32, max_codepoint=126),
	2192	+ min_size=1,
	2193	+ max_size=50,
	2194	+ ),
	2195	+ )
	2196	+ def prepare_service_name(self, name: str) -> str:
	2197	+ return name
	2198	+
	2199	+ @stateful.rule(
	2200	+ target=configurations,
	2201	+ settings_obj=stateful.consumes(settings),
	2202	+ )
	2203	+ def prepare_global_config(
	2204	+ self,
	2205	+ settings_obj: dict[str, int],
	2206	+ ) -> _types.VaultConfig:
	2207	+ return {
	2208	+ 'global': cast(_types.VaultConfigGlobalSettings, settings_obj),
	2209	+ 'services': {},
	2210	+ }
	2211	+
	2212	+ @stateful.rule(
	2213	+ target=configurations,
	2214	+ service=known_services,
	2215	+ settings_obj=stateful.consumes(settings),
	2216	+ )
	2217	+ def prepare_service_config(
	2218	+ self,
	2219	+ service: str,
	2220	+ settings_obj: dict[str, int],
	2221	+ ) -> _types.VaultConfig:
	2222	+ return {
	2223	+ 'services': {
	2224	+ service: cast(
	2225	+ _types.VaultConfigServicesSettings, settings_obj
	2226	+ ),
	2227	+ },
	2228	+ }
	2229	+
	2230	+ @staticmethod
	2231	+ def fold_configs(
	2232	+ c1: _types.VaultConfig, c2: _types.VaultConfig
	2233	+ ) -> _types.VaultConfig:
	2234	+ new_global_dict = c1.get('global', c2.get('global'))
	2235	+ if new_global_dict is not None:
	2236	+ return {
	2237	+ 'global': new_global_dict,
	2238	+ 'services': {c2['services'], c1['services']},
	2239	+ }
	2240	+ return {
	2241	+ 'services': {c2['services'], c1['services']},
	2242	+ }
	2243	+
	2244	+ @stateful.rule(
	2245	+ target=configurations,
	2246	+ config_base=stateful.consumes(configurations),
	2247	+ config_folded=stateful.consumes(configurations),
	2248	+ )
	2249	+ def fold_configuration_into(
	2250	+ self,
	2251	+ config_base: _types.VaultConfig,
	2252	+ config_folded: _types.VaultConfig,
	2253	+ ) -> _types.VaultConfig:
	2254	+ return self.fold_configs(config_folded, config_base)
	2255	+
	2256	+ @stateful.rule(
	2257	+ settings_obj=stateful.consumes(settings),
	2258	+ )
	2259	+ def set_globals(
	2260	+ self,
	2261	+ settings_obj: _types.VaultConfigGlobalSettings,
	2262	+ ) -> None:
	2263	+ self.current_config['global'] = settings_obj
	2264	+ assert _types.is_vault_config(self.current_config)
	2265	+ # NOTE: This relies on settings_obj containing only the keys
	2266	+ # "length", "repeat", "upper", "lower", "number", "space",
	2267	+ # "dash" and "symbol".
	2268	+ _result = self.runner.invoke(
	2269	+ cli.derivepassphrase_vault,
	2270	+ ['--config']
	2271	+ + [f'--{key}={value}' for key, value in settings_obj.items()],
	2272	+ catch_exceptions=False,
	2273	+ )
	2274	+ result = tests.ReadableResult.parse(_result)
	2275	+ assert result.clean_exit(empty_stderr=False)
	2276	+
	2277	+ # No check for whether the service settings currently exist. This
	2278	+ # may therefore actually "create" the settings, not merely "modify"
	2279	+ # them.
	2280	+ #
	2281	+ # (There is no check because this appears to be hard or impossible
	2282	+ # to express as a hypothesis strategy: it would depend on the
	2283	+ # current state of the state machine instance. This could be
	2284	+ # circumvented with `hypothesis.assume`, but that may likely trigger
	2285	+ # health check errors.)
	2286	+ @stateful.rule(
	2287	+ service=known_services,
	2288	+ settings_obj=stateful.consumes(settings),
	2289	+ )
	2290	+ def set_service(
	2291	+ self,
	2292	+ service: str,
	2293	+ settings_obj: _types.VaultConfigServicesSettings,
	2294	+ ) -> None:
	2295	+ self.current_config['services'][service] = settings_obj
	2296	+ assert _types.is_vault_config(self.current_config)
	2297	+ # NOTE: This relies on settings_obj containing only the keys
	2298	+ # "length", "repeat", "upper", "lower", "number", "space",
	2299	+ # "dash" and "symbol".
	2300	+ _result = self.runner.invoke(
	2301	+ cli.derivepassphrase_vault,
	2302	+ ['--config']
	2303	+ + [f'--{key}={value}' for key, value in settings_obj.items()]
	2304	+ + ['--', service],
	2305	+ catch_exceptions=False,
	2306	+ )
	2307	+ result = tests.ReadableResult.parse(_result)
	2308	+ assert result.clean_exit(empty_stderr=False)
	2309	+
	2310	+ @stateful.precondition(lambda self: 'global' in self.current_config)
	2311	+ @stateful.rule()
	2312	+ def purge_global(self) -> None:
	2313	+ self.current_config.pop('global')
	2314	+ _result = self.runner.invoke(
	2315	+ cli.derivepassphrase_vault,
	2316	+ ['--delete-globals'],
	2317	+ input='y',
	2318	+ catch_exceptions=False,
	2319	+ )
	2320	+ result = tests.ReadableResult.parse(_result)
	2321	+ assert result.clean_exit(empty_stderr=False)
	2322	+
	2323	+ # No check for whether the service settings currently exist. This
	2324	+ # may therefore actually be almost a no-op, purging settings that
	2325	+ # aren't set in the first place.
	2326	+ #
	2327	+ # (There is no check because this appears to be hard or impossible
	2328	+ # to express as a hypothesis strategy: it would depend on the
	2329	+ # current state of the state machine instance. This could be
	2330	+ # circumvented with `hypothesis.assume`, but that may likely trigger
	2331	+ # health check errors.)
	2332	+ @stateful.precondition(lambda self: bool(self.current_config['services']))
	2333	+ @stateful.rule(service=stateful.consumes(known_services))
	2334	+ def purge_service(self, service: str) -> None:
	2335	+ ret = self.current_config['services'].pop(service, None)
	2336	+ if ret is not None:
	2337	+ _result = self.runner.invoke(
	2338	+ cli.derivepassphrase_vault,
	2339	+ ['--delete', service],
	2340	+ input='y',
	2341	+ catch_exceptions=False,
	2342	+ )
	2343	+ result = tests.ReadableResult.parse(_result)
	2344	+ assert result.clean_exit(empty_stderr=False)
	2345	+
	2346	+ @stateful.rule()
	2347	+ def purge_all(self) -> None:
	2348	+ self.current_config = {'services': {}}
	2349	+ _result = self.runner.invoke(
	2350	+ cli.derivepassphrase_vault,
	2351	+ ['--clear'],
	2352	+ input='y',
	2353	+ catch_exceptions=False,
	2354	+ )
	2355	+ result = tests.ReadableResult.parse(_result)
	2356	+ assert result.clean_exit(empty_stderr=False)
	2357	+
	2358	+ @stateful.rule(
	2359	+ config=stateful.consumes(configurations),
	2360	+ )
	2361	+ def import_configuraton(self, config: _types.VaultConfig) -> None:
	2362	+ self.current_config = self.fold_configs(config, self.current_config)
	2363	+ assert _types.is_vault_config(self.current_config)
	2364	+ _result = self.runner.invoke(
	2365	+ cli.derivepassphrase_vault,
	2366	+ ['--import', '-'],
	2367	+ input=json.dumps(config),
	2368	+ catch_exceptions=False,
	2369	+ )
	2370	+ assert tests.ReadableResult.parse(_result).clean_exit(
	2371	+ empty_stderr=False
	2372	+ )
	2373	+
	2374	+ def teardown(self) -> None:
	2375	+ self.exit_stack.close()
	2376	+
	2377	+
	2378	+TestConfigMerging = ConfigMergingStateMachine.TestCase
2103	2379