diff --git a/tests/qemu-iotests/255 b/tests/qemu-iotests/255
index 4a4818bafb..513e9ebb58 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 2a81f9e30c..e73b0c20b3 100755
--- a/tests/qemu-iotests/257
+++ b/tests/qemu-iotests/257
@@ -265,9 +265,15 @@ def test_bitmap_sync(bsync_mode, msync_mode='bitmap', failure=None):
                 ebitmap.clear()
             ebitmap.dirty_group(2)
 
-        vm.run_job(job, auto_dismiss=True, auto_finalize=False,
-                   pre_finalize=_callback,
-                   cancel=(failure == 'simulated'))
+        class TestJobRunner(iotests.JobRunner):
+            def on_pending(self, event):
+                _callback()
+                super().on_pending(event)
+
+        runner = TestJobRunner(vm, job, cancel=(failure == 'simulated'),
+                               auto_finalize=False, auto_dismiss=True)
+        runner.run()
+
         bitmaps = vm.query_bitmaps()
         log({'bitmaps': bitmaps}, indent=2)
         log('')
diff --git a/tests/qemu-iotests/287 b/tests/qemu-iotests/287
index 0ab58dc011..f06e6ff084 100755
--- a/tests/qemu-iotests/287
+++ b/tests/qemu-iotests/287
@@ -165,13 +165,22 @@ def test_bitmap_populate(config):
                 if not config.disabled:
                     ebitmap.dirty_group(2)
 
+
+        class TestJobRunner(iotests.JobRunner):
+            def on_pending(self, event):
+                if config.mid_writes:
+                    perform_writes(drive0, 2)
+                    if not config.disabled:
+                        ebitmap.dirty_group(2)
+                super().on_pending(event)
+
         job = populate(drive0, 'target', 'bitpop0')
         assert job['return'] == {'return': {}}
-        vm.run_job(job['id'],
-                   auto_dismiss=job['auto-dismiss'],
-                   auto_finalize=job['auto-finalize'],
-                   pre_finalize=pre_finalize,
-                   cancel=config.cancel)
+        job_runner = TestJobRunner(vm, job['id'],
+                                   auto_dismiss=job['auto-dismiss'],
+                                   auto_finalize=job['auto-finalize'],
+                                   cancel=config.cancel)
+        job_runner.run()
         log('')
 
 
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index 3390fab021..37a8b4d649 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -460,6 +460,130 @@ def remote_filename(path):
     else:
         raise Exception("Protocol %s not supported" % (imgproto))
 
+
+class JobRunner:
+    def __init__(self, vm, job,
+                 use_log=True,
+                 cancel=False,
+                 auto_finalize=True,
+                 auto_dismiss=False):
+        self._vm = vm
+        self._id = job
+        self.logging = use_log
+        self.cancel = cancel
+
+        self._auto_finalize = auto_finalize
+        self._auto_dismiss = auto_dismiss
+        self._exited = False
+        self._error = None
+
+        match_device = {'data': {'device': job}}
+        match_id = {'data': {'id': job}}
+        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,
+        }
+
+    # Job events -- state changes.
+
+    def on_create(self, event):
+        pass
+
+    def on_run(self, event):
+        pass
+
+    def on_pause(self, event):
+        pass
+
+    def on_ready(self, event):
+        if self.logging:
+            self._vm.qmp_log('job-complete', id=self._id)
+        else:
+            self._vm.qmp('job-complete', id=self._id)
+
+    def on_standby(self, event):
+        pass
+
+    def on_waiting(self, event):
+        pass
+
+    def on_pending(self, event):
+        if self._auto_finalize:
+            return
+
+        if self.cancel:
+            if self.logging:
+                self._vm.qmp_log('job-cancel', id=self._id)
+            else:
+                self._vm.qmp('job-cancel', id=self._id)
+        else:
+            if self.logging:
+                self._vm.qmp_log('job-finalize', id=self._id)
+            else:
+                self._vm.qmp('job-finalize', id=self._id)
+
+    def on_abort(self, event):
+        result = self._vm.qmp('query-jobs')
+        for j in result['return']:
+            if j['id'] == self._id:
+                self.error = j['error']
+                if self.logging:
+                    log('Job failed: %s' % (j['error']))
+
+    def on_conclude(self, event):
+        if self._auto_dismiss:
+            return
+
+        if self.logging:
+            self._vm.qmp_log('job-dismiss', id=self._id)
+        else:
+            self._vm.qmp('job-dismiss', id=self._id)
+
+    def on_null(self, event):
+        self._exited = True
+
+    # Macro events -- QAPI events
+
+    def on_change(self, event):
+        status = event['data']['status']
+        assert status in self._dispatch
+        self._dispatch[status](event)
+
+    def on_block_job_event(self, event):
+        if self.logging:
+            log(event)
+
+    def _event(self, event):
+        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=60.0):
+        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'''
 
@@ -585,7 +709,7 @@ def qmp_log(self, cmd, filters=[], indent=None, **kwargs):
 
     # 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, use_log=True, wait=60.0):
+                cancel=False, use_log=True, wait=60.0):
         """
         run_job moves a job from creation through to dismissal.
 
@@ -594,59 +718,15 @@ def run_job(self, job, auto_finalize=True, auto_dismiss=False,
                               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 use_log: Bool. When false, does not log QMP messages.
         :param wait: Float. Timeout value specifying how long to wait for any
                      event, in seconds. Defaults to 60.0.
         """
-        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':
-                if use_log:
-                    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']
-                        if use_log:
-                            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 and use_log:
-                    self.qmp_log('job-cancel', id=job)
-                elif cancel:
-                    self.qmp('job-cancel', id=job)
-                elif use_log:
-                    self.qmp_log('job-finalize', id=job)
-                else:
-                    self.qmp('job-finalize', id=job)
-            elif status == 'concluded' and not auto_dismiss:
-                if use_log:
-                    self.qmp_log('job-dismiss', id=job)
-                else:
-                    self.qmp('job-dismiss', id=job)
-            elif status == 'null':
-                return error
+        job_runner = JobRunner(self, job, use_log, cancel,
+                               auto_finalize, auto_dismiss)
+        return job_runner.run(wait=wait)
 
     # Returns None on success, and an error string on failure
     def blockdev_create(self, options, job_id='job0', filters=None):
