Fix browser recording playback and web management consistency

test_app.py

@@ -0,0 +1,133 @@
+"""Isolated regression tests: python -m unittest -v test_app."""
+import hashlib
+import http.client
+import json
+import os
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+_data = tempfile.TemporaryDirectory()
+os.environ['VISION_DATA'] = _data.name
+os.environ['VISION_RECORDINGS'] = _data.name + '/recordings'
+import app
+import playback
+
+
+class WebTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        app.initialize()
+        cls.server = app.ThreadingHTTPServer(('127.0.0.1', 0), app.Handler)
+        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
+        cls.thread.start()
+        app.DB.execute('INSERT INTO sessions VALUES (?,?)',
+                       (hashlib.sha256(b'f' * 64).hexdigest(), time.time() + 60))
+        app.DB.commit()
+        app.save_object('cameras', {'id': 'a' * 16, 'name': 'Example', 'password': 'private'})
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.server.shutdown()
+        cls.server.server_close()
+        cls.thread.join()
+        app.DB.close()
+        _data.cleanup()
+
+    def request(self, path, authenticated=True, data=None, origin=None):
+        c = http.client.HTTPConnection(*self.server.server_address, timeout=3)
+        headers = {'Cookie': 'vision=' + 'f' * 64} if authenticated else {}
+        if origin:
+            headers['Origin'] = origin
+        if data is not None:
+            headers['Content-Type'] = 'application/json'
+        c.request('GET' if data is None else 'POST', path,
+                  None if data is None else json.dumps(data), headers)
+        r = c.getresponse()
+        result = r.status, dict(r.getheaders()), r.read()
+        c.close()
+        return result
+
+    def test_media_and_api_require_login(self):
+        for path in ['/api/state', '/api/recordings', '/media/playback',
+                     '/media/live/cam_' + 'a' * 16 + '_hd/index.m3u8']:
+            with self.subTest(path=path):
+                self.assertEqual(self.request(path, False)[0], 401)
+
+    def test_static_security_headers(self):
+        code, headers, _ = self.request('/', False)
+        self.assertEqual(code, 200)
+        self.assertEqual(headers['X-Frame-Options'], 'DENY')
+        self.assertIn("frame-ancestors 'none'", headers['Content-Security-Policy'])
+
+    def test_origin_rejected(self):
+        self.assertEqual(self.request('/api/storage', data={}, origin='https://example.org')[0], 403)
+
+    def test_invalid_ranges_and_paths(self):
+        base = '/media/playback?camera=' + 'a' * 16 + '&start=2026-01-01T00:00:00Z&duration='
+        for duration in ['nan', 'inf', '-1', '3601']:
+            self.assertEqual(self.request(base + duration)[0], 400)
+        self.assertEqual(self.request('/media/live/../../vision.db')[0], 400)
+        self.assertEqual(self.request('/vision.db', False)[0], 404)
+
+    def test_compatible_playback_dispatch(self):
+        calls = []
+        def compatible(handler, path, duration):
+            calls.append((path, duration))
+            handler.answer({'ok': True})
+        with patch.object(app.Handler, 'compatible_playback', compatible):
+            code, _, _ = self.request('/media/playback?camera=' + 'a' * 16 + '&start=2026-01-01T00:00:00Z&duration=10')
+        self.assertEqual(code, 200)
+        self.assertEqual(calls[0][1], 10)
+        self.assertIn('format=fmp4', calls[0][0])
+
+    def test_original_download_dispatch(self):
+        calls = []
+        def proxy(handler, port, path, download=False):
+            calls.append((path, download))
+            handler.answer({'ok': True})
+        with patch.object(app.Handler, 'proxy', proxy):
+            code, _, _ = self.request('/media/playback?camera=' + 'a' * 16 + '&start=2026-01-01T00:00:00Z&duration=10&download=1')
+        self.assertEqual(code, 200)
+        self.assertIn('format=mp4', calls[0][0])
+        self.assertTrue(calls[0][1])
+
+    def test_stable_object_order(self):
+        app.save_object('sites', {'id': 'b', 'name': 'Second'})
+        app.save_object('sites', {'id': 'a', 'name': 'First'})
+        first = app.objects('sites')
+        app.save_object('sites', first[0])
+        self.assertEqual(app.objects('sites'), first)
+
+    def test_password_redacted(self):
+        self.assertNotIn('password', app.public_camera(app.get_object('cameras', 'a' * 16)))
+
+    def test_playback_has_bound_and_cleanup(self):
+        playback.SLOTS.acquire()
+        playback.SLOTS.acquire()
+        try:
+            with self.assertRaises(playback.PlaybackError):
+                with playback.stream('unused', 'private', 10):
+                    self.fail('No third conversion allowed')
+        finally:
+            playback.SLOTS.release()
+            playback.SLOTS.release()
+        with patch('playback.shutil.which', return_value=None):
+            with self.assertRaises(playback.PlaybackError):
+                with playback.stream('unused', 'private', 10):
+                    pass
+        self.assertTrue(playback.SLOTS.acquire(blocking=False))
+        playback.SLOTS.release()
+
+    def test_vaapi_uses_supported_cqp(self):
+        with patch('playback.shutil.which', return_value='ffmpeg'), patch('playback.os.access', return_value=True):
+            command = playback.command('http://127.0.0.1/source', 'private', 10)
+        self.assertEqual(command[command.index('-rc_mode') + 1], 'CQP')
+        self.assertIn('h264_vaapi', command)
+        self.assertIn('pipe:1', command)
+
+
+if __name__ == '__main__':
+    unittest.main()