73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[1]
|
|
RELEASE_ROOT = META_ROOT / "tools" / "release"
|
|
if str(RELEASE_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(RELEASE_ROOT))
|
|
|
|
from govoplan_release.http_fetch import ( # noqa: E402
|
|
DEFAULT_MAX_RESPONSE_BYTES,
|
|
fetch_http,
|
|
)
|
|
|
|
|
|
class ReleaseHttpFetchTests(unittest.TestCase):
|
|
def test_content_length_over_default_limit_blocks_before_read(self) -> None:
|
|
response = FakeResponse(
|
|
body=b"ignored",
|
|
headers={"Content-Length": str(DEFAULT_MAX_RESPONSE_BYTES + 1)},
|
|
)
|
|
|
|
with mock.patch(
|
|
"govoplan_release.http_fetch.urllib.request.urlopen",
|
|
return_value=response,
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "exceeds the configured size"):
|
|
fetch_http("https://example.invalid/catalog.json", timeout=1)
|
|
|
|
self.assertEqual([], response.read_sizes)
|
|
|
|
def test_streamed_body_is_capped_when_length_is_absent(self) -> None:
|
|
response = FakeResponse(body=b"x" * 9, headers={})
|
|
|
|
with mock.patch(
|
|
"govoplan_release.http_fetch.urllib.request.urlopen",
|
|
return_value=response,
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "exceeds the configured size"):
|
|
fetch_http(
|
|
"https://example.invalid/catalog.json",
|
|
timeout=1,
|
|
max_response_bytes=8,
|
|
)
|
|
|
|
self.assertEqual([9], response.read_sizes)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, *, body: bytes, headers: dict[str, str]) -> None:
|
|
self.status = 200
|
|
self.headers = headers
|
|
self.body = body
|
|
self.read_sizes: list[int] = []
|
|
|
|
def __enter__(self) -> FakeResponse:
|
|
return self
|
|
|
|
def __exit__(self, *args: object) -> None:
|
|
return None
|
|
|
|
def read(self, size: int) -> bytes:
|
|
self.read_sizes.append(size)
|
|
return self.body[:size]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|