Coverage for tests / test_derivepassphrase_ssh_agent / test_heavy_duty.py: 100.000%
78 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 `derivepassphrase.ssh_agent`: heavy-duty tests."""
7from __future__ import annotations
9import contextlib
11import pytest
12from hypothesis import stateful, strategies
14from derivepassphrase import _types
15from derivepassphrase.ssh_agent import socketprovider
16from tests.machinery import pytest as pytest_machinery
18# All tests in this module are heavy-duty tests.
19pytestmark = [pytest_machinery.heavy_duty]
22@strategies.composite
23def draw_alias_chain(
24 draw: strategies.DrawFn,
25 *,
26 known_keys_strategy: strategies.SearchStrategy[str],
27 new_keys_strategy: strategies.SearchStrategy[str],
28 chain_size: strategies.SearchStrategy[int] = strategies.integers( # noqa: B008
29 min_value=1,
30 max_value=5,
31 ),
32 existing: bool = False,
33) -> tuple[str, ...]:
34 """Draw names for alias chains in the SSH agent socket provider registry.
36 Depending on arguments, draw a set of names from the new keys bundle
37 that do not yet exist in the registry, to insert as a new alias
38 chain. Alternatively, draw a non-alias name from the known keys
39 bundle, then draw other names that either don't exist yet in the
40 registry, or that alias the first name directly or indirectly. The
41 chain length, and whether to target existing registry entries or
42 not, may be set statically, or may be drawn from a respective
43 strategy.
45 Args:
46 draw:
47 The `hypothesis` draw function.
48 chain_size:
49 A strategy for determining the correct alias chain length.
50 Must not yield any integers less than 1.
51 existing:
52 If true, target an existing registry entry in the alias
53 chain, and permit rewriting existing aliases of that same
54 entry to the new alias. Otherwise, draw only new names.
55 known_keys_strategy:
56 A strategy for generating provider registry keys already
57 contained in the registry. Typically, this is
58 a [Bundle][hypothesis.stateful.Bundle].
59 new_keys_strategy:
60 A strategy for generating provider registry keys not yet
61 contained in the registry with high probability. Typically,
62 this is a [consuming][hypothesis.stateful.consumes]
63 [Bundle][hypothesis.stateful.Bundle].
65 Returns:
66 A tuple of names forming an alias chain, each entry pointing to
67 or intending to point to the previous entry in the tuple.
69 """
70 registry = socketprovider.SocketProvider.registry 1a
72 def not_an_alias(key: str) -> bool: 1a
73 return key in registry and not isinstance(registry[key], str) 1a
75 def is_indirect_alias_of( 1a
76 key: str, target: str 1a
77 ) -> bool: # pragma: no cover 1a
78 if key == target: 1a
79 return False # not an alias 1a
80 seen = set() # loop detection 1a
81 while key not in seen: 1a
82 seen.add(key) 1a
83 if key not in registry: 1a
84 return False
85 if not isinstance(registry[key], str): 1a
86 return False 1a
87 if key == target: 1a
88 return True
89 tmp = registry[key] 1a
90 assert isinstance(tmp, str) 1a
91 key = tmp 1a
92 return False # loop
94 err_msg_chain_size = "Chain sizes must always be 1 or larger." 1a
96 size = draw(chain_size, label="chain_size") 1a
97 if size < 1: # pragma: no cover 1a
98 raise ValueError(err_msg_chain_size)
99 names: list[str] = [] 1a
100 base: str | None = None 1a
101 if existing: 1a
102 names.append( 1a
103 draw(known_keys_strategy.filter(not_an_alias), label="base")
104 )
105 base = names[0] 1a
106 size -= 1 1a
107 new_key_strategy = new_keys_strategy.filter( 1a
108 lambda key: key not in registry
109 )
110 old_key_strategy = known_keys_strategy.filter( 1a
111 lambda key: is_indirect_alias_of(key, target=base)
112 )
113 list_strategy_source = strategies.one_of( 1a
114 new_key_strategy, old_key_strategy
115 )
116 else:
117 list_strategy_source = new_keys_strategy.filter( 1a
118 lambda key: key not in registry
119 )
120 list_strategy = strategies.lists( 1a
121 list_strategy_source.filter(lambda candidate: candidate != base),
122 min_size=size,
123 max_size=size,
124 unique=True,
125 )
126 names.extend(draw(list_strategy, label="others")) 1a
127 return tuple(names) 1a
130class SSHAgentSocketProviderRegistryStateMachine(
131 stateful.RuleBasedStateMachine
132):
133 """A state machine for the SSH agent socket provider registry.
135 Record possible changes to the socket provider registry, keeping track
136 of true entries, aliases, and reservations.
138 """
140 def __init__(self) -> None: 1ba
141 """Initialize self, set up context managers and enter them."""
142 super().__init__() 1a
143 self.exit_stack = contextlib.ExitStack().__enter__() 1a
144 self.monkeypatch = self.exit_stack.enter_context( 1a
145 pytest.MonkeyPatch.context()
146 )
147 self.orig_registry = socketprovider.SocketProvider.registry 1a
148 self.registry: dict[ 1a
149 str, _types.SSHAgentSocketProvider | str | None
150 ] = {
151 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX: (
152 self.orig_registry[
153 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX
154 ]
155 ),
156 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS: (
157 self.orig_registry[
158 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS
159 ]
160 ),
161 _types.BuiltinSSHAgentSocketProvider.NATIVE: (
162 _types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE
163 if self.orig_registry[
164 _types.BuiltinSSHAgentSocketProvider.NATIVE
165 ]
166 == _types.BuiltinSSHAgentSocketProvider.WINDOWS
167 else _types.BuiltinSSHAgentSocketProvider.UNIX_DOMAIN
168 ),
169 _types.BuiltinSSHAgentSocketProvider.UNIX_DOMAIN: (
170 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX
171 ),
172 _types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE: (
173 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS
174 ),
175 }
176 self.monkeypatch.setattr( 1a
177 socketprovider.SocketProvider, "registry", self.registry
178 )
179 self.model: dict[str, _types.SSHAgentSocketProvider | None] = {} 1a
181 known_keys: stateful.Bundle[str] = stateful.Bundle("known_keys")
182 """"""
183 new_keys: stateful.Bundle[str] = stateful.Bundle("new_keys")
184 """"""
186 def sample_provider(self) -> _types.SSHAgentSocket:
187 raise AssertionError
189 @stateful.initialize(
190 target=known_keys,
191 )
192 def get_registry_keys(self) -> stateful.MultipleResults[str]:
193 """Read the standard keys from the registry."""
194 self.model.update({ 1a
195 k: socketprovider.SocketProvider.lookup(k) for k in self.registry
196 })
197 return stateful.multiple(*self.registry.keys()) 1a
199 @stateful.rule(
200 target=new_keys,
201 k=strategies.text("abcdefghijklmnopqrstuvwxyz0123456789_").filter(
202 lambda s: s not in socketprovider.SocketProvider.registry
203 ),
204 )
205 def new_key(self, k: str) -> str:
206 return k 1a
208 @stateful.invariant()
209 def check_consistency(self) -> None:
210 lookup = socketprovider.SocketProvider.lookup 1a
211 assert self.registry.keys() == self.model.keys() 1a
212 for k in self.model: 1a
213 resolved = lookup(k) 1a
214 modelled = self.model[k] 1a
215 step1 = self.registry[k] 1a
216 manually = lookup(step1) if isinstance(step1, str) else step1 1a
217 assert resolved == modelled 1a
218 assert resolved == manually 1a
220 @stateful.rule(
221 target=known_keys,
222 chain=draw_alias_chain(
223 known_keys_strategy=known_keys,
224 new_keys_strategy=stateful.consumes(new_keys),
225 existing=True,
226 ),
227 )
228 def alias_existing(
229 self, chain: tuple[str, ...]
230 ) -> stateful.MultipleResults[str]:
231 try: 1a
232 provider = socketprovider.SocketProvider.resolve(chain[0]) 1a
233 except NotImplementedError: # pragma: no cover [failsafe]
234 provider = self.sample_provider
235 assert ( 1a
236 socketprovider.SocketProvider.register(*chain)(provider)
237 == provider
238 )
239 for k in chain: 1a
240 self.model[k] = provider 1a
241 return stateful.multiple(*chain[1:]) 1a
243 @stateful.rule(
244 target=known_keys,
245 chain=draw_alias_chain(
246 known_keys_strategy=known_keys,
247 new_keys_strategy=stateful.consumes(new_keys),
248 existing=False,
249 ),
250 )
251 def alias_new(self, chain: list[str]) -> stateful.MultipleResults[str]:
252 provider = self.sample_provider 1a
253 assert ( 1a
254 socketprovider.SocketProvider.register(*chain)(provider)
255 == provider
256 )
257 for k in chain: 1a
258 self.model[k] = provider 1a
259 return stateful.multiple(*chain) 1a
261 def teardown(self) -> None:
262 """Upon teardown, exit all contexts entered in `__init__`."""
263 self.exit_stack.close() 1a
266TestSSHAgentSocketProviderRegistry = (
267 SSHAgentSocketProviderRegistryStateMachine.TestCase
268)