tests.fixtures
Define fixtures for the test suite.
1"""Define fixtures for the test suite.""" 2 3from __future__ import annotations 4 5import contextlib 6import json 7from collections.abc import Iterator 8from pathlib import Path 9from typing import Any, TypeVar 10from unittest import mock 11 12import pytest 13import pytest_mock 14from boto3 import Session 15from botocore.exceptions import ClientError 16from botocore.paginate import PageIterator, Paginator 17from botocore.response import StreamingBody 18from mypy_boto3_appconfig import AppConfigClient 19from mypy_boto3_appconfigdata import AppConfigDataClient 20from mypy_boto3_appconfigdata.type_defs import GetLatestConfigurationResponseTypeDef 21from mypy_boto3_secretsmanager import SecretsManagerClient 22from mypy_boto3_secretsmanager.type_defs import ListSecretVersionIdsResponseTypeDef, SecretVersionsListEntryTypeDef 23from pytest_mock import MockerFixture 24 25from config_ninja import systemd 26 27# pylint: disable=redefined-outer-name 28 29T = TypeVar('T') 30 31MOCK_PYPI_RESPONSE = {'releases': {'1.0': 'ignore', '1.1': 'ignore', '1.2a0': 'ignore'}} 32MOCK_YAML_CONFIG = b""" 33key_0: value_0 34key_1: 1 35key_2: true 36key_3: 37 - 1 38 - 2 39 - 3 40""".strip() 41 42 43class MockFile(mock.MagicMock): 44 """Mock the file object returned by `contextlib.closing`.""" 45 46 mock_bytes: bytes 47 48 def read(self) -> bytes: 49 """Mock the `read` method to return data used in tests.""" 50 return self.mock_bytes 51 52 53def mock_file(mock_bytes: bytes) -> MockFile: 54 """Mock the file object returned by `contextlib.closing`.""" 55 mock_file = MockFile() 56 mock_file.mock_bytes = mock_bytes 57 return mock_file 58 59 60@pytest.fixture 61def _mock_contextlib_closing(mocker: MockerFixture) -> None: # pyright: ignore[reportUnusedFunction] 62 """Mock `contextlib.closing`.""" 63 64 @contextlib.contextmanager 65 def _mocked(request: Any) -> Iterator[Any]: 66 """Pass the input parameter straight through.""" 67 yield request 68 69 mocker.patch('contextlib.closing', new=_mocked) 70 71 72@pytest.fixture 73def _mock_urlopen_for_pypi(mocker: MockerFixture) -> None: # pyright: ignore[reportUnusedFunction] 74 """Mock `urllib.request.urlopen` for PyPI requests.""" 75 76 def _mocked(_: Any) -> MockFile: 77 return mock_file(json.dumps(MOCK_PYPI_RESPONSE).encode('utf-8')) 78 79 mocker.patch('urllib.request.urlopen', new=_mocked) 80 81 82@pytest.fixture 83def mock_appconfig_client() -> AppConfigClient: 84 """Mock the `boto3` client for the `AppConfig` service.""" 85 return mock.MagicMock(name='mock_appconfig_client', spec_set=AppConfigClient) 86 87 88@pytest.fixture 89def _mock_install_io(mocker: MockerFixture) -> None: # pyright: ignore[reportUnusedFunction] 90 """Mock various I/O utilities used by the `install` script.""" 91 mocker.patch('shutil.rmtree') 92 mocker.patch('subprocess.run') 93 mocker.patch('venv.EnvBuilder') 94 mocker.patch('runpy.run_path') 95 96 97@pytest.fixture 98def mock_session(mocker: MockerFixture) -> Session: 99 """Mock the `boto3.Session` class.""" 100 mock_session = mock.MagicMock(name='mock_session', spec_set=Session) 101 mocker.patch('boto3.Session', return_value=mock_session) 102 return mock_session 103 104 105@pytest.fixture 106def mock_session_with_0_ids(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 107 """Mock the `boto3` client for the `AppConfig` service to return no IDs.""" 108 mock_page_iterator = mock.MagicMock(spec_set=PageIterator) 109 mock_page_iterator.search.return_value = [] 110 111 mock_paginator = mock.MagicMock(spec_set=Paginator) 112 mock_paginator.paginate.return_value = mock_page_iterator 113 114 mock_appconfig_client.get_paginator.return_value = mock_paginator 115 mock_session.client.return_value = mock_appconfig_client 116 return mock_session 117 118 119@pytest.fixture 120def mock_session_with_1_id(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 121 """Mock the `boto3` client for the `AppConfig` service to return a single ID.""" 122 mock_page_iterator = mock.MagicMock(name='mock_page_iterator', spec_set=PageIterator) 123 mock_page_iterator.search.return_value = ['id-1'] 124 125 mock_paginator = mock.MagicMock(name='mock_page_iterator', spec_set=Paginator) 126 mock_paginator.paginate.return_value = mock_page_iterator 127 128 mock_appconfig_client.get_paginator.return_value = mock_paginator 129 130 mock_session.client.return_value = mock_appconfig_client 131 return mock_session 132 133 134@pytest.fixture 135def mock_session_with_2_ids(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 136 """Mock the `boto3` client for the `AppConfig` service to return two IDs.""" 137 mock_page_iterator = mock.MagicMock(spec_set=PageIterator) 138 mock_page_iterator.search.return_value = ['id-1', 'id-2'] 139 140 mock_paginator = mock.MagicMock(spec_set=Paginator) 141 mock_paginator.paginate.return_value = mock_page_iterator 142 143 mock_appconfig_client.get_paginator.return_value = mock_paginator 144 145 mock_session.client.return_value = mock_appconfig_client 146 return mock_session 147 148 149@pytest.fixture 150def mock_latest_config() -> GetLatestConfigurationResponseTypeDef: 151 """Mock the response from `get_latest_configuration`.""" 152 mock_config_stream = mock.MagicMock(spec_set=StreamingBody) 153 mock_config_stream.read.return_value = MOCK_YAML_CONFIG 154 return { 155 'NextPollConfigurationToken': 'token', 156 'NextPollIntervalInSeconds': 1, 157 'ContentType': 'application/json', 158 'Configuration': mock_config_stream, 159 'VersionLabel': 'v1', 160 'ResponseMetadata': { 161 'RequestId': '', 162 'HostId': '', 163 'HTTPStatusCode': 200, 164 'HTTPHeaders': {}, 165 'RetryAttempts': 3, 166 }, 167 } 168 169 170@pytest.fixture 171def mock_latest_config_first_empty() -> GetLatestConfigurationResponseTypeDef: 172 """Mock the response from `get_latest_configuration`. 173 174 Return an empty `bytes` on the first iteration, and `MOCK_YAML_CONFIG` on the second. This supports testing 175 `config_ninja.contrib.appconfig.AppConfig`'s response to an empty return value. 176 """ 177 was_called: list[bool] = [] 178 179 def mock_read(*_: Any, **__: Any) -> bytes: 180 if was_called: 181 return MOCK_YAML_CONFIG 182 was_called.append(True) 183 return b'' 184 185 mock_config_stream = mock.MagicMock(spec_set=StreamingBody) 186 mock_config_stream.read = mock_read 187 return { 188 'NextPollConfigurationToken': 'token', 189 'NextPollIntervalInSeconds': 0, 190 'ContentType': 'application/json', 191 'Configuration': mock_config_stream, 192 'VersionLabel': 'v1', 193 'ResponseMetadata': { 194 'RequestId': '', 195 'HostId': '', 196 'HTTPStatusCode': 200, 197 'HTTPHeaders': {}, 198 'RetryAttempts': 3, 199 }, 200 } 201 202 203@pytest.fixture 204def mock_appconfigdata_client(mock_latest_config: mock.MagicMock) -> AppConfigDataClient: 205 """Mock the low-level `boto3` client for the `AppConfigData` service.""" 206 mock_client = mock.MagicMock(name='mock_appconfigdata_client', spec_set=AppConfigDataClient) 207 mock_client.get_latest_configuration.return_value = mock_latest_config 208 return mock_client 209 210 211@pytest.fixture 212def mock_appconfigdata_client_first_empty(mock_latest_config_first_empty: mock.MagicMock) -> AppConfigDataClient: 213 """Mock the low-level `boto3` client for the `AppConfigData` service.""" 214 mock_client = mock.MagicMock(name='mock_appconfigdata_client', spec_set=AppConfigDataClient) 215 mock_client.get_latest_configuration.return_value = mock_latest_config_first_empty 216 return mock_client 217 218 219@pytest.fixture 220def mock_secretsmanager_client() -> SecretsManagerClient: 221 """Mock the `boto3` client for the `SecretsManager` service.""" 222 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 223 mock_client.get_secret_value.return_value = { 224 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 225 'VersionId': 'v1', 226 } 227 mock_client.list_secret_version_ids.return_value = { 228 'Versions': [{'VersionId': 'v1'}, {'VersionId': 'v2', 'VersionStages': ['AWSCURRENT']}] 229 } 230 231 return mock_client 232 233 234@pytest.fixture 235def mock_secretsmanager_client_no_current() -> SecretsManagerClient: 236 """Mock the `boto3` client for the `SecretsManager` service.""" 237 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 238 mock_client.get_secret_value.return_value = { 239 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 240 'VersionId': 'v3', 241 } 242 mock_client.list_secret_version_ids.return_value = { 243 'Versions': [{'VersionId': 'v4'}, {'VersionId': 'v5', 'VersionStages': ['AWSPREVIOUS']}] 244 } 245 246 return mock_client 247 248 249@pytest.fixture 250def mock_secretsmanager_client_no_current_initially() -> SecretsManagerClient: 251 """Mock the `boto3` client for the `SecretsManager` service.""" 252 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 253 mock_client.get_secret_value.return_value = { 254 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 255 'VersionId': 'v6', 256 } 257 258 def _mock_response(versions: list[SecretVersionsListEntryTypeDef]) -> ListSecretVersionIdsResponseTypeDef: 259 return { 260 'ARN': 'arn:aws:secretsmanager:us-west-2:123456789012:secret/my-secret-1-a1b2c3', 261 'Name': 'my-secret-1', 262 'ResponseMetadata': { 263 'HTTPHeaders': {}, 264 'HTTPStatusCode': 200, 265 'RequestId': '12345678-1234-1234-1234-123456789012', 266 'RetryAttempts': 0, 267 }, 268 'Versions': versions, 269 } 270 271 versions_per_call = [ 272 _mock_response([{'VersionId': 'v6', 'VersionStages': ['AWSCURRENT']}]), 273 _mock_response([{'VersionId': 'v6'}, {'VersionId': 'v7'}]), 274 _mock_response( 275 [ 276 {'VersionId': 'v6', 'VersionStages': ['AWSPREVIOUS']}, 277 {'VersionId': 'v7', 'VersionStages': ['AWSCURRENT']}, 278 ] 279 ), 280 ] 281 282 class Counter: 283 count = 0 284 285 def increment(self) -> None: 286 self.count += 1 287 288 num_calls = Counter() 289 290 def mock_list_secret_version_ids(*_: Any, **__: Any) -> ListSecretVersionIdsResponseTypeDef: 291 versions = versions_per_call[num_calls.count] 292 num_calls.increment() 293 return versions 294 295 mock_client.list_secret_version_ids = mock_list_secret_version_ids 296 return mock_client 297 298 299@pytest.fixture 300def mock_full_session( 301 mock_session_with_1_id: mock.MagicMock, 302 mock_appconfig_client: mock.MagicMock, 303 mock_appconfigdata_client: mock.MagicMock, 304) -> Session: 305 """Mock the `boto3.Session` class with a full AppConfig client.""" 306 307 def client(service: str) -> mock.MagicMock: 308 if service == 'appconfig': 309 return mock_appconfig_client 310 if service == 'appconfigdata': 311 return mock_appconfigdata_client 312 raise ValueError(f'Unknown service: {service}') 313 314 mock_session_with_1_id.client = client 315 return mock_session_with_1_id 316 317 318@pytest.fixture 319def mock_poll_too_early( 320 mock_latest_config: GetLatestConfigurationResponseTypeDef, 321) -> AppConfigDataClient: 322 """Raise a `BadRequestException` when polling for configuration changes.""" 323 mock_client = mock.MagicMock(spec_set=AppConfigDataClient) 324 mock_client.exceptions.BadRequestException = ClientError 325 call_count = 0 326 327 def side_effect(*_: Any, **__: Any) -> GetLatestConfigurationResponseTypeDef: 328 nonlocal call_count 329 call_count += 1 330 if call_count == 1: 331 raise mock_client.exceptions.BadRequestException( 332 { 333 'Error': { 334 'Code': 'BadRequestException', 335 'Message': 'Request too early', 336 }, 337 'ResponseMetadata': {}, 338 }, 339 'GetLatestConfiguration', 340 ) 341 return mock_latest_config 342 343 mock_client.get_latest_configuration.side_effect = side_effect 344 345 return mock_client 346 347 348@pytest.fixture 349def monkeypatch_systemd(mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[Path, Path]: 350 """Monkeypatch various utilities for interfacing with `systemd` and the shell. 351 352 Returns: 353 tuple[pathlib.Path, pathlib.Path]: the patched `SYSTEM_INSTALL_PATH` and `USER_INSTALL_PATH` 354 """ 355 mocker.patch('config_ninja.systemd.sh') 356 mocker.patch.context_manager(systemd, 'sudo') 357 mocker.patch('config_ninja.systemd.sdnotify') 358 359 system_install_path = tmp_path / 'system' 360 user_install_path = tmp_path / 'user' 361 362 monkeypatch.setattr(systemd, 'AVAILABLE', True) 363 monkeypatch.setattr(systemd, 'SYSTEM_INSTALL_PATH', system_install_path) 364 monkeypatch.setattr(systemd, 'USER_INSTALL_PATH', user_install_path) 365 366 return (system_install_path, user_install_path) 367 368 369@pytest.fixture 370def example_file(tmp_path: Path) -> Path: 371 """Write the test configuration to a file in the temporary directory.""" 372 path = tmp_path / 'example.yaml' 373 path.write_bytes(MOCK_YAML_CONFIG) 374 return path 375 376 377example_file.__doc__ = f"""Write the test configuration to a file in the temporary directory. 378 379```yaml 380{MOCK_YAML_CONFIG.decode('utf-8')} 381``` 382""" 383 384 385@pytest.fixture(autouse=True) 386def mock_logging_dict_config(mocker: pytest_mock.MockerFixture) -> mock.MagicMock: 387 """Mock the `logging.config.dictConfig()` function.""" 388 return mocker.patch('logging.config.dictConfig') 389 390 391@pytest.fixture(autouse=True) 392def mock_stop_coverage_func(mocker: pytest_mock.MockerFixture) -> mock.MagicMock: 393 """Mock the `coverage` module to stop coverage collection.""" 394 return mocker.patch('poethepoet.executor.base._stop_coverage')
44class MockFile(mock.MagicMock): 45 """Mock the file object returned by `contextlib.closing`.""" 46 47 mock_bytes: bytes 48 49 def read(self) -> bytes: 50 """Mock the `read` method to return data used in tests.""" 51 return self.mock_bytes
Mock the file object returned by contextlib.closing
.
49 def read(self) -> bytes: 50 """Mock the `read` method to return data used in tests.""" 51 return self.mock_bytes
Mock the read
method to return data used in tests.
Inherited Members
- unittest.mock.MagicMixin
- MagicMixin
- unittest.mock.MagicMock
- mock_add_spec
- reset_mock
- unittest.mock.CallableMixin
- side_effect
- unittest.mock.NonCallableMock
- attach_mock
- return_value
- called
- call_count
- call_args
- call_args_list
- mock_calls
- configure_mock
- assert_not_called
- assert_called
- assert_called_once
- assert_called_with
- assert_called_once_with
- assert_has_calls
- assert_any_call
54def mock_file(mock_bytes: bytes) -> MockFile: 55 """Mock the file object returned by `contextlib.closing`.""" 56 mock_file = MockFile() 57 mock_file.mock_bytes = mock_bytes 58 return mock_file
Mock the file object returned by contextlib.closing
.
83@pytest.fixture 84def mock_appconfig_client() -> AppConfigClient: 85 """Mock the `boto3` client for the `AppConfig` service.""" 86 return mock.MagicMock(name='mock_appconfig_client', spec_set=AppConfigClient)
Mock the boto3
client for the AppConfig
service.
98@pytest.fixture 99def mock_session(mocker: MockerFixture) -> Session: 100 """Mock the `boto3.Session` class.""" 101 mock_session = mock.MagicMock(name='mock_session', spec_set=Session) 102 mocker.patch('boto3.Session', return_value=mock_session) 103 return mock_session
Mock the boto3.Session
class.
106@pytest.fixture 107def mock_session_with_0_ids(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 108 """Mock the `boto3` client for the `AppConfig` service to return no IDs.""" 109 mock_page_iterator = mock.MagicMock(spec_set=PageIterator) 110 mock_page_iterator.search.return_value = [] 111 112 mock_paginator = mock.MagicMock(spec_set=Paginator) 113 mock_paginator.paginate.return_value = mock_page_iterator 114 115 mock_appconfig_client.get_paginator.return_value = mock_paginator 116 mock_session.client.return_value = mock_appconfig_client 117 return mock_session
Mock the boto3
client for the AppConfig
service to return no IDs.
120@pytest.fixture 121def mock_session_with_1_id(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 122 """Mock the `boto3` client for the `AppConfig` service to return a single ID.""" 123 mock_page_iterator = mock.MagicMock(name='mock_page_iterator', spec_set=PageIterator) 124 mock_page_iterator.search.return_value = ['id-1'] 125 126 mock_paginator = mock.MagicMock(name='mock_page_iterator', spec_set=Paginator) 127 mock_paginator.paginate.return_value = mock_page_iterator 128 129 mock_appconfig_client.get_paginator.return_value = mock_paginator 130 131 mock_session.client.return_value = mock_appconfig_client 132 return mock_session
Mock the boto3
client for the AppConfig
service to return a single ID.
135@pytest.fixture 136def mock_session_with_2_ids(mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock) -> AppConfigClient: 137 """Mock the `boto3` client for the `AppConfig` service to return two IDs.""" 138 mock_page_iterator = mock.MagicMock(spec_set=PageIterator) 139 mock_page_iterator.search.return_value = ['id-1', 'id-2'] 140 141 mock_paginator = mock.MagicMock(spec_set=Paginator) 142 mock_paginator.paginate.return_value = mock_page_iterator 143 144 mock_appconfig_client.get_paginator.return_value = mock_paginator 145 146 mock_session.client.return_value = mock_appconfig_client 147 return mock_session
Mock the boto3
client for the AppConfig
service to return two IDs.
150@pytest.fixture 151def mock_latest_config() -> GetLatestConfigurationResponseTypeDef: 152 """Mock the response from `get_latest_configuration`.""" 153 mock_config_stream = mock.MagicMock(spec_set=StreamingBody) 154 mock_config_stream.read.return_value = MOCK_YAML_CONFIG 155 return { 156 'NextPollConfigurationToken': 'token', 157 'NextPollIntervalInSeconds': 1, 158 'ContentType': 'application/json', 159 'Configuration': mock_config_stream, 160 'VersionLabel': 'v1', 161 'ResponseMetadata': { 162 'RequestId': '', 163 'HostId': '', 164 'HTTPStatusCode': 200, 165 'HTTPHeaders': {}, 166 'RetryAttempts': 3, 167 }, 168 }
Mock the response from get_latest_configuration
.
171@pytest.fixture 172def mock_latest_config_first_empty() -> GetLatestConfigurationResponseTypeDef: 173 """Mock the response from `get_latest_configuration`. 174 175 Return an empty `bytes` on the first iteration, and `MOCK_YAML_CONFIG` on the second. This supports testing 176 `config_ninja.contrib.appconfig.AppConfig`'s response to an empty return value. 177 """ 178 was_called: list[bool] = [] 179 180 def mock_read(*_: Any, **__: Any) -> bytes: 181 if was_called: 182 return MOCK_YAML_CONFIG 183 was_called.append(True) 184 return b'' 185 186 mock_config_stream = mock.MagicMock(spec_set=StreamingBody) 187 mock_config_stream.read = mock_read 188 return { 189 'NextPollConfigurationToken': 'token', 190 'NextPollIntervalInSeconds': 0, 191 'ContentType': 'application/json', 192 'Configuration': mock_config_stream, 193 'VersionLabel': 'v1', 194 'ResponseMetadata': { 195 'RequestId': '', 196 'HostId': '', 197 'HTTPStatusCode': 200, 198 'HTTPHeaders': {}, 199 'RetryAttempts': 3, 200 }, 201 }
Mock the response from get_latest_configuration
.
Return an empty bytes
on the first iteration, and MOCK_YAML_CONFIG
on the second. This supports testing
config_ninja.contrib.appconfig.AppConfig
's response to an empty return value.
204@pytest.fixture 205def mock_appconfigdata_client(mock_latest_config: mock.MagicMock) -> AppConfigDataClient: 206 """Mock the low-level `boto3` client for the `AppConfigData` service.""" 207 mock_client = mock.MagicMock(name='mock_appconfigdata_client', spec_set=AppConfigDataClient) 208 mock_client.get_latest_configuration.return_value = mock_latest_config 209 return mock_client
Mock the low-level boto3
client for the AppConfigData
service.
212@pytest.fixture 213def mock_appconfigdata_client_first_empty(mock_latest_config_first_empty: mock.MagicMock) -> AppConfigDataClient: 214 """Mock the low-level `boto3` client for the `AppConfigData` service.""" 215 mock_client = mock.MagicMock(name='mock_appconfigdata_client', spec_set=AppConfigDataClient) 216 mock_client.get_latest_configuration.return_value = mock_latest_config_first_empty 217 return mock_client
Mock the low-level boto3
client for the AppConfigData
service.
220@pytest.fixture 221def mock_secretsmanager_client() -> SecretsManagerClient: 222 """Mock the `boto3` client for the `SecretsManager` service.""" 223 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 224 mock_client.get_secret_value.return_value = { 225 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 226 'VersionId': 'v1', 227 } 228 mock_client.list_secret_version_ids.return_value = { 229 'Versions': [{'VersionId': 'v1'}, {'VersionId': 'v2', 'VersionStages': ['AWSCURRENT']}] 230 } 231 232 return mock_client
Mock the boto3
client for the SecretsManager
service.
235@pytest.fixture 236def mock_secretsmanager_client_no_current() -> SecretsManagerClient: 237 """Mock the `boto3` client for the `SecretsManager` service.""" 238 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 239 mock_client.get_secret_value.return_value = { 240 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 241 'VersionId': 'v3', 242 } 243 mock_client.list_secret_version_ids.return_value = { 244 'Versions': [{'VersionId': 'v4'}, {'VersionId': 'v5', 'VersionStages': ['AWSPREVIOUS']}] 245 } 246 247 return mock_client
Mock the boto3
client for the SecretsManager
service.
250@pytest.fixture 251def mock_secretsmanager_client_no_current_initially() -> SecretsManagerClient: 252 """Mock the `boto3` client for the `SecretsManager` service.""" 253 mock_client = mock.MagicMock(name='mock_secretsmanager_client', spec_set=SecretsManagerClient) 254 mock_client.get_secret_value.return_value = { 255 'SecretString': json.dumps({'username': 'admin', 'password': 1234}), 256 'VersionId': 'v6', 257 } 258 259 def _mock_response(versions: list[SecretVersionsListEntryTypeDef]) -> ListSecretVersionIdsResponseTypeDef: 260 return { 261 'ARN': 'arn:aws:secretsmanager:us-west-2:123456789012:secret/my-secret-1-a1b2c3', 262 'Name': 'my-secret-1', 263 'ResponseMetadata': { 264 'HTTPHeaders': {}, 265 'HTTPStatusCode': 200, 266 'RequestId': '12345678-1234-1234-1234-123456789012', 267 'RetryAttempts': 0, 268 }, 269 'Versions': versions, 270 } 271 272 versions_per_call = [ 273 _mock_response([{'VersionId': 'v6', 'VersionStages': ['AWSCURRENT']}]), 274 _mock_response([{'VersionId': 'v6'}, {'VersionId': 'v7'}]), 275 _mock_response( 276 [ 277 {'VersionId': 'v6', 'VersionStages': ['AWSPREVIOUS']}, 278 {'VersionId': 'v7', 'VersionStages': ['AWSCURRENT']}, 279 ] 280 ), 281 ] 282 283 class Counter: 284 count = 0 285 286 def increment(self) -> None: 287 self.count += 1 288 289 num_calls = Counter() 290 291 def mock_list_secret_version_ids(*_: Any, **__: Any) -> ListSecretVersionIdsResponseTypeDef: 292 versions = versions_per_call[num_calls.count] 293 num_calls.increment() 294 return versions 295 296 mock_client.list_secret_version_ids = mock_list_secret_version_ids 297 return mock_client
Mock the boto3
client for the SecretsManager
service.
300@pytest.fixture 301def mock_full_session( 302 mock_session_with_1_id: mock.MagicMock, 303 mock_appconfig_client: mock.MagicMock, 304 mock_appconfigdata_client: mock.MagicMock, 305) -> Session: 306 """Mock the `boto3.Session` class with a full AppConfig client.""" 307 308 def client(service: str) -> mock.MagicMock: 309 if service == 'appconfig': 310 return mock_appconfig_client 311 if service == 'appconfigdata': 312 return mock_appconfigdata_client 313 raise ValueError(f'Unknown service: {service}') 314 315 mock_session_with_1_id.client = client 316 return mock_session_with_1_id
Mock the boto3.Session
class with a full AppConfig client.
319@pytest.fixture 320def mock_poll_too_early( 321 mock_latest_config: GetLatestConfigurationResponseTypeDef, 322) -> AppConfigDataClient: 323 """Raise a `BadRequestException` when polling for configuration changes.""" 324 mock_client = mock.MagicMock(spec_set=AppConfigDataClient) 325 mock_client.exceptions.BadRequestException = ClientError 326 call_count = 0 327 328 def side_effect(*_: Any, **__: Any) -> GetLatestConfigurationResponseTypeDef: 329 nonlocal call_count 330 call_count += 1 331 if call_count == 1: 332 raise mock_client.exceptions.BadRequestException( 333 { 334 'Error': { 335 'Code': 'BadRequestException', 336 'Message': 'Request too early', 337 }, 338 'ResponseMetadata': {}, 339 }, 340 'GetLatestConfiguration', 341 ) 342 return mock_latest_config 343 344 mock_client.get_latest_configuration.side_effect = side_effect 345 346 return mock_client
Raise a BadRequestException
when polling for configuration changes.
349@pytest.fixture 350def monkeypatch_systemd(mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[Path, Path]: 351 """Monkeypatch various utilities for interfacing with `systemd` and the shell. 352 353 Returns: 354 tuple[pathlib.Path, pathlib.Path]: the patched `SYSTEM_INSTALL_PATH` and `USER_INSTALL_PATH` 355 """ 356 mocker.patch('config_ninja.systemd.sh') 357 mocker.patch.context_manager(systemd, 'sudo') 358 mocker.patch('config_ninja.systemd.sdnotify') 359 360 system_install_path = tmp_path / 'system' 361 user_install_path = tmp_path / 'user' 362 363 monkeypatch.setattr(systemd, 'AVAILABLE', True) 364 monkeypatch.setattr(systemd, 'SYSTEM_INSTALL_PATH', system_install_path) 365 monkeypatch.setattr(systemd, 'USER_INSTALL_PATH', user_install_path) 366 367 return (system_install_path, user_install_path)
Monkeypatch various utilities for interfacing with systemd
and the shell.
Returns:
tuple[pathlib.Path, pathlib.Path]: the patched
SYSTEM_INSTALL_PATH
andUSER_INSTALL_PATH
370@pytest.fixture 371def example_file(tmp_path: Path) -> Path: 372 """Write the test configuration to a file in the temporary directory.""" 373 path = tmp_path / 'example.yaml' 374 path.write_bytes(MOCK_YAML_CONFIG) 375 return path
Write the test configuration to a file in the temporary directory.
386@pytest.fixture(autouse=True) 387def mock_logging_dict_config(mocker: pytest_mock.MockerFixture) -> mock.MagicMock: 388 """Mock the `logging.config.dictConfig()` function.""" 389 return mocker.patch('logging.config.dictConfig')
Mock the logging.config.dictConfig()
function.
392@pytest.fixture(autouse=True) 393def mock_stop_coverage_func(mocker: pytest_mock.MockerFixture) -> mock.MagicMock: 394 """Mock the `coverage` module to stop coverage collection.""" 395 return mocker.patch('poethepoet.executor.base._stop_coverage')
Mock the coverage
module to stop coverage collection.