77 lines
4.1 KiB
Python
77 lines
4.1 KiB
Python
|
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),[])
|