xref: /openbmc/qemu/tests/avocado/migration.py (revision f0ccce6a) 
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
13import tempfile
14from avocado_qemu import QemuSystemTest
15from avocado import skipUnless
16
17from avocado.utils.network import ports
18from avocado.utils import wait
19from avocado.utils.path import find_command
20
21
22class 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)
82