diff --git a/perfact/zodbsync/tests/test_sync.py b/perfact/zodbsync/tests/test_sync.py index 5ddd767..d1e9348 100644 --- a/perfact/zodbsync/tests/test_sync.py +++ b/perfact/zodbsync/tests/test_sync.py @@ -2329,3 +2329,27 @@ def test_layer_info_datafs(self): ) self.run('record', '/') assert getattr(self.app.blob, 'zodbsync_layer', None) is None + + def test_fail_when_meta_is_missing(self): + """ + Check that playing back a structure where no layer has a meta file for + a given folder does not work. + """ + root = f'{self.repo.path}/__root__' + os.mkdir(f'{root}/newfolder') + os.mkdir(f'{root}/newobj') + with open(f'{root}/newobj/__source__.py', 'w'): + pass + with pytest.raises(AssertionError): + self.run('playback', '/') + + def test_fail_when_meta_missing_layers(self): + """ + Check that playing back a structure where no layer has a meta file for + a given folder does not work (multi-layer). + """ + with self.addlayer() as layer: + os.mkdir(f'{self.repo.path}/__root__/newfolder') + os.mkdir(f'{layer}/workdir/__root__/newfolder') + with pytest.raises(AssertionError): + self.run('playback', '/') diff --git a/perfact/zodbsync/zodbsync.py b/perfact/zodbsync/zodbsync.py index 907b951..e626f8c 100644 --- a/perfact/zodbsync/zodbsync.py +++ b/perfact/zodbsync/zodbsync.py @@ -79,7 +79,7 @@ def mod_read(obj=None, onerrorstop=False, default_owner=None, if meta_type not in known_types: if onerrorstop: - assert False, "Unsupported type: %s" % meta_type + raise AssertionError(f"Unsupported type: {meta_type}") else: meta['unsupported'] = meta_type return meta @@ -139,7 +139,8 @@ def mod_write(data, parent=None, obj_id=None, override=False, root=None, temp_obj = None # ID exists? Check for type if obj and obj.meta_type != meta_type: - assert override, "Type mismatch for object " + repr(data) + if not override: + raise AssertionError(f"Type mismatch for object {data!r}") contents = obj_contents(obj) if contents: # Rename so we can cut+paste the children @@ -322,10 +323,11 @@ def start_transaction(self, note=''): # Log in as a manager uf = self.app.acl_users user = uf.getUser(self.manager_user).__of__(uf) - assert user is not None, ( - 'User %s is not available in database. Perhaps you need to set' - ' create_manager_user in config.py?' % self.manager_user - ) + if user is None: + raise AssertionError( + f'User {self.manager_user} is not available in database.' + ' Perhaps you need to set create_manager_user in config.py?' + ) self.logger.info('Using user %s' % self.manager_user) AccessControl.SecurityManagement.newSecurityManager(None, user) @@ -424,7 +426,8 @@ def fs_pathinfo(self, path): 'layeridx': None, } path = path.lstrip('/') - children = set() + candidates = set() # subfolders on any layer + children = set() # those with a __meta__ file on a some layer for idx, layer in enumerate(layers): fspath = os.path.join(layer['workdir'], self.site, path) if not os.path.isdir(fspath): @@ -437,8 +440,14 @@ def fs_pathinfo(self, path): for entry in os.listdir(fspath): if entry in children or entry.startswith('__'): continue + candidates.add(entry) if os.path.exists(os.path.join(fspath, entry, '__meta__')): children.add(entry) + missing = candidates - children + if missing: + raise AssertionError( + f"No __meta__ file on any layer: {path}/{children}" + ) result['children'] = sorted(children) return result @@ -632,11 +641,13 @@ def fs_parse(self, fspath, data=None): if data is None: data = self.fs_read(fspath) - assert 'meta' in data, 'Missing meta file: ' + fspath + if 'meta' not in data: + raise AssertionError(f"Missing meta file: {fspath}") src_fnames = data.get('src_fnames', []) - assert len(src_fnames) <= 1, ( - "Multiple source files in " + fspath - ) + if len(src_fnames) > 1: + raise AssertionError( + f"Multiple source files in {fspath}" + ) result = dict(literal_eval(data['meta'])) if src_fnames: src_fname = src_fnames[0]