Codex
553 lines
34 KiB
Python
553 lines
34 KiB
Python
"""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, token=None):
|
|
c = http.client.HTTPConnection(*self.server.server_address, timeout=3)
|
|
headers = {'Cookie': 'vision=' + (token or '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_asset_urls_follow_release_version(self):
|
|
code, headers, body = self.request('/', False)
|
|
self.assertEqual(code, 200)
|
|
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), 15)
|
|
version = (app.ROOT / 'VERSION').read_text().strip()
|
|
for asset in assets:
|
|
self.assertTrue(asset.endswith('?v=' + version))
|
|
self.assertEqual(self.request(asset, False)[0], 200)
|
|
|
|
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','/api/family/event']:
|
|
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_family_event_draft_publish_archive_and_restore(self):
|
|
def save(data):
|
|
return app.FAMILY.save('event',dict(data,revision=app.setting('familyRevision',0)),self.admin)['item']
|
|
p=app.FAMILY.save('person',{'name':'Event person','revision':app.setting('familyRevision',0)},self.admin)['item']
|
|
reader={'role':'member','familyAccess':'read','personId':p['id']}
|
|
e=save({'title':'First event','eventDate':'2024-02-10','summary':'Summary','body':'Line one\nLine two','personIds':[p['id'],p['id']]})
|
|
self.assertEqual(e['status'],'draft');self.assertEqual(e['personIds'],[p['id']])
|
|
self.assertNotIn(e['id'],[r['id'] for r in app.FAMILY.snapshot(reader)['events']])
|
|
e=save(dict(e,status='published'))
|
|
self.assertIn('publishedAt',e)
|
|
self.assertIn(e['id'],[r['id'] for r in app.FAMILY.snapshot(reader)['events']])
|
|
e=save(dict(e,archived=True))
|
|
self.assertNotIn(e['id'],[r['id'] for r in app.FAMILY.snapshot(reader)['events']])
|
|
self.assertIn(e['id'],[r['id'] for r in app.FAMILY.snapshot(self.admin)['events']])
|
|
e=save(dict(e,archived=False))
|
|
self.assertIn(e['id'],[r['id'] for r in app.FAMILY.snapshot(reader)['events']])
|
|
with self.assertRaises(app.Problem):app.FAMILY.save('event',dict(e,revision=0),self.admin)
|
|
with self.assertRaises(app.Problem):app.FAMILY.save('event',dict(e,revision=app.setting('familyRevision',0)),reader)
|
|
with self.assertRaises(app.Problem):app.FAMILY.snapshot({'role':'member','familyAccess':'none'})
|
|
|
|
def test_family_event_validation(self):
|
|
good={'title':'Event','eventDate':'2020-02-29','body':'Text','status':'published'}
|
|
for changes in [{'eventDate':'2023-02-29'},{'title':''},{'sourceUrl':'javascript:alert(1)'},
|
|
{'sourceUrl':'https://user:secret@example.com'},{'personIds':['missing']},
|
|
{'body':'x'*12001},{'status':'public'},{'archived':'false'},{'personIds':'wrong'},
|
|
{'body':'','summary':'','sourceUrl':''}]:
|
|
with self.subTest(changes=list(changes)):
|
|
with self.assertRaises(app.Problem):app.FAMILY.event(dict(good,**changes),None)
|
|
event=app.FAMILY.event(dict(good,sourceUrl='https://example.org/report'),None)
|
|
self.assertEqual(event['sourceUrl'],'https://example.org/report')
|
|
|
|
def test_event_accepts_full_chinese_body_and_multiline_summary(self):
|
|
payload={'title':'中文长文','eventDate':'2026-10-03','body':'家'*12000,'summary':'第一段\n第二段',
|
|
'status':'published','revision':app.setting('familyRevision',0)}
|
|
self.assertGreater(len(json.dumps(payload).encode()),16384)
|
|
code,_,body=self.request('/api/family/event',data=payload)
|
|
self.assertEqual(code,200)
|
|
saved=json.loads(body)['item']
|
|
self.assertEqual(saved['body'],payload['body'])
|
|
self.assertEqual(saved['summary'],payload['summary'])
|
|
for changes in [{'body':'家'*12001},{'summary':'家'*401},{'summary':'bad\x00text'}]:
|
|
invalid=dict(payload,**changes,revision=app.setting('familyRevision',0))
|
|
self.assertEqual(self.request('/api/family/event',data=invalid)[0],400)
|
|
|
|
def test_event_keeps_publication_history_and_omitted_status(self):
|
|
def save(data):return app.FAMILY.save('event',dict(data,revision=app.setting('familyRevision',0)),self.admin)['item']
|
|
event=save({'title':'记事','eventDate':'2026-10-03','body':'正文','status':'published'})
|
|
published=event['publishedAt'];created=event['createdAt']
|
|
edited=dict(event,title='更新标题');edited.pop('status');edited.pop('publishedAt')
|
|
event=save(edited)
|
|
self.assertEqual(event['status'],'published');self.assertEqual(event['publishedAt'],published)
|
|
event=save(dict(event,status='draft'))
|
|
self.assertEqual(event['publishedAt'],published)
|
|
reader={'role':'member','familyAccess':'read'}
|
|
self.assertNotIn(event['id'],[e['id'] for e in app.FAMILY.snapshot(reader)['events']])
|
|
event=save(dict(event,status='published'))
|
|
self.assertEqual(event['publishedAt'],published);self.assertEqual(event['createdAt'],created)
|
|
|
|
def test_biological_birth_order_checked_on_link_and_person_changes(self):
|
|
def save(kind,data):return app.FAMILY.save(kind,dict(data,revision=app.setting('familyRevision',0)),self.admin)['item']
|
|
parent=save('person',{'name':'Parent','birthDate':'1980-01-01'})
|
|
child=save('person',{'name':'Child','birthDate':'2000-01-01'})
|
|
relation=save('link',{'fromId':parent['id'],'toId':child['id'],'kind':'parent','lineage':'biological'})
|
|
for person,date in [(parent,'2000-01-01'),(parent,'2001-01-01'),(child,'1979-01-01')]:
|
|
revision=app.setting('familyRevision',0)
|
|
with self.assertRaises(app.Problem):save('person',dict(person,birthDate=date))
|
|
self.assertEqual(app.setting('familyRevision',0),revision)
|
|
self.assertEqual(app.get_object('people',person['id'])['birthDate'],person['birthDate'])
|
|
save('link',dict(relation,active=False))
|
|
save('person',dict(parent,birthDate='2001-01-01'))
|
|
with self.assertRaises(app.Problem):save('link',dict(relation,active=True))
|
|
save('link',dict(relation,active=True,lineage='adoptive'))
|
|
|
|
def test_request_body_limit_is_bounded_and_invalid_length_is_400(self):
|
|
from email.message import Message
|
|
from types import SimpleNamespace
|
|
for length in ['invalid','-1','0','131073']:
|
|
headers=Message();headers['Content-Type']='application/json';headers['Content-Length']=length
|
|
fake=SimpleNamespace(headers=headers,rfile=io.BytesIO(b'{}'))
|
|
with self.assertRaises(app.Problem) as caught:app.Handler.body(fake,131072)
|
|
self.assertEqual(caught.exception.status,400)
|
|
|
|
def test_event_endpoint_and_static_modules(self):
|
|
self.assertEqual(self.request('/api/family/event',False,{'title':'Denied'})[0],401)
|
|
payload={'title':'API event','eventDate':'2026-01-01','body':'Test only','status':'published','revision':app.setting('familyRevision',0)}
|
|
self.assertEqual(self.request('/api/family/event',data=payload)[0],200)
|
|
for path in ['/family-graph.js','/family-map.js','/family-map.css','/family-events.js']:
|
|
self.assertEqual(self.request(path,False)[0],200)
|
|
|
|
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)
|
|
|
|
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()
|