Merge pull request #122 from perfact/playback-hook

Implementation for hook to define phases of playback during git commands

.github/workflows/check.yml

@@ -13,9 +13,9 @@ jobs:
         python-version: ['3.8', '3.9', '3.10']
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies

CHANGELOG

@@ -1,6 +1,7 @@
 unreleased
   * Update README
   * Explicitly name git branch in tests
+  * Add configuration option playback_hook
 
 22.2.5
   * Omit title attributes if they are callable.

README.md

@@ -84,6 +84,23 @@ Path to a script that is executed after a successful (non-recursive) playback,
 including indirect calls from `reset` or `pick`. If the script exists, it is
 called and fed the list of changed objects in a JSON format.
 
+### `playback_hook`
+Path to script which is called to define the phases of playback to be
+executed, Recieves a json dictionary in the form of `{"paths": [...]}`
+and should output a json dictionary in the form of
+
+```json
+[
+  {
+    "paths": [...],
+    "cmd": [...]
+  },
+  {
+    "paths": [...],
+  }
+]
+```
+
 ## Usage
 
 The executable `zodbsync` provides several subcommands

config.py

@@ -31,5 +31,9 @@
 #codechange_mail = "zope-devel@example.de"
 #codechange_sender = "no-reply-zodbsync-changes@example.de"
 
+# Path to script which is called to define the phases of playback to be
+# executed.
+# playback_hook = '/usr/share/perfact/zope4-tools/zodbsync-playback-hook'
+
 # Path to script that is called for postprocessing after a playback if it exists
 # run_after_playback = '/usr/share/perfact/zope4-tools/zodbsync-postproc'

perfact/zodbsync/subcommand.py

@@ -5,6 +5,7 @@
 import os
 
 import filelock
+import json
 
 from .helpers import Namespace
 
@@ -139,6 +140,75 @@ def check_repo(self):
         # The commit to compare to with regards to changed files
         self.orig_commit = self.branches[self.orig_branch]
 
+    def _playback_paths(self, paths):
+        paths = self.sync.prepare_paths(paths)
+        dryrun = self.args.dry_run
+
+        playback_hook = self.config.get('playback_hook', None)
+        if playback_hook and os.path.isfile(playback_hook):
+            proc = subprocess.Popen(
+                playback_hook, stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                universal_newlines=True)
+            out, _ = proc.communicate(json.dumps({'paths': paths}))
+            returncode = proc.returncode
+            if returncode:
+                raise AssertionError(
+                    "Error calling playback hook, returncode "
+                    "{}, [[{}]] on {}".format(
+                        returncode, playback_hook, out
+                    )
+                )
+            phases = json.loads(out)
+        else:
+            phases = [{'name': 'playback', 'paths': paths}]
+            if self.config.get('run_after_playback', None):
+                phases[-1]['cmd'] = self.config['run_after_playback']
+
+        for ix, phase in enumerate(phases):
+            phase_name = phase.get('name') or str(ix)
+            phase_cmd = phase.get('cmd')
+
+            self.sync.playback_paths(
+                paths=phase['paths'],
+                recurse=False,
+                override=True,
+                skip_errors=self.args.skip_errors,
+                dryrun=dryrun,
+            )
+
+            if dryrun or not (phase_cmd and os.path.isfile(phase_cmd)):
+                continue
+
+            self.logger.info(
+                'Calling phase %s, command: %s', phase_name, phase_cmd
+            )
+            proc = subprocess.Popen(
+                phase_cmd, stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                universal_newlines=True)
+            out, _ = proc.communicate(json.dumps(
+                {'paths': phase['paths']}
+            ))
+            returncode = proc.returncode
+
+            if returncode:
+                self.logger.error(
+                    "Error during phase command %s, %s",
+                    returncode, out
+                )
+                if sys.stdin.isatty():
+                    print("Enter 'y' to continue, other to rollback")
+                    res = input()
+                    if res == 'y':
+                        continue
+
+                raise AssertionError(
+                    "Unrecoverable error in phase command"
+                )
+            else:
+                self.logger.info(out)
+
     @staticmethod
     def gitexec(func):
         """
@@ -150,12 +220,14 @@ def gitexec(func):
         - play back changed objects (diff between old and new HEAD)
         - unstash
         """
