[PATCH v3 0/3] iotests: add JobRunner framework

[PATCH v3 0/3] iotests: add JobRunner framework

[John Snow]

Requires: 20200317004105.27059-1-jsnow@redhat.com

(This requires the iotests pylint & logging series.)

The basic idea is to make a generic job runtime manager and allow
callers to subclass the manager. Then, instead of adding callback
arguments to the function all the time, we have à la carte customization
of the loop.

To showcase this a little bit, I removed the pre_finalization argument
and made existing callers use a custom JobRunner; and then converted
test 040 to use this style of job runner.

Is it a simplification? No. Is it cool? Maybe. Did it remove the
duplicated job-running code in 040? yes.

V3:
 - Rebased on logging series v8
 - Converted 155's new usage of job_run

V2:
 - Rebased on logging series; logging conditionals are pretty now.
 - Inlined callback login in 257
 - No longer based on bitmap-populate job (no test 287)
 - Moved super() call to the beginning of test 040's callback
 - Added docstrings and type annotations

John Snow (3):
  qmp.py: change event_wait to use a dict
  iotests: add JobRunner class
  iotests: modify test 040 to use JobRunner

 python/qemu/machine.py        |  10 +-
 tests/qemu-iotests/040        |  51 +++++----
 tests/qemu-iotests/155        |  15 ++-
 tests/qemu-iotests/255        |   9 +-
 tests/qemu-iotests/257        |  54 +++++----
 tests/qemu-iotests/260        |   5 +-
 tests/qemu-iotests/iotests.py | 201 +++++++++++++++++++++++++---------
 tests/qemu-iotests/pylintrc   |  11 ++
 8 files changed, 246 insertions(+), 110 deletions(-)

-- 
2.21.1

[PATCH v3 1/3] qmp.py: change event_wait to use a dict

[John Snow]

It's easier to work with than a list of tuples, because we can check the
keys for membership.

Signed-off-by: John Snow <jsnow@redhat.com>
---
 python/qemu/machine.py        | 10 +++++-----
 tests/qemu-iotests/040        | 12 ++++++------
 tests/qemu-iotests/260        |  5 +++--
 tests/qemu-iotests/iotests.py | 16 ++++++++--------
 4 files changed, 22 insertions(+), 21 deletions(-)

diff --git a/python/qemu/machine.py b/python/qemu/machine.py
index 183d8f3d38..748de5f322 100644
--- a/python/qemu/machine.py
+++ b/python/qemu/machine.py
@@ -476,21 +476,21 @@ def event_wait(self, name, timeout=60.0, match=None):
         timeout: QEMUMonitorProtocol.pull_event timeout parameter.
         match: Optional match criteria. See event_match for details.
         """
-        return self.events_wait([(name, match)], timeout)
+        return self.events_wait({name: match}, timeout)
 
     def events_wait(self, events, timeout=60.0):
         """
         events_wait waits for and returns a named event from QMP with a timeout.
 
