Private
Public Access
138 lines
6.6 KiB
Python
138 lines
6.6 KiB
Python
"""Local trigger pictures: synchronized crop, persistence, API and deletion."""
|
|
import asyncio
|
|
from io import BytesIO
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from PIL import Image
|
|
|
|
from halloween.event_store import EventStore
|
|
from halloween.web import create_app
|
|
|
|
|
|
def sample():
|
|
image = Image.new('RGB', (100, 80), (20, 40, 180))
|
|
# Distinct central crop: outside the chosen area must not be recorded.
|
|
image.paste((180, 40, 20), (20, 16, 80, 64))
|
|
stream = BytesIO()
|
|
image.save(stream, 'JPEG', quality=95)
|
|
return stream.getvalue()
|
|
|
|
|
|
REGION = [.2, .2, .8, .8]
|
|
|
|
|
|
class StoreImageTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.folder = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.folder.cleanup)
|
|
self.root = Path(self.folder.name)
|
|
self.store = EventStore(self.root)
|
|
|
|
def test_automatic_kinds_persist_cropped_picture_after_restart(self):
|
|
for method in (self.store.record_detection_trigger, self.store.record_motion_trigger):
|
|
event_id = method('sanft', jpeg=sample(), region=REGION)
|
|
path = EventStore(self.root).image_path(event_id)
|
|
with Image.open(path) as image:
|
|
self.assertEqual(image.size, (60, 48))
|
|
self.assertEqual(image.format, 'JPEG')
|
|
self.assertGreater(image.getpixel((30, 24))[0], 150)
|
|
self.assertLess(image.getpixel((30, 24))[2], 50)
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(path.parent.stat().st_mode & 0o777, 0o700)
|
|
event = self.store.recent()[0]
|
|
self.assertEqual(event['image_url'], f'/api/events/{event_id}/image')
|
|
self.assertNotIn('image_filename', event)
|
|
self.assertIsNone(event['profile_id'])
|
|
|
|
def test_delete_image_preserves_event_and_profile(self):
|
|
event_id = self.store.record_motion_trigger('sanft', jpeg=sample(), region=REGION)
|
|
profile = self.store.create_profile('Fiktiv')
|
|
self.store.assign_profile(event_id, profile['id'])
|
|
path = self.store.image_path(event_id)
|
|
self.assertTrue(self.store.delete_image(event_id))
|
|
self.assertFalse(path.exists())
|
|
event = EventStore(self.root).recent()[0]
|
|
self.assertIsNone(event['image_url'])
|
|
self.assertEqual(event['profile_id'], profile['id'])
|
|
self.assertFalse(self.store.delete_image(event_id))
|
|
with self.assertRaises(KeyError):
|
|
self.store.image_path(event_id)
|
|
|
|
def test_manual_events_and_migrated_old_events_have_no_picture(self):
|
|
event_id = self.store.record_test_trigger('sanft')
|
|
self.assertIsNone(self.store.recent()[0]['image_url'])
|
|
with self.assertRaises(KeyError):
|
|
self.store.image_path(event_id)
|
|
old_root = self.root / 'legacy'
|
|
old_root.mkdir()
|
|
with sqlite3.connect(old_root / 'events.sqlite3') as connection:
|
|
connection.execute('CREATE TABLE events (id INTEGER PRIMARY KEY, occurred_at TEXT, kind TEXT, mode TEXT)')
|
|
connection.execute("INSERT INTO events VALUES (1,'2026-10-14T10:00:00+00:00','test_trigger','sanft')")
|
|
old = EventStore(old_root)
|
|
self.assertIsNone(old.recent()[0]['image_url'])
|
|
self.assertEqual(old.recent()[0]['id'], 1)
|
|
new_id = old.record_motion_trigger('sanft', jpeg=sample(), region=REGION)
|
|
self.assertTrue(old.image_path(new_id).exists())
|
|
|
|
def test_database_failure_removes_written_image(self):
|
|
with patch.object(self.store, '_connect', side_effect=sqlite3.OperationalError('synthetic DB failure')):
|
|
with self.assertRaises(sqlite3.Error):
|
|
self.store.record_motion_trigger('sanft', jpeg=sample(), region=REGION)
|
|
self.assertEqual(list(self.store.image_dir.iterdir()), [])
|
|
self.assertEqual(self.store.recent(), [])
|
|
|
|
def test_image_failure_leaves_no_temp_file_or_database_event(self):
|
|
jpeg = sample()
|
|
with patch.object(Image.Image, 'save', side_effect=OSError('synthetic disk full')):
|
|
with self.assertRaises(OSError):
|
|
self.store.record_motion_trigger('sanft', jpeg=jpeg, region=REGION)
|
|
self.assertEqual(self.store.recent(), [])
|
|
if self.store.image_dir.exists():
|
|
self.assertEqual(list(self.store.image_dir.iterdir()), [])
|
|
|
|
def test_tampered_database_cannot_read_or_delete_outside_image_dir(self):
|
|
outside = self.root / 'keep.txt'
|
|
outside.write_text('keep')
|
|
event_id = self.store.record_test_trigger('sanft')
|
|
with sqlite3.connect(self.store.path) as connection:
|
|
connection.execute('UPDATE events SET image_filename = ? WHERE id = ?', ('../keep.txt', event_id))
|
|
with self.assertRaises(KeyError):
|
|
self.store.image_path(event_id)
|
|
with self.assertRaises(KeyError):
|
|
self.store.delete_image(event_id)
|
|
self.assertEqual(outside.read_text(), 'keep')
|
|
self.assertIsNone(self.store.recent()[0]['image_url'])
|
|
|
|
|
|
class ImageApiTest(unittest.TestCase):
|
|
def test_view_and_delete_image_without_affecting_event(self):
|
|
async def scenario():
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
app = create_app(demo=True, state_dir=Path(folder))
|
|
event_id = app.state.events.record_motion_trigger('sanft', jpeg=sample(), region=REGION)
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
|
|
url = (await client.get('/api/events')).json()['events'][0]['image_url']
|
|
image = await client.get(url)
|
|
self.assertEqual(image.status_code, 200)
|
|
self.assertEqual(image.headers['content-type'], 'image/jpeg')
|
|
self.assertEqual(image.headers['cache-control'], 'no-store')
|
|
self.assertEqual(image.headers['x-content-type-options'], 'nosniff')
|
|
self.assertEqual((await client.get('/api/events/999/image')).status_code, 404)
|
|
self.assertEqual((await client.delete(url)).status_code, 403)
|
|
self.assertEqual((await client.delete(url, headers={'X-Halloween-UI':'1'})).status_code, 200)
|
|
self.assertEqual((await client.get(url)).status_code, 404)
|
|
event = (await client.get('/api/events')).json()['events'][0]
|
|
self.assertEqual(event['id'], event_id)
|
|
self.assertIsNone(event['image_url'])
|
|
asyncio.run(scenario())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|