summaryrefslogtreecommitdiff
path: root/cloudinit/tests
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit/tests')
-rw-r--r--cloudinit/tests/test_dhclient_hook.py105
-rw-r--r--cloudinit/tests/test_url_helper.py25
-rw-r--r--cloudinit/tests/test_util.py99
3 files changed, 211 insertions, 18 deletions
diff --git a/cloudinit/tests/test_dhclient_hook.py b/cloudinit/tests/test_dhclient_hook.py
new file mode 100644
index 00000000..7aab8dd5
--- /dev/null
+++ b/cloudinit/tests/test_dhclient_hook.py
@@ -0,0 +1,105 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+"""Tests for cloudinit.dhclient_hook."""
+
+from cloudinit import dhclient_hook as dhc
+from cloudinit.tests.helpers import CiTestCase, dir2dict, populate_dir
+
+import argparse
+import json
+import mock
+import os
+
+
+class TestDhclientHook(CiTestCase):
+
+ ex_env = {
+ 'interface': 'eth0',
+ 'new_dhcp_lease_time': '3600',
+ 'new_host_name': 'x1',
+ 'new_ip_address': '10.145.210.163',
+ 'new_subnet_mask': '255.255.255.0',
+ 'old_host_name': 'x1',
+ 'PATH': '/usr/sbin:/usr/bin:/sbin:/bin',
+ 'pid': '614',
+ 'reason': 'BOUND',
+ }
+
+ # some older versions of dhclient put the same content,
+ # but in upper case with DHCP4_ instead of new_
+ ex_env_dhcp4 = {
+ 'REASON': 'BOUND',
+ 'DHCP4_dhcp_lease_time': '3600',
+ 'DHCP4_host_name': 'x1',
+ 'DHCP4_ip_address': '10.145.210.163',
+ 'DHCP4_subnet_mask': '255.255.255.0',
+ 'INTERFACE': 'eth0',
+ 'PATH': '/usr/sbin:/usr/bin:/sbin:/bin',
+ 'pid': '614',
+ }
+
+ expected = {
+ 'dhcp_lease_time': '3600',
+ 'host_name': 'x1',
+ 'ip_address': '10.145.210.163',
+ 'subnet_mask': '255.255.255.0'}
+
+ def setUp(self):
+ super(TestDhclientHook, self).setUp()
+ self.tmp = self.tmp_dir()
+
+ def test_handle_args(self):
+ """quick test of call to handle_args."""
+ nic = 'eth0'
+ args = argparse.Namespace(event=dhc.UP, interface=nic)
+ with mock.patch.dict("os.environ", clear=True, values=self.ex_env):
+ dhc.handle_args(dhc.NAME, args, data_d=self.tmp)
+ found = dir2dict(self.tmp + os.path.sep)
+ self.assertEqual([nic + ".json"], list(found.keys()))
+ self.assertEqual(self.expected, json.loads(found[nic + ".json"]))
+
+ def test_run_hook_up_creates_dir(self):
+ """If dir does not exist, run_hook should create it."""
+ subd = self.tmp_path("subdir", self.tmp)
+ nic = 'eth1'
+ dhc.run_hook(nic, 'up', data_d=subd, env=self.ex_env)
+ self.assertEqual(
+ set([nic + ".json"]), set(dir2dict(subd + os.path.sep)))
+
+ def test_run_hook_up(self):
+ """Test expected use of run_hook_up."""
+ nic = 'eth0'
+ dhc.run_hook(nic, 'up', data_d=self.tmp, env=self.ex_env)
+ found = dir2dict(self.tmp + os.path.sep)
+ self.assertEqual([nic + ".json"], list(found.keys()))
+ self.assertEqual(self.expected, json.loads(found[nic + ".json"]))
+
+ def test_run_hook_up_dhcp4_prefix(self):
+ """Test run_hook filters correctly with older DHCP4_ data."""
+ nic = 'eth0'
+ dhc.run_hook(nic, 'up', data_d=self.tmp, env=self.ex_env_dhcp4)
+ found = dir2dict(self.tmp + os.path.sep)
+ self.assertEqual([nic + ".json"], list(found.keys()))
+ self.assertEqual(self.expected, json.loads(found[nic + ".json"]))
+
+ def test_run_hook_down_deletes(self):
+ """down should delete the created json file."""
+ nic = 'eth1'
+ populate_dir(
+ self.tmp, {nic + ".json": "{'abcd'}", 'myfile.txt': 'text'})
+ dhc.run_hook(nic, 'down', data_d=self.tmp, env={'old_host_name': 'x1'})
+ self.assertEqual(
+ set(['myfile.txt']),
+ set(dir2dict(self.tmp + os.path.sep)))
+
+ def test_get_parser(self):
+ """Smoke test creation of get_parser."""
+ # cloud-init main uses 'action'.
+ event, interface = (dhc.UP, 'mynic0')
+ self.assertEqual(
+ argparse.Namespace(event=event, interface=interface,
+ action=(dhc.NAME, dhc.handle_args)),
+ dhc.get_parser().parse_args([event, interface]))
+
+
+# vi: ts=4 expandtab
diff --git a/cloudinit/tests/test_url_helper.py b/cloudinit/tests/test_url_helper.py
index 113249d9..aa9f3ec1 100644
--- a/cloudinit/tests/test_url_helper.py
+++ b/cloudinit/tests/test_url_helper.py
@@ -1,10 +1,12 @@
# This file is part of cloud-init. See LICENSE file for license information.
-from cloudinit.url_helper import oauth_headers, read_file_or_url
+from cloudinit.url_helper import (
+ NOT_FOUND, UrlError, oauth_headers, read_file_or_url, retry_on_url_exc)
from cloudinit.tests.helpers import CiTestCase, mock, skipIf
from cloudinit import util
import httpretty
+import requests
try:
@@ -64,3 +66,24 @@ class TestReadFileOrUrl(CiTestCase):
result = read_file_or_url(url)
self.assertEqual(result.contents, data)
self.assertEqual(str(result), data.decode('utf-8'))
+
+
+class TestRetryOnUrlExc(CiTestCase):
+
+ def test_do_not_retry_non_urlerror(self):
+ """When exception is not UrlError return False."""
+ myerror = IOError('something unexcpected')
+ self.assertFalse(retry_on_url_exc(msg='', exc=myerror))
+
+ def test_perform_retries_on_not_found(self):
+ """When exception is UrlError with a 404 status code return True."""
+ myerror = UrlError(cause=RuntimeError(
+ 'something was not found'), code=NOT_FOUND)
+ self.assertTrue(retry_on_url_exc(msg='', exc=myerror))
+
+ def test_perform_retries_on_timeout(self):
+ """When exception is a requests.Timout return True."""
+ myerror = UrlError(cause=requests.Timeout('something timed out'))
+ self.assertTrue(retry_on_url_exc(msg='', exc=myerror))
+
+# vi: ts=4 expandtab
diff --git a/cloudinit/tests/test_util.py b/cloudinit/tests/test_util.py
index edb0c18f..e3d2dbaa 100644
--- a/cloudinit/tests/test_util.py
+++ b/cloudinit/tests/test_util.py
@@ -18,25 +18,51 @@ MOUNT_INFO = [
]
OS_RELEASE_SLES = dedent("""\
- NAME="SLES"\n
- VERSION="12-SP3"\n
- VERSION_ID="12.3"\n
- PRETTY_NAME="SUSE Linux Enterprise Server 12 SP3"\n
- ID="sles"\nANSI_COLOR="0;32"\n
- CPE_NAME="cpe:/o:suse:sles:12:sp3"\n
+ NAME="SLES"
+ VERSION="12-SP3"
+ VERSION_ID="12.3"
+ PRETTY_NAME="SUSE Linux Enterprise Server 12 SP3"
+ ID="sles"
+ ANSI_COLOR="0;32"
+ CPE_NAME="cpe:/o:suse:sles:12:sp3"
""")
OS_RELEASE_OPENSUSE = dedent("""\
-NAME="openSUSE Leap"
-VERSION="42.3"
-ID=opensuse
-ID_LIKE="suse"
-VERSION_ID="42.3"
-PRETTY_NAME="openSUSE Leap 42.3"
-ANSI_COLOR="0;32"
-CPE_NAME="cpe:/o:opensuse:leap:42.3"
-BUG_REPORT_URL="https://bugs.opensuse.org"
-HOME_URL="https://www.opensuse.org/"
+ NAME="openSUSE Leap"
+ VERSION="42.3"
+ ID=opensuse
+ ID_LIKE="suse"
+ VERSION_ID="42.3"
+ PRETTY_NAME="openSUSE Leap 42.3"
+ ANSI_COLOR="0;32"
+ CPE_NAME="cpe:/o:opensuse:leap:42.3"
+ BUG_REPORT_URL="https://bugs.opensuse.org"
+ HOME_URL="https://www.opensuse.org/"
+""")
+
+OS_RELEASE_OPENSUSE_L15 = dedent("""\
+ NAME="openSUSE Leap"
+ VERSION="15.0"
+ ID="opensuse-leap"
+ ID_LIKE="suse opensuse"
+ VERSION_ID="15.0"
+ PRETTY_NAME="openSUSE Leap 15.0"
+ ANSI_COLOR="0;32"
+ CPE_NAME="cpe:/o:opensuse:leap:15.0"
+ BUG_REPORT_URL="https://bugs.opensuse.org"
+ HOME_URL="https://www.opensuse.org/"
+""")
+
+OS_RELEASE_OPENSUSE_TW = dedent("""\
+ NAME="openSUSE Tumbleweed"
+ ID="opensuse-tumbleweed"
+ ID_LIKE="opensuse suse"
+ VERSION_ID="20180920"
+ PRETTY_NAME="openSUSE Tumbleweed"
+ ANSI_COLOR="0;32"
+ CPE_NAME="cpe:/o:opensuse:tumbleweed:20180920"
+ BUG_REPORT_URL="https://bugs.opensuse.org"
+ HOME_URL="https://www.opensuse.org/"
""")
OS_RELEASE_CENTOS = dedent("""\
@@ -447,12 +473,35 @@ class TestGetLinuxDistro(CiTestCase):
@mock.patch('cloudinit.util.load_file')
def test_get_linux_opensuse(self, m_os_release, m_path_exists):
- """Verify we get the correct name and machine arch on OpenSUSE."""
+ """Verify we get the correct name and machine arch on openSUSE
+ prior to openSUSE Leap 15.
+ """
m_os_release.return_value = OS_RELEASE_OPENSUSE
m_path_exists.side_effect = TestGetLinuxDistro.os_release_exists
dist = util.get_linux_distro()
self.assertEqual(('opensuse', '42.3', platform.machine()), dist)
+ @mock.patch('cloudinit.util.load_file')
+ def test_get_linux_opensuse_l15(self, m_os_release, m_path_exists):
+ """Verify we get the correct name and machine arch on openSUSE
+ for openSUSE Leap 15.0 and later.
+ """
+ m_os_release.return_value = OS_RELEASE_OPENSUSE_L15
+ m_path_exists.side_effect = TestGetLinuxDistro.os_release_exists
+ dist = util.get_linux_distro()
+ self.assertEqual(('opensuse-leap', '15.0', platform.machine()), dist)
+
+ @mock.patch('cloudinit.util.load_file')
+ def test_get_linux_opensuse_tw(self, m_os_release, m_path_exists):
+ """Verify we get the correct name and machine arch on openSUSE
+ for openSUSE Tumbleweed
+ """
+ m_os_release.return_value = OS_RELEASE_OPENSUSE_TW
+ m_path_exists.side_effect = TestGetLinuxDistro.os_release_exists
+ dist = util.get_linux_distro()
+ self.assertEqual(
+ ('opensuse-tumbleweed', '20180920', platform.machine()), dist)
+
@mock.patch('platform.dist')
def test_get_linux_distro_no_data(self, m_platform_dist, m_path_exists):
"""Verify we get no information if os-release does not exist"""
@@ -478,4 +527,20 @@ class TestGetLinuxDistro(CiTestCase):
dist = util.get_linux_distro()
self.assertEqual(('foo', '1.1', 'aarch64'), dist)
+
+@mock.patch('os.path.exists')
+class TestIsLXD(CiTestCase):
+
+ def test_is_lxd_true_on_sock_device(self, m_exists):
+ """When lxd's /dev/lxd/sock exists, is_lxd returns true."""
+ m_exists.return_value = True
+ self.assertTrue(util.is_lxd())
+ m_exists.assert_called_once_with('/dev/lxd/sock')
+
+ def test_is_lxd_false_when_sock_device_absent(self, m_exists):
+ """When lxd's /dev/lxd/sock is absent, is_lxd returns false."""
+ m_exists.return_value = False
+ self.assertFalse(util.is_lxd())
+ m_exists.assert_called_once_with('/dev/lxd/sock')
+
# vi: ts=4 expandtab
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-10 13:40:12 +0000