168 lines
7.7 KiB
Python
168 lines
7.7 KiB
Python
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)
|