Move some of the code from test_migration.py to a new class so it can
be reused to invoke migrations from other tests.
Signed-off-by: Fabiano Rosas <faro...@suse.de>
---
I see this conflicts with Thomas' series, I'll update accordingly.
---
tests/functional/qemu_test/migration.py | 40 +++++++++++++++++++++++++
tests/functional/test_migration.py | 24 ++-------------
2 files changed, 43 insertions(+), 21 deletions(-)
create mode 100644 tests/functional/qemu_test/migration.py
diff --git a/tests/functional/qemu_test/migration.py
b/tests/functional/qemu_test/migration.py
new file mode 100644
index 0000000000..37988704e8
--- /dev/null
+++ b/tests/functional/qemu_test/migration.py
@@ -0,0 +1,40 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Migration test
+#
+# Copyright (c) 2019 Red Hat, Inc.
+#
+# Authors:
+# Cleber Rosa <cr...@redhat.com>
+# Caio Carrara <ccarr...@redhat.com>
+#
+# This work is licensed under the terms of the GNU GPL, version 2 or
+# later. See the COPYING file in the top-level directory.
+
+import time
+
+
+class Migration():
+
+ @staticmethod
+ def migration_finished(vm):
+ return vm.cmd('query-migrate')['status'] in ('completed', 'failed')
+
+ def assert_migration(self, test, src_vm, dst_vm, timeout):
+
+ end = time.monotonic() + timeout
+ while time.monotonic() < end and not self.migration_finished(src_vm):
+ time.sleep(0.1)
+
+ end = time.monotonic() + timeout
+ while time.monotonic() < end and not self.migration_finished(dst_vm):
+ time.sleep(0.1)
+
+ test.assertEqual(src_vm.cmd('query-migrate')['status'], 'completed')
+ test.assertEqual(dst_vm.cmd('query-migrate')['status'], 'completed')
+ test.assertEqual(dst_vm.cmd('query-status')['status'], 'running')
+ test.assertEqual(src_vm.cmd('query-status')['status'],'postmigrate')
+
+ def migrate(self, test, source_vm, dest_vm, src_uri, timeout):
+ source_vm.qmp('migrate', uri=src_uri)
+ self.assert_migration(test, source_vm, dest_vm, timeout)
diff --git a/tests/functional/test_migration.py
b/tests/functional/test_migration.py
index c4393c3543..1c75a98330 100755
--- a/tests/functional/test_migration.py
+++ b/tests/functional/test_migration.py
@@ -15,6 +15,7 @@
import time
from qemu_test import QemuSystemTest, skipIfMissingCommands
+from qemu_test.migration import Migration
from qemu_test.ports import Ports
@@ -22,25 +23,6 @@ class MigrationTest(QemuSystemTest):
timeout = 10
- @staticmethod
- def migration_finished(vm):
- return vm.cmd('query-migrate')['status'] in ('completed', 'failed')
-
- def assert_migration(self, src_vm, dst_vm):
-
- end = time.monotonic() + self.timeout
- while time.monotonic() < end and not self.migration_finished(src_vm):
- time.sleep(0.1)
-
- end = time.monotonic() + self.timeout
- while time.monotonic() < end and not self.migration_finished(dst_vm):
- time.sleep(0.1)
-
- self.assertEqual(src_vm.cmd('query-migrate')['status'], 'completed')
- self.assertEqual(dst_vm.cmd('query-migrate')['status'], 'completed')
- self.assertEqual(dst_vm.cmd('query-status')['status'], 'running')
- self.assertEqual(src_vm.cmd('query-status')['status'],'postmigrate')
-
def select_machine(self):
target_machine = {
'aarch64': 'quanta-gsj',
@@ -67,8 +49,8 @@ def do_migrate(self, dest_uri, src_uri=None):
source_vm = self.get_vm(name="source-qemu")
source_vm.add_args('-nodefaults')
source_vm.launch()
- source_vm.qmp('migrate', uri=src_uri)
- self.assert_migration(source_vm, dest_vm)
+
+ Migration().migrate(self, source_vm, dest_vm, src_uri, self.timeout)
def _get_free_port(self, ports):
port = ports.find_free_port()
--
2.35.3