+
         @SubCommand.with_lock
         def wrapper(self, *args, **kwargs):
             # Check for unstaged changes
             self.check_repo()
 
             try:
+                self.paths = []
                 func(self, *args, **kwargs)
 
                 # Fail and roll back for any of the markers of an interrupted
@@ -173,20 +245,10 @@ def wrapper(self, *args, **kwargs):
                 conflicts = files & set(self.unstaged_changes)
                 assert not conflicts, "Change in unstaged files, aborting"
 
-                # Strip site name from the start
-                files = [fname[len(self.sync.site):] for fname in files]
-                # Strip filename to get the object path
-                dirs = [fname.rsplit('/', 1)[0] for fname in files]
                 # Make unique and sort
-                paths = sorted(set(dirs))
-
-                self.sync.playback_paths(
-                    paths=paths,
-                    recurse=False,
-                    override=True,
-                    skip_errors=self.args.skip_errors,
-                    dryrun=self.args.dry_run,
-                )
+                self.paths = sorted(set(files))
+
+                self._playback_paths(self.paths)
 
                 if self.args.dry_run:
                     self.abort()
@@ -223,6 +285,17 @@ def wrapper(self, *args, **kwargs):
                         self.logger.exception("Unable to show diff")
 
                 self.abort()
+                # if we are not in dryrun we can't be sure we havent already
+                # committed some stuff to the data-fs so playback all paths
+                # abort
+                if not self.args.dry_run and self.paths:
+                    self.sync.playback_paths(
+                        paths=self.paths,
+                        recurse=False,
+                        override=True,
+                        skip_errors=True,
+                        dryrun=False,
+                    )
                 raise
 
         return wrapper

perfact/zodbsync/tests/test_sync.py