-        events: a sequence of (name, match_criteria) tuples.
+        events: a mapping containing {name: match_criteria}.
                 The match criteria are optional and may be None.
                 See event_match for details.
         timeout: QEMUMonitorProtocol.pull_event timeout parameter.
         """
         def _match(event):
-            for name, match in events:
-                if event['event'] == name and self.event_match(event, match):
-                    return True
+            name = event['event']
+            if name in events:
+                return self.event_match(event, events[name])
             return False
 
         # Search cached events
diff --git a/tests/qemu-iotests/040 b/tests/qemu-iotests/040
index 32c82b4ec6..90b59081ff 100755
--- a/tests/qemu-iotests/040
+++ b/tests/qemu-iotests/040
@@ -485,12 +485,12 @@ class TestErrorHandling(iotests.QMPTestCase):
 
     def run_job(self, expected_events, error_pauses_job=False):
         match_device = {'data': {'device': 'job0'}}
-        events = [
-            ('BLOCK_JOB_COMPLETED', match_device),
-            ('BLOCK_JOB_CANCELLED', match_device),
-            ('BLOCK_JOB_ERROR', match_device),
-            ('BLOCK_JOB_READY', match_device),
-        ]
+        events = {
+            'BLOCK_JOB_COMPLETED': match_device,
+            'BLOCK_JOB_CANCELLED': match_device,
+            'BLOCK_JOB_ERROR': match_device,
+            'BLOCK_JOB_READY': match_device,
+        }
 
         completed = False
         log = []
diff --git a/tests/qemu-iotests/260 b/tests/qemu-iotests/260
index 804a7addb9..729f031122 100755
--- a/tests/qemu-iotests/260
+++ b/tests/qemu-iotests/260
@@ -67,8 +67,9 @@ def test(persistent, restart):
 
     vm.qmp_log('block-commit', device='drive0', top=top,
                filters=[iotests.filter_qmp_testfiles])
-    ev = vm.events_wait((('BLOCK_JOB_READY', None),
-                         ('BLOCK_JOB_COMPLETED', None)))
+    ev = vm.events_wait({
+        'BLOCK_JOB_READY': None,
+        'BLOCK_JOB_COMPLETED': None })
     log(filter_qmp_event(ev))
     if (ev['event'] == 'BLOCK_JOB_COMPLETED'):
         vm.shutdown()
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index 9304066471..bc96f1ce27 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -617,14 +617,14 @@ def run_job(self, job, auto_finalize=True, auto_dismiss=False,
         """
         match_device = {'data': {'device': job}}
         match_id = {'data': {'id': job}}
-        events = [
-            ('BLOCK_JOB_COMPLETED', match_device),
-            ('BLOCK_JOB_CANCELLED', match_device),
-            ('BLOCK_JOB_ERROR', match_device),
-            ('BLOCK_JOB_READY', match_device),
-            ('BLOCK_JOB_PENDING', match_id),
-            ('JOB_STATUS_CHANGE', match_id)
-        ]
+        events = {
+            'BLOCK_JOB_COMPLETED': match_device,
+            'BLOCK_JOB_CANCELLED': match_device,
+            'BLOCK_JOB_ERROR': match_device,
+            'BLOCK_JOB_READY': match_device,
+            'BLOCK_JOB_PENDING': match_id,
+            'JOB_STATUS_CHANGE': match_id,
+        }
         error = None
         while True:
             ev = filter_qmp_event(self.events_wait(events, timeout=wait))
-- 
2.21.1

[PATCH v3 2/3] iotests: add JobRunner class

[John Snow]

The idea is that instead of increasing the arguments to job_run all the
time, create a more general-purpose job runner that can be subclassed to
do interesting things with.

pylint note: the 'callbacks' option guards against unused warning
arguments in functions designated as callbacks. It does not currently
guard against "no-self-use" though; hence a once-off ignore.

mypy note: QapiEvent is only a weak alias; it's fully interchangable
with the type it's declared as. In the future, we may wish to tighten
these types. For now, this communicates the rough shape of the type and
(more importantly) the intent.

Signed-off-by: John Snow <jsnow@redhat.com>
---
 tests/qemu-iotests/155        |  15 ++-
 tests/qemu-iotests/255        |   9 +-
 tests/qemu-iotests/257        |  54 +++++----
 tests/qemu-iotests/iotests.py | 201 +++++++++++++++++++++++++---------
 tests/qemu-iotests/pylintrc   |  11 ++
 5 files changed, 213 insertions(+), 77 deletions(-)

diff --git a/tests/qemu-iotests/155 b/tests/qemu-iotests/155
index cb371d4649..e2a013e774 100755
--- a/tests/qemu-iotests/155
+++ b/tests/qemu-iotests/155
@@ -163,6 +163,16 @@ class BaseClass(iotests.QMPTestCase):
             self.assert_qmp_absent(node, 'image/backing-image')
 
 
