admin/ucvl-home-vision
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 43 Wiki Activity

Add verified camera settings, family version recovery and account self service

Browse Source
This commit is contained in:
Codex
2026-10-04 04:52:39 +08:00
parent 56a24c80e2
commit 96f0ef31fc
18 changed files with 969 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_app.py
+205 -3
View File
@@ -39,9 +39,9 @@ class WebTests(unittest.TestCase):
app.DB.close()
_data.cleanup()
def request(self, path, authenticated=True, data=None, origin=None):
def request(self, path, authenticated=True, data=None, origin=None, token=None):
c = http.client.HTTPConnection(*self.server.server_address, timeout=3)
headers = {'Cookie': 'vision=' + 'f' * 64} if authenticated else {}
headers = {'Cookie': 'vision=' + (token or 'f' * 64)} if authenticated else {}
if origin:
headers['Origin'] = origin
if data is not None:
@@ -78,7 +78,7 @@ class WebTests(unittest.TestCase):
self.assertEqual(headers['Cache-Control'], 'no-store')
self.assertNotIn(b'__ASSET_VERSION__', body)
assets = re.findall(r'(?:src|href)="(/[^"?]+\.(?:js|css)\?v=[^"]+)"', body.decode())
self.assertEqual(len(assets), 13)
self.assertEqual(len(assets), 15)
version = (app.ROOT / 'VERSION').read_text().strip()
for asset in assets:
self.assertTrue(asset.endswith('?v=' + version))
@@ -345,6 +345,208 @@ class WebTests(unittest.TestCase):
with self.assertRaises(app.Problem):
app.media_json('/list?path=example', playback=True)
def family_save(self, kind, data):
return app.FAMILY.save(kind, dict(data, revision=app.setting('familyRevision', 0)), self.admin)
def test_history_records_before_after_and_restore_as_new_version(self):
first = self.family_save('person', {'name': 'History original', 'biography': 'Original story'})
second = self.family_save('person', dict(first['item'], name='History changed'))
change = app.FAMILY.change(self.admin, second['revision'])
self.assertEqual(change['before']['name'], 'History original')
self.assertEqual(change['after']['name'], 'History changed')
restored = app.FAMILY.restore({'changeRevision': second['revision'], 'side': 'before',
'revision': app.setting('familyRevision')}, self.admin)
self.assertEqual(restored['item']['name'], 'History original')
self.assertGreater(restored['revision'], second['revision'])
entries = app.FAMILY.history(self.admin, 'person', first['item']['id'])['items']
self.assertEqual(len(entries), 3)
self.assertEqual(entries[0]['restoredFrom'], second['revision'])
self.assertEqual(app.FAMILY.change(self.admin, second['revision']), change)
def test_history_rejects_stale_restore_without_writing(self):
first = self.family_save('person', {'name': 'Stale restore'})
second = self.family_save('person', dict(first['item'], name='Latest'))
with self.assertRaises(app.Problem) as caught:
app.FAMILY.restore(dict(changeRevision=first['revision'], side='after', revision=first['revision']), self.admin)
self.assertEqual(caught.exception.status, 409)
self.assertEqual(app.get_object('people', first['item']['id'])['name'], 'Latest')
self.assertEqual(app.setting('familyRevision'), second['revision'])
def test_restore_revalidates_current_birth_order(self):
old = self.family_save('person', {'name': 'Restore parent', 'birthDate': '1980-01-01'})
parent = self.family_save('person', dict(old['item'], birthDate='1960-01-01'))['item']
child = self.family_save('person', {'name': 'Restore child', 'birthDate': '1970-01-01'})['item']
self.family_save('link', dict(fromId=parent['id'], toId=child['id'], kind='parent', lineage='biological'))
with self.assertRaises(app.Problem):
app.FAMILY.restore(dict(changeRevision=old['revision'], side='after', revision=app.setting('familyRevision')), self.admin)
self.assertEqual(app.get_object('people', parent['id'])['birthDate'], '1960-01-01')
def test_restore_link_cannot_create_new_cycle(self):
a = self.family_save('person', {'name': 'Cycle A'})['item']
b = self.family_save('person', {'name': 'Cycle B'})['item']
old = self.family_save('link', dict(fromId=a['id'], toId=b['id'], kind='parent'))
self.family_save('link', dict(old['item'], active=False))
self.family_save('link', dict(fromId=b['id'], toId=a['id'], kind='parent'))
with self.assertRaises(app.Problem):
app.FAMILY.restore(dict(changeRevision=old['revision'], side='after', revision=app.setting('familyRevision')), self.admin)
self.assertFalse(app.get_object('family_links', old['item']['id'])['active'])
def test_restore_event_preserves_first_publication_and_validates_scope(self):
draft = self.family_save('event', dict(title='Restored story', eventDate='2020-01-01', body='Draft', status='draft'))
published = self.family_save('event', dict(draft['item'], body='Published', status='published'))
restored = app.FAMILY.restore(dict(changeRevision=draft['revision'], side='after', revision=app.setting('familyRevision')), self.admin)
self.assertEqual(restored['item']['status'], 'draft')
self.assertEqual(restored['item']['publishedAt'], published['item']['publishedAt'])
self.assertNotIn(draft['item']['id'], [e['id'] for e in app.FAMILY.snapshot(dict(role='member', familyAccess='read'))['events']])
for side in ['before', 'invalid']:
with self.assertRaises(app.Problem):
app.FAMILY.restore(dict(changeRevision=draft['revision'], side=side, revision=app.setting('familyRevision')), self.admin)
def test_history_pagination_is_complete_and_omits_story_bodies(self):
item = self.family_save('person', {'name': 'Pagination', 'biography': 'Private full story'})['item']
for i in range(23):item = self.family_save('person', dict(item, note=str(i)))['item']
first = app.FAMILY.history(self.admin, 'person', item['id'])
second = app.FAMILY.history(self.admin, 'person', item['id'], first['nextBefore'])
revisions = [v['revision'] for v in first['items'] + second['items']]
self.assertEqual(len(revisions), 24)
self.assertEqual(len(set(revisions)), 24)
self.assertIsNone(second['nextBefore'])
self.assertNotIn('Private full story', json.dumps(first))
def test_history_and_restore_endpoints_enforce_edit_permission(self):
item = self.family_save('person', {'name': 'Permission history'})
reader = app.ACCOUNTS.save(dict(username='history-reader', password='test-reader', familyAccess='read'), self.admin)
token = 'd' * 64
with app.LOCK:
app.DB.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(), time.time()+3600, reader['id']))
app.DB.commit()
path = '/api/family/history?kind=person&id=' + item['item']['id']
for target in [path, '/api/family/change?revision=' + str(item['revision'])]:
self.assertEqual(self.request(target, token=token)[0], 403)
self.assertEqual(self.request(target, authenticated=False)[0], 401)
self.assertEqual(self.request(target)[0], 200)
self.assertEqual(self.request('/api/family/restore', token=token, data={})[0], 403)
for number in ['bad', '-1', '0', '99999999999999999999999']:
self.assertEqual(self.request('/api/family/change?revision=' + number)[0], 400)
def test_family_history_transaction_rolls_back_on_storage_failure(self):
before = app.setting('familyRevision', 0)
with app.LOCK:
app.DB.execute("CREATE TRIGGER fail_history BEFORE INSERT ON family_changes BEGIN SELECT RAISE(ABORT, 'test failure'); END")
app.DB.commit()
try:
import sqlite3
with self.assertRaises(sqlite3.IntegrityError):self.family_save('person', {'name': 'Must not persist'})
self.assertEqual(app.setting('familyRevision', 0), before)
self.assertFalse(any(p['name']=='Must not persist' for p in app.objects('people')))
finally:
with app.LOCK:app.DB.execute('DROP TRIGGER fail_history');app.DB.commit()
def test_concurrent_family_saves_accept_only_one_revision(self):
from concurrent.futures import ThreadPoolExecutor
revision = app.setting('familyRevision', 0); barrier = threading.Barrier(2)
def save(index):
barrier.wait()
try:app.FAMILY.save('person', dict(name='Concurrent '+str(index), revision=revision), self.admin);return 200
except app.Problem as e:return e.status
with ThreadPoolExecutor(max_workers=2) as pool:self.assertEqual(sorted(pool.map(save, range(2))), [200, 409])
self.assertEqual(app.setting('familyRevision'), revision+1)
def test_password_change_revokes_all_own_sessions_and_keeps_other_accounts(self):
member = app.ACCOUNTS.save(dict(username='password-member', password='old-password', familyAccess='read'), self.admin)
tokens = ['1'*64, '2'*64]
with app.LOCK:
for token in tokens:app.DB.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(), time.time()+3600, member['id']))
app.DB.commit()
code, headers, _ = self.request('/api/account/password', token=tokens[0], data=dict(currentPassword='old-password', newPassword='new-password', id=self.admin['id'], role='admin'))
self.assertEqual(code, 200);self.assertIn('Max-Age=0', headers['Set-Cookie'])
self.assertIsNone(app.ACCOUNTS.verify(member['username'], 'old-password'))
self.assertEqual(app.ACCOUNTS.verify(member['username'], 'new-password')['role'], 'member')
for token in tokens:self.assertEqual(self.request('/api/family', token=token)[0], 401)
self.assertEqual(self.request('/api/users')[0], 200)
def test_password_validation_and_rate_limit(self):
actor = app.ACCOUNTS.save(dict(username='password-validation', password='old-password'), self.admin)
for data in [dict(currentPassword='old-password', newPassword='tiny'),dict(currentPassword='old-password', newPassword='old-password')]:
with self.assertRaises(app.Problem):app.ACCOUNTS.change_password(data, actor)
for _ in range(8):
with self.assertRaises(app.Problem) as caught:app.ACCOUNTS.change_password(dict(currentPassword='wrong', newPassword='new-password'), actor)
self.assertEqual(caught.exception.status, 403)
with self.assertRaises(app.Problem) as caught:app.ACCOUNTS.change_password(dict(currentPassword='old-password', newPassword='new-password'), actor)
self.assertEqual(caught.exception.status, 429)
self.assertIsNotNone(app.ACCOUNTS.verify(actor['username'], 'old-password'))
self.assertEqual(self.request('/api/account/password', authenticated=False, data={})[0], 401)
self.assertEqual(self.request('/api/account/password', data={}, origin='http://[bad')[0], 403)
def test_object_ids_and_hosts_reject_malformed_input_without_500(self):
for endpoint in ['sites', 'assets', 'cameras', 'recorders']:
for value in [[], {}, 17, True]:
self.assertEqual(self.request('/api/'+endpoint, data={'id':value})[0], 400)
self.assertEqual(self.request('/api/'+endpoint, data={'id':'9'*16})[0], 404)
for value in [None, 123, [], {}, True]:
with self.assertRaises(app.Problem):app.host_address(value)
def test_concurrent_camera_capacity_and_asset_space_move(self):
from concurrent.futures import ThreadPoolExecutor
site = app.save_site({'name':'Capacity space'})
other = app.save_site({'name':'Moved space'})
asset = app.save_asset(dict(name='Single lens', siteId=site['id'], lensCount=1))
barrier = threading.Barrier(2)
def save(index):
barrier.wait()
try:app.save_camera(dict(name='Concurrent lens '+str(index), assetId=asset['id'], host='192.168.0.2'));return 200
except app.Problem as e:return e.status
with patch.object(app, 'write_media_config'):
with ThreadPoolExecutor(max_workers=2) as pool:self.assertEqual(sorted(pool.map(save, range(2))), [200, 400])
attached = [c for c in app.objects('cameras') if c.get('assetId')==asset['id']]
self.assertEqual(len(attached), 1)
app.save_asset(dict(asset, siteId=other['id']))
self.assertEqual(app.get_object('cameras', attached[0]['id'])['siteId'], other['id'])
def test_recorder_updates_are_reflected_in_channel_endpoint_without_password(self):
site = app.save_site({'name':'Recorder test'})
asset = app.save_asset(dict(name='Recorder lens', siteId=site['id'], lensCount=1))
with patch.object(app, 'write_media_config'):
r = app.save_recorder(dict(name='NVR test', siteId=site['id'], host='192.168.0.8', username='test', password='private'))
camera = app.save_camera(dict(name='Channel', assetId=asset['id'], sourceKind='recorder', recorderId=r['id']))
app.save_recorder(dict(r, host='192.168.0.9', username='updated'))
current = app.public_camera(app.get_object('cameras', camera['id']))
self.assertEqual(current['host'], '192.168.0.9')
self.assertEqual(current['username'], 'updated')
self.assertTrue(current['passwordConfigured']);self.assertNotIn('password', current)
def test_camera_settings_admin_only_and_errors_are_explicit(self):
user = app.ACCOUNTS.save(dict(username='settings-member', password='settings-pass', siteIds=['test-site'] if app.get_object('sites','test-site') else []), self.admin)
token = '3'*64
with app.LOCK:
app.DB.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(),time.time()+3600,user['id']));app.DB.commit()
path = '/api/camera/settings?camera='+'a'*16
self.assertEqual(self.request(path,token=token)[0],403)
self.assertEqual(self.request('/api/camera/settings',data={},token=token)[0],403)
self.assertEqual(self.request(path,authenticated=False)[0],401)
with patch.object(app.camera_settings.CONTROLLER,'read',return_value=({'device':{'Model':'Example'},'revision':'one'},{})):
self.assertEqual(self.request(path)[0],200)
with patch.object(app.camera_settings.CONTROLLER,'save',side_effect=ValueError('unsupported')):
self.assertEqual(self.request('/api/camera/settings',data={'camera':'a'*16})[0],400)
with patch.object(app.camera_settings.CONTROLLER,'save',side_effect=OSError()):
self.assertEqual(self.request('/api/camera/settings',data={'camera':'a'*16})[0],502)
def test_asset_move_rolls_back_all_channels_on_failure(self):
import sqlite3
site = app.save_site({'name':'Original site atomic'})
other = app.save_site({'name':'New site atomic'})
asset = app.save_asset(dict(name='Atomic move',siteId=site['id'],lensCount=1))
with patch.object(app,'write_media_config'):
camera = app.save_camera(dict(name='Atomic lens',assetId=asset['id'],host='192.168.0.2'))
with app.LOCK:
app.DB.execute("CREATE TRIGGER fail_camera_move BEFORE UPDATE ON cameras BEGIN SELECT RAISE(ABORT,'test'); END");app.DB.commit()
try:
with self.assertRaises(sqlite3.IntegrityError):app.save_asset(dict(asset,siteId=other['id']))
self.assertEqual(app.get_object('assets',asset['id'])['siteId'],site['id'])
self.assertEqual(app.get_object('cameras',camera['id'])['siteId'],site['id'])
finally:
with app.LOCK:app.DB.execute('DROP TRIGGER fail_camera_move');app.DB.commit()
if __name__ == '__main__':
unittest.main()