@@ -304,7 +304,7 @@ def prepare_pick(self, name='TestFolder', msg='Second commit'):
         initialized repository. Returns the commit ID.
         '''
         # Add a folder, commit it
-        self.add_folder('TestFolder', 'Second commit')
+        self.add_folder(name, msg)
         commit = self.get_head_id()
 
         # Reset the commit
@@ -1541,3 +1541,105 @@ def test_playback_postprocess(self):
         with open(self.config.path, 'w') as f:
             f.write(orig_config)
         del self.runner
+
+    def test_playback_hook(self):
+        """
+        Add configuration option for a playback hook script and check that
+        only the paths returned are played back
+        """
+        self.add_folder('NewFolder', 'First Folder')
+        self.add_folder('NewFolder2', 'Second Folder')
+        commit = self.get_head_id()
+        # Reset the commit
+        self.gitrun('reset', '--hard', 'HEAD~2')
+
+        playback_cmd = "{}/playback_cmd".format(self.zeo.path)
+        cmd_script = '\n'.join([
+            "#!/bin/bash",
+            "cat > {}"
+        ]).format('{}.out'.format(playback_cmd))
+        with open(playback_cmd, 'w') as f:
+            f.write(cmd_script)
+        os.chmod(playback_cmd, 0o700)
+
+        fname = "{}/playback_hook".format(self.zeo.path)
+        playback_dict = [{
+            "paths": ["/NewFolder"],
+            "cmd": playback_cmd
+        }]
+
+        script = '\n'.join([
+            "#!/bin/bash",
+            "echo '{}'".format(json.dumps(playback_dict)),
+        ])
+        with open(fname, 'w') as f:
+            f.write(script)
+        os.chmod(fname, 0o700)
+        with open(self.config.path) as f:
+            orig_config = f.read()
+        with open(self.config.path, 'a') as f:
+            f.write('\nplayback_hook = "{}"\n'.format(fname))
+
+        # Avoid error regarding reusing runner with changed config
+        del self.runner
+        self.run('pick', 'HEAD..{}'.format(commit))
+
+        assert 'NewFolder' in self.app.objectIds()
+        assert 'NewFolder2' not in self.app.objectIds()
+        assert os.path.isfile('{}.out'.format(playback_cmd))
+
+        with open(self.config.path, 'w') as f:
+            f.write(orig_config)
+        del self.runner
+
+    def test_playback_hook_failed(self):
+        """
+        Add configuration option for a playback hook script with a
+        failing cmd and check that all changes are rolled back
+        """
+        self.add_folder('NewFolder', 'First Folder')
+        self.add_folder('NewFolder2', 'Second Folder')
+        commit = self.get_head_id()
+        # Reset the commit
+        self.gitrun('reset', '--hard', 'HEAD~2')
+
+        playback_cmd = "{}/playback_cmd".format(self.zeo.path)
+        cmd_script = '\n'.join([
+            "#!/bin/bash",
+            "exit 42"
+        ])
+        with open(playback_cmd, 'w') as f:
+            f.write(cmd_script)
+        os.chmod(playback_cmd, 0o700)
+
+        fname = "{}/playback_hook".format(self.zeo.path)
+        playback_dict = [{
+            "paths": ["/NewFolder"],
+            "cmd": playback_cmd
+            }, {
+            "paths": ["/NewFolder2"],
+            },
+        ]
+        script = '\n'.join([
+            "#!/bin/bash",
+            "echo '{}'".format(json.dumps(playback_dict)),
+        ])
+        with open(fname, 'w') as f:
+            f.write(script)
+        os.chmod(fname, 0o700)
+        with open(self.config.path) as f:
+            orig_config = f.read()
+        with open(self.config.path, 'a') as f:
+            f.write('\nplayback_hook = "{}"\n'.format(fname))
+
+        # Avoid error regarding reusing runner with changed config
+        del self.runner
+        with pytest.raises(AssertionError):
+            self.run('pick', 'HEAD..{}'.format(commit))
+
+        assert 'NewFolder' not in self.app.objectIds()
+        assert 'NewFolder2' not in self.app.objectIds()
+
+        with open(self.config.path, 'w') as f:
+            f.write(orig_config)
+        del self.runner

perfact/zodbsync/zodbsync.py

@@ -5,8 +5,6 @@
 import shutil
 import time  # for periodic output
 import sys
-import json
-import subprocess
 
 # for using an explicit transaction manager
 import transaction
@@ -660,11 +658,7 @@ def _playback_fixorder(self, path):
         object_handlers[fs_data['type']].fix_order(obj, fs_data)
         del self.fs_data[path]
 
-    def playback_paths(self, paths, recurse=True, override=False,
-                       skip_errors=False, dryrun=False):
-        self.recurse = recurse
-        self.override = override
-        self.skip_errors = skip_errors
+    def prepare_paths(self, paths):
         # normalize paths - cut off filenames and the site name
         paths = {
             path.rsplit('/', 1)[0] if (
@@ -679,9 +673,17 @@ def playback_paths(self, paths, recurse=True, override=False,
         })
 
         if not len(paths):
-            return
+            return []
 
         paths = [path.rstrip('/') + '/' for path in paths]
+        return paths
+
+    def playback_paths(self, paths, recurse=True, override=False,
+                       skip_errors=False, dryrun=False):
+        self.recurse = recurse
+        self.override = override
+        self.skip_errors = skip_errors
+        paths = self.prepare_paths(paths)
 
         if recurse:
             paths = remove_redundant_paths(paths)
@@ -744,12 +746,6 @@ def playback_paths(self, paths, recurse=True, override=False,
             txn_mgr.abort()
         else:
             txn_mgr.commit()
-            postproc = self.config.get('run_after_playback', None)
-            if postproc and os.path.isfile(postproc):
-                self.logger.info('Calling postprocessing script ' + postproc)
-                proc = subprocess.Popen(postproc, stdin=subprocess.PIPE,
-                                        universal_newlines=True)
-                proc.communicate(json.dumps({'paths': paths}))
 
     def recent_changes(self, since_secs=None, txnid=None, limit=50,
                        search_limit=100):