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')
MOCK_PYPI_RESPONSE = {'releases': {'1.0': 'ignore', '1.1': 'ignore', '1.2a0': 'ignore'}}
MOCK_YAML_CONFIG = b'key_0: value_0\nkey_1: 1\nkey_2: true\nkey_3:\n - 1\n - 2\n - 3'
class MockFile(unittest.mock.MagicMock):
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.

mock_bytes: bytes
def read(self) -> bytes:
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
def mock_file(mock_bytes: bytes) -> MockFile:
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.

@pytest.fixture
def mock_appconfig_client() -> mypy_boto3_appconfig.AppConfigClient:
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.

@pytest.fixture
def mock_session(mocker: pytest_mock.MockerFixture) -> boto3.session.Session:
 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.

@pytest.fixture
def mock_session_with_0_ids( mock_appconfig_client: unittest.mock.MagicMock, mock_session: unittest.mock.MagicMock) -> mypy_boto3_appconfig.AppConfigClient:
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.

@pytest.fixture
def mock_session_with_1_id( mock_appconfig_client: unittest.mock.MagicMock, mock_session: unittest.mock.MagicMock) -> mypy_boto3_appconfig.AppConfigClient:
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.

@pytest.fixture
def mock_session_with_2_ids( mock_appconfig_client: unittest.mock.MagicMock, mock_session: unittest.mock.MagicMock) -> mypy_boto3_appconfig.AppConfigClient:
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.

@pytest.fixture
def mock_latest_config() -> mypy_boto3_appconfigdata.type_defs.GetLatestConfigurationResponseTypeDef:
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.

@pytest.fixture
def mock_latest_config_first_empty() -> mypy_boto3_appconfigdata.type_defs.GetLatestConfigurationResponseTypeDef:
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.

@pytest.fixture
def mock_appconfigdata_client( mock_latest_config: unittest.mock.MagicMock) -> mypy_boto3_appconfigdata.AppConfigDataClient:
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.

@pytest.fixture
def mock_appconfigdata_client_first_empty( mock_latest_config_first_empty: unittest.mock.MagicMock) -> mypy_boto3_appconfigdata.AppConfigDataClient:
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.

@pytest.fixture
def mock_secretsmanager_client() -> mypy_boto3_secretsmanager.client.SecretsManagerClient:
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.

@pytest.fixture
def mock_secretsmanager_client_no_current() -> mypy_boto3_secretsmanager.client.SecretsManagerClient:
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.

@pytest.fixture
def mock_secretsmanager_client_no_current_initially() -> mypy_boto3_secretsmanager.client.SecretsManagerClient:
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.

@pytest.fixture
def mock_full_session( mock_session_with_1_id: unittest.mock.MagicMock, mock_appconfig_client: unittest.mock.MagicMock, mock_appconfigdata_client: unittest.mock.MagicMock) -> boto3.session.Session:
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.

@pytest.fixture
def mock_poll_too_early( mock_latest_config: mypy_boto3_appconfigdata.type_defs.GetLatestConfigurationResponseTypeDef) -> mypy_boto3_appconfigdata.AppConfigDataClient:
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.

@pytest.fixture
def monkeypatch_systemd( mocker: pytest_mock.MockerFixture, monkeypatch: _pytest.monkeypatch.MonkeyPatch, tmp_path: pathlib.Path) -> tuple[pathlib.Path, pathlib.Path]:
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 and USER_INSTALL_PATH

@pytest.fixture
def example_file(tmp_path: pathlib.Path) -> pathlib.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.

@pytest.fixture(autouse=True)
def mock_logging_dict_config(mocker: pytest_mock.MockerFixture) -> unittest.mock.MagicMock:
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.

@pytest.fixture(autouse=True)
def mock_stop_coverage_func(mocker: pytest_mock.MockerFixture) -> unittest.mock.MagicMock:
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.