migration.py (2752B)
1 # Migration test 2 # 3 # Copyright (c) 2019 Red Hat, Inc. 4 # 5 # Authors: 6 # Cleber Rosa <crosa@redhat.com> 7 # Caio Carrara <ccarrara@redhat.com> 8 # 9 # This work is licensed under the terms of the GNU GPL, version 2 or 10 # later. See the COPYING file in the top-level directory. 11 12 13 import tempfile 14 from avocado_qemu import QemuSystemTest 15 from avocado import skipUnless 16 17 from avocado.utils.network import ports 18 from avocado.utils import wait 19 from avocado.utils.path import find_command 20 21 22 class Migration(QemuSystemTest): 23 """ 24 :avocado: tags=migration 25 """ 26 27 timeout = 10 28 29 @staticmethod 30 def migration_finished(vm): 31 return vm.command('query-migrate')['status'] in ('completed', 'failed') 32 33 def assert_migration(self, src_vm, dst_vm): 34 wait.wait_for(self.migration_finished, 35 timeout=self.timeout, 36 step=0.1, 37 args=(src_vm,)) 38 wait.wait_for(self.migration_finished, 39 timeout=self.timeout, 40 step=0.1, 41 args=(dst_vm,)) 42 self.assertEqual(src_vm.command('query-migrate')['status'], 'completed') 43 self.assertEqual(dst_vm.command('query-migrate')['status'], 'completed') 44 self.assertEqual(dst_vm.command('query-status')['status'], 'running') 45 self.assertEqual(src_vm.command('query-status')['status'],'postmigrate') 46 47 def do_migrate(self, dest_uri, src_uri=None): 48 dest_vm = self.get_vm('-incoming', dest_uri) 49 dest_vm.add_args('-nodefaults') 50 dest_vm.launch() 51 if src_uri is None: 52 src_uri = dest_uri 53 source_vm = self.get_vm() 54 source_vm.add_args('-nodefaults') 55 source_vm.launch() 56 source_vm.qmp('migrate', uri=src_uri) 57 self.assert_migration(source_vm, dest_vm) 58 59 def _get_free_port(self): 60 port = ports.find_free_port() 61 if port is None: 62 self.cancel('Failed to find a free port') 63 return port 64 65 66 def test_migration_with_tcp_localhost(self): 67 dest_uri = 'tcp:localhost:%u' % self._get_free_port() 68 self.do_migrate(dest_uri) 69 70 def test_migration_with_unix(self): 71 with tempfile.TemporaryDirectory(prefix='socket_') as socket_path: 72 dest_uri = 'unix:%s/qemu-test.sock' % socket_path 73 self.do_migrate(dest_uri) 74 75 @skipUnless(find_command('nc', default=False), "'nc' command not found") 76 def test_migration_with_exec(self): 77 """The test works for both netcat-traditional and netcat-openbsd packages.""" 78 free_port = self._get_free_port() 79 dest_uri = 'exec:nc -l localhost %u' % free_port 80 src_uri = 'exec:nc localhost %u' % free_port 81 self.do_migrate(dest_uri, src_uri)