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