feat: add scoped device catalog and unlimited custom area hierarchy

test_households.py

@@ -55,7 +55,7 @@ class HouseholdTests(unittest.TestCase):
             db.execute('INSERT INTO sessions VALUES (?,?,?)', (hashlib.sha256(token.encode()).hexdigest(), time.time()+600, user['id']))
         db.commit()
         def state():
-            return {**{t: app.objects(t) for t in ('sites','assets','recorders','cameras')}, 'scan': {'network':'private'}, 'storage': {'maxGB':40}}
+            return {**{t: app.objects(t) for t in ('sites','assets','recorders','cameras','devices')}, 'scan': {'network':'private'}, 'storage': {'maxGB':40}}
         p = patch.object(app, 'status', state); p.start(); self.addCleanup(p.stop)
         self.server = app.ThreadingHTTPServer(('127.0.0.1', 0), app.Handler)
         self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
@@ -207,5 +207,60 @@ class HouseholdTests(unittest.TestCase):
         self.assertEqual(saved,user)
         self.assertEqual([tuple(r) for r in before],[tuple(r) for r in app.DB.execute('SELECT * FROM sessions ORDER BY hash')])
 
+    def test_office_factory_and_custom_grouping_are_independent(self):
+        for kind in ('office','factory'):
+            code, scope=self.request('/api/households',self.root,dict(name=kind,kind=kind))
+            self.assertEqual(code,200);self.assertEqual(scope['kind'],kind)
+        self.assertEqual(self.request('/api/device-view',data={'groups':['manager','type','space']})[0],200)
+        self.assertEqual(app.get_object('households',self.aid)['deviceGrouping'],['manager','type','space'])
+        self.assertNotIn('deviceGrouping',app.get_object('households',self.bid))
+        self.assertEqual(self.request('/api/device-view',data={'groups':[]})[0],200)
+        for groups in (['type','type'],['unknown'],'type',[{}]):
+            self.assertEqual(self.request('/api/device-view',data={'groups':groups})[0],400)
+        self.assertEqual(self.request('/api/device-view',self.members[self.aid],{'groups':['type']})[0],403)
+
+    def test_region_hierarchy_has_no_fixed_depth_and_rejects_cycles(self):
+        parent=self.sites[self.aid]
+        for n in range(160):
+            parent=app.save_site(dict(name='Region '+str(n),parentId=parent['id'],category='custom '+str(n)),self.owner)
+        self.assertEqual(len(app.HOUSEHOLDS.objects(self.owner,'sites')),161)
+        for child in (parent,self.sites[self.aid]):
+            self.assertEqual(self.request('/api/sites',data=dict(child,parentId=parent['id']))[0],400)
+        self.assertEqual(self.request('/api/sites',data=dict(parent,parentId=self.sites[self.bid]['id']))[0],404)
+
+    def test_subregion_grants_are_explicit_and_checked_per_request(self):
+        child=app.save_site(dict(name='Bedroom',parentId=self.sites[self.aid]['id']),self.owner)
+        asset=app.save_asset(dict(self.assets[self.aid],siteId=child['id']),self.owner)
+        camera=app.get_object('cameras',self.cameras[self.aid]['id'])
+        member=app.get_object('users',self.members[self.aid]['id'])
+        with self.assertRaises(app.Problem):app.ACCOUNTS.camera(member,camera)
+        app.save_object('users',dict(member,includeSubspaces=True))
+        self.assertEqual(len(self.request('/api/state',member)[1]['cameras']),1)
+        # Moving the region to another branch immediately removes inherited access.
+        app.save_site(dict(child,parentId=''),self.owner)
+        self.assertEqual(len(self.request('/api/state',member)[1]['cameras']),0)
+        self.assertEqual(app.get_object('assets',asset['id'])['siteId'],child['id'])
+
+    def test_generic_devices_responsibilities_and_cross_family_access(self):
+        data=dict(name='卧室台灯',deviceType='灯',siteId=self.sites[self.aid]['id'],ownerName='家人甲',managerName='家人乙',tags=['照明','常用'])
+        code,device=self.request('/api/devices',data=data)
+        self.assertEqual(code,200);self.assertEqual(device['managerName'],'家人乙')
+        self.assertEqual([d['id'] for d in self.request('/api/state')[1]['devices']],[device['id']])
+        self.assertEqual(self.request('/api/state',self.other)[1]['devices'],[])
+        self.assertIn(self.request('/api/devices',self.other,dict(device,name='hijack'))[0],(403,404))
+        self.assertEqual(self.request('/api/devices',data=dict(data,siteId=self.sites[self.bid]['id']))[0],404)
+        self.assertEqual(self.request('/api/devices',self.members[self.aid],data)[0],403)
+        for invalid in (dict(data,tags='tag'),dict(data,deviceType='录像机'),dict(data,tags=['x']*13)):
+            self.assertEqual(self.request('/api/devices',data=invalid)[0],400)
+
+    def test_video_and_other_devices_share_responsibility_metadata(self):
+        body=dict(self.assets[self.aid],ownerName='设备所有人',managerName='设备负责人',tags=['视频'])
+        asset=app.save_asset(body,self.owner)
+        recorder=app.save_recorder(dict(name='Recorder',siteId=self.sites[self.aid]['id'],host='192.168.3.90',ownerName='设备所有人',managerName='设备负责人',tags=['视频']),self.owner)
+        self.assertEqual(asset['managerName'],recorder['managerName'])
+        self.assertEqual(asset['tags'],recorder['tags'])
+        body.pop('ownerName');body.pop('managerName');body.pop('tags')
+        self.assertEqual(app.save_asset(body,self.owner)['managerName'],'设备负责人')
+
 
 if __name__=='__main__': unittest.main()