1# This file is part of cloud-init. See LICENSE file for license information.
2from enum import Enum
3import logging
4import os
5import uuid
6from tempfile import NamedTemporaryFile
7
8from pycloudlib.instance import BaseInstance
9from pycloudlib.result import Result
10
11from tests.integration_tests import integration_settings
12from tests.integration_tests.util import retry
13
14try:
15    from typing import TYPE_CHECKING
16    if TYPE_CHECKING:
17        from tests.integration_tests.clouds import (  # noqa: F401
18            IntegrationCloud
19        )
20except ImportError:
21    pass
22
23
24log = logging.getLogger('integration_testing')
25
26
27def _get_tmp_path():
28    tmp_filename = str(uuid.uuid4())
29    return '/var/tmp/{}.tmp'.format(tmp_filename)
30
31
32class CloudInitSource(Enum):
33    """Represents the cloud-init image source setting as a defined value.
34
35    Values here represent all possible values for CLOUD_INIT_SOURCE in
36    tests/integration_tests/integration_settings.py. See that file for an
37    explanation of these values. If the value set there can't be parsed into
38    one of these values, an exception will be raised
39    """
40    NONE = 1
41    IN_PLACE = 2
42    PROPOSED = 3
43    PPA = 4
44    DEB_PACKAGE = 5
45    UPGRADE = 6
46
47    def installs_new_version(self):
48        if self.name in [self.NONE.name, self.IN_PLACE.name]:
49            return False
50        return True
51
52
53class IntegrationInstance:
54    def __init__(self, cloud: 'IntegrationCloud', instance: BaseInstance,
55                 settings=integration_settings):
56        self.cloud = cloud
57        self.instance = instance
58        self.settings = settings
59
60    def destroy(self):
61        self.instance.delete()
62
63    def restart(self):
64        """Restart this instance (via cloud mechanism) and wait for boot.
65
66        This wraps pycloudlib's `BaseInstance.restart`
67        """
68        log.info("Restarting instance and waiting for boot")
69        self.instance.restart()
70
71    def execute(self, command, *, use_sudo=True) -> Result:
72        if self.instance.username == 'root' and use_sudo is False:
73            raise Exception('Root user cannot run unprivileged')
74        return self.instance.execute(command, use_sudo=use_sudo)
75
76    def pull_file(self, remote_path, local_path):
77        # First copy to a temporary directory because of permissions issues
78        tmp_path = _get_tmp_path()
79        self.instance.execute('cp {} {}'.format(str(remote_path), tmp_path))
80        self.instance.pull_file(tmp_path, str(local_path))
81
82    def push_file(self, local_path, remote_path):
83        # First push to a temporary directory because of permissions issues
84        tmp_path = _get_tmp_path()
85        self.instance.push_file(str(local_path), tmp_path)
86        self.execute('mv {} {}'.format(tmp_path, str(remote_path)))
87
88    def read_from_file(self, remote_path) -> str:
89        result = self.execute('cat {}'.format(remote_path))
90        if result.failed:
91            # TODO: Raise here whatever pycloudlib raises when it has
92            # a consistent error response
93            raise IOError(
94                'Failed reading remote file via cat: {}\n'
95                'Return code: {}\n'
96                'Stderr: {}\n'
97                'Stdout: {}'.format(
98                    remote_path, result.return_code,
99                    result.stderr, result.stdout)
100            )
101        return result.stdout
102
103    def write_to_file(self, remote_path, contents: str):
104        # Writes file locally and then pushes it rather
105        # than writing the file directly on the instance
106        with NamedTemporaryFile('w', delete=False) as tmp_file:
107            tmp_file.write(contents)
108
109        try:
110            self.push_file(tmp_file.name, remote_path)
111        finally:
112            os.unlink(tmp_file.name)
113
114    def snapshot(self):
115        image_id = self.cloud.snapshot(self.instance)
116        log.info('Created new image: %s', image_id)
117        return image_id
118
119    def install_new_cloud_init(
120        self,
121        source: CloudInitSource,
122        take_snapshot=True,
123        clean=True,
124    ):
125        if source == CloudInitSource.DEB_PACKAGE:
126            self.install_deb()
127        elif source == CloudInitSource.PPA:
128            self.install_ppa()
129        elif source == CloudInitSource.PROPOSED:
130            self.install_proposed_image()
131        elif source == CloudInitSource.UPGRADE:
132            self.upgrade_cloud_init()
133        else:
134            raise Exception(
135                "Specified to install {} which isn't supported here".format(
136                    source)
137            )
138        version = self.execute('cloud-init -v').split()[-1]
139        log.info('Installed cloud-init version: %s', version)
140        if clean:
141            self.instance.clean()
142        if take_snapshot:
143            snapshot_id = self.snapshot()
144            self.cloud.snapshot_id = snapshot_id
145
146    # assert with retry because we can compete with apt already running in the
147    # background and get: E: Could not get lock /var/lib/apt/lists/lock - open
148    # (11: Resource temporarily unavailable)
149
150    @retry(tries=30, delay=1)
151    def install_proposed_image(self):
152        log.info('Installing proposed image')
153        assert self.execute(
154            'echo deb "http://archive.ubuntu.com/ubuntu '
155            '$(lsb_release -sc)-proposed main" >> '
156            '/etc/apt/sources.list.d/proposed.list'
157        ).ok
158        assert self.execute('apt-get update -q').ok
159        assert self.execute('apt-get install -qy cloud-init').ok
160
161    @retry(tries=30, delay=1)
162    def install_ppa(self):
163        log.info('Installing PPA')
164        assert self.execute('add-apt-repository {} -y'.format(
165            self.settings.CLOUD_INIT_SOURCE)
166        ).ok
167        assert self.execute('apt-get update -q').ok
168        assert self.execute('apt-get install -qy cloud-init').ok
169
170    @retry(tries=30, delay=1)
171    def install_deb(self):
172        log.info('Installing deb package')
173        deb_path = integration_settings.CLOUD_INIT_SOURCE
174        deb_name = os.path.basename(deb_path)
175        remote_path = '/var/tmp/{}'.format(deb_name)
176        self.push_file(
177            local_path=integration_settings.CLOUD_INIT_SOURCE,
178            remote_path=remote_path)
179        assert self.execute('dpkg -i {path}'.format(path=remote_path)).ok
180
181    @retry(tries=30, delay=1)
182    def upgrade_cloud_init(self):
183        log.info('Upgrading cloud-init to latest version in archive')
184        assert self.execute("apt-get update -q").ok
185        assert self.execute("apt-get install -qy cloud-init").ok
186
187    def __enter__(self):
188        return self
189
190    def __exit__(self, exc_type, exc_val, exc_tb):
191        if not self.settings.KEEP_INSTANCE:
192            self.destroy()
193
194
195class IntegrationEc2Instance(IntegrationInstance):
196    pass
197
198
199class IntegrationGceInstance(IntegrationInstance):
200    pass
201
202
203class IntegrationAzureInstance(IntegrationInstance):
204    pass
205
206
207class IntegrationOciInstance(IntegrationInstance):
208    pass
209
210
211class IntegrationLxdInstance(IntegrationInstance):
212    pass
213