Add personal accounts, scoped access and editable family genealogy

test_app.py

@@ -22,13 +22,14 @@ 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() + 60))
+        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', 'password': 'private'})
+        app.save_object('cameras', {'id': 'a' * 16, 'name': 'Example', 'siteId': 'test-site', 'password': 'private'})
 
     @classmethod
     def tearDownClass(cls):
@@ -113,6 +114,84 @@ class WebTests(unittest.TestCase):
     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()