verify: Support updating entry data from path

3 files changed, 235 insertions, 0 deletions

diff --git a/gemato/exceptions.py b/gemato/exceptions.py
index 950d6a5..7c75f46 100644
--- a/gemato/exceptions.py
+++ b/gemato/exceptions.py
@@ -88,3 +88,15 @@ class OpenPGPNoImplementation(Exception):
     def __init__(self):
         super(OpenPGPNoImplementation, self).__init__(
                 "No supported OpenPGP implementation found (install gnupg)")
+
+
+class ManifestInvalidPath(Exception):
+    """
+    An exception raised when an invalid path tries to be added to
+    Manifest.
+    """
+
+    def __init__(self, path, detail):
+        super(ManifestInvalidPath, self).__init__(
+                "Attempting to add invalid path {} to Manifest: {} must not be {}"
+                .format(path, detail[0], detail[1]))
diff --git a/gemato/verify.py b/gemato/verify.py
index 046c575..04edc91 100644
--- a/gemato/verify.py
+++ b/gemato/verify.py
@@ -198,6 +198,61 @@ def verify_path(path, e, expected_dev=None):
     return (True, [])
 
 
+def update_entry_for_path(path, e, hashes=None, expected_dev=None):
+    """
+    Update the data in entry @e to match the current state of file
+    at path @path. Uses hashes listed in @hashes (using Manifest names),
+    or the current set of hashes in @e if @hashes is None.
+
+    The file must exist and be a regular file, and the entry must be
+    of MISC, DATA, MANIFEST or a derived type. The path/filename
+    is not updated nor checked.
+
+    If @expected_dev is not None, verifies that the file resides
+    on specified device. If the device does not match, raises
+    ManifestCrossDevice exception. It can be used to verify that
+    the files do not cross filesystem boundaries.
+    """
+
+    assert isinstance(e, gemato.manifest.ManifestPathEntry)
+    assert not isinstance(e, gemato.manifest.ManifestEntryIGNORE)
+    assert not isinstance(e, gemato.manifest.ManifestEntryOPTIONAL)
+
+    if hashes is None:
+        hashes = list(e.checksums)
+
+    with contextlib.closing(get_file_metadata(path, hashes)) as g:
+        # 1. verify whether the file existed in the first place
+        exists = next(g)
+        if not exists:
+            raise gemato.exceptions.ManifestInvalidPath(path,
+                    ('__exists__', exists))
+
+        # 2. check for xdev condition
+        st_dev = next(g)
+        if expected_dev is not None and st_dev != expected_dev:
+            raise gemato.exceptions.ManifestCrossDevice(path)
+
+        # 3. verify whether the file is a regular file
+        ifmt, ftype = next(g)
+        if not stat.S_ISREG(ifmt):
+            raise gemato.exceptions.ManifestInvalidPath(path,
+                    ('__type__', ftype))
+
+        # 4. get the apparent file size
+        st_size = next(g)
+
+        # 5. get the checksums and real size
+        checksums = next(g)
+        size = checksums.pop('__size__')
+        if st_size != 0:
+            assert st_size == size, ('Apparent size (st_size = {}) and real size ({}) are different!'
+                    .format(st_size, size))
+
+        e.size = size
+        e.checksums = checksums
+
+
 def verify_entry_compatibility(e1, e2):
     """
     Verify that the two entries @e1 and @e2 are compatible.
diff --git a/tests/test_verify.py b/tests/test_verify.py
index 5393b4c..a9e7a5e 100644
--- a/tests/test_verify.py
+++ b/tests/test_verify.py
@@ -49,6 +49,12 @@ class NonExistingFileVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(os.path.join(self.dir, 'test'), None),
                 (True, []))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA('test', 0, {})
+        self.assertRaises(gemato.exceptions.ManifestInvalidPath,
+                gemato.verify.update_entry_for_path,
+                os.path.join(self.dir, 'test'), e)
+
 
 class DirectoryVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -85,6 +91,12 @@ class DirectoryVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.dir, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.dir), 0, {})
+        self.assertRaises(gemato.exceptions.ManifestInvalidPath,
+                gemato.verify.update_entry_for_path, self.dir, e)
+
 
 class CharacterDeviceVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -118,6 +130,12 @@ class CharacterDeviceVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.path, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        self.assertRaises(gemato.exceptions.ManifestInvalidPath,
+                gemato.verify.update_entry_for_path, self.path, e)
+
 
 class NamedPipeVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -157,6 +175,12 @@ class NamedPipeVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.path, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        self.assertRaises(gemato.exceptions.ManifestInvalidPath,
+                gemato.verify.update_entry_for_path, self.path, e)
+
 
 class UNIXSocketVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -199,6 +223,12 @@ class UNIXSocketVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.path, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        self.assertRaises(gemato.exceptions.ManifestInvalidPath,
+                gemato.verify.update_entry_for_path, self.path, e)
+
 
 class EmptyFileVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -306,6 +336,78 @@ class EmptyFileVerificationTest(unittest.TestCase):
                 gemato.verify.verify_path, self.path, e,
                 expected_dev=st.st_dev)
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 0)
+        self.assertDictEqual(e.checksums, {})
+
+    def test_update_with_hashes(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e, ['MD5', 'SHA1'])
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 0)
+        self.assertDictEqual(e.checksums, {
+                'MD5': 'd41d8cd98f00b204e9800998ecf8427e',
+                'SHA1': 'da39a3ee5e6b4b0d3255bfef95601890afd80709',
+            })
+
+    def test_update_with_hashes_from_manifest(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {'MD5': '', 'SHA1': ''})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 0)
+        self.assertDictEqual(e.checksums, {
+                'MD5': 'd41d8cd98f00b204e9800998ecf8427e',
+                'SHA1': 'da39a3ee5e6b4b0d3255bfef95601890afd80709',
+            })
+
+    def test_update_cross_filesystem(self):
+        try:
+            st = os.stat('/proc')
+        except OSError:
+            raise unittest.SkipTest('Unable to stat /proc')
+
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        self.assertRaises(gemato.exceptions.ManifestCrossDevice,
+                gemato.verify.update_entry_for_path, self.path, e,
+                expected_dev=st.st_dev)
+
+    def test_update_MISC(self):
+        e = gemato.manifest.ManifestEntryMISC(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 0)
+        self.assertDictEqual(e.checksums, {})
+
+    def test_update_OPTIONAL(self):
+        e = gemato.manifest.ManifestEntryOPTIONAL(
+                os.path.basename(self.path))
+        self.assertRaises(AssertionError,
+                gemato.verify.update_entry_for_path, self.path, e)
+
+    def test_update_IGNORE(self):
+        e = gemato.manifest.ManifestEntryIGNORE(
+                os.path.basename(self.path))
+        self.assertRaises(AssertionError,
+                gemato.verify.update_entry_for_path, self.path, e)
+
+    def test_update_AUX(self):
+        e = gemato.manifest.ManifestEntryAUX(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path,
+                os.path.join('files', os.path.basename(self.path)))
+        self.assertEqual(e.size, 0)
+        self.assertDictEqual(e.checksums, {})
+
+
 
 class NonEmptyFileVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -390,6 +492,36 @@ class NonEmptyFileVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.path, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 43)
+        self.assertDictEqual(e.checksums, {})
+
+    def test_update_with_hashes(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e, ['MD5', 'SHA1'])
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 43)
+        self.assertDictEqual(e.checksums, {
+                'MD5': '9e107d9d372bb6826bd81d3542a419d6',
+                'SHA1': '2fd4e1c67a2d28fced849ee1bb76e7391b93eb12',
+            })
+
+    def test_update_with_hashes_from_manifest(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {'MD5': '', 'SHA1': ''})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, 43)
+        self.assertDictEqual(e.checksums, {
+                'MD5': '9e107d9d372bb6826bd81d3542a419d6',
+                'SHA1': '2fd4e1c67a2d28fced849ee1bb76e7391b93eb12',
+            })
+
 
 class SymbolicLinkVerificationTest(NonEmptyFileVerificationTest):
     """
