-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconftest.py
More file actions
62 lines (49 loc) · 2.11 KB
/
Copy pathconftest.py
File metadata and controls
62 lines (49 loc) · 2.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""Shared deterministic support for central CI regression tests."""
from __future__ import annotations
from collections.abc import Iterator
import pytest
from scripts.ci import materialize_base_python_requirements as materializer
@pytest.fixture(autouse=True)
def clear_trusted_uv_process_caches() -> Iterator[None]:
"""Isolate the original process-global trusted uv caches across monkeypatches."""
install_cache_clear = materializer._install_trusted_uv.cache_clear
opener_cache_clear = materializer._install_trusted_uv_url_opener.cache_clear
install_cache_clear()
opener_cache_clear()
yield
install_cache_clear()
opener_cache_clear()
class FakeHttpResponse:
"""Expose bounded context-managed reads from one deterministic final URL."""
def __init__(
self,
final_url: str,
payload: bytes = b"archive",
*,
maximum_chunk_size: int | None = None,
) -> None:
"""Store response bytes, final URL, and an optional short-read bound."""
if maximum_chunk_size is not None and maximum_chunk_size < 1:
raise ValueError("maximum_chunk_size must be positive when provided")
self._final_url = final_url
self._payload = payload
self._maximum_chunk_size = maximum_chunk_size
self._offset = 0
def __enter__(self) -> "FakeHttpResponse":
"""Return this response from the context manager."""
return self
def __exit__(self, *_args: object) -> None:
"""Leave the synthetic response context without suppressing errors."""
def geturl(self) -> str:
"""Return the final URL observed by the downloader."""
return self._final_url
def read(self, size: int) -> bytes:
"""Return the next bounded response chunk and advance the stream cursor."""
if size < 0:
size = len(self._payload) - self._offset
if self._maximum_chunk_size is not None:
size = min(size, self._maximum_chunk_size)
start = self._offset
end = min(len(self._payload), start + size)
self._offset = end
return self._payload[start:end]