tests.fixtures

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