tests.fixtures

  1"""Define fixtures for the test suite."""
  2
  3from __future__ import annotations
  4
  5import contextlib
  6import json
  7from pathlib import Path
  8from typing import Any, Iterator, TypeVar
  9from unittest import mock
 10
 11import pytest
 12from boto3 import Session
 13from botocore.exceptions import ClientError
 14from botocore.paginate import PageIterator, Paginator
 15from botocore.response import StreamingBody
 16from mypy_boto3_appconfig import AppConfigClient
 17from mypy_boto3_appconfigdata import AppConfigDataClient
 18from mypy_boto3_appconfigdata.type_defs import GetLatestConfigurationResponseTypeDef
 19from pytest_mock import MockerFixture
 20
 21from config_ninja import cli, systemd
 22
 23# pylint: disable=redefined-outer-name
 24
 25T = TypeVar('T')
 26
 27MOCK_PYPI_RESPONSE = {'releases': {'1.0': 'ignore', '1.1': 'ignore', '1.2a0': 'ignore'}}
 28MOCK_YAML_CONFIG = b"""
 29key_0: value_0
 30key_1: 1
 31key_2: true
 32key_3:
 33    - 1
 34    - 2
 35    - 3
 36""".strip()
 37
 38
 39class MockFile(mock.MagicMock):
 40    """Mock the file object returned by `contextlib.closing`."""
 41
 42    mock_bytes: bytes
 43
 44    def read(self) -> bytes:
 45        """Mock the `read` method to return data used in tests."""
 46        return self.mock_bytes
 47
 48
 49def mock_file(mock_bytes: bytes) -> MockFile:
 50    """Mock the file object returned by `contextlib.closing`."""
 51    mock_file = MockFile()
 52    mock_file.mock_bytes = mock_bytes
 53    return mock_file
 54
 55
 56@pytest.fixture()
 57def _mock_contextlib_closing(mocker: MockerFixture) -> None:  # pyright: ignore[reportUnusedFunction]
 58    """Mock `contextlib.closing`."""
 59
 60    @contextlib.contextmanager
 61    def _mocked(request: Any) -> Iterator[Any]:
 62        """Pass the input parameter straight through."""
 63        yield request
 64
 65    mocker.patch('contextlib.closing', new=_mocked)
 66
 67
 68@pytest.fixture()
 69def _mock_urlopen_for_pypi(mocker: MockerFixture) -> None:  # pyright: ignore[reportUnusedFunction]
 70    """Mock `urllib.request.urlopen` for PyPI requests."""
 71
 72    def _mocked(_: Any) -> MockFile:
 73        return mock_file(json.dumps(MOCK_PYPI_RESPONSE).encode('utf-8'))
 74
 75    mocker.patch('urllib.request.urlopen', new=_mocked)
 76
 77
 78@pytest.fixture()
 79def mock_appconfig_client() -> AppConfigClient:
 80    """Mock the `boto3` client for the `AppConfig` service."""
 81    return mock.MagicMock(name='mock_appconfig_client', spec_set=AppConfigClient)
 82
 83
 84@pytest.fixture()
 85def _mock_install_io(mocker: MockerFixture) -> None:  # pyright: ignore[reportUnusedFunction]
 86    """Mock various I/O utilities used by the `install` script."""
 87    mocker.patch('shutil.rmtree')
 88    mocker.patch('subprocess.run')
 89    mocker.patch('venv.EnvBuilder')
 90    mocker.patch('runpy.run_path')
 91
 92
 93@pytest.fixture()
 94def mock_session(mocker: MockerFixture) -> Session:
 95    """Mock the `boto3.Session` class."""
 96    mock_session = mock.MagicMock(name='mock_session', spec_set=Session)
 97    mocker.patch('boto3.Session', return_value=mock_session)
 98    return mock_session
 99
