Add hierarchical geography and cloud-hosted profile images

test_cloud_images.py

@@ -0,0 +1,76 @@
+import io
+import hashlib
+import unittest
+from unittest.mock import patch
+from test_app import app
+import test_households as fixture
+from cloud_images import CloudImages
+
+
+def install_cloud_fixture(test, app):
+    """Exercise the real image store over an isolated in-memory Baidu transport."""
+    files={};blocks={}
+    def prepare(remote,size,hashes):
+        if remote in files:return {'existing':dict(fs_id=files[remote][0],path=remote,size=len(files[remote][1]))}
+        return {'uploadid':'mock-upload','block_list':[0]}
+    def block(remote,upload,index,raw,md5):
+        assert hashlib.md5(raw).hexdigest()==md5;blocks[remote]=raw
+    def complete(item):
+        remote=item['remotePath'];files[remote]=(len(files)+1,blocks[remote])
+        return dict(fs_id=files[remote][0],path=remote,size=len(files[remote][1]))
+    def open_file(key,item,byte_range=''):
+        return io.BytesIO(files[item['remotePath']][1])
+    for obj,name,value in [(app,'CLOUD_IMAGES',CloudImages(app)),(app.BAIDU,'token',lambda *a:'fixture'),
+         (app.BAIDU,'public_config',lambda:dict(appFolder='测试应用')),(app.BAIDU,'prepare',prepare),
+         (app.BAIDU,'send_block',block),(app.BAIDU,'complete',complete),(app.BAIDU,'open_file',open_file)]:
+        p=patch.object(obj,name,value);p.start();test.addCleanup(p.stop)
+    return files
+
+
+class CloudImageTests(unittest.TestCase):
+    stop_server=fixture.HouseholdTests.stop_server
+    request=fixture.HouseholdTests.request
+    def setUp(self):
+        fixture.HouseholdTests.setUp(self)
+        self.files=install_cloud_fixture(self,app)
+
+    def test_cover_is_uploaded_checked_and_db_contains_only_reference(self):
+        from test_appearance import image_data
+        result=app.HOUSEHOLDS.save(dict(id=self.aid,name='云封面',coverData=image_data()),self.root)
+        self.assertNotIn('coverData',result)
+        self.assertIn('/家庭与场所/',result['coverRef']['remotePath'])
+        ref=result['coverRef'];app.CLOUD_IMAGES.cache.clear()
+        self.assertEqual(hashlib.sha256(app.CLOUD_IMAGES.get(ref)).hexdigest(),ref['sha256'])
+        self.assertEqual(len(self.files),1)
+        app.HOUSEHOLDS.save(dict(id=self.aid,name='重复上传',profileRevision=1,coverData=image_data()),self.root)
+        self.assertEqual(len(self.files),1)
+
+    def test_failed_upload_does_not_save_profile_or_replace_existing_cover(self):
+        from test_appearance import image_data
+        before=app.get_object('households',self.aid)
+        with patch.object(app.BAIDU,'prepare',side_effect=app.Problem('网盘断线',502)):
+            with self.assertRaises(app.Problem):app.HOUSEHOLDS.save(dict(id=self.aid,name='不应保存',coverData=image_data()),self.root)
+        self.assertEqual(app.get_object('households',self.aid),before)
+
+    def test_conflict_during_upload_preserves_newer_profile(self):
+        from test_appearance import image_data
+        put=app.CLOUD_IMAGES.put
+        def concurrent(*args):
+            ref=put(*args);home=app.get_object('households',self.aid)
+            home.update(name='他人刚修改',profileRevision=1);app.save_object('households',home);return ref
+        with patch.object(app.CLOUD_IMAGES,'put',side_effect=concurrent):
+            with self.assertRaises(app.Problem):app.HOUSEHOLDS.save(dict(id=self.aid,name='旧编辑',coverData=image_data()),self.root)
+        self.assertEqual(app.get_object('households',self.aid)['name'],'他人刚修改')
+
+    def test_legacy_migration_is_verified_and_idempotent(self):
+        from test_appearance import image_data
+        home=app.get_object('households',self.aid);home['coverData']=image_data();app.save_object('households',home)
+        self.assertEqual(len(app.CLOUD_IMAGES.migrate_legacy()),1)
+        self.assertNotIn('coverData',app.get_object('households',self.aid))
+        self.assertEqual(app.CLOUD_IMAGES.migrate_legacy(),[])
+
+    def test_corrupt_remote_fails_before_commit_and_no_disk_cache(self):
+        from test_appearance import image_data
+        with patch.object(app.BAIDU,'open_file',return_value=io.BytesIO(b'corrupt')):
+            with self.assertRaises(app.Problem):app.CLOUD_IMAGES.put(image_data(),self.aid,'封面')
+        self.assertEqual(list(app.CLOUD_IMAGES.cache),[])