+class MirrorJob(iotests.JobRunner):
+    def __init__(self, *args, test, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.test = test
+
+    def on_pending(self, event):
+        self.test.openBacking()
+        super().on_pending(event)
+
+
 # Class variables for controlling its behavior:
 #
 # cmd: Mirroring command to execute, either drive-mirror or blockdev-mirror
@@ -188,8 +198,9 @@ class MirrorBaseClass(BaseClass):
 
         self.assert_qmp(result, 'return', {})
 
-        self.vm.run_job('mirror-job', auto_finalize=False,
-                        pre_finalize=self.openBacking, auto_dismiss=True)
+        job = MirrorJob(self.vm, 'mirror-job', test=self,
+                        auto_finalize=False, auto_dismiss=True)
+        job.run()
 
     def testFull(self):
         self.runMirror('full')
diff --git a/tests/qemu-iotests/255 b/tests/qemu-iotests/255
index 8f08f741da..e66cdfd672 100755
--- a/tests/qemu-iotests/255
+++ b/tests/qemu-iotests/255
@@ -71,8 +71,13 @@ with iotests.FilePath('t.qcow2') as disk_path, \
     result = vm.qmp_log('block-commit', job_id='job0', auto_finalize=False,
                         device='overlay', top_node='mid')
 
-    vm.run_job('job0', auto_finalize=False, pre_finalize=start_requests,
-                auto_dismiss=True)
+    class TestJobRunner(iotests.JobRunner):
+        def on_pending(self, event):
+            start_requests()
+            super().on_pending(event)
+
+    runner = TestJobRunner(vm, 'job0', auto_finalize=False, auto_dismiss=True)
+    runner.run()
 
     vm.shutdown()
 
diff --git a/tests/qemu-iotests/257 b/tests/qemu-iotests/257
index 004a433b8b..95341c330f 100755
--- a/tests/qemu-iotests/257
+++ b/tests/qemu-iotests/257
@@ -352,30 +352,40 @@ def test_bitmap_sync(bsync_mode, msync_mode='bitmap', failure=None):
         job = backup(drive0, 1, bsync1, msync_mode,
                      bitmap="bitmap0", bitmap_mode=bsync_mode)
 
-        def _callback():
-            """Issue writes while the job is open to test bitmap divergence."""
-            # Note: when `failure` is 'intermediate', this isn't called.
-            log('')
-            bitmaps = perform_writes(drive0, 2, filter_node_name='backup-top')
-            # Named bitmap (static, should be unchanged)
-            ebitmap.compare(vm.get_bitmap(drive0.node, 'bitmap0',
-                                          bitmaps=bitmaps))
-            # Anonymous bitmap (dynamic, shows new writes)
-            anonymous = EmulatedBitmap()
-            anonymous.dirty_group(2)
-            anonymous.compare(vm.get_bitmap(drive0.node, '', recording=True,
-                                            bitmaps=bitmaps))
 
-            # Simulate the order in which this will happen:
-            # group 1 gets cleared first, then group two gets written.
-            if ((bsync_mode == 'on-success' and not failure) or
-                (bsync_mode == 'always')):
-                ebitmap.clear()
-            ebitmap.dirty_group(2)
+        class TestJobRunner(iotests.JobRunner):
+            def on_pending(self, event):
+                """
+                Issue writes while the job is open to test bitmap divergence.
+                """
+
+                # Note: when `failure` is 'intermediate', this isn't called.
+                log('')
+                bitmaps = perform_writes(drive0, 2,
+                                         filter_node_name='backup-top')
+                # Named bitmap (static, should be unchanged)
+                ebitmap.compare(vm.get_bitmap(drive0.node, 'bitmap0',
+                                              bitmaps=bitmaps))
+                # Anonymous bitmap (dynamic, shows new writes)
+                anonymous = EmulatedBitmap()
+                anonymous.dirty_group(2)
+                anonymous.compare(vm.get_bitmap(drive0.node, '', recording=True,
+                                                bitmaps=bitmaps))
+
+                # Simulate the order in which this will happen:
+                # group 1 gets cleared first, then group two gets written.
+                if ((bsync_mode == 'on-success' and not failure) or
+                    (bsync_mode == 'always')):
+                    ebitmap.clear()
+                ebitmap.dirty_group(2)
+
+                super().on_pending(event)
+
+
+        runner = TestJobRunner(vm, job, cancel=(failure == 'simulated'),
+                               auto_finalize=False, auto_dismiss=True)
+        runner.run()
 
-        vm.run_job(job, auto_dismiss=True, auto_finalize=False,
-                   pre_finalize=_callback,
-                   cancel=(failure == 'simulated'))
         bitmaps = vm.query_bitmaps()
         log({'bitmaps': bitmaps}, indent=2)
         log('')
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index bc96f1ce27..67c3de380a 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -29,7 +29,7 @@
 import atexit
 import io
 from collections import OrderedDict
-from typing import Collection
+from typing import Any, Collection, Dict, Optional
 
 # pylint: disable=import-error, wrong-import-position
 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
@@ -37,6 +37,9 @@
 
 assert sys.version_info >= (3, 6)
 
+# Type aliases
+QapiEvent = Dict[str, Any]
+
 # Use this logger for logging messages directly from the iotests module
 logger = logging.getLogger('qemu.iotests')
 logger.addHandler(logging.NullHandler())
@@ -476,6 +479,141 @@ def remote_filename(path):
     else:
         raise Exception("Protocol %s not supported" % (imgproto))
 
+
+class JobRunner:
+    """
+    JobRunner offers a job-lifetime management framework.
+
+    It can be used as-is for a no-frills run-to-completion module,
+    or subclassed to gain access to per-event callbacks for
+    customizable behavior.
+
+    :param vm: The VM the job is running on
+    :param job: Job ID of a recently created job
+    :param cancel: When true, cancels the job prior to finalization.
+    :param auto_finalize: True if the job was configured to finalize itself.
+    :param auto_dismiss: True if the job will dismiss itself post-finalization.
+    """
+    def __init__(self,
+                 vm: 'VM',
+                 job: str,
+                 cancel: bool = False,
+                 auto_finalize: bool = True,
+                 auto_dismiss: bool = False):
+        self._vm = vm
+        self._id = job
+        self.cancel = cancel
+
+        self._auto_finalize = auto_finalize
+        self._auto_dismiss = auto_dismiss
+        self._exited = False
+        self._error: Optional[str] = None
+
+        match_device = {'data': {'device': self._id}}
+        match_id = {'data': {'id': self._id}}
+
+        # Listen for these events with these parameters:
+        self._events = {
+            'BLOCK_JOB_COMPLETED': match_device,
+            'BLOCK_JOB_CANCELLED': match_device,
+            'BLOCK_JOB_ERROR': match_device,
+            'BLOCK_JOB_READY': match_device,
+            'BLOCK_JOB_PENDING': match_id,
+            'JOB_STATUS_CHANGE': match_id
+        }
+
+        self._dispatch = {
+            'created': self.on_create,
+            'running': self.on_run,
+            'paused': self.on_pause,
+            'ready': self.on_ready,
+            'standby': self.on_standby,
+            'waiting': self.on_waiting,
+            'pending': self.on_pending,
+            'aborting': self.on_abort,
+            'concluded': self.on_conclude,
+            'null': self.on_null,
+        }
+
+    # These are Job state change callbacks.
+    # Subclass and override these for custom workflows.
+
+    def on_create(self, event: QapiEvent) -> None:
+        pass
+
+    def on_run(self, event: QapiEvent) -> None:
+        pass
+
+    def on_pause(self, event: QapiEvent) -> None:
+        pass
+
+    def on_ready(self, event: QapiEvent) -> None:
+        self._vm.qmp_log('job-complete', id=self._id)
+
+    def on_standby(self, event: QapiEvent) -> None:
+        pass
+
+    def on_waiting(self, event: QapiEvent) -> None:
+        pass
+
+    def on_pending(self, event: QapiEvent) -> None:
+        if self._auto_finalize:
+            return
+
+        if self.cancel:
+            self._vm.qmp_log('job-cancel', id=self._id)
+        else:
+            self._vm.qmp_log('job-finalize', id=self._id)
+
+    def on_abort(self, event: QapiEvent) -> None:
+        result = self._vm.qmp('query-jobs')
+        for j in result['return']:
+            if j['id'] == self._id:
+                self._error = j['error']
+                log('Job failed: %s' % (j['error']))
+
+    def on_conclude(self, event: QapiEvent) -> None:
+        if self._auto_dismiss:
+            return
+
+        self._vm.qmp_log('job-dismiss', id=self._id)
+
+    def on_null(self, event: QapiEvent) -> None:
+        self._exited = True
+
+    # Macro events -- QAPI events.
+    # These are callbacks for individual top-level events.
+
+    def on_change(self, event: QapiEvent) -> None:
+        status = event['data']['status']
+        assert status in self._dispatch
+        self._dispatch[status](event)
+
+    def on_block_job_event(self, event: QapiEvent) -> None:
+        # pylint: disable=no-self-use
+        log(event)
+
+    def event(self, event: QapiEvent) -> None:
+        assert event['event'] in self._events.keys()
+        if event['event'] == 'JOB_STATUS_CHANGE':
+            self.on_change(event)
+        else:
+            self.on_block_job_event(event)
+
+    def run(self, wait: float = 60.0) -> Optional[str]:
+        """
+        Run the event loop for this job.
+
+        :param wait: Timeout in seconds specifying how long to wait
+                     for an event. Defaults to 60.0.
+        :return: Error string on failure, Nothing on success.
+        """
+        while not self._exited:
+            raw_event = self._vm.events_wait(self._events, timeout=wait)
+            self.event(filter_qmp_event(raw_event))
+        return self._error
+
+
 class VM(qtest.QEMUQtestMachine):
     '''A QEMU VM'''
 
@@ -597,60 +735,21 @@ def qmp_log(self, cmd, filters=(), indent=None, **kwargs):
         log(result, filters, indent=indent)
         return result
 
-    # Returns None on success, and an error string on failure
-    def run_job(self, job, auto_finalize=True, auto_dismiss=False,
-                pre_finalize=None, cancel=False, wait=60.0):
+    def run_job(self, job, **kwargs) -> Optional[str]:
         """
         run_job moves a job from creation through to dismissal.
 
-        :param job: String. ID of recently-launched job
-        :param auto_finalize: Bool. True if the job was launched with
-                              auto_finalize. Defaults to True.
-        :param auto_dismiss: Bool. True if the job was launched with
-                             auto_dismiss=True. Defaults to False.
-        :param pre_finalize: Callback. A callable that takes no arguments to be
-                             invoked prior to issuing job-finalize, if any.
-        :param cancel: Bool. When true, cancels the job after the pre_finalize
-                       callback.
-        :param wait: Float. Timeout value specifying how long to wait for any
-                     event, in seconds. Defaults to 60.0.
+        :param job: Job ID of a recently created job.
+        :param kwargs: See JobRunner.__init__() and JobRunner.run().
+
+        :return: Error string on failure, Nothing on success.
         """
-        match_device = {'data': {'device': job}}
-        match_id = {'data': {'id': job}}
-        events = {
-            'BLOCK_JOB_COMPLETED': match_device,
-            'BLOCK_JOB_CANCELLED': match_device,
-            'BLOCK_JOB_ERROR': match_device,
-            'BLOCK_JOB_READY': match_device,
-            'BLOCK_JOB_PENDING': match_id,
-            'JOB_STATUS_CHANGE': match_id,
-        }
-        error = None
-        while True:
-            ev = filter_qmp_event(self.events_wait(events, timeout=wait))
-            if ev['event'] != 'JOB_STATUS_CHANGE':
-                log(ev)
-                continue
-            status = ev['data']['status']
-            if status == 'aborting':
-                result = self.qmp('query-jobs')
-                for j in result['return']:
-                    if j['id'] == job:
-                        error = j['error']
-                        log('Job failed: %s' % (j['error']))
-            elif status == 'ready':
-                self.qmp_log('job-complete', id=job)
-            elif status == 'pending' and not auto_finalize:
-                if pre_finalize:
-                    pre_finalize()
-                if cancel:
-                    self.qmp_log('job-cancel', id=job)
-                else:
-                    self.qmp_log('job-finalize', id=job)
-            elif status == 'concluded' and not auto_dismiss:
-                self.qmp_log('job-dismiss', id=job)
-            elif status == 'null':
-                return error
+        if 'wait' in kwargs:
+            run_kwargs = {'wait': kwargs.pop('wait')}
+        else:
+            run_kwargs = {}
+        job_runner = JobRunner(self, job, **kwargs)
+        return job_runner.run(**run_kwargs)
 
     # Returns None on success, and an error string on failure
     def blockdev_create(self, options, job_id='job0', filters=None):
diff --git a/tests/qemu-iotests/pylintrc b/tests/qemu-iotests/pylintrc
index 8d02f00607..6dae97d916 100644
--- a/tests/qemu-iotests/pylintrc
+++ b/tests/qemu-iotests/pylintrc
@@ -17,9 +17,20 @@ disable=invalid-name,
         too-many-locals,
         too-many-branches,
         too-many-public-methods,
+        too-many-instance-attributes,
         # These are temporary, and should be removed:
         missing-docstring,
 
+
+[VARIABLES]
+
+# List of strings which can identify a callback function by name. A callback
+# name must start or end with one of those strings.
+callbacks=cb_,
+          _cb,
+          on_,
+
+
 [FORMAT]
 
 # Maximum number of characters on a single line.
-- 
2.21.1

[PATCH v3 3/3] iotests: modify test 040 to use JobRunner

[John Snow]

Instead of having somewhat reproduced it for itself.

Signed-off-by: John Snow <jsnow@redhat.com>
---
 tests/qemu-iotests/040 | 51 +++++++++++++++++++++---------------------
 1 file changed, 25 insertions(+), 26 deletions(-)

diff --git a/tests/qemu-iotests/040 b/tests/qemu-iotests/040
index 90b59081ff..e2ef3bb812 100755
--- a/tests/qemu-iotests/040
+++ b/tests/qemu-iotests/040
@@ -483,34 +483,33 @@ class TestErrorHandling(iotests.QMPTestCase):
                           file=('top-dbg' if top_debug else 'top-file'),
                           backing='mid-fmt')
 