100
101@pytest.fixture()
102def mock_session_with_0_ids(
103    mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock
104) -> AppConfigClient:
105    """Mock the `boto3` client for the `AppConfig` service to return no IDs."""
106    mock_page_iterator = mock.MagicMock(spec_set=PageIterator)
107    mock_page_iterator.search.return_value = []
108
109    mock_paginator = mock.MagicMock(spec_set=Paginator)
110    mock_paginator.paginate.return_value = mock_page_iterator
111
112    mock_appconfig_client.get_paginator.return_value = mock_paginator
113    mock_session.client.return_value = mock_appconfig_client
114    return mock_session
115
116
117@pytest.fixture()
118def mock_session_with_1_id(
119    mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock
120) -> 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(
136    mock_appconfig_client: mock.MagicMock, mock_session: mock.MagicMock
137) -> AppConfigClient:
138    """Mock the `boto3` client for the `AppConfig` service to return two IDs."""
139    mock_page_iterator = mock.MagicMock(spec_set=PageIterator)
140    mock_page_iterator.search.return_value = ['id-1', 'id-2']
141
142    mock_paginator = mock.MagicMock(spec_set=Paginator)
143    mock_paginator.paginate.return_value = mock_page_iterator
144
145    mock_appconfig_client.get_paginator.return_value = mock_paginator
146
147    mock_session.client.return_value = mock_appconfig_client
148    return mock_session
149
150
151@pytest.fixture()
152def mock_latest_config() -> GetLatestConfigurationResponseTypeDef:
153    """Mock the response from `get_latest_configuration`."""
154    mock_config_stream = mock.MagicMock(spec_set=StreamingBody)
155    mock_config_stream.read.return_value = MOCK_YAML_CONFIG
156    return {
157        'NextPollConfigurationToken': 'token',
158        'NextPollIntervalInSeconds': 1,
159        'ContentType': 'application/json',
160        'Configuration': mock_config_stream,
161        'VersionLabel': 'v1',
162        'ResponseMetadata': {
163            'RequestId': '',
164            'HostId': '',
165            'HTTPStatusCode': 200,
166            'HTTPHeaders': {},
167            'RetryAttempts': 3,
168        },
169    }
170
171
172@pytest.fixture()
173def mock_appconfigdata_client(mock_latest_config: mock.MagicMock) -> AppConfigDataClient:
174    """Mock the low-level `boto3` client for the `AppConfigData` service."""
175    mock_client = mock.MagicMock(name='mock_appconfigdata_client', spec_set=AppConfigDataClient)
176    mock_client.get_latest_configuration.return_value = mock_latest_config
177    return mock_client
178
179
180@pytest.fixture()
181def mock_full_session(
182    mock_session_with_1_id: mock.MagicMock,
183    mock_appconfig_client: mock.MagicMock,
184    mock_appconfigdata_client: mock.MagicMock,
185) -> Session:
186    """Mock the `boto3.Session` class with a full AppConfig client."""
187
188    def client(service: str) -> mock.MagicMock:
189        if service == 'appconfig':
190            return mock_appconfig_client
191        if service == 'appconfigdata':
192            return mock_appconfigdata_client
193        raise ValueError(f'Unknown service: {service}')
194
195    mock_session_with_1_id.client = client
196    return mock_session_with_1_id
197
198
199@pytest.fixture()
200def mock_poll_too_early(
201    mock_latest_config: GetLatestConfigurationResponseTypeDef,
202) -> AppConfigDataClient:
203    """Raise a `BadRequestException` when polling for configuration changes."""
204    mock_client = mock.MagicMock(spec_set=AppConfigDataClient)
205    mock_client.exceptions.BadRequestException = ClientError
206    call_count = 0
207
208    def side_effect(*_: Any, **__: Any) -> GetLatestConfigurationResponseTypeDef:
209        nonlocal call_count
210        call_count += 1
211        if call_count == 1:
212            raise mock_client.exceptions.BadRequestException(
213                {
214                    'Error': {
215                        'Code': 'BadRequestException',
216                        'Message': 'Request too early',
217                    },
218                    'ResponseMetadata': {},
219                },
220                'GetLatestConfiguration',
221            )
222        return mock_latest_config
223
224    mock_client.get_latest_configuration.side_effect = side_effect
225
226    return mock_client
227
228
229@pytest.fixture()
230def monkeypatch_systemd(
231    mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
232) -> tuple[Path, Path]:
233    """Monkeypatch various utilities for interfacing with `systemd` and the shell.
234
235    Returns:
236        tuple[pathlib.Path, pathlib.Path]: the patched `SYSTEM_INSTALL_PATH` and `USER_INSTALL_PATH`
237    """
238    mocker.patch('config_ninja.systemd.sh')
239    mocker.patch.context_manager(systemd, 'sudo')
240    mocker.patch('config_ninja.systemd.sdnotify')
241
242    system_install_path = tmp_path / 'system'
243    user_install_path = tmp_path / 'user'
244
245    monkeypatch.setattr(cli, 'SYSTEMD_AVAILABLE', True)
246    monkeypatch.setattr(systemd, 'SYSTEM_INSTALL_PATH', system_install_path)
247    monkeypatch.setattr(systemd, 'USER_INSTALL_PATH', user_install_path)
248
249    return (system_install_path, user_install_path)
250
251
252@pytest.fixture()
253def example_file(tmp_path: Path) -> Path:
254    """Write the test configuration to a file in the temporary directory."""
255    path = tmp_path / 'example.yaml'
256    path.write_bytes(MOCK_YAML_CONFIG)
257    return path
258
259
260example_file.__doc__ = f"""Write the test configuration to a file in the temporary directory.
261
262```yaml
263{MOCK_YAML_CONFIG.decode('utf-8')}
264```
265"""