@@ -538,6 +670,36 @@ class ProcFileVerificationTest(unittest.TestCase):
         self.assertEqual(gemato.verify.verify_path(self.path, None),
                 (False, [('__exists__', False, True)]))
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, self.size)
+        self.assertDictEqual(e.checksums, {})
+
+    def test_update_with_hashes(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        gemato.verify.update_entry_for_path(self.path, e, ['MD5', 'SHA1'])
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, self.size)
+        self.assertDictEqual(e.checksums, {
+                'MD5': self.md5,
+                'SHA1': self.sha1,
+            })
+
+    def test_update_with_hashes_from_manifest(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {'MD5': '', 'SHA1': ''})
+        gemato.verify.update_entry_for_path(self.path, e)
+        self.assertEqual(e.path, os.path.basename(self.path))
+        self.assertEqual(e.size, self.size)
+        self.assertDictEqual(e.checksums, {
+                'MD5': self.md5,
+                'SHA1': self.sha1,
+            })
+
 
 class UnreadableFileVerificationTest(unittest.TestCase):
     def setUp(self):
@@ -562,6 +724,12 @@ class UnreadableFileVerificationTest(unittest.TestCase):
         self.assertRaises(OSError, gemato.verify.verify_path,
                 os.path.join(self.dir, e.path), e)
 
+    def test_update(self):
+        e = gemato.manifest.ManifestEntryDATA(
+                os.path.basename(self.path), 0, {})
+        self.assertRaises(OSError,
+                gemato.verify.update_entry_for_path, self.path, e)
+
 
 class EntryCompatibilityVerificationTest(unittest.TestCase):
     def test_matching_entry(self):