from __future__ import annotations import hashlib import json import base64 import re import unittest from cmbuyer_client.core.models import DeviceCredentials, EvidenceUpload, SecretToken from cmbuyer_client.core.errors import AmbiguousRemoteError, ValidationError from cmbuyer_client.remote.evidence_sink import HttpEvidenceSink from cmbuyer_client.remote.http_transport import HttpResponse from tests.core.test_models import ATTEMPT_ID, TASK_ID, TOKEN from tests.remote.test_task_source import DEVICE_ID, FakeTransport UPLOAD_ID = "43c9f507-7473-4fa6-8d71-8786c34c6301" ASSET_ID = "63c9f507-7473-4fa6-8d71-8786c34c6301" PNG = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=" ) class EvidenceSinkTests(unittest.TestCase): def test_upload_has_fixed_fields_and_never_contains_local_path(self) -> None: digest = hashlib.sha256(PNG).hexdigest() upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, digest, "2026-08-04T09:01:00Z", PNG) payload = { "asset_id": ASSET_ID, "task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "kind": "SKU_PANEL_GATE_1", "privacy_tier": "INTERNAL_RAW", "sha256": digest, "byte_size": len(PNG), "content_type": "image/png", "width_px": 1, "height_px": 1, "captured_at": "2026-08-04T09:01:00Z", } raw = json.dumps(payload, separators=(",", ":")).encode() transport = FakeTransport(HttpResponse(201, (("Content-Type", "application/json"),), raw)) receipt = HttpEvidenceSink(transport).upload(DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload) self.assertEqual(receipt.asset_id, ASSET_ID) self.assertEqual(len(transport.calls), 1) body = transport.calls[0][3] self.assertIn(b'filename="evidence.png"', body) self.assertNotIn(b"C:\\", body) self.assertNotIn(b"manifest", body) self.assertNotIn(b".xml", body) names = re.findall(br'Content-Disposition: form-data; name="([^"]+)"', body) self.assertEqual( names, [b"upload_key", b"attempt_id", b"kind", b"privacy_tier", b"sha256", b"captured_at", b"file"], ) self.assertEqual(body.count(b'filename="evidence.png"'), 1) self.assertNotIn(b"claim_token", body) self.assertNotIn(b"session_id", body) boundary = dict(transport.calls[0][2])["Content-Type"].split("boundary=", 1)[1] self.assertEqual(boundary, "cmbuyer-" + UPLOAD_ID.replace("-", "")) def test_captured_at_equivalent_trailing_zeros_are_accepted(self) -> None: digest = hashlib.sha256(PNG).hexdigest() upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, digest, "2026-08-04T09:01:00.120000Z", PNG) payload = { "asset_id": ASSET_ID, "task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "kind": "SKU_PANEL_GATE_1", "privacy_tier": "INTERNAL_RAW", "sha256": digest, "byte_size": len(PNG), "content_type": "image/png", "width_px": 1, "height_px": 1, "captured_at": "2026-08-04T09:01:00.12Z", } raw = json.dumps(payload, separators=(",", ":")).encode() for status in (200, 201): with self.subTest(status=status): transport = FakeTransport(HttpResponse(status, (("Content-Type", "application/json"),), raw)) receipt = HttpEvidenceSink(transport).upload(DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload) self.assertEqual(receipt.captured_at, "2026-08-04T09:01:00.12Z") def test_wrong_content_hash_fails_before_http_object_can_be_built(self) -> None: transport = FakeTransport(HttpResponse(500, (), b"")) with self.assertRaises(ValidationError): upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, "0" * 64, "2026-08-04T09:01:00Z", PNG) HttpEvidenceSink(transport).upload(DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload) self.assertEqual(transport.calls, []) def test_unknown_2xx_is_ambiguous(self) -> None: digest = hashlib.sha256(PNG).hexdigest() upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, digest, "2026-08-04T09:01:00Z", PNG) for status in (202, 204, 206): with self.subTest(status=status): transport = FakeTransport(HttpResponse(status, (), b"")) with self.assertRaises(AmbiguousRemoteError): HttpEvidenceSink(transport).upload(DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload) self.assertEqual(len(transport.calls), 1) def test_boundary_collision_and_receipt_mismatch_fail_closed(self) -> None: marker = ("--cmbuyer-" + UPLOAD_ID.replace("-", "")).encode() collision_content = PNG + marker collision = EvidenceUpload( TASK_ID, UPLOAD_ID, ATTEMPT_ID, hashlib.sha256(collision_content).hexdigest(), "2026-08-04T09:01:00Z", collision_content, ) transport = FakeTransport(HttpResponse(500, (), b"")) from cmbuyer_client.core.errors import ProtocolRemoteError with self.assertRaises(ProtocolRemoteError): HttpEvidenceSink(transport).upload(DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), collision) self.assertEqual(transport.calls, []) digest = hashlib.sha256(PNG).hexdigest() upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, digest, "2026-08-04T09:01:00Z", PNG) mismatch = { "asset_id": ASSET_ID, "task_id": TASK_ID, "attempt_id": ATTEMPT_ID, "kind": "SKU_PANEL_GATE_1", "privacy_tier": "INTERNAL_RAW", "sha256": "f" * 64, "byte_size": len(PNG), "content_type": "image/png", "width_px": 1, "height_px": 1, "captured_at": "2026-08-04T09:01:00Z", } raw = json.dumps(mismatch, separators=(",", ":")).encode() with self.assertRaises(AmbiguousRemoteError): HttpEvidenceSink(FakeTransport(HttpResponse(201, (("Content-Type", "application/json"),), raw))).upload( DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload ) dimension_mismatch = dict(mismatch) dimension_mismatch["sha256"] = digest dimension_mismatch["width_px"] = 2 raw = json.dumps(dimension_mismatch, separators=(",", ":")).encode() with self.assertRaises(AmbiguousRemoteError): HttpEvidenceSink(FakeTransport(HttpResponse(201, (("Content-Type", "application/json"),), raw))).upload( DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)), upload ) def test_evidence_error_status_matrix(self) -> None: digest = hashlib.sha256(PNG).hexdigest() upload = EvidenceUpload(TASK_ID, UPLOAD_ID, ATTEMPT_ID, digest, "2026-08-04T09:01:00Z", PNG) credentials = DeviceCredentials(DEVICE_ID, SecretToken(TOKEN)) from cmbuyer_client.core.errors import CredentialRemoteError, ManualRemoteError, ProtocolRemoteError cases = ( (401, CredentialRemoteError), (400, ProtocolRemoteError), (403, ProtocolRemoteError), (409, ManualRemoteError), (413, ProtocolRemoteError), (415, ProtocolRemoteError), (500, AmbiguousRemoteError), (503, AmbiguousRemoteError), ) for status, expected in cases: with self.subTest(status=status), self.assertRaises(expected): HttpEvidenceSink(FakeTransport(HttpResponse(status, (), b""))).upload(credentials, upload)