diff options
| -rw-r--r-- | gemato/recursiveloader.py | 19 | ||||
| -rw-r--r-- | tests/test_recursiveloader.py | 48 |
2 files changed, 29 insertions, 38 deletions
diff --git a/gemato/recursiveloader.py b/gemato/recursiveloader.py index bbc298f..15d78d7 100644 --- a/gemato/recursiveloader.py +++ b/gemato/recursiveloader.py @@ -181,7 +181,7 @@ class ManifestRecursiveLoader(object): hashes=None, allow_create=False, sort=None, compress_watermark=None, compress_format=None, profile=gemato.profile.DefaultProfile(), - max_jobs=None): + max_jobs=None, allow_xdev=True): """ Instantiate the loader for a Manifest tree starting at top-level Manifest @top_manifest_path. @@ -227,6 +227,11 @@ class ManifestRecursiveLoader(object): to optimize some operations. If None (the default), the number will automatically be determined based on CPU count. Otherwise, the specified number will be used. + + If @allow_xdev is true, Manifest can contain files located + across different filesystem. If it is false, gemato will raise + an exception upon crossing filesystem boundaries. It defaults + to true. """ self.root_directory = os.path.dirname(top_manifest_path) @@ -255,10 +260,11 @@ class ManifestRecursiveLoader(object): top_manifest_path) self.loaded_manifests = {} self.updated_manifests = set() + self.manifest_device = None # TODO: allow catching OpenPGP exceptions somehow? m = self.load_manifest(self.top_level_manifest_filename, - allow_create=allow_create, store_dev=True) + allow_create=allow_create, store_dev=not allow_xdev) self.openpgp_signed = m.openpgp_signed self.openpgp_signature = m.openpgp_signature @@ -601,7 +607,8 @@ class ManifestRecursiveLoader(object): for dirpath, dirnames, filenames in it: dir_st = os.stat(dirpath) - if dir_st.st_dev != self.manifest_device: + if (self.manifest_device is not None + and dir_st.st_dev != self.manifest_device): raise gemato.exceptions.ManifestCrossDevice(dirpath) dir_id = (dir_st.st_dev, dir_st.st_ino) @@ -986,7 +993,8 @@ class ManifestRecursiveLoader(object): for dirpath, dirnames, filenames in it: dir_st = os.stat(dirpath) - if dir_st.st_dev != self.manifest_device: + if (self.manifest_device is not None + and dir_st.st_dev != self.manifest_device): raise gemato.exceptions.ManifestCrossDevice(dirpath) dir_id = (dir_st.st_dev, dir_st.st_ino) @@ -1098,7 +1106,8 @@ class ManifestRecursiveLoader(object): for dirpath, dirnames, filenames in it: dir_st = os.stat(dirpath) - if dir_st.st_dev != self.manifest_device: + if (self.manifest_device is not None + and dir_st.st_dev != self.manifest_device): raise gemato.exceptions.ManifestCrossDevice(dirpath) dir_id = (dir_st.st_dev, dir_st.st_ino) diff --git a/tests/test_recursiveloader.py b/tests/test_recursiveloader.py index 19646df..019c29f 100644 --- a/tests/test_recursiveloader.py +++ b/tests/test_recursiveloader.py @@ -1553,41 +1553,34 @@ DATA sub/version 0 MD5 d41d8cd98f00b204e9800998ecf8427e def test_assert_directory_verifies(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.assert_directory_verifies, '') def test_assert_directory_verifies_nonstrict(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.assert_directory_verifies, '', fail_handler=callback_return_true) def test_assert_directory_verifies_subdir(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.assert_directory_verifies, 'sub') - def test_cli_verifies(self): - self.assertEqual( - gemato.cli.main(['gemato', 'verify', self.dir]), - 1) - def test_update_entries_for_directory(self): m = gemato.recursiveloader.ManifestRecursiveLoader( os.path.join(self.dir, 'Manifest'), + allow_xdev=False, hashes=['SHA256', 'SHA512']) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.update_entries_for_directory, '') - def test_cli_update(self): - self.assertEqual( - gemato.cli.main(['gemato', 'update', '--hashes=SHA256 SHA512', - self.dir]), - 1) - class CrossDeviceEmptyManifestTest(TempDirTestCase): """ @@ -1605,35 +1598,27 @@ class CrossDeviceEmptyManifestTest(TempDirTestCase): def test_assert_directory_verifies(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.assert_directory_verifies, '') def test_assert_directory_verifies_nonstrict(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.assert_directory_verifies, '', fail_handler=callback_return_true) - def test_cli_verifies(self): - self.assertEqual( - gemato.cli.main(['gemato', 'verify', self.dir]), - 1) - def test_update_entries_for_directory(self): m = gemato.recursiveloader.ManifestRecursiveLoader( os.path.join(self.dir, 'Manifest'), + allow_xdev=False, hashes=['SHA256', 'SHA512']) self.assertRaises(gemato.exceptions.ManifestCrossDevice, m.update_entries_for_directory, '') - def test_cli_update(self): - self.assertEqual( - gemato.cli.main(['gemato', 'update', '--hashes=SHA256 SHA512', - self.dir]), - 1) - class CrossDeviceIgnoreManifestTest(TempDirTestCase): """ @@ -1653,17 +1638,14 @@ IGNORE sub def test_assert_directory_verifies(self): m = gemato.recursiveloader.ManifestRecursiveLoader( - os.path.join(self.dir, 'Manifest')) + os.path.join(self.dir, 'Manifest'), + allow_xdev=False) m.assert_directory_verifies('') - def test_cli_verifies(self): - self.assertEqual( - gemato.cli.main(['gemato', 'verify', self.dir]), - 0) - def test_update_entries_for_directory(self): m = gemato.recursiveloader.ManifestRecursiveLoader( os.path.join(self.dir, 'Manifest'), + allow_xdev=False, hashes=['SHA256', 'SHA512']) m.update_entries_for_directory('') self.assertEqual(len(m.loaded_manifests['Manifest'].entries), 1) |