Commits

Thomas Waldmann  committed cdb9c30

backend: .store() now checks size and hash declared in meta (if present)

  • Participants
  • Parent commits bea1424

Comments (0)

Files changed (2)

File backend/_tests/__init__.py

 
 import pytest
 
+from config import SIZE, HASH_ALGORITHM
+
 class BackendTestBase(object):
     def setup_method(self, method):
         """
         with pytest.raises(KeyError):
             self.be.retrieve(metaid)
 
+    def test_store_check_size(self):
+        # no size
+        meta = dict(name='foo')
+        data = 'barbaz'
+        metaid = self.be.store(meta, StringIO(data))
+        m, d = self.be.retrieve(metaid)
+        assert meta[SIZE] == 6
+        # correct size
+        meta = dict(name='foo', size=6)
+        data = 'barbaz'
+        metaid = self.be.store(meta, StringIO(data))
+        m, d = self.be.retrieve(metaid)
+        assert meta[SIZE] == 6
+        # wrong size (less data than size declared in meta)
+        meta = dict(name='foo', size=42)
+        data = 'barbaz'
+        with pytest.raises(ValueError):
+            metaid = self.be.store(meta, StringIO(data))
+        # wrong size (more data than size declared in meta)
+        meta = dict(name='foo', size=3)
+        data = 'barbaz'
+        with pytest.raises(ValueError):
+            metaid = self.be.store(meta, StringIO(data))
+
+    def test_store_check_hash(self):
+        # no hash
+        meta = dict(name='foo')
+        data = 'barbaz'
+        metaid = self.be.store(meta, StringIO(data))
+        m, d = self.be.retrieve(metaid)
+        hashcode = meta[HASH_ALGORITHM]
+        # correct hash
+        meta = dict(name='foo')
+        meta[HASH_ALGORITHM] = hashcode
+        data = 'barbaz'
+        metaid = self.be.store(meta, StringIO(data))
+        m, d = self.be.retrieve(metaid)
+        assert meta[HASH_ALGORITHM] == hashcode
+        # wrong data -> hash mismatch
+        meta = dict(name='foo')
+        meta[HASH_ALGORITHM] = hashcode
+        data = 'brrbrr'
+        with pytest.raises(ValueError):
+            metaid = self.be.store(meta, StringIO(data))
+
     def test_iter(self):
         mds = [#(metadata items, data str)
                 (dict(name='one'), 'ONE'),

File backend/storages.py

             dataid = make_uuid()
             self.data_store[dataid] = tfw
             meta['dataid'] = dataid
-            meta['size'] = tfw.size
-            meta[HASH_ALGORITHM] = tfw.hash.hexdigest()
+            size_expected = meta.get('size')
+            size_real = tfw.size
+            if size_expected is not None and size_expected != size_real:
+                raise ValueError("computed data size (%d) does not match data size declared in metadata (%d)" % (
+                                 size_real, size_expected))
+            meta['size'] = size_real
+            hash_expected = meta.get(HASH_ALGORITHM)
+            hash_real = tfw.hash.hexdigest()
+            if hash_expected is not None and hash_expected != hash_real:
+                raise ValueError("computed data hash (%s) does not match data hash declared in metadata (%s)" % (
+                                 hash_real, hash_expected))
+            meta[HASH_ALGORITHM] = hash_real
         else:
             dataid = meta['dataid']
             # we will just asume stuff is correct if you pass it with a data id