|
Packit Service |
a04d08 |
# This file is part of cloud-init. See LICENSE file for license information.
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
from cloudinit import distros
|
|
Packit Service |
a04d08 |
from cloudinit import util
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
from cloudinit.tests import helpers
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
import os
|
|
Packit Service |
751c4a |
import pytest
|
|
Packit Service |
a04d08 |
import shutil
|
|
Packit Service |
a04d08 |
import tempfile
|
|
Packit Service |
751c4a |
from unittest import mock
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
unknown_arch_info = {
|
|
Packit Service |
a04d08 |
'arches': ['default'],
|
|
Packit Service |
a04d08 |
'failsafe': {'primary': 'http://fs-primary-default',
|
|
Packit Service |
a04d08 |
'security': 'http://fs-security-default'}
|
|
Packit Service |
a04d08 |
}
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
package_mirrors = [
|
|
Packit Service |
a04d08 |
{'arches': ['i386', 'amd64'],
|
|
Packit Service |
a04d08 |
'failsafe': {'primary': 'http://fs-primary-intel',
|
|
Packit Service |
a04d08 |
'security': 'http://fs-security-intel'},
|
|
Packit Service |
a04d08 |
'search': {
|
|
Packit Service |
a04d08 |
'primary': ['http://%(ec2_region)s.ec2/',
|
|
Packit Service |
a04d08 |
'http://%(availability_zone)s.clouds/'],
|
|
Packit Service |
a04d08 |
'security': ['http://security-mirror1-intel',
|
|
Packit Service |
a04d08 |
'http://security-mirror2-intel']}},
|
|
Packit Service |
a04d08 |
{'arches': ['armhf', 'armel'],
|
|
Packit Service |
a04d08 |
'failsafe': {'primary': 'http://fs-primary-arm',
|
|
Packit Service |
a04d08 |
'security': 'http://fs-security-arm'}},
|
|
Packit Service |
a04d08 |
unknown_arch_info
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
gpmi = distros._get_package_mirror_info
|
|
Packit Service |
a04d08 |
gapmi = distros._get_arch_package_mirror_info
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
class TestGenericDistro(helpers.FilesystemMockingTestCase):
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def setUp(self):
|
|
Packit Service |
a04d08 |
super(TestGenericDistro, self).setUp()
|
|
Packit Service |
a04d08 |
# Make a temp directoy for tests to use.
|
|
Packit Service |
a04d08 |
self.tmp = tempfile.mkdtemp()
|
|
Packit Service |
a04d08 |
self.addCleanup(shutil.rmtree, self.tmp)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _write_load_sudoers(self, _user, rules):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
os.makedirs(os.path.join(self.tmp, "etc"))
|
|
Packit Service |
a04d08 |
os.makedirs(os.path.join(self.tmp, "etc", 'sudoers.d'))
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
d.write_sudo_rules("harlowja", rules)
|
|
Packit Service |
a04d08 |
contents = util.load_file(d.ci_sudoers_fn)
|
|
Packit Service |
a04d08 |
return contents
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def _count_in(self, lines_look_for, text_content):
|
|
Packit Service |
a04d08 |
found_amount = 0
|
|
Packit Service |
a04d08 |
for e in lines_look_for:
|
|
Packit Service |
a04d08 |
for line in text_content.splitlines():
|
|
Packit Service |
a04d08 |
line = line.strip()
|
|
Packit Service |
a04d08 |
if line == e:
|
|
Packit Service |
a04d08 |
found_amount += 1
|
|
Packit Service |
a04d08 |
return found_amount
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sudoers_ensure_rules(self):
|
|
Packit Service |
a04d08 |
rules = 'ALL=(ALL:ALL) ALL'
|
|
Packit Service |
a04d08 |
contents = self._write_load_sudoers('harlowja', rules)
|
|
Packit Service |
a04d08 |
expected = ['harlowja ALL=(ALL:ALL) ALL']
|
|
Packit Service |
a04d08 |
self.assertEqual(len(expected), self._count_in(expected, contents))
|
|
Packit Service |
a04d08 |
not_expected = [
|
|
Packit Service |
a04d08 |
'harlowja A',
|
|
Packit Service |
a04d08 |
'harlowja L',
|
|
Packit Service |
a04d08 |
'harlowja L',
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
self.assertEqual(0, self._count_in(not_expected, contents))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sudoers_ensure_rules_list(self):
|
|
Packit Service |
a04d08 |
rules = [
|
|
Packit Service |
a04d08 |
'ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
'B-ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
'C-ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
contents = self._write_load_sudoers('harlowja', rules)
|
|
Packit Service |
a04d08 |
expected = [
|
|
Packit Service |
a04d08 |
'harlowja ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
'harlowja B-ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
'harlowja C-ALL=(ALL:ALL) ALL',
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
self.assertEqual(len(expected), self._count_in(expected, contents))
|
|
Packit Service |
a04d08 |
not_expected = [
|
|
Packit Service |
a04d08 |
'harlowja A',
|
|
Packit Service |
a04d08 |
'harlowja L',
|
|
Packit Service |
a04d08 |
'harlowja L',
|
|
Packit Service |
a04d08 |
]
|
|
Packit Service |
a04d08 |
self.assertEqual(0, self._count_in(not_expected, contents))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sudoers_ensure_new(self):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
d.ensure_sudo_dir("/b")
|
|
Packit Service |
a04d08 |
contents = util.load_file("/etc/sudoers")
|
|
Packit Service |
a04d08 |
self.assertIn("includedir /b", contents)
|
|
Packit Service |
a04d08 |
self.assertTrue(os.path.isdir("/b"))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_sudoers_ensure_append(self):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
util.write_file("/etc/sudoers", "josh, josh\n")
|
|
Packit Service |
a04d08 |
d.ensure_sudo_dir("/b")
|
|
Packit Service |
a04d08 |
contents = util.load_file("/etc/sudoers")
|
|
Packit Service |
a04d08 |
self.assertIn("includedir /b", contents)
|
|
Packit Service |
a04d08 |
self.assertTrue(os.path.isdir("/b"))
|
|
Packit Service |
a04d08 |
self.assertIn("josh", contents)
|
|
Packit Service |
a04d08 |
self.assertEqual(2, contents.count("josh"))
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_arch_package_mirror_info_unknown(self):
|
|
Packit Service |
a04d08 |
"""for an unknown arch, we should get back that with arch 'default'."""
|
|
Packit Service |
a04d08 |
arch_mirrors = gapmi(package_mirrors, arch="unknown")
|
|
Packit Service |
a04d08 |
self.assertEqual(unknown_arch_info, arch_mirrors)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_arch_package_mirror_info_known(self):
|
|
Packit Service |
a04d08 |
arch_mirrors = gapmi(package_mirrors, arch="amd64")
|
|
Packit Service |
a04d08 |
self.assertEqual(package_mirrors[0], arch_mirrors)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_systemd_in_use(self):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
os.makedirs('/run/systemd/system')
|
|
Packit Service |
a04d08 |
self.assertTrue(d.uses_systemd())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_systemd_not_in_use(self):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
self.assertFalse(d.uses_systemd())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_systemd_symlink(self):
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
self.patchOS(self.tmp)
|
|
Packit Service |
a04d08 |
self.patchUtils(self.tmp)
|
|
Packit Service |
a04d08 |
os.makedirs('/run/systemd')
|
|
Packit Service |
a04d08 |
os.symlink('/', '/run/systemd/system')
|
|
Packit Service |
a04d08 |
self.assertFalse(d.uses_systemd())
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
@mock.patch('cloudinit.distros.debian.read_system_locale')
|
|
Packit Service |
a04d08 |
def test_get_locale_ubuntu(self, m_locale):
|
|
Packit Service |
a04d08 |
"""Test ubuntu distro returns locale set to C.UTF-8"""
|
|
Packit Service |
a04d08 |
m_locale.return_value = 'C.UTF-8'
|
|
Packit Service |
a04d08 |
cls = distros.fetch("ubuntu")
|
|
Packit Service |
a04d08 |
d = cls("ubuntu", {}, None)
|
|
Packit Service |
a04d08 |
locale = d.get_locale()
|
|
Packit Service |
a04d08 |
self.assertEqual('C.UTF-8', locale)
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_get_locale_rhel(self):
|
|
Packit Service |
a04d08 |
"""Test rhel distro returns NotImplementedError exception"""
|
|
Packit Service |
a04d08 |
cls = distros.fetch("rhel")
|
|
Packit Service |
a04d08 |
d = cls("rhel", {}, None)
|
|
Packit Service |
a04d08 |
with self.assertRaises(NotImplementedError):
|
|
Packit Service |
a04d08 |
d.get_locale()
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_expire_passwd_uses_chpasswd(self):
|
|
Packit Service |
a04d08 |
"""Test ubuntu.expire_passwd uses the passwd command."""
|
|
Packit Service |
a04d08 |
for d_name in ("ubuntu", "rhel"):
|
|
Packit Service |
a04d08 |
cls = distros.fetch(d_name)
|
|
Packit Service |
a04d08 |
d = cls(d_name, {}, None)
|
|
Packit Service |
751c4a |
with mock.patch("cloudinit.subp.subp") as m_subp:
|
|
Packit Service |
a04d08 |
d.expire_passwd("myuser")
|
|
Packit Service |
a04d08 |
m_subp.assert_called_once_with(["passwd", "--expire", "myuser"])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
def test_expire_passwd_freebsd_uses_pw_command(self):
|
|
Packit Service |
a04d08 |
"""Test FreeBSD.expire_passwd uses the pw command."""
|
|
Packit Service |
a04d08 |
cls = distros.fetch("freebsd")
|
|
Packit Service |
a04d08 |
d = cls("freebsd", {}, None)
|
|
Packit Service |
751c4a |
with mock.patch("cloudinit.subp.subp") as m_subp:
|
|
Packit Service |
a04d08 |
d.expire_passwd("myuser")
|
|
Packit Service |
a04d08 |
m_subp.assert_called_once_with(
|
|
Packit Service |
a04d08 |
["pw", "usermod", "myuser", "-p", "01-Jan-1970"])
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
a04d08 |
|
|
Packit Service |
751c4a |
class TestGetPackageMirrors:
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def return_first(self, mlist):
|
|
Packit Service |
751c4a |
if not mlist:
|
|
Packit Service |
751c4a |
return None
|
|
Packit Service |
751c4a |
return mlist[0]
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def return_second(self, mlist):
|
|
Packit Service |
751c4a |
if not mlist:
|
|
Packit Service |
751c4a |
return None
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
return mlist[1] if len(mlist) > 1 else None
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def return_none(self, _mlist):
|
|
Packit Service |
751c4a |
return None
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def return_last(self, mlist):
|
|
Packit Service |
751c4a |
if not mlist:
|
|
Packit Service |
751c4a |
return None
|
|
Packit Service |
751c4a |
return(mlist[-1])
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
@pytest.mark.parametrize(
|
|
Packit Service |
751c4a |
"allow_ec2_mirror, platform_type, mirrors",
|
|
Packit Service |
751c4a |
[
|
|
Packit Service |
751c4a |
(True, "ec2", [
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1.ec2/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'},
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1a.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
]),
|
|
Packit Service |
751c4a |
(True, "other", [
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1.ec2/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'},
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1a.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
]),
|
|
Packit Service |
751c4a |
(False, "ec2", [
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1.ec2/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'},
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1a.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
]),
|
|
Packit Service |
751c4a |
(False, "other", [
|
|
Packit Service |
751c4a |
{'primary': 'http://us-east-1a.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'},
|
|
Packit Service |
751c4a |
{'primary': 'http://fs-primary-intel',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
])
|
|
Packit Service |
751c4a |
])
|
|
Packit Service |
751c4a |
def test_get_package_mirror_info_az_ec2(self,
|
|
Packit Service |
751c4a |
allow_ec2_mirror,
|
|
Packit Service |
751c4a |
platform_type,
|
|
Packit Service |
751c4a |
mirrors):
|
|
Packit Service |
751c4a |
flag_path = "cloudinit.distros." \
|
|
Packit Service |
751c4a |
"ALLOW_EC2_MIRRORS_ON_NON_AWS_INSTANCE_TYPES"
|
|
Packit Service |
751c4a |
with mock.patch(flag_path, allow_ec2_mirror):
|
|
Packit Service |
751c4a |
arch_mirrors = gapmi(package_mirrors, arch="amd64")
|
|
Packit Service |
751c4a |
data_source_mock = mock.Mock(
|
|
Packit Service |
751c4a |
availability_zone="us-east-1a",
|
|
Packit Service |
751c4a |
platform_type=platform_type)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_first)
|
|
Packit Service |
751c4a |
assert(results == mirrors[0])
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_second)
|
|
Packit Service |
751c4a |
assert(results == mirrors[1])
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_none)
|
|
Packit Service |
751c4a |
assert(results == package_mirrors[0]['failsafe'])
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def test_get_package_mirror_info_az_non_ec2(self):
|
|
Packit Service |
751c4a |
arch_mirrors = gapmi(package_mirrors, arch="amd64")
|
|
Packit Service |
751c4a |
data_source_mock = mock.Mock(availability_zone="nova.cloudvendor")
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_first)
|
|
Packit Service |
751c4a |
assert(results == {
|
|
Packit Service |
751c4a |
'primary': 'http://nova.cloudvendor.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'}
|
|
Packit Service |
751c4a |
)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_last)
|
|
Packit Service |
751c4a |
assert(results == {
|
|
Packit Service |
751c4a |
'primary': 'http://nova.cloudvendor.clouds/',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
def test_get_package_mirror_info_none(self):
|
|
Packit Service |
751c4a |
arch_mirrors = gapmi(package_mirrors, arch="amd64")
|
|
Packit Service |
751c4a |
data_source_mock = mock.Mock(availability_zone=None)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
# because both search entries here replacement based on
|
|
Packit Service |
751c4a |
# availability-zone, the filter will be called with an empty list and
|
|
Packit Service |
751c4a |
# failsafe should be taken.
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_first)
|
|
Packit Service |
751c4a |
assert(results == {
|
|
Packit Service |
751c4a |
'primary': 'http://fs-primary-intel',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror1-intel'}
|
|
Packit Service |
751c4a |
)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
751c4a |
results = gpmi(arch_mirrors, data_source=data_source_mock,
|
|
Packit Service |
751c4a |
mirror_filter=self.return_last)
|
|
Packit Service |
751c4a |
assert(results == {
|
|
Packit Service |
751c4a |
'primary': 'http://fs-primary-intel',
|
|
Packit Service |
751c4a |
'security': 'http://security-mirror2-intel'}
|
|
Packit Service |
751c4a |
)
|
|
Packit Service |
751c4a |
|
|
Packit Service |
a04d08 |
# vi: ts=4 expandtab
|