86 lines
4.6 KiB
Python
86 lines
4.6 KiB
Python
"""Offline validation regressions; no camera, server or credentials are contacted."""
|
|
import base64
|
|
from datetime import datetime, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import zlib
|
|
|
|
spec = importlib.util.spec_from_file_location('camera_capture', Path(__file__).with_name('camera-capture.py'))
|
|
camera = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(camera)
|
|
|
|
|
|
def chunk(kind, data):
|
|
return struct.pack('>I', len(data)) + kind + data + struct.pack('>I', zlib.crc32(kind + data) & 0xffffffff)
|
|
|
|
|
|
PNG = b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', 1, 1, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(b'\0\0\0\0')) + chunk(b'IEND', b'')
|
|
TOKEN = 'private-test-token-' * 3
|
|
CAPTURE_ID = '11111111-1111-4111-8111-111111111111'
|
|
POSE = dict(x=0, y=100, z=0, yaw=0, pitch=30, dimension='minecraft:overworld', fov=80, width=1280, height=720)
|
|
|
|
|
|
class CameraCaptureTest(unittest.TestCase):
|
|
def reply(self):
|
|
return dict(status='completed', captureId=CAPTURE_ID, mimeType='image/png', dimension=POSE['dimension'],
|
|
capturedAt=datetime.now(timezone.utc).isoformat(), width=1, height=1, imageBase64=base64.b64encode(PNG).decode())
|
|
|
|
def test_invalid_request_fails_before_config_or_transport(self):
|
|
for change in ({'x': float('nan')}, {'pitch': 91}, {'yaw': -361}, {'y': 2049}, {'fov': 29}, {'width': True}, {'dimension': 'https://example.com'}, {'x': '0'}):
|
|
with self.subTest(change=change), patch.object(camera.checks, 'request_json') as request:
|
|
with self.assertRaises(camera.checks.TestFailure):
|
|
camera.capture('/does/not/exist', POSE | change, '/unused')
|
|
request.assert_not_called()
|
|
|
|
def test_raw_image_is_unchanged_and_metadata_allowlisted(self):
|
|
result = self.reply() | {'arbitrary': {'secret': TOKEN}}
|
|
raw, receipt = camera.validated_image(result, CAPTURE_ID, POSE, TOKEN, time.time())
|
|
self.assertEqual(raw, PNG)
|
|
metadata = json.loads(receipt)
|
|
self.assertTrue(metadata['operator_direct_worker'])
|
|
self.assertEqual(metadata['imageBytes'], len(PNG))
|
|
self.assertNotIn('arbitrary', metadata)
|
|
self.assertNotIn('imageBase64', metadata)
|
|
|
|
def test_wrong_identity_dimension_timestamp_and_size_rejected(self):
|
|
for change in ({'captureId': '22222222-2222-4222-8222-222222222222'}, {'dimension': 'minecraft:the_nether'}, {'width': 2}, {'mimeType': 'image/jpeg'}, {'capturedAt': '2020-01-01T00:00:00Z'}):
|
|
with self.subTest(change=change), self.assertRaises(camera.checks.TestFailure):
|
|
camera.validated_image(self.reply() | change, CAPTURE_ID, POSE, TOKEN, time.time())
|
|
|
|
def test_bad_png_and_private_metadata_rejected(self):
|
|
corrupt = bytearray(PNG);corrupt[30] ^= 1
|
|
for change in ({'imageBase64': 'bad!'}, {'imageBase64': base64.b64encode(corrupt).decode()}, {'readiness': TOKEN}, {'readiness': {}}, {'eyeY': float('nan')}):
|
|
with self.subTest(change=change), self.assertRaises(camera.checks.TestFailure):
|
|
camera.validated_image(self.reply() | change, CAPTURE_ID, POSE, TOKEN, time.time())
|
|
|
|
def test_polled_capture_cannot_switch_identity(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp);config = root / 'config.yml';output = root / 'frame.png'
|
|
config.write_text(f'camera-token: {TOKEN}\ncamera-port: 8766\n')
|
|
responses = [dict(status='pending', captureId=CAPTURE_ID), self.reply() | {'captureId': '22222222-2222-4222-8222-222222222222'}]
|
|
with patch.object(camera.checks, 'request_json', side_effect=responses), patch.object(camera.time, 'sleep'):
|
|
with self.assertRaises(camera.checks.TestFailure):camera.capture(config, POSE, output)
|
|
self.assertFalse(output.exists())
|
|
|
|
def test_oversized_and_nonobject_http_response_rejected(self):
|
|
class Response:
|
|
def __init__(self, raw):self.raw=raw
|
|
def __enter__(self):return self
|
|
def __exit__(self, *args):pass
|
|
def read(self, limit):return self.raw[:limit]
|
|
class Opener:
|
|
def __init__(self, raw):self.raw=raw
|
|
def open(self, *args, **kwargs):return Response(self.raw)
|
|
for raw in (b'[]', b'x' * (camera.checks.MAX_RESPONSE + 1)):
|
|
with self.assertRaises(camera.checks.TestFailure):
|
|
camera.checks.request_json(Opener(raw), 'http://127.0.0.1:8766/health', TOKEN)
|
|
|
|
|
|
if __name__ == '__main__':unittest.main()
|