ucvl-home-vision/test_app.py

"""Isolated regression tests: python -m unittest -v test_app."""
import hashlib
import http.client
import io
import json
import os
import re
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.admin = app.ACCOUNTS.save({'username': 'admin', 'password': 'test-password', 'role': 'admin'}, None)
        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() + 3600, cls.admin['id']))
        app.DB.commit()
        app.save_object('cameras', {'id': 'a' * 16, 'name': 'Example', 'siteId': 'test-site', '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_live_quality_has_only_main_and_sub(self):
        code, _, body = self.request('/', False)
        self.assertEqual(code, 200)
        select = re.search(r'<select id="quality">(.*?)</select>', body.decode()).group(1)
        self.assertEqual(re.findall(r'<option value="(.*?)">(.*?)</option>', select),
                         [('hd', '主码流'), ('compat', '子码流')])

    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_accounts_and_authorization(self):
        member = app.ACCOUNTS.save({'username': 'member', 'password': 'password8', 'familyAccess': 'read'}, self.admin)
        self.assertNotIn('hash', member)
        self.assertNotIn('salt', member)
        self.assertIsNotNone(app.ACCOUNTS.verify('member', 'password8'))
        self.assertIsNone(app.ACCOUNTS.verify('member', 'incorrect'))
        code, _, body = self.request('/api/users')
        self.assertEqual(code, 200)
        self.assertNotIn(b'"hash"', body)
        token='e'*64
        app.DB.execute('INSERT INTO sessions VALUES (?,?,?)',(hashlib.sha256(token.encode()).hexdigest(),time.time()+3600,member['id']))
        app.DB.commit()
        def request(path,data=None):
            c=http.client.HTTPConnection(*self.server.server_address,timeout=3)
            c.request('POST' if data is not None else 'GET',path,None if data is None else json.dumps(data),
                      {'Cookie':'vision='+token,'Content-Type':'application/json'})
            r=c.getresponse();status=r.status;r.read();c.close();return status
        for path in ['/api/users','/api/audit','/media/live/cam_'+'a'*16+'_hd/index.m3u8',
                     '/api/recordings?camera='+'a'*16]:
            self.assertEqual(request(path),403)
        for path in ['/api/storage','/api/users','/api/family/person']:
            self.assertEqual(request(path,{}),403)
        self.assertEqual(request('/api/family'),200)
        app.ACCOUNTS.save(dict(member,disabled=True),self.admin)
        self.assertEqual(request('/api/family'),401)

    def test_account_constraints(self):
        with self.assertRaises(app.Problem):
            app.ACCOUNTS.save(dict(self.admin,disabled=True),self.admin)
        with self.assertRaises(app.Problem):
            app.ACCOUNTS.save({'username':'admin','password':'password8'},self.admin)
        with self.assertRaises(app.Problem):
            app.ACCOUNTS.save({'username':'short','password':'123'},self.admin)

    def test_family_graph_validation_and_revision(self):
        def save(kind,data):
            return app.FAMILY.save(kind,dict(data,revision=app.setting('familyRevision',0)),self.admin)['item']
        p=save('person',{'name':'祖辈','biography':'第一行\n第二行'})
        q=save('person',{'name':'子辈'})
        r=save('person',{'name':'孙辈'})
        first=save('link',{'fromId':p['id'],'toId':q['id'],'kind':'parent'})
        save('link',{'fromId':q['id'],'toId':r['id'],'kind':'parent'})
        for value in [{'fromId':r['id'],'toId':p['id'],'kind':'parent'},
                      {'fromId':p['id'],'toId':p['id'],'kind':'parent'},
                      {'fromId':p['id'],'toId':r['id'],'kind':'spouse'},
                      {'fromId':p['id'],'toId':q['id'],'kind':'parent'}]:
            with self.assertRaises(app.Problem):save('link',value)
        with self.assertRaises(app.Problem):
            app.FAMILY.save('person',{'name':'过期编辑','revision':0},self.admin)
        with self.assertRaises(app.Problem):
            save('person',{'name':'错误日期','birthDate':'2026-02-30'})
        with self.assertRaises(app.Problem):
            save('person',{'name':'错误日期','birthDate':'2020-01-01','deathDate':'1900-01-01','lifeStatus':'deceased'})
        save('link',dict(first,active=False))
        self.assertFalse(app.get_object('family_links',first['id'])['active'])
        save('link',dict(first,active=True))
        save('person',dict(p,archived=True))
        self.assertTrue(app.get_object('people',p['id'])['archived'])
        save('person',dict(p,archived=False))

    def test_legacy_migration_preserves_password_and_sessions(self):
        # Exercise migration in a separate database without touching the HTTP fixture.
        import sqlite3
        from pathlib import Path
        with tempfile.TemporaryDirectory() as folder:
            db=sqlite3.connect(Path(folder)/'migration.db');db.row_factory=sqlite3.Row
            db.executescript('CREATE TABLE settings(key TEXT PRIMARY KEY,value TEXT NOT NULL);CREATE TABLE sessions(hash TEXT PRIMARY KEY,expires REAL NOT NULL);')
            credentials=app.ACCOUNTS.password('old-password')
            db.execute('INSERT INTO settings VALUES (?,?)',('account',json.dumps(credentials)))
            db.execute('INSERT INTO sessions VALUES (?,?)',('legacy',time.time()+60));db.commit()
            with patch.object(app,'DB',db):
                app.ACCOUNTS.initialize();app.ACCOUNTS.initialize()
                self.assertEqual(len(app.objects('users')),1)
                self.assertIsNotNone(app.ACCOUNTS.verify('admin','old-password'))
                self.assertEqual(db.execute('SELECT user_id FROM sessions').fetchone()[0],app.objects('users')[0]['id'])
                self.assertIsNone(app.setting('account'))
            db.close()

    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)

    def test_ptz_requires_both_space_and_control_permission(self):
        camera = dict(id='e'*16, name='Test', siteId='permitted', enabled=True)
        viewer = dict(role='member', siteIds=['permitted'], ptzControl=False)
        with self.assertRaises(app.Problem):app.ACCOUNTS.control(viewer, camera)
        viewer['ptzControl'] = True
        self.assertEqual(app.ACCOUNTS.control(viewer, camera), camera)
        viewer['siteIds'] = []
        with self.assertRaises(app.Problem):app.ACCOUNTS.control(viewer, camera)
        self.assertEqual(self.request('/api/ptz?camera='+'e'*16,False)[0],401)

    def test_family_defaults_alive_and_rejects_death_for_alive(self):
        p=app.FAMILY.person({'name':'Example'},None)
        self.assertEqual(p['lifeStatus'],'alive')
        with self.assertRaises(app.Problem):app.FAMILY.person({'name':'Example','deathDate':'2020-01-01'},None)
        self.assertEqual(app.FAMILY.person({'name':'Example','lifeStatus':'deceased','deathDate':'2020-01-01'},None)['deathDate'],'2020-01-01')

    def test_empty_recording_day_is_not_service_failure(self):
        error = app.urllib.error.HTTPError('http://localhost/list', 404, 'Not Found', {},
                                           io.BytesIO(b'{"error":"no recording segments found"}'))
        with patch('app.urllib.request.urlopen', side_effect=error):
            self.assertEqual(app.media_json('/list?path=example', playback=True), [])

    def test_real_media_service_error_is_preserved(self):
        error = app.urllib.error.HTTPError('http://localhost/list', 500, 'Error', {}, io.BytesIO(b'{}'))
        with patch('app.urllib.request.urlopen', side_effect=error):
            with self.assertRaises(app.Problem):
                app.media_json('/list?path=example', playback=True)


if __name__ == '__main__':
    unittest.main()