From e120b65218a3ba00999b7239392a82336c2092b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:09:43 +0000 Subject: [PATCH 1/2] Upload raw files above 300 MiB to R2 through its S3 API, and register UCL's house price per square metre (#409, #408 item 6) wrangler r2 object put refuses objects above 300 MiB. chronicle/r2_s3.py signs requests to R2's S3-compatible API (Signature Version 4, region auto) and sends such files as multipart uploads in 100 MiB parts. A failed part is retried on its own, and the upload is aborted if a part still fails. Each part is freed, and the reference cycle httpx leaves around its request collected, before the next is read, so memory holds one part. Credentials come from R2_ACCESS_KEY_ID and R2_SECRET_ACCESS_KEY, or else from the CLOUDFLARE_API_TOKEN Wrangler uses. publish-raw, inventory-artifacts and publish-derived hash files a chunk at a time, and fetch-artifact streams a download to a .partial file. UCL's House Price Per Square Metre in England and Wales, 1995-2023 (UK Data Service ReShare 856542, CC BY 4.0) is registered raw-only, with both bundles (3.2 and 4.4 GB) in R2 and git-ignored here. No build parses it. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 4 + .../r2-large-raw-uploads-409.changed.md | 1 + ...use-price-per-square-metre-856542.added.md | 1 + chronicle/artifacts.py | 112 +++-- chronicle/r2_s3.py | 398 ++++++++++++++++++ .../manifest.yaml | 64 +++ docs/agent-source-package-harness.md | 8 +- docs/data-sources/uk-high-value-housing.md | 6 +- docs/data-sources/uk-land-value-inputs-408.md | 45 +- docs/pe-uk-source-checklist.md | 8 +- tests/test_chronicle_artifacts.py | 107 +++++ tests/test_chronicle_r2_s3.py | 342 +++++++++++++++ ...ucl_house_price_per_square_metre_856542.py | 122 ++++++ 13 files changed, 1178 insertions(+), 40 deletions(-) create mode 100644 changelog.d/r2-large-raw-uploads-409.changed.md create mode 100644 changelog.d/ucl-house-price-per-square-metre-856542.added.md create mode 100644 chronicle/r2_s3.py create mode 100644 db/data/ucl/house_price_per_square_metre_856542/manifest.yaml create mode 100644 tests/test_chronicle_r2_s3.py create mode 100644 tests/test_source_ucl_house_price_per_square_metre_856542.py diff --git a/.gitignore b/.gitignore index 3ee158df..26b35ac2 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,7 @@ db/data/voa/ctsop*_lsoa_2026/CTSOP_*.zip db/data/tpa/lvt_model_2026/config.py db/data/tpa/lvt_model_2026/cells.json db/data/tpa/lvt_model_2026/LICENSE +# UCL's house price per square metre deposit (ReShare 856542, chronicle#409) is two +# zips of 3.2 and 4.4 GB, above GitHub's file limit and carrying Royal Mail address +# data licensed for non-commercial use. R2 holds them; the manifest pins the bytes. +db/data/ucl/house_price_per_square_metre_856542/856542_bundle_*.zip diff --git a/changelog.d/r2-large-raw-uploads-409.changed.md b/changelog.d/r2-large-raw-uploads-409.changed.md new file mode 100644 index 00000000..fe2eb1ca --- /dev/null +++ b/changelog.d/r2-large-raw-uploads-409.changed.md @@ -0,0 +1 @@ +`publish-raw`, `fetch-artifact` and `publish-derived` upload files above Wrangler's 300 MiB limit to R2 through its S3-compatible API (chronicle#409). Each file goes up as a multipart upload in 100 MiB parts, with one part in memory at a time, and a failed part is retried on its own. Every file is hashed a chunk at a time instead of being read whole, and `fetch-artifact` streams downloads to disk. diff --git a/changelog.d/ucl-house-price-per-square-metre-856542.added.md b/changelog.d/ucl-house-price-per-square-metre-856542.added.md new file mode 100644 index 00000000..e9491077 --- /dev/null +++ b/changelog.d/ucl-house-price-per-square-metre-856542.added.md @@ -0,0 +1 @@ +Register UCL's House Price Per Square Metre in England and Wales, 1995-2023 (UK Data Service ReShare 856542), both bundles, raw-only in R2 (chronicle#408, item 6). diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index c3f10266..6eef806b 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -15,12 +15,13 @@ import sqlite3 import subprocess from threading import Thread -from typing import Any, BinaryIO, Literal, get_args +from typing import Any, BinaryIO, Iterator, Literal, get_args from urllib.parse import unquote, urlparse import httpx import yaml +from chronicle import r2_s3 from chronicle.epoch import EMIT_EPOCH, Epoch, canonicalize_key, hash_domain from chronicle.yaml_io import safe_load as safe_load_yaml @@ -29,6 +30,9 @@ DEFAULT_R2_DERIVED_BUCKET = "ledger-derived" DEFAULT_R2_PREFIX = "raw" DEFAULT_R2_DERIVED_PREFIX = "derived" +#: ``wrangler r2 object put`` refuses objects above 300 MiB. Larger files go to R2 +#: through its S3-compatible API as multipart uploads (chronicle#409). +WRANGLER_PUT_LIMIT_BYTES = 300 * 1024 * 1024 # New UK and New Zealand uploads are namespaced by country. US objects predate # the country segment and deliberately keep their legacy ``raw/{source_id}`` @@ -53,6 +57,7 @@ "ons", "scotgov", "slc", + "ucl", "voa", "welshgov", } @@ -525,17 +530,13 @@ def fetch_source_artifact( package_path=output, ) fetched_at = datetime.now(UTC).replace(microsecond=0).isoformat() - content, inferred_filename = _read_artifact(source_url) - artifact_filename = filename or inferred_filename + artifact_filename = filename or _filename_from_url(source_url) if not artifact_filename: raise ValueError("Could not infer artifact filename; pass --filename.") output.mkdir(parents=True, exist_ok=True) local_path = output / artifact_filename - local_path.write_bytes(content) - - sha256 = hashlib.sha256(content).hexdigest() - size_bytes = len(content) + sha256, size_bytes = _read_artifact(source_url, local_path) manifest_path = output / "manifest.yaml" r2_location = ArtifactStorageLocation( @@ -682,8 +683,7 @@ def publish_derived_artifacts( relative_path = artifact_path.relative_to(input_path).as_posix() if relative_path == "build_artifacts.jsonl": continue - content = artifact_path.read_bytes() - sha256 = hashlib.sha256(content).hexdigest() + sha256, size_bytes = _hash_file(artifact_path) location = ArtifactStorageLocation( provider="r2", bucket=r2_bucket, @@ -708,7 +708,7 @@ def publish_derived_artifacts( artifact_name=relative_path, local_path=str(artifact_path), sha256=sha256, - size_bytes=len(content), + size_bytes=size_bytes, r2_location=location, upload=upload, ) @@ -1016,6 +1016,12 @@ def _hash_stream(stream: BinaryIO) -> tuple[str, int]: return digest.hexdigest(), size_bytes +def _hash_file(path: Path) -> tuple[str, int]: + """Hash a file a chunk at a time: raw files can be several GB (chronicle#409).""" + with path.open("rb") as stream: + return _hash_stream(stream) + + _ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") _MISSING_OBJECT_MARKER = re.compile( r"(? str | None: return None -def _read_artifact(source_url: str) -> tuple[bytes, str]: +def _read_artifact(source_url: str, local_path: Path) -> tuple[str, int]: + """Copy a source artifact to ``local_path`` a chunk at a time, hashing it. + + The bytes go to a ``.partial`` file that takes the final name only once the + copy is complete, so a failed download leaves any earlier copy in place. + """ + parsed = urlparse(source_url) + if parsed.scheme not in ("http", "https", "file", ""): + raise ValueError(f"Unsupported source URL scheme: {parsed.scheme}") + partial_path = local_path.with_name(f"{local_path.name}.partial") + digest = hashlib.sha256() + size_bytes = 0 + try: + with partial_path.open("wb") as output: + for chunk in _artifact_chunks(source_url): + output.write(chunk) + digest.update(chunk) + size_bytes += len(chunk) + partial_path.replace(local_path) + except BaseException: + partial_path.unlink(missing_ok=True) + raise + return digest.hexdigest(), size_bytes + + +def _artifact_chunks(source_url: str) -> Iterator[bytes]: parsed = urlparse(source_url) if parsed.scheme in ("http", "https"): - response = httpx.get(source_url, follow_redirects=True, timeout=60) - response.raise_for_status() - return response.content, _filename_from_url(source_url) - if parsed.scheme == "file": - path = Path(unquote(parsed.path)) - return path.read_bytes(), path.name - if not parsed.scheme: - path = Path(source_url) - return path.read_bytes(), path.name - raise ValueError(f"Unsupported source URL scheme: {parsed.scheme}") + with httpx.stream( + "GET", source_url, follow_redirects=True, timeout=60 + ) as response: + response.raise_for_status() + yield from response.iter_bytes(1024 * 1024) + return + path = Path(unquote(parsed.path)) if parsed.scheme == "file" else Path(source_url) + with path.open("rb") as stream: + yield from iter(lambda: stream.read(1024 * 1024), b"") def _filename_from_url(source_url: str) -> str: @@ -1492,6 +1522,8 @@ def _upload_r2_object( wrangler_command: str, ) -> ArtifactCommandResult: content_type, _ = mimetypes.guess_type(local_path.name) + if local_path.stat().st_size > WRANGLER_PUT_LIMIT_BYTES: + return _upload_r2_object_in_parts(location, local_path, content_type) command = [ *shlex.split(wrangler_command), "r2", @@ -1507,6 +1539,36 @@ def _upload_r2_object( return _run_command(command) +def _upload_r2_object_in_parts( + location: ArtifactStorageLocation, + local_path: Path, + content_type: str | None, +) -> ArtifactCommandResult: + """Upload a file too large for Wrangler through R2's S3 API (chronicle#409).""" + command = ( + "r2-s3-multipart-put", + f"{location.bucket}/{location.key}", + str(local_path), + ) + try: + with _r2_s3_client() as client: + parts = client.put_file( + location.bucket, location.key, local_path, content_type=content_type + ) + except (OSError, r2_s3.R2S3Error) as exc: + return ArtifactCommandResult( + command=command, returncode=1, stdout="", stderr=str(exc) + ) + return ArtifactCommandResult( + command=command, returncode=0, stdout=f"uploaded in {parts} parts", stderr="" + ) + + +def _r2_s3_client() -> r2_s3.R2Client: + """Return the S3 API client for large uploads; tests substitute their own.""" + return r2_s3.R2Client(r2_s3.credentials_from_environment()) + + def _r2_only_storage(spec: dict[str, Any]) -> tuple[bool, list[str]]: """Return whether a file entry keeps its bytes in R2 only, and any issues. @@ -1578,9 +1640,7 @@ def _publish_raw_manifest_entry( elif not artifact_path.exists(): errors.append("missing_file") else: - content = artifact_path.read_bytes() - sha256_actual = hashlib.sha256(content).hexdigest() - size_bytes = len(content) + sha256_actual, size_bytes = _hash_file(artifact_path) if sha256_expected and sha256_actual != sha256_expected: errors.append("checksum_mismatch") @@ -1702,9 +1762,7 @@ def _inventory_entry( if not r2_only: errors.append("missing_file") else: - content = artifact_path.read_bytes() - sha256_actual = hashlib.sha256(content).hexdigest() - size_bytes = len(content) + sha256_actual, size_bytes = _hash_file(artifact_path) if sha256_expected and sha256_actual != sha256_expected: errors.append("checksum_mismatch") storage = spec.get("storage") if isinstance(spec, dict) else None diff --git a/chronicle/r2_s3.py b/chronicle/r2_s3.py new file mode 100644 index 00000000..2caffb69 --- /dev/null +++ b/chronicle/r2_s3.py @@ -0,0 +1,398 @@ +"""Upload large R2 objects through R2's S3-compatible API (chronicle#409). + +``wrangler r2 object put`` refuses objects above 300 MiB, so Chronicle sends +larger files here. R2 accepts requests signed with AWS Signature Version 4 in +region ``auto``. A file goes up as a multipart upload in fixed-size parts: each +part is retried on its own, and the upload is aborted if a part still fails, so +no partial object is left under the key. + +Credentials come from ``R2_ACCESS_KEY_ID`` and ``R2_SECRET_ACCESS_KEY`` when +both are set, and otherwise from the ``CLOUDFLARE_API_TOKEN`` that Wrangler +already uses: Cloudflare's R2 token documentation gives that token's S3 key ID +as the token's ID and its secret as the SHA-256 of the token's value. The +account is ``CLOUDFLARE_ACCOUNT_ID``, or else the ``account_id`` that +``wrangler.toml`` pins. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from datetime import UTC, datetime +import gc +import hashlib +import hmac +import os +from pathlib import Path +import time +import tomllib +from types import TracebackType +from urllib.parse import quote +from xml.etree import ElementTree +from xml.sax.saxutils import escape + +import httpx + + +#: R2 takes up to 10,000 parts per upload, so 100 MiB parts reach 976 GiB. +DEFAULT_PART_SIZE_BYTES = 100 * 1024 * 1024 +CLOUDFLARE_API_URL = "https://api.cloudflare.com/client/v4" +_RETRY_STATUSES = frozenset({429, 500, 502, 503, 504}) + + +class R2S3Error(RuntimeError): + """An R2 S3 API request or credential lookup failed.""" + + +@dataclass(frozen=True) +class R2Credentials: + """An S3 API key pair for one Cloudflare account's R2 buckets.""" + + account_id: str + access_key_id: str + secret_access_key: str = field(repr=False) + + @property + def host(self) -> str: + """Return the account's S3 API host.""" + return f"{self.account_id}.r2.cloudflarestorage.com" + + +def credentials_from_environment( + environ: Mapping[str, str] | None = None, + *, + wrangler_toml: str | Path = "wrangler.toml", + transport: httpx.BaseTransport | None = None, +) -> R2Credentials: + """Return R2 S3 credentials from the environment (see the module docstring).""" + environ = os.environ if environ is None else environ + account_id = environ.get("CLOUDFLARE_ACCOUNT_ID") or _wrangler_account_id( + Path(wrangler_toml) + ) + if not account_id: + raise R2S3Error( + "No Cloudflare account: set CLOUDFLARE_ACCOUNT_ID, or run from the " + "directory whose wrangler.toml pins account_id." + ) + access_key_id = environ.get("R2_ACCESS_KEY_ID") + secret_access_key = environ.get("R2_SECRET_ACCESS_KEY") + if access_key_id and secret_access_key: + return R2Credentials(account_id, access_key_id, secret_access_key) + token = environ.get("CLOUDFLARE_API_TOKEN") + if not token: + raise R2S3Error( + "No R2 credentials: set R2_ACCESS_KEY_ID and R2_SECRET_ACCESS_KEY, " + "or CLOUDFLARE_API_TOKEN." + ) + return R2Credentials( + account_id, + _api_token_id(token, account_id, transport=transport), + hashlib.sha256(token.encode("utf-8")).hexdigest(), + ) + + +def sign_v4( + method: str, + *, + host: str, + path: str, + query: Mapping[str, str], + headers: Mapping[str, str], + payload_sha256: str, + access_key_id: str, + secret_access_key: str, + now: datetime, + region: str = "auto", +) -> dict[str, str]: + """Return a request's headers with its AWS Signature Version 4. + + ``path`` is the request path already URI-encoded, as it is sent. Every + header returned is signed: ``host``, ``x-amz-content-sha256``, + ``x-amz-date`` and the ``headers`` given. + """ + amz_date = now.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ") + signed = { + **{name.lower(): " ".join(value.split()) for name, value in headers.items()}, + "host": host, + "x-amz-content-sha256": payload_sha256, + "x-amz-date": amz_date, + } + names = sorted(signed) + signed_names = ";".join(names) + canonical_request = "\n".join( + [ + method, + path, + canonical_query(query), + "".join(f"{name}:{signed[name]}\n" for name in names), + signed_names, + payload_sha256, + ] + ) + scope = f"{amz_date[:8]}/{region}/s3/aws4_request" + string_to_sign = "\n".join( + [ + "AWS4-HMAC-SHA256", + amz_date, + scope, + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(), + ] + ) + key = f"AWS4{secret_access_key}".encode("utf-8") + for part in (amz_date[:8], region, "s3", "aws4_request"): + key = hmac.new(key, part.encode("utf-8"), hashlib.sha256).digest() + signature = hmac.new( + key, string_to_sign.encode("utf-8"), hashlib.sha256 + ).hexdigest() + return { + **signed, + "authorization": ( + f"AWS4-HMAC-SHA256 Credential={access_key_id}/{scope}, " + f"SignedHeaders={signed_names}, Signature={signature}" + ), + } + + +def canonical_query(query: Mapping[str, str]) -> str: + """Return a query string in Signature Version 4's canonical form.""" + return "&".join( + f"{_uri_encode(name)}={_uri_encode(value)}" + for name, value in sorted(query.items()) + ) + + +class R2Client: + """Signed S3 API requests to one Cloudflare account's R2 buckets.""" + + def __init__( + self, + credentials: R2Credentials, + *, + transport: httpx.BaseTransport | None = None, + part_size: int = DEFAULT_PART_SIZE_BYTES, + attempts: int = 4, + clock: Callable[[], datetime] | None = None, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + if part_size <= 0 or attempts <= 0: + raise ValueError("part_size and attempts must be positive") + self.credentials = credentials + self.part_size = part_size + self.attempts = attempts + self._clock = clock or (lambda: datetime.now(UTC)) + self._sleep = sleep + self._http = httpx.Client( + transport=transport, + timeout=httpx.Timeout(60.0, read=600.0, write=600.0), + ) + + def __enter__(self) -> R2Client: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.close() + + def close(self) -> None: + """Close the HTTP connection pool.""" + self._http.close() + + def put_file( + self, + bucket: str, + key: str, + path: str | Path, + *, + content_type: str | None = None, + ) -> int: + """Upload a file as a multipart upload and return its part count. + + Parts are read one at a time, so memory holds one part, not the file. + """ + path = Path(path) + if path.stat().st_size == 0: + raise R2S3Error(f"A multipart upload needs a non-empty file: {path}") + upload_id = self._create_upload(bucket, key, content_type) + parts: list[tuple[int, str]] = [] + try: + with path.open("rb") as stream: + while chunk := stream.read(self.part_size): + number = len(parts) + 1 + parts.append( + ( + number, + self._upload_part(bucket, key, upload_id, number, chunk), + ) + ) + # httpx keeps each response in a reference cycle with its + # request, which holds the part it sent. The cycle + # collector can leave those for many parts, and so most of + # a file in memory, so free the part before the next read. + del chunk + gc.collect() + self._complete_upload(bucket, key, upload_id, parts) + except BaseException: + self._abort_upload(bucket, key, upload_id) + raise + return len(parts) + + def _create_upload(self, bucket: str, key: str, content_type: str | None) -> str: + response = self._request( + "POST", + bucket, + key, + {"uploads": ""}, + headers={"content-type": content_type} if content_type else {}, + ) + upload_id = _xml_text(response, "UploadId") + if not upload_id: + raise R2S3Error(f"CreateMultipartUpload for {key} returned no UploadId") + return upload_id + + def _upload_part( + self, bucket: str, key: str, upload_id: str, number: int, chunk: bytes + ) -> str: + response = self._request( + "PUT", + bucket, + key, + {"partNumber": str(number), "uploadId": upload_id}, + body=chunk, + ) + etag = response.headers.get("etag") + if not etag: + raise R2S3Error(f"UploadPart {number} of {key} returned no ETag") + return etag + + def _complete_upload( + self, bucket: str, key: str, upload_id: str, parts: list[tuple[int, str]] + ) -> None: + body = "".join( + f"{number}{escape(etag)}" + for number, etag in parts + ) + response = self._request( + "POST", + bucket, + key, + {"uploadId": upload_id}, + body=f"{body}".encode( + "utf-8" + ), + headers={"content-type": "application/xml"}, + ) + # S3 can report a failed completion inside a 200 response. + if _xml_root_tag(response) == "Error": + raise R2S3Error( + f"CompleteMultipartUpload for {key} failed: {response.text[:300]}" + ) + + def _abort_upload(self, bucket: str, key: str, upload_id: str) -> None: + try: + self._request("DELETE", bucket, key, {"uploadId": upload_id}) + except R2S3Error: + # The caller's error matters more; R2 expires abandoned uploads. + pass + + def _request( + self, + method: str, + bucket: str, + key: str, + query: Mapping[str, str], + *, + body: bytes = b"", + headers: Mapping[str, str] | None = None, + ) -> httpx.Response: + path = f"/{_uri_encode(bucket)}/{_uri_encode(key, keep_slash=True)}" + url = f"https://{self.credentials.host}{path}" + if query: + url = f"{url}?{canonical_query(query)}" + payload_sha256 = hashlib.sha256(body).hexdigest() + for attempt in range(1, self.attempts + 1): + signed = sign_v4( + method, + host=self.credentials.host, + path=path, + query=query, + headers=headers or {}, + payload_sha256=payload_sha256, + access_key_id=self.credentials.access_key_id, + secret_access_key=self.credentials.secret_access_key, + now=self._clock(), + ) + try: + response = self._http.request(method, url, content=body, headers=signed) + except httpx.TransportError as exc: + if attempt == self.attempts: + raise R2S3Error(f"{method} {key} failed: {exc}") from exc + else: + if response.is_success: + return response + if ( + response.status_code not in _RETRY_STATUSES + or attempt == self.attempts + ): + raise R2S3Error( + f"{method} {key} returned HTTP {response.status_code}: " + f"{response.text[:300]}" + ) + self._sleep(2.0**attempt) + raise AssertionError("unreachable") + + +def _api_token_id( + token: str, account_id: str, *, transport: httpx.BaseTransport | None +) -> str: + """Return an API token's ID: user tokens and account tokens verify apart.""" + try: + with httpx.Client(transport=transport, timeout=30.0) as client: + for url in ( + f"{CLOUDFLARE_API_URL}/user/tokens/verify", + f"{CLOUDFLARE_API_URL}/accounts/{account_id}/tokens/verify", + ): + response = client.get(url, headers={"Authorization": f"Bearer {token}"}) + if response.status_code == 200: + token_id = (response.json().get("result") or {}).get("id") + if token_id: + return str(token_id) + except (httpx.HTTPError, ValueError) as exc: + raise R2S3Error(f"Could not verify CLOUDFLARE_API_TOKEN: {exc}") from exc + raise R2S3Error("CLOUDFLARE_API_TOKEN did not verify, so its R2 key ID is unknown.") + + +def _wrangler_account_id(path: Path) -> str | None: + try: + payload = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError): + return None + account_id = payload.get("account_id") + return account_id if isinstance(account_id, str) and account_id else None + + +def _uri_encode(value: str, *, keep_slash: bool = False) -> str: + return quote(value, safe="-_.~/" if keep_slash else "-_.~") + + +def _xml_text(response: httpx.Response, tag: str) -> str | None: + try: + root = ElementTree.fromstring(response.content) + except ElementTree.ParseError: + return None + # ``{*}`` matches the tag in S3's namespace or in none. + element = root.find(f".//{{*}}{tag}") + return element.text if element is not None else None + + +def _xml_root_tag(response: httpx.Response) -> str | None: + if not response.content.strip(): + return None + try: + tag = ElementTree.fromstring(response.content).tag + except ElementTree.ParseError: + return None + return tag.rsplit("}", 1)[-1] diff --git a/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml new file mode 100644 index 00000000..eac5cbda --- /dev/null +++ b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml @@ -0,0 +1,64 @@ +source_id: ucl +package_id: ucl-house-price-per-square-metre-856542 +dataset: ucl_ucl-house-price-per-square-metre-856542 +source_name: House Price Per Square Metre in England and Wales, 1995-2023 +publisher: University College London, deposited in UK Data Service ReShare +source_page: https://reshare.ukdataservice.ac.uk/856542/ +table: UK Data Service ReShare 856542, bundles 1 and 2 +license: Creative Commons Attribution 4.0 International (CC BY 4.0) +license_url: https://creativecommons.org/licenses/by/4.0/ +attribution: Chi, B., Dennett, A., Oléron-Evans, T. and Morphet, R. (2025). House Price Per Square + Metre in England and Wales, 1995-2023. UK Data Service ReShare. DOI 10.5255/UKDA-SN-856542 +license_evidence: 'ReShare''s record for 856542 states Creative Commons Attribution 4.0 International + (CC BY 4.0). UCL''s Licence_Information_for_data_sources.docx in bundle 1 adds that tranall_link_02122023.zip, + LRPPD_02122023.zip and NSPL_NOV_2023_UK.zip contain Royal Mail copyright postcode and address data, + limited to non-commercial use, with all non-address data under the Open Government Licence v3.0. + It quotes the EPC terms, which say the data contains personal data and that Royal Mail''s copyright + notice must go with any address data that is shared, and the Price Paid terms. So the bytes stay + out of this public repository: the zips are git-ignored and the private ledger-raw bucket holds + them, bundle 1 with that notice.' +license_verified_at: '2026-11-06' +files: + bundle_1: + filename: 856542_bundle_1.zip + source_url: https://reshare.ukdataservice.ac.uk/856542/1/856542_bundle_1.zip + source_table: 'ReShare 856542 bundle 1: the linked file tranall_link_02122023.zip, with UCL''s + readme and licence note' + sha256: e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296 + size_bytes: 3214161126 + fetched_at: '2026-11-06T11:17:39+00:00' + storage: + local: false + r2: + provider: r2 + bucket: ledger-raw + key: raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_1/e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296/856542_bundle_1.zip + uri: r2://ledger-raw/raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_1/e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296/856542_bundle_1.zip + bundle_2: + filename: 856542_bundle_2.zip + source_url: https://reshare.ukdataservice.ac.uk/856542/2/856542_bundle_2.zip + source_table: 'ReShare 856542 bundle 2: the files UCL linked, epc20_id.zip (domestic EPCs with + addresses removed), LRPPD_02122023.zip (Price Paid as at 2 December 2023) and NSPL_NOV_2023_UK.zip' + sha256: 61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97 + size_bytes: 4426805029 + fetched_at: '2026-11-06T11:14:45+00:00' + storage: + local: false + r2: + provider: r2 + bucket: ledger-raw + key: raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_2/61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97/856542_bundle_2.zip + uri: r2://ledger-raw/raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_2/61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97/856542_bundle_2.zip +registration_kind: raw_only +provenance_notes: 'UK Data Service ReShare 856542 (DOI 10.5255/UKDA-SN-856542), published on 22 August + 2025 by Bin Chi, Adam Dennett, Thomas Oléron-Evans and Robin Morphet of UCL, the latest deposit + of the series 854240, 855033, 856204 and 856542. Bundle 1 holds the linked file tranall_link_02122023.zip + (3,214,108,822 bytes, stored), which UCL''s readme describes as 21,438,854 transactions in England + and Wales from 1 January 1995 to 31 October 2023 with 107 fields, plus the readme and the licence + note. Bundle 2 holds the files UCL linked: the domestic EPCs downloaded on 2 December 2023 with + their address fields removed (epc20_id.zip), Price Paid with field names as at 2 December 2023 (LRPPD_02122023.zip) + and the November 2023 NSPL (NSPL_NOV_2023_UK.zip). ReShare''s ETags equal the MD5s of the bytes + fetched on 6 November 2026 (239aebb08e10939b7cba4c5b2d5f89c9 and 5fa5f769f3061aef86c455a4ed23c97e), + and both bundles were last modified on 22 August 2025. Raw only: no build parses either bundle (chronicle#408, + item 6). Both are above the 300 MiB that wrangler r2 object put accepts, so publish-raw uploads + them through R2''s S3 API in 100 MiB parts (chronicle#409).' diff --git a/docs/agent-source-package-harness.md b/docs/agent-source-package-harness.md index 261c1dd5..de1e56c3 100644 --- a/docs/agent-source-package-harness.md +++ b/docs/agent-source-package-harness.md @@ -19,7 +19,13 @@ metadata in `manifest.yaml`, and can upload the exact bytes to the private artifact registry with `uv run chronicle inventory-artifacts --root db/data`. For already-downloaded manifest artifacts, agents should run `uv run chronicle publish-raw --root db/data` to upload checksum-verified bytes to -R2 and write `storage.r2` metadata back into each manifest entry. A manifest +R2 and write `storage.r2` metadata back into each manifest entry. A file above the +300 MiB that `wrangler r2 object put` accepts goes through R2's S3-compatible API +instead, as a multipart upload in 100 MiB parts +([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)). That path +signs with `R2_ACCESS_KEY_ID` and `R2_SECRET_ACCESS_KEY` when both are set, and +otherwise with the S3 key pair Cloudflare derives from `CLOUDFLARE_API_TOKEN`. It +takes the account from `CLOUDFLARE_ACCOUNT_ID` or `wrangler.toml`. A manifest entry marked `storage.local: false` keeps its bytes in R2 only and never in Git: both commands accept it without a local copy, so its absence is not an error to fix by downloading, and `uv run chronicle verify-raw` checks the R2 object. diff --git a/docs/data-sources/uk-high-value-housing.md b/docs/data-sources/uk-high-value-housing.md index bc7af85c..ade857d1 100644 --- a/docs/data-sources/uk-high-value-housing.md +++ b/docs/data-sources/uk-high-value-housing.md @@ -72,9 +72,11 @@ Each of their file entries says so with `storage.local: false`: `.gitignore` keeps the local copies out of commits. Price Paid is registered as HM Land Registry's yearly files, not its single -`pp-complete.csv` (5,525,539,417 bytes). `publish-raw` uploads through +`pp-complete.csv` (5,525,539,417 bytes). `publish-raw` then uploaded only through `wrangler r2 object put`, which refuses files above 300 MiB; the largest yearly -file (`pp-2002.csv`) is 239,344,429 bytes. The 32 yearly files of a release hold +file (`pp-2002.csv`) is 239,344,429 bytes. Larger files now go through R2's S3 +API ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)), but +the yearly files stay: they are the release's own split. The 32 yearly files of a release hold exactly the complete file's rows, split by year of transfer. For this release, order-independent row digests by year agree between the two (31,610,742 rows). diff --git a/docs/data-sources/uk-land-value-inputs-408.md b/docs/data-sources/uk-land-value-inputs-408.md index 29a6a5e6..17c64ff2 100644 --- a/docs/data-sources/uk-land-value-inputs-408.md +++ b/docs/data-sources/uk-land-value-inputs-408.md @@ -396,13 +396,44 @@ These rows give Microcosm council tax by authority for all three nations, which - **Table 9a against Tables 9d and 2e.** 9a's 155,400 claims equal 9d's total. 2e's quarters from 2024 Q2 to 2025 Q1 sum to 155,368.45. -## 6. UCL's house price per square metre: not in this PR - -ReShare 856542's two bundles are 3,214,161,126 and 4,426,805,029 bytes. Each holds stored -nested zips, the linked file `tranall_link_02122023.zip` alone being 3.2 GB. `publish-raw` -uploads through `wrangler r2 object put`, which refuses objects over 300 MiB. Item 6 therefore -waits for a separate change to how Chronicle stores large raw files, which -[chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409) asks for a ruling on. +## 6. UCL's house price per square metre: both bundles, raw-only + +**`ucl-house-price-per-square-metre-856542`** registers +[UK Data Service ReShare 856542](https://reshare.ukdataservice.ac.uk/856542/), *House Price +Per Square Metre in England and Wales, 1995-2023*. It is by Bin Chi, Adam Dennett, Thomas +Oléron-Evans and Robin Morphet (UCL), published on 22 August 2025 under CC BY 4.0. No build +parses it. It came after the rest of this issue, once Chronicle could store files above +300 MiB ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)). + +- **Bundle 1**, `856542_bundle_1.zip`: 3,214,161,126 bytes, sha256 `e397cdac…34b296`. + - It holds the linked file `tranall_link_02122023.zip` (3,214,108,822 bytes, stored), with + UCL's readme and licence note. + - The readme describes the linked file as 21,438,854 transactions in England and Wales from + 1 January 1995 to 31 October 2023, with 107 fields. +- **Bundle 2**, `856542_bundle_2.zip`: 4,426,805,029 bytes, sha256 `61cbf6a7…02da97`. It + holds the files UCL linked: + - the domestic EPCs downloaded on 2 December 2023, with their address fields removed + (`epc20_id.zip`); + - Price Paid with field names as at 2 December 2023 (`LRPPD_02122023.zip`); + - the November 2023 NSPL (`NSPL_NOV_2023_UK.zip`). +- ReShare's ETags equal the MD5s of the bytes fetched on 6 November 2026, and both bundles + were last modified on 22 August 2025. + +**Licence.** ReShare states CC BY 4.0. UCL's licence note adds that the linked file, Price +Paid and the NSPL carry Royal Mail address data, limited to non-commercial use, with +everything else under the Open Government Licence v3.0. It also quotes the EPC terms: the data +contains personal data, and Royal Mail's notice must go with any address data that is shared. +As with Price Paid, the zips are git-ignored and the private `ledger-raw` bucket holds them. + +**Storage.** Both bundles are above the 300 MiB that `wrangler r2 object put` accepts, so +`publish-raw` sent them through R2's S3 API, in 31 and 43 parts of 100 MiB. R2's ETags for +them, `"7eb31fba…44b1df-31"` and `"d04abcef…798c87-43"`, are the MD5 of the parts' MD5s, +which can be recomputed from the pinned bytes without downloading the objects. `verify-raw` +reads them back in full through `wrangler r2 object get --pipe`, which streams. + +Microcosm needs bundle 1's linked file for each sale's floor area; bundle 2 keeps what UCL +linked from. The London Datastore's 2024 edition is not registered, because its record states +no licence. ## 7. Floor area and dwelling density by authority diff --git a/docs/pe-uk-source-checklist.md b/docs/pe-uk-source-checklist.md index e403b860..4b7666d1 100644 --- a/docs/pe-uk-source-checklist.md +++ b/docs/pe-uk-source-checklist.md @@ -457,7 +457,9 @@ facts; field contracts and caveats are in Published inputs for Microcosm's land value work ([PolicyEngine/uk-land-value-tax](https://github.com/PolicyEngine/uk-land-value-tax)). Eight of -the issue's nine items are here: about 326,500 new facts, plus three raw-only registrations. +the issue's nine items came first: about 326,500 new facts, plus three raw-only registrations. +The ninth, UCL's house price per square metre, followed once `publish-raw` could upload files +above 300 MiB ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)). Field contracts and caveats are in [`docs/data-sources/uk-land-value-inputs-408.md`](data-sources/uk-land-value-inputs-408.md). @@ -484,8 +486,8 @@ Field contracts and caveats are in `hmrc-stamp-tax-statistics-2024-25` (1,135 facts). - **Tax Policy Associates' model.** Three files pinned raw at one commit, with no facts (`tpa-lvt-model-2026`). -- **Not in this PR.** UCL's house price per square metre file (item 6) is above the 300 MiB that - `publish-raw` can upload ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)). +- **UCL's house price per square metre (item 6).** UK Data Service ReShare 856542's two + bundles, 3.2 and 4.4 GB, pinned raw with no facts (`ucl-house-price-per-square-metre-856542`). ## Consumer contracts diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index ff407ec5..eca99ec4 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -28,6 +28,7 @@ from chronicle.bundle import NZ_BUNDLE_SOURCE_PREFIXES, NZ_BUNDLE_SOURCES from chronicle.epoch import Epoch from chronicle.harness import main as harness_main +from chronicle.r2_s3 import R2S3Error from chronicle.source_package import load_source_package @@ -677,6 +678,112 @@ def test_publish_raw_refuses_r2_only_bytes_without_an_r2_location(tmp_path): assert not log.exists() +def _fetch_release(tmp_path: Path, content: bytes) -> Path: + source = tmp_path / "release.csv" + source.write_bytes(content) + output_dir = tmp_path / "data" / "hmlr" / "release" + fetch_source_artifact( + str(source), + source_id="hmlr", + package_id="hmlr-release", + year=2026, + output_dir=output_dir, + ) + return output_dir + + +def test_fetch_publish_and_inventory_hash_files_without_reading_them_whole( + tmp_path, monkeypatch +): + # Raw files can be several GB (chronicle#409), so none is read whole. + def refuse_whole_read(path): + raise AssertionError(f"read whole: {path}") + + monkeypatch.setattr(Path, "read_bytes", refuse_whole_read) + wrangler, _ = _logging_wrangler(tmp_path) + output_dir = _fetch_release(tmp_path, b"raw release bytes") + + published = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + inventory = inventory_source_artifacts(output_dir) + + assert published.valid and inventory.valid + (entry,) = published.entries + assert entry.sha256 == hashlib.sha256(b"raw release bytes").hexdigest() + assert entry.size_bytes == len(b"raw release bytes") + assert inventory.entries[0].sha256_actual == entry.sha256 + + +class _RecordingS3Client: + uploads: list[tuple[str, str, bytes, str | None]] = [] + error: Exception | None = None + + def __enter__(self): + return self + + def __exit__(self, *exc_info): + return None + + def put_file(self, bucket, key, path, *, content_type=None): + if self.error is not None: + raise self.error + self.uploads.append((bucket, key, Path(path).read_bytes(), content_type)) + return 1 + + +def test_files_above_the_wrangler_limit_go_through_the_s3_api(tmp_path, monkeypatch): + wrangler, log = _logging_wrangler(tmp_path) + output_dir = _fetch_release(tmp_path, b"above the limit") + monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4) + monkeypatch.setattr(_RecordingS3Client, "uploads", []) + monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client) + + report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text()) + r2 = manifest["files"][2026]["storage"]["r2"] + + assert report.valid + assert report.counts["uploaded_count"] == 1 + assert report.entries[0].upload.command[0] == "r2-s3-multipart-put" + assert _RecordingS3Client.uploads == [ + ("ledger-raw", r2["key"], b"above the limit", "text/csv") + ] + assert r2["key"].startswith("raw/uk/hmlr/hmlr-release/2026/") + assert not log.exists() + + +def test_a_failed_s3_upload_records_no_r2_location(tmp_path, monkeypatch): + wrangler, log = _logging_wrangler(tmp_path) + output_dir = _fetch_release(tmp_path, b"above the limit") + monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4) + monkeypatch.setattr(_RecordingS3Client, "error", R2S3Error("HTTP 403: denied")) + monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client) + + report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text()) + + assert not report.valid + assert report.entries[0].errors == ("r2_upload_failed",) + assert "HTTP 403" in report.entries[0].upload.stderr + assert "storage" not in manifest["files"][2026] + assert not log.exists() + + +def test_fetch_keeps_the_earlier_copy_when_a_download_fails(tmp_path): + output_dir = _fetch_release(tmp_path, b"first copy") + + with pytest.raises(FileNotFoundError): + fetch_source_artifact( + str(tmp_path / "gone" / "release.csv"), + source_id="hmlr", + package_id="hmlr-release", + year=2026, + output_dir=output_dir, + ) + + assert (output_dir / "release.csv").read_bytes() == b"first copy" + assert not (output_dir / "release.csv.partial").exists() + + def test_artifact_cli_commands_emit_json(tmp_path, capsys): source = tmp_path / "source.xls" source.write_bytes(b"cli artifact") diff --git a/tests/test_chronicle_r2_s3.py b/tests/test_chronicle_r2_s3.py new file mode 100644 index 00000000..0a3c7e60 --- /dev/null +++ b/tests/test_chronicle_r2_s3.py @@ -0,0 +1,342 @@ +"""Tests for Chronicle's large-file path to R2 through its S3 API (chronicle#409).""" + +from __future__ import annotations + +from datetime import UTC, datetime +import gc +import hashlib +from pathlib import Path +import tracemalloc + +import httpx +import pytest + +from chronicle.r2_s3 import ( + R2Client, + R2Credentials, + R2S3Error, + canonical_query, + credentials_from_environment, + sign_v4, +) + + +EMPTY_SHA256 = hashlib.sha256(b"").hexdigest() +CREDENTIALS = R2Credentials("account", "key-id", "secret") + + +@pytest.mark.parametrize( + ("method", "path", "query", "headers", "payload", "signature"), + [ + # The three Signature Version 4 examples in AWS's S3 API reference + # ("Authenticating Requests: Using the Authorization Header"). + ( + "GET", + "/test.txt", + {}, + {"Range": "bytes=0-9"}, + b"", + "f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41", + ), + ( + "GET", + "/", + {"max-keys": "2", "prefix": "J"}, + {}, + b"", + "34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7", + ), + ( + "PUT", + "/test%24file.text", + {}, + { + "Date": "Fri, 24 May 2013 00:00:00 GMT", + "x-amz-storage-class": "REDUCED_REDUNDANCY", + }, + b"Welcome to Amazon S3.", + "98ad721746da40c64f1a55b78f14c238d841ea1380cd77a1b5971af0ece108bd", + ), + ], +) +def test_sign_v4_reproduces_the_aws_examples( + method, path, query, headers, payload, signature +): + signed = sign_v4( + method, + host="examplebucket.s3.amazonaws.com", + path=path, + query=query, + headers=headers, + payload_sha256=hashlib.sha256(payload).hexdigest(), + access_key_id="AKIAIOSFODNN7EXAMPLE", + secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + now=datetime(2013, 5, 24, tzinfo=UTC), + region="us-east-1", + ) + + assert signed["authorization"].endswith(f"Signature={signature}") + assert signed["authorization"].startswith( + "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/" + "aws4_request, SignedHeaders=" + ) + + +def test_canonical_query_sorts_names_and_encodes_values(): + assert canonical_query({"uploadId": "a+b/c=", "partNumber": "2"}) == ( + "partNumber=2&uploadId=a%2Bb%2Fc%3D" + ) + assert canonical_query({"uploads": ""}) == "uploads=" + + +class FakeR2: + """An in-memory S3 multipart endpoint that records every request.""" + + def __init__(self, fail_part: dict[int, list[int]] | None = None): + self.requests: list[httpx.Request] = [] + self.parts: dict[int, bytes] = {} + self.fail_part = fail_part or {} + self.complete_status = 200 + self.complete_body = b"" + + def handler(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + query = dict(request.url.params) + assert ( + request.headers["x-amz-content-sha256"] + == hashlib.sha256(request.content).hexdigest() + ) + assert request.headers["authorization"].startswith( + "AWS4-HMAC-SHA256 Credential=key-id/" + ) + if request.method == "POST" and "uploads" in query: + return httpx.Response( + 200, + content=( + b'up+1/=' + b"" + ), + ) + if request.method == "PUT": + number = int(query["partNumber"]) + statuses = self.fail_part.get(number, []) + if statuses: + return httpx.Response(statuses.pop(0)) + self.parts[number] = request.content + return httpx.Response(200, headers={"ETag": f'"etag-{number}"'}) + if request.method == "POST": + return httpx.Response(self.complete_status, content=self.complete_body) + if request.method == "DELETE": + return httpx.Response(204) + raise AssertionError(f"unexpected request {request.method} {request.url}") + + def client(self, **kwargs) -> R2Client: + return R2Client( + CREDENTIALS, + transport=httpx.MockTransport(self.handler), + part_size=5, + sleep=lambda seconds: None, + **kwargs, + ) + + +def _file(tmp_path: Path, content: bytes) -> Path: + path = tmp_path / "bundle.zip" + path.write_bytes(content) + return path + + +def test_put_file_uploads_fixed_size_parts_and_completes_them_in_order(tmp_path): + fake = FakeR2() + + with fake.client() as client: + parts = client.put_file( + "ledger-raw", + "raw/uk/ucl/pkg/bundle_1/digest/bundle.zip", + _file(tmp_path, b"abcdefghijkl"), + content_type="application/zip", + ) + + assert parts == 3 + assert fake.parts == {1: b"abcde", 2: b"fghij", 3: b"kl"} + create, *uploads, complete = fake.requests + assert create.url.path == "/ledger-raw/raw/uk/ucl/pkg/bundle_1/digest/bundle.zip" + assert create.url.query == b"uploads=" + assert create.headers["content-type"] == "application/zip" + assert [request.url.query for request in uploads] == [ + f"partNumber={number}&uploadId=up%2B1%2F%3D".encode() for number in (1, 2, 3) + ] + assert complete.url.query == b"uploadId=up%2B1%2F%3D" + assert complete.content == ( + b"" + b'1"etag-1"' + b'2"etag-2"' + b'3"etag-3"' + b"" + ) + + +def test_put_file_retries_a_failed_part_on_its_own(tmp_path): + fake = FakeR2(fail_part={2: [503, 500]}) + + with fake.client() as client: + client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl")) + + part_numbers = [ + request.url.params["partNumber"] + for request in fake.requests + if request.method == "PUT" + ] + assert part_numbers == ["1", "2", "2", "2", "3"] + assert fake.parts == {1: b"abcde", 2: b"fghij", 3: b"kl"} + + +@pytest.mark.parametrize("statuses", [[403], [503, 503, 503, 503]]) +def test_put_file_aborts_the_upload_when_a_part_fails(tmp_path, statuses): + fake = FakeR2(fail_part={2: statuses}) + + with fake.client() as client, pytest.raises(R2S3Error, match="HTTP"): + client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl")) + + assert [request.method for request in fake.requests][-1] == "DELETE" + assert fake.requests[-1].url.query == b"uploadId=up%2B1%2F%3D" + assert 3 not in fake.parts + assert not any( + request.method == "POST" and "uploadId" in request.url.params + for request in fake.requests + ) + + +def test_put_file_fails_on_an_error_inside_a_successful_completion(tmp_path): + fake = FakeR2() + fake.complete_body = b"InternalError" + + with fake.client() as client, pytest.raises(R2S3Error, match="InternalError"): + client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl")) + + assert fake.requests[-1].method == "DELETE" + + +def test_put_file_frees_each_part_before_reading_the_next(tmp_path): + # httpx keeps each response in a reference cycle with its request, which + # holds the part it sent. With the cycle collector off, every part would + # stay in memory unless put_file frees it. + part_size = 1 << 20 + sizes: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "PUT": + sizes.append(len(request.content)) + return httpx.Response(200, headers={"ETag": '"etag"'}) + if "uploads" in request.url.params: + return httpx.Response(200, content=b"up") + return httpx.Response(200, content=b"") + + path = _file(tmp_path, bytes(8 * part_size)) + client = R2Client( + CREDENTIALS, transport=httpx.MockTransport(handler), part_size=part_size + ) + gc.disable() + tracemalloc.start() + try: + with client: + client.put_file("ledger-raw", "key", path) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + gc.enable() + + assert sizes == [part_size] * 8 + assert peak < 3 * part_size + + +def test_put_file_refuses_an_empty_file(tmp_path): + fake = FakeR2() + + with fake.client() as client, pytest.raises(R2S3Error, match="non-empty"): + client.put_file("ledger-raw", "key", _file(tmp_path, b"")) + + assert fake.requests == [] + + +def test_explicit_r2_keys_need_no_token_lookup(tmp_path): + credentials = credentials_from_environment( + { + "CLOUDFLARE_ACCOUNT_ID": "account", + "R2_ACCESS_KEY_ID": "key-id", + "R2_SECRET_ACCESS_KEY": "secret", + "CLOUDFLARE_API_TOKEN": "unused", + }, + wrangler_toml=tmp_path / "absent.toml", + transport=httpx.MockTransport( + lambda request: pytest.fail("no Cloudflare API call expected") + ), + ) + + assert credentials == R2Credentials("account", "key-id", "secret") + assert credentials.host == "account.r2.cloudflarestorage.com" + assert "secret" not in repr(credentials) + + +@pytest.mark.parametrize("user_status", [200, 401]) +def test_api_token_gives_the_key_pair_and_wrangler_toml_the_account( + tmp_path, user_status +): + wrangler_toml = tmp_path / "wrangler.toml" + wrangler_toml.write_text('name = "chronicle"\naccount_id = "acct-123"\n') + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.url.path) + assert request.headers["authorization"] == "Bearer token-value" + if request.url.path == "/client/v4/user/tokens/verify" and user_status != 200: + return httpx.Response(user_status, json={"success": False}) + return httpx.Response(200, json={"result": {"id": "token-id"}}) + + credentials = credentials_from_environment( + {"CLOUDFLARE_API_TOKEN": "token-value"}, + wrangler_toml=wrangler_toml, + transport=httpx.MockTransport(handler), + ) + + assert credentials == R2Credentials( + "acct-123", "token-id", hashlib.sha256(b"token-value").hexdigest() + ) + expected_paths = ["/client/v4/user/tokens/verify"] + if user_status != 200: + expected_paths.append("/client/v4/accounts/acct-123/tokens/verify") + assert seen == expected_paths + + +def test_a_token_lookup_that_cannot_reach_cloudflare_is_reported(tmp_path): + def unreachable(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("offline", request=request) + + with pytest.raises(R2S3Error, match="Could not verify"): + credentials_from_environment( + {"CLOUDFLARE_ACCOUNT_ID": "account", "CLOUDFLARE_API_TOKEN": "token"}, + wrangler_toml=tmp_path / "absent.toml", + transport=httpx.MockTransport(unreachable), + ) + + +def test_an_unreadable_create_response_is_an_r2_error(tmp_path): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"not xml") + + client = R2Client(CREDENTIALS, transport=httpx.MockTransport(handler)) + with client, pytest.raises(R2S3Error, match="no UploadId"): + client.put_file("ledger-raw", "key", _file(tmp_path, b"abc")) + + +@pytest.mark.parametrize( + ("environ", "message"), + [ + ({"CLOUDFLARE_API_TOKEN": "token"}, "No Cloudflare account"), + ({"CLOUDFLARE_ACCOUNT_ID": "account"}, "No R2 credentials"), + ], +) +def test_missing_credentials_are_reported(tmp_path, environ, message): + with pytest.raises(R2S3Error, match=message): + credentials_from_environment(environ, wrangler_toml=tmp_path / "absent.toml") diff --git a/tests/test_source_ucl_house_price_per_square_metre_856542.py b/tests/test_source_ucl_house_price_per_square_metre_856542.py new file mode 100644 index 00000000..bc524b38 --- /dev/null +++ b/tests/test_source_ucl_house_price_per_square_metre_856542.py @@ -0,0 +1,122 @@ +"""UCL's house price per square metre deposit, registered raw-only (chronicle#408, item 6). + +UK Data Service ReShare 856542 publishes two stored zips, of 3.2 and 4.4 GB, under CC BY +4.0. Both carry Royal Mail address data licensed for non-commercial use. They are pinned and +held in R2 only, uploaded through R2's S3 API because Wrangler takes at most 300 MiB +(chronicle#409), and no build parses them. +""" + +from __future__ import annotations + +from pathlib import Path +import subprocess + +import yaml + +from chronicle.artifacts import ( + WRANGLER_PUT_LIMIT_BYTES, + build_r2_key, + inventory_source_artifacts, + publish_source_artifacts, +) +from chronicle.bundle import UK_BUNDLE_SOURCES +from chronicle.source_package import SOURCE_PACKAGE_ALIASES + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +DIRECTORY = "ucl/house_price_per_square_metre_856542" +PACKAGE_DIR = REPOSITORY_ROOT / "db" / "data" / DIRECTORY +PACKAGE_ID = "ucl-house-price-per-square-metre-856542" +# file key -> filename, sha256, bytes, and ReShare's ETag, which is the file's MD5. +BUNDLES = { + "bundle_1": ( + "856542_bundle_1.zip", + "e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296", + 3_214_161_126, + "239aebb08e10939b7cba4c5b2d5f89c9", + ), + "bundle_2": ( + "856542_bundle_2.zip", + "61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97", + 4_426_805_029, + "5fa5f769f3061aef86c455a4ed23c97e", + ), +} + + +def _manifest() -> dict: + return yaml.safe_load((PACKAGE_DIR / "manifest.yaml").read_text(encoding="utf-8")) + + +def test_both_bundles_are_pinned_raw_only_in_r2(): + manifest = _manifest() + assert manifest["source_id"] == "ucl" + assert manifest["package_id"] == PACKAGE_ID + assert manifest["registration_kind"] == "raw_only" + assert manifest["license"] == ( + "Creative Commons Attribution 4.0 International (CC BY 4.0)" + ) + assert manifest["license_verified_at"] == "2026-11-06" + assert "non-commercial" in manifest["license_evidence"] + assert "10.5255/UKDA-SN-856542" in manifest["attribution"] + assert list(manifest["files"]) == list(BUNDLES) + for file_key, (filename, sha256, size_bytes, md5) in BUNDLES.items(): + entry = manifest["files"][file_key] + number = file_key.removeprefix("bundle_") + assert entry["filename"] == filename + assert entry["source_url"] == ( + f"https://reshare.ukdataservice.ac.uk/856542/{number}/{filename}" + ) + assert entry["sha256"] == sha256 + assert entry["size_bytes"] == size_bytes + assert md5 in manifest["provenance_notes"] + # Too large for Wrangler, so publish-raw sends it through the S3 API. + assert size_bytes > WRANGLER_PUT_LIMIT_BYTES + + key = build_r2_key( + source_id="ucl", + package_id=PACKAGE_ID, + year=file_key, + sha256=sha256, + filename=filename, + package_path=PACKAGE_DIR / "manifest.yaml", + ) + assert key == f"raw/uk/ucl/{PACKAGE_ID}/{file_key}/{sha256}/{filename}" + assert entry["storage"] == { + "local": False, + "r2": { + "provider": "r2", + "bucket": "ledger-raw", + "key": key, + "uri": f"r2://ledger-raw/{key}", + }, + } + ignored = subprocess.run( + ["git", "check-ignore", "--quiet", str(PACKAGE_DIR / filename)], + cwd=REPOSITORY_ROOT, + check=False, + ) + assert ignored.returncode == 0 + + assert PACKAGE_ID not in SOURCE_PACKAGE_ALIASES + assert PACKAGE_ID not in UK_BUNDLE_SOURCES + assert not ( + REPOSITORY_ROOT / "packages" / DIRECTORY / "source_package.yaml" + ).exists() + + +def test_a_checkout_without_the_bundles_stays_valid(tmp_path): + target = tmp_path / "db" / "data" / DIRECTORY + target.mkdir(parents=True) + (target / "manifest.yaml").write_bytes((PACKAGE_DIR / "manifest.yaml").read_bytes()) + + inventory = inventory_source_artifacts(tmp_path / "db" / "data") + published = publish_source_artifacts( + tmp_path / "db" / "data", wrangler_command="false" + ) + + assert inventory.valid + assert inventory.counts["missing_count"] == 0 + assert inventory.counts["r2_only_count"] == 2 + assert published.valid + assert published.counts["skipped_count"] == 2 + assert published.counts["uploaded_count"] == 0 From 555c440959a3723e3c6211c72eed5f944c45153d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:41:50 +0000 Subject: [PATCH 2/2] Skip a large upload when R2 already holds the object, and pin the 300 MiB boundary Review on #426 (Vahid, round 1): - The S3 path now sends HEAD for the key first. The key carries the file's sha256, so an object of the file's size there already holds these bytes, and the file is not sent again. After a failed upload, such as a completion that committed but lost its response, the same check counts the upload once the object is there, and otherwise reports the upload's own error. - A test puts sparse files of exactly 300 MiB and one byte more through _upload_r2_object: the first goes to Wrangler, the second to the S3 path. - The manifest and doc note ReShare's temporal coverage, to 23 October 2023, beside the readme's and abstract's 31 October 2023. Co-Authored-By: Claude Opus 5.5 --- chronicle/artifacts.py | 33 +++++- chronicle/r2_s3.py | 13 ++- .../manifest.yaml | 15 +-- docs/data-sources/uk-land-value-inputs-408.md | 2 + tests/test_chronicle_artifacts.py | 108 +++++++++++++++++- tests/test_chronicle_r2_s3.py | 32 ++++++ 6 files changed, 189 insertions(+), 14 deletions(-) diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index 6eef806b..0ff134fa 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -1544,17 +1544,42 @@ def _upload_r2_object_in_parts( local_path: Path, content_type: str | None, ) -> ArtifactCommandResult: - """Upload a file too large for Wrangler through R2's S3 API (chronicle#409).""" + """Upload a file too large for Wrangler through R2's S3 API (chronicle#409). + + The key carries the file's sha256, so an object of the file's size under it + already holds these bytes. Such a file is not sent again, and a failed + upload, such as a completion whose response was lost, counts once the + object is there. + """ command = ( "r2-s3-multipart-put", f"{location.bucket}/{location.key}", str(local_path), ) try: + size_bytes = local_path.stat().st_size with _r2_s3_client() as client: - parts = client.put_file( - location.bucket, location.key, local_path, content_type=content_type - ) + if client.object_size(location.bucket, location.key) == size_bytes: + return ArtifactCommandResult( + command=command, returncode=0, stdout="already in R2", stderr="" + ) + try: + parts = client.put_file( + location.bucket, location.key, local_path, content_type=content_type + ) + except r2_s3.R2S3Error as error: + try: + landed = client.object_size(location.bucket, location.key) + except r2_s3.R2S3Error: + landed = None + if landed != size_bytes: + raise error + return ArtifactCommandResult( + command=command, + returncode=0, + stdout=f"in R2 although the upload reported: {error}", + stderr="", + ) except (OSError, r2_s3.R2S3Error) as exc: return ArtifactCommandResult( command=command, returncode=1, stdout="", stderr=str(exc) diff --git a/chronicle/r2_s3.py b/chronicle/r2_s3.py index 2caffb69..b1642610 100644 --- a/chronicle/r2_s3.py +++ b/chronicle/r2_s3.py @@ -240,6 +240,16 @@ def put_file( raise return len(parts) + def object_size(self, bucket: str, key: str) -> int | None: + """Return the size in bytes of the object under ``key``, or None if absent.""" + response = self._request("HEAD", bucket, key, {}, missing_ok=True) + if response.status_code == 404: + return None + size = response.headers.get("content-length") + if size is None: + raise R2S3Error(f"HEAD {key} returned no Content-Length") + return int(size) + def _create_upload(self, bucket: str, key: str, content_type: str | None) -> str: response = self._request( "POST", @@ -307,6 +317,7 @@ def _request( *, body: bytes = b"", headers: Mapping[str, str] | None = None, + missing_ok: bool = False, ) -> httpx.Response: path = f"/{_uri_encode(bucket)}/{_uri_encode(key, keep_slash=True)}" url = f"https://{self.credentials.host}{path}" @@ -331,7 +342,7 @@ def _request( if attempt == self.attempts: raise R2S3Error(f"{method} {key} failed: {exc}") from exc else: - if response.is_success: + if response.is_success or (missing_ok and response.status_code == 404): return response if ( response.status_code not in _RETRY_STATUSES diff --git a/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml index eac5cbda..e4392f25 100644 --- a/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml +++ b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml @@ -55,10 +55,11 @@ provenance_notes: 'UK Data Service ReShare 856542 (DOI 10.5255/UKDA-SN-856542), of the series 854240, 855033, 856204 and 856542. Bundle 1 holds the linked file tranall_link_02122023.zip (3,214,108,822 bytes, stored), which UCL''s readme describes as 21,438,854 transactions in England and Wales from 1 January 1995 to 31 October 2023 with 107 fields, plus the readme and the licence - note. Bundle 2 holds the files UCL linked: the domestic EPCs downloaded on 2 December 2023 with - their address fields removed (epc20_id.zip), Price Paid with field names as at 2 December 2023 (LRPPD_02122023.zip) - and the November 2023 NSPL (NSPL_NOV_2023_UK.zip). ReShare''s ETags equal the MD5s of the bytes - fetched on 6 November 2026 (239aebb08e10939b7cba4c5b2d5f89c9 and 5fa5f769f3061aef86c455a4ed23c97e), - and both bundles were last modified on 22 August 2025. Raw only: no build parses either bundle (chronicle#408, - item 6). Both are above the 300 MiB that wrangler r2 object put accepts, so publish-raw uploads - them through R2''s S3 API in 100 MiB parts (chronicle#409).' + note. ReShare''s record gives the temporal coverage as 1 January 1995 to 23 October 2023, while + its abstract, like the readme, says 31 October 2023. Bundle 2 holds the files UCL linked: the domestic + EPCs downloaded on 2 December 2023 with their address fields removed (epc20_id.zip), Price Paid + with field names as at 2 December 2023 (LRPPD_02122023.zip) and the November 2023 NSPL (NSPL_NOV_2023_UK.zip). + ReShare''s ETags equal the MD5s of the bytes fetched on 6 November 2026 (239aebb08e10939b7cba4c5b2d5f89c9 + and 5fa5f769f3061aef86c455a4ed23c97e), and both bundles were last modified on 22 August 2025. Raw + only: no build parses either bundle (chronicle#408, item 6). Both are above the 300 MiB that wrangler + r2 object put accepts, so publish-raw uploads them through R2''s S3 API in 100 MiB parts (chronicle#409).' diff --git a/docs/data-sources/uk-land-value-inputs-408.md b/docs/data-sources/uk-land-value-inputs-408.md index 17c64ff2..1846232c 100644 --- a/docs/data-sources/uk-land-value-inputs-408.md +++ b/docs/data-sources/uk-land-value-inputs-408.md @@ -410,6 +410,8 @@ parses it. It came after the rest of this issue, once Chronicle could store file UCL's readme and licence note. - The readme describes the linked file as 21,438,854 transactions in England and Wales from 1 January 1995 to 31 October 2023, with 107 fields. + - ReShare's record gives the temporal coverage as 1 January 1995 to 23 October 2023, while + its abstract, like the readme, says 31 October 2023. - **Bundle 2**, `856542_bundle_2.zip`: 4,426,805,029 bytes, sha256 `61cbf6a7…02da97`. It holds the files UCL linked: - the domestic EPCs downloaded on 2 December 2023, with their address fields removed diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index eca99ec4..733aad6d 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -13,6 +13,9 @@ from chronicle.cli import main as cli_main from chronicle.artifacts import ( R2_COUNTRY_PUBLISHERS, + WRANGLER_PUT_LIMIT_BYTES, + ArtifactStorageLocation, + _upload_r2_object, build_artifact_key, build_artifact_rows, build_derived_r2_key, @@ -716,6 +719,9 @@ def refuse_whole_read(path): class _RecordingS3Client: uploads: list[tuple[str, str, bytes, str | None]] = [] error: Exception | None = None + # What HEAD finds under the key before an upload, and after one that failed. + existing_size: int | Exception | None = None + size_after_error: int | Exception | None = None def __enter__(self): return self @@ -723,8 +729,14 @@ def __enter__(self): def __exit__(self, *exc_info): return None + def object_size(self, bucket, key): + if isinstance(self.existing_size, Exception): + raise self.existing_size + return self.existing_size + def put_file(self, bucket, key, path, *, content_type=None): if self.error is not None: + type(self).existing_size = self.size_after_error raise self.error self.uploads.append((bucket, key, Path(path).read_bytes(), content_type)) return 1 @@ -751,11 +763,20 @@ def test_files_above_the_wrangler_limit_go_through_the_s3_api(tmp_path, monkeypa assert not log.exists() -def test_a_failed_s3_upload_records_no_r2_location(tmp_path, monkeypatch): +@pytest.mark.parametrize( + "size_after_error", + [None, R2S3Error("HEAD failed")], + ids=["nothing-under-the-key", "head-fails-too"], +) +def test_a_failed_s3_upload_records_no_r2_location( + tmp_path, monkeypatch, size_after_error +): wrangler, log = _logging_wrangler(tmp_path) output_dir = _fetch_release(tmp_path, b"above the limit") monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4) monkeypatch.setattr(_RecordingS3Client, "error", R2S3Error("HTTP 403: denied")) + monkeypatch.setattr(_RecordingS3Client, "existing_size", None) + monkeypatch.setattr(_RecordingS3Client, "size_after_error", size_after_error) monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client) report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) @@ -763,11 +784,94 @@ def test_a_failed_s3_upload_records_no_r2_location(tmp_path, monkeypatch): assert not report.valid assert report.entries[0].errors == ("r2_upload_failed",) - assert "HTTP 403" in report.entries[0].upload.stderr + # The upload's own error is reported, not the check that followed it. + assert report.entries[0].upload.stderr == "HTTP 403: denied" assert "storage" not in manifest["files"][2026] assert not log.exists() +@pytest.mark.parametrize( + ("existing_size", "sent"), + [(15, False), (14, True)], + ids=["same-size", "other-size"], +) +def test_an_object_of_the_files_size_under_its_key_is_not_sent_again( + tmp_path, monkeypatch, existing_size, sent +): + # The key carries the file's sha256, so an object of its size is the file. + wrangler, log = _logging_wrangler(tmp_path) + output_dir = _fetch_release(tmp_path, b"above the limit") + monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4) + monkeypatch.setattr(_RecordingS3Client, "uploads", []) + monkeypatch.setattr(_RecordingS3Client, "existing_size", existing_size) + monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client) + + report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text()) + entry = report.entries[0] + + assert report.valid + assert manifest["files"][2026]["storage"]["r2"]["key"] == entry.r2_location.key + assert len(_RecordingS3Client.uploads) == int(sent) + assert (entry.upload.stdout == "already in R2") is not sent + assert not log.exists() + + +def test_an_upload_whose_completion_landed_counts_as_uploaded(tmp_path, monkeypatch): + # A completion can commit and lose its response: the retry then finds no + # upload and put_file raises, though the object is in R2. + wrangler, log = _logging_wrangler(tmp_path) + output_dir = _fetch_release(tmp_path, b"above the limit") + monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4) + monkeypatch.setattr( + _RecordingS3Client, + "error", + R2S3Error("POST key returned HTTP 404: NoSuchUpload"), + ) + monkeypatch.setattr(_RecordingS3Client, "existing_size", None) + monkeypatch.setattr(_RecordingS3Client, "size_after_error", 15) + monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client) + + report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler)) + manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text()) + entry = report.entries[0] + + assert report.valid + assert entry.errors == () + assert "NoSuchUpload" in entry.upload.stdout + assert manifest["files"][2026]["storage"]["r2"]["key"] == entry.r2_location.key + assert not log.exists() + + +def test_the_s3_path_starts_one_byte_above_the_wrangler_limit(tmp_path, monkeypatch): + # Wrangler refuses only files above its limit (sizeBytes > 300 MiB). + wrangler, log = _logging_wrangler(tmp_path) + sent: list[int] = [] + + class SizeRecordingClient(_RecordingS3Client): + def put_file(self, bucket, key, path, *, content_type=None): + sent.append(Path(path).stat().st_size) + return 4 + + monkeypatch.setattr(_RecordingS3Client, "existing_size", None) + monkeypatch.setattr("chronicle.artifacts._r2_s3_client", SizeRecordingClient) + location = ArtifactStorageLocation(provider="r2", bucket="ledger-raw", key="raw/k") + results = {} + for size in (WRANGLER_PUT_LIMIT_BYTES, WRANGLER_PUT_LIMIT_BYTES + 1): + path = tmp_path / f"{size}.zip" + with path.open("wb") as stream: + stream.truncate(size) # sparse, so nothing is written + results[size] = _upload_r2_object( + location, path, wrangler_command=str(wrangler) + ) + + assert WRANGLER_PUT_LIMIT_BYTES == 300 * 1024 * 1024 + assert results[WRANGLER_PUT_LIMIT_BYTES].command[0] == str(wrangler) + assert results[WRANGLER_PUT_LIMIT_BYTES + 1].command[0] == "r2-s3-multipart-put" + assert sent == [WRANGLER_PUT_LIMIT_BYTES + 1] + assert log.read_text().count("r2 object put") == 1 + + def test_fetch_keeps_the_earlier_copy_when_a_download_fails(tmp_path): output_dir = _fetch_release(tmp_path, b"first copy") diff --git a/tests/test_chronicle_r2_s3.py b/tests/test_chronicle_r2_s3.py index 0a3c7e60..6f9acce9 100644 --- a/tests/test_chronicle_r2_s3.py +++ b/tests/test_chronicle_r2_s3.py @@ -98,6 +98,8 @@ def __init__(self, fail_part: dict[int, list[int]] | None = None): self.fail_part = fail_part or {} self.complete_status = 200 self.complete_body = b"" + self.objects: dict[str, int] = {} + self.head_status: int | None = None def handler(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) @@ -109,6 +111,15 @@ def handler(self, request: httpx.Request) -> httpx.Response: assert request.headers["authorization"].startswith( "AWS4-HMAC-SHA256 Credential=key-id/" ) + if request.method == "HEAD": + if self.head_status is not None: + return httpx.Response(self.head_status) + key = request.url.path.split("/", 2)[2] + if key not in self.objects: + return httpx.Response(404) + return httpx.Response( + 200, headers={"Content-Length": str(self.objects[key])} + ) if request.method == "POST" and "uploads" in query: return httpx.Response( 200, @@ -251,6 +262,27 @@ def handler(request: httpx.Request) -> httpx.Response: assert peak < 3 * part_size +def test_object_size_reads_head_and_is_none_for_a_missing_key(): + fake = FakeR2() + fake.objects["raw/uk/ucl/pkg/bundle_2.zip"] = 4_426_805_029 + + with fake.client() as client: + present = client.object_size("ledger-raw", "raw/uk/ucl/pkg/bundle_2.zip") + missing = client.object_size("ledger-raw", "raw/uk/ucl/pkg/bundle_3.zip") + + assert present == 4_426_805_029 + assert missing is None + assert [request.method for request in fake.requests] == ["HEAD", "HEAD"] + + +def test_object_size_reports_a_refused_head_rather_than_a_missing_object(): + fake = FakeR2() + fake.head_status = 403 + + with fake.client() as client, pytest.raises(R2S3Error, match="HTTP 403"): + client.object_size("ledger-raw", "key") + + def test_put_file_refuses_an_empty_file(tmp_path): fake = FakeR2()