impuls.tools.testing_mocks¶

class impuls.tools.testing_mocks.MockResource(content: bytes, fetchtime: datetime = datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc), last_modified: datetime = datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc), clock: DatetimeNowLike = datetime.now, persist_last_modified: bool = False)¶

Bases: ConcreteResource

MockResource mocks a Resource, returning a predefined content and allowing control over the last_modified attribute.

fetch(conditional: bool) → Iterator[bytes]¶

fetch returns the content of the resource; preferably in chunks of FETCH_CHUNK_SIZE length.

last_modified and fetch_time attributes of the should be updated right before the first chunk is returned.

If the conditional is set, the Resource must raise InputNotModified if the resource was not modified since last_modified. In this case, last_modified and fetch_time must not be updated.

load_extra_metadata(metadata: dict[str, Any]) → None¶: Invoked by Impuls resource mechanism to load extra metadata returned by save_extra_metadata(). Not called if a resource has no extra metadata.

refresh() → None¶

save_extra_metadata() → dict[str, Any] | None¶

Serializes any extra metadata into JSON to be preserved across runs.

If an empty dictionary or None is returned, extra metadata is not saved.

clock: DatetimeNowLike¶

content: bytes¶

extra_metadata: dict[str, Any] | None¶

persistent_last_modified: datetime | None¶

class impuls.tools.testing_mocks.DatetimeNowLike(*args, **kwargs)¶

Bases: Protocol

__call__(tz: tzinfo | None = Ellipsis) → datetime¶: Call self as a function.

class impuls.tools.testing_mocks.MockDatetimeNow(times: Iterable[datetime])¶

Bases: object

MockDatetimeNow is a helper for mocking datetime.now, by returning datetimes from a provided iterator.

Once the provided iterator runs out, StopIteration is raised.

>>> fake_dt_now = MockDatetimeNow([
...     datetime(2020, 1, 30, 5, 10),
...     datetime(2020, 1, 30, 5, 20),
...     datetime(2020, 1, 30, 5, 30),
...     datetime(2020, 1, 30, 5, 40),
... ]).now
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 10)
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 20)
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 30)
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 40)
>>> fake_dt_now()
Traceback (most recent call last):
    ...
StopIteration

classmethod constant(t: datetime) → Self¶

classmethod evenly_spaced(start: datetime, delta: timedelta) → Self¶

evenly_spaced provides an infinite MockDatetimeNow which returns (start, start + delta, start + 2*delta, …).

>>> fake_dt_now = (
...     MockDatetimeNow
...     .evenly_spaced(datetime(2020, 1, 30, 5, 10), timedelta(minutes=10))
...     .now
... )
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 10)
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 20)
>>> fake_dt_now()
datetime.datetime(2020, 1, 30, 5, 30)

now(tz: tzinfo | None = None) → datetime¶

patch(*datetime_targets: str) → Generator[None, None, None]¶

class impuls.tools.testing_mocks.MockFile(prefix: str = 'impuls-test', suffix: str | None = None, directory: bool = False)¶

Bases: object

MockFile creates a temporary file for testing purposes. The file must be removed after usage by calling mock_file.cleanup(). This action is automatically performed if MockFile is used in a with statement.

>>> with MockFile() as f:
...     _ = f.write_text("Hello, world!")
...     f.read_text()
'Hello, world!'

__enter__() → Path¶

__exit__(*_: Any) → bool¶

cleanup() → None¶

path: Path¶

class impuls.tools.testing_mocks.MockHTTPResponse(status_code: int, content: bytes = b'', headers: Mapping[str, str] | None = {})¶

Bases: object

MockHTTPResponse tries to mimic the requests.Response object. Only methods and attributes required for the tests are implemented.

>>> r = MockHTTPResponse(200, b"Hello!")
>>> r.status_code
200
>>> r.content
b'Hello!'
>>> r.headers
{}

__enter__() → MockHTTPResponse¶: The context manager for MockHTTPResponse does nothing. >>> with MockHTTPResponse(200, b”Hello!”) as r: … r.status_code, r.content (200, b’Hello!’)

__exit__(*_: Any) → bool¶

iter_content(chunk_size: int = 16, decode_unicode: bool = False) → Iterable[bytes]¶

iter_content generates self.content in chunks of the provided size. For now decode_unicode must be False.

>>> r = MockHTTPResponse(200, b"Lorem ipsum dolor sit")
>>> list(r.iter_content(8))
[b'Lorem ip', b'sum dolo', b'r sit']

raise_for_status() → None¶

Raises requests.HTTPError if the status_code is bigger than or equal to 400.

>>> MockHTTPResponse(200).raise_for_status()
>>> MockHTTPResponse(404).raise_for_status()
Traceback (most recent call last):
    ...
requests.exceptions.HTTPError: 404