+
+    class TestJobRunner(iotests.JobRunner):
+        expected_events = ('BLOCK_JOB_COMPLETED',
+                           'BLOCK_JOB_ERROR',
+                           'BLOCK_JOB_READY')
+
+        def __init__(self, *args, test, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.log = []
+            self.test = test
+
+        def on_pause(self, event):
+            super().on_pause(event)
+            result = self._vm.qmp('block-job-resume', device=self._id)
+            self.test.assert_qmp(result, 'return', {})
+
+        def on_block_job_event(self, event):
+            if event['event'] not in self.expected_events:
+                self.test.fail("Unexpected event: %s" % event)
+            super().on_block_job_event(event)
+            self.log.append(event)
+
     def run_job(self, expected_events, error_pauses_job=False):
-        match_device = {'data': {'device': 'job0'}}
-        events = {
-            'BLOCK_JOB_COMPLETED': match_device,
-            'BLOCK_JOB_CANCELLED': match_device,
-            'BLOCK_JOB_ERROR': match_device,
-            'BLOCK_JOB_READY': match_device,
-        }
-
-        completed = False
-        log = []
-        while not completed:
-            ev = self.vm.events_wait(events, timeout=5.0)
-            if ev['event'] == 'BLOCK_JOB_COMPLETED':
-                completed = True
-            elif ev['event'] == 'BLOCK_JOB_ERROR':
-                if error_pauses_job:
-                    result = self.vm.qmp('block-job-resume', device='job0')
-                    self.assert_qmp(result, 'return', {})
-            elif ev['event'] == 'BLOCK_JOB_READY':
-                result = self.vm.qmp('block-job-complete', device='job0')
-                self.assert_qmp(result, 'return', {})
-            else:
-                self.fail("Unexpected event: %s" % ev)
-            log.append(iotests.filter_qmp_event(ev))
-
+        job = self.TestJobRunner(self.vm, 'job0', test=self)
+        job.run()
         self.maxDiff = None
-        self.assertEqual(expected_events, log)
+        self.assertEqual(expected_events, job.log)
 
     def event_error(self, op, action):
         return {
-- 
2.21.1