From 6b7e65e4f57902c25363c78a7e47aa2caa579b7b Mon Sep 17 00:00:00 2001 From: Ben Howard Date: Thu, 18 Jul 2013 15:37:18 -0600 Subject: Added SmartOS datasource and unit tests. --- tests/unittests/test_datasource/test_smartos.py | 191 ++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 tests/unittests/test_datasource/test_smartos.py (limited to 'tests/unittests') diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py new file mode 100644 index 00000000..494f9828 --- /dev/null +++ b/tests/unittests/test_datasource/test_smartos.py @@ -0,0 +1,191 @@ +# vi: ts=4 expandtab +# +# Copyright (C) 2013 Canonical Ltd. +# +# Author: Ben Howard +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# +# This is a testcase for the SmartOS datasource. It replicates a serial +# console and acts like the SmartOS console does in order to validate +# return responses. +# + +from cloudinit import helpers +from cloudinit.sources import DataSourceSmartOS + +from mocker import MockerTestCase +import uuid + +mock_returns = { + 'hostname': 'test-host', + 'root_authorized_keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname', + 'disable_iptables_flag': False, + 'enable_motd_sys_info': False, + 'system_uuid': str(uuid.uuid4()), + 'smartdc': 'smartdc', + 'userdata': """ +#!/bin/sh +/bin/true +""", +} + + +class MockSerial(object): + """Fake a serial terminal for testing the code that + interfaces with the serial""" + + port = None + + def __init__(self): + self.last = None + self.last = None + self.new = True + self.count = 0 + self.mocked_out = [] + + def open(self): + return True + + def close(self): + return True + + def isOpen(self): + return True + + def write(self, line): + line = line.replace('GET ', '') + self.last = line.rstrip() + + def readline(self): + if self.new: + self.new = False + if self.last in mock_returns: + return 'SUCCESS\n' + else: + return 'NOTFOUND %s\n' % self.last + + if self.last in mock_returns: + if not self.mocked_out: + self.mocked_out = [x for x in self._format_out()] + print self.mocked_out + + if len(self.mocked_out) > self.count: + self.count += 1 + return self.mocked_out[self.count - 1] + + def _format_out(self): + if self.last in mock_returns: + try: + for l in mock_returns[self.last].splitlines(): + yield "%s\n" % l + except: + yield "%s\n" % mock_returns[self.last] + + yield '\n' + yield '.' + + +class TestSmartOSDataSource(MockerTestCase): + def setUp(self): + # makeDir comes from MockerTestCase + self.tmp = self.makeDir() + + # patch cloud_dir, so our 'seed_dir' is guaranteed empty + self.paths = helpers.Paths({'cloud_dir': self.tmp}) + + self.unapply = [] + super(TestSmartOSDataSource, self).setUp() + + def tearDown(self): + apply_patches([i for i in reversed(self.unapply)]) + super(TestSmartOSDataSource, self).tearDown() + + def apply_patches(self, patches): + ret = apply_patches(patches) + self.unapply += ret + + def _get_ds(self): + + def _get_serial(): + return MockSerial() + + def _dmi_data(): + return mock_returns['system_uuid'], 'smartdc' + + data = {'sys_cfg': {}} + mod = DataSourceSmartOS + self.apply_patches([(mod, 'get_serial', _get_serial)]) + self.apply_patches([(mod, 'dmi_data', _dmi_data)]) + dsrc = mod.DataSourceSmartOS( + data.get('sys_cfg', {}), distro=None, paths=self.paths) + return dsrc + + def test_seed(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals('/dev/ttyS1', dsrc.seed) + + def test_issmartdc(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertTrue(dsrc.is_smartdc) + + def test_uuid(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(mock_returns['system_uuid'], + dsrc.metadata['instance-id']) + + def test_root_keys(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(mock_returns['root_authorized_keys'], + dsrc.metadata['public-keys']) + + def test_hostname(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(mock_returns['hostname'], + dsrc.metadata['local-hostname']) + + def test_disable_iptables_flag(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(str(mock_returns['disable_iptables_flag']), + dsrc.metadata['iptables_disable']) + + def test_motd_sys_info(self): + dsrc = self._get_ds() + ret = dsrc.get_data() + self.assertTrue(ret) + self.assertEquals(str(mock_returns['enable_motd_sys_info']), + dsrc.metadata['motd_sys_info']) + + +def apply_patches(patches): + ret = [] + for (ref, name, replace) in patches: + if replace is None: + continue + orig = getattr(ref, name) + setattr(ref, name, replace) + ret.append((ref, name, orig)) + return ret -- cgit v1.2.3 From 4b41f7dc3d37d5bf7397bbc34d8a5e0c56798ac7 Mon Sep 17 00:00:00 2001 From: Ben Howard Date: Tue, 23 Jul 2013 16:33:46 -0600 Subject: Changed get_serial to be fully parameterized and return the serial initialized. Added a mapping of attributes between cloud-init and smartos. --- cloudinit/sources/DataSourceSmartOS.py | 64 ++++++++++++++----------- cloudinit/util.py | 5 +- tests/unittests/test_datasource/test_smartos.py | 10 ++-- 3 files changed, 43 insertions(+), 36 deletions(-) (limited to 'tests/unittests') diff --git a/cloudinit/sources/DataSourceSmartOS.py b/cloudinit/sources/DataSourceSmartOS.py index 129020ec..d6589f57 100644 --- a/cloudinit/sources/DataSourceSmartOS.py +++ b/cloudinit/sources/DataSourceSmartOS.py @@ -27,25 +27,37 @@ # -import os -import os.path -import serial from cloudinit import log as logging from cloudinit import sources from cloudinit import util +import os +import os.path +import serial DEF_TTY_LOC = '/dev/ttyS1' -TTY_LOC = None +DEF_TTY_TIMEOUT = 60 LOG = logging.getLogger(__name__) +SMARTOS_ATTRIB_MAP = { + #Cloud-init Key : (SmartOS Key, Strip line endings) + 'local-hostname': ('hostname', True), + 'public-keys': ('root_authorized_keys', True), + 'user-script': ('user-script', False), + 'user-data': ('user-data', False), + 'iptables_disable': ('iptables_disable', True), + 'motd_sys_info': ('motd_sys_info', True), +} + class DataSourceSmartOS(sources.DataSource): def __init__(self, sys_cfg, distro, paths): sources.DataSource.__init__(self, sys_cfg, distro, paths) self.seed_dir = os.path.join(paths.seed_dir, 'sdc') - self.seed = None self.is_smartdc = None + self.seed = self.sys_cfg.get("serial_device", DEF_TTY_LOC) + self.seed_timeout = self.sys_cfg.get("serial_timeout", + DEF_TTY_TIMEOUT) def __str__(self): root = sources.DataSource.__str__(self) @@ -55,30 +67,25 @@ class DataSourceSmartOS(sources.DataSource): md = {} ud = "" - TTY_LOC = self.sys_cfg.get("serial_device", DEF_TTY_LOC) - if not os.path.exists(TTY_LOC): + if not os.path.exists(self.seed): LOG.debug("Host does not appear to be on SmartOS") return False - self.seed = TTY_LOC + self.seed = self.seed system_uuid, system_type = dmi_data() if 'smartdc' not in system_type.lower(): LOG.debug("Host is not on SmartOS") return False self.is_smartdc = True + md['instance-id'] = system_uuid - hostname = query_data("hostname", strip=True) - if not hostname: - hostname = system_uuid + for ci_noun, attribute in SMARTOS_ATTRIB_MAP.iteritems(): + smartos_noun, strip = attribute + md[ci_noun] = query_data(smartos_noun, self.seed, + self.seed_timeout, strip=strip) - md['local-hostname'] = hostname - md['instance-id'] = system_uuid - md['public-keys'] = query_data("root_authorized_keys", strip=True) - md['user-script'] = query_data("user-script") - md['user-data'] = query_data("user-script") - md['iptables_disable'] = query_data("disable_iptables_flag", - strip=True) - md['motd_sys_info'] = query_data("enable_motd_sys_info", strip=True) + if not md['local-hostname']: + md['local-hostname'] = system_uuid if md['user-data']: ud = md['user-data'] @@ -93,7 +100,7 @@ class DataSourceSmartOS(sources.DataSource): return self.metadata['instance-id'] -def get_serial(): +def get_serial(seed_device, seed_timeout): """This is replaced in unit testing, allowing us to replace serial.Serial with a mocked class @@ -102,18 +109,17 @@ def get_serial(): each line individually up until the single ".", the transfer is usually very fast (i.e. microseconds) to get the response. """ - if not TTY_LOC: - raise AttributeError("TTY_LOC value is not set") - - _ret = serial.Serial(TTY_LOC, timeout=60) - if not _ret.isOpen(): - raise SystemError("Unable to open %s" % TTY_LOC) + if not seed_device: + raise AttributeError("seed_device value is not set") - return _ret + ser = serial.Serial(seed_device, timeout=seed_timeout) + if not ser.isOpen(): + raise SystemError("Unable to open %s" % seed_device) + return ser -def query_data(noun, strip=False): +def query_data(noun, seed_device, seed_timeout, strip=False): """Makes a request to via the serial console via "GET " In the response, the first line is the status, while subsequent lines @@ -124,7 +130,7 @@ def query_data(noun, strip=False): if not noun: return False - ser = get_serial() + ser = get_serial(seed_device, seed_timeout) ser.write("GET %s\n" % noun.rstrip()) status = str(ser.readline()).rstrip() response = [] diff --git a/cloudinit/util.py b/cloudinit/util.py index 7163225f..a2fbc004 100644 --- a/cloudinit/util.py +++ b/cloudinit/util.py @@ -1744,13 +1744,14 @@ def get_mount_info(path, log=LOG): lines = load_file(mountinfo_path).splitlines() return parse_mount_info(path, lines, log) + def which(program): # Return path of program for execution if found in path def is_exe(fpath): return os.path.isfile(fpath) and os.access(fpath, os.X_OK) - fpath, fname = os.path.split(program) - if fpath: + _fpath, _ = os.path.split(program) + if _fpath: if is_exe(program): return program else: diff --git a/tests/unittests/test_datasource/test_smartos.py b/tests/unittests/test_datasource/test_smartos.py index 494f9828..6c12f1e2 100644 --- a/tests/unittests/test_datasource/test_smartos.py +++ b/tests/unittests/test_datasource/test_smartos.py @@ -31,8 +31,8 @@ import uuid mock_returns = { 'hostname': 'test-host', 'root_authorized_keys': 'ssh-rsa AAAAB3Nz...aC1yc2E= keyname', - 'disable_iptables_flag': False, - 'enable_motd_sys_info': False, + 'disable_iptables_flag': None, + 'enable_motd_sys_info': None, 'system_uuid': str(uuid.uuid4()), 'smartdc': 'smartdc', 'userdata': """ @@ -118,7 +118,7 @@ class TestSmartOSDataSource(MockerTestCase): def _get_ds(self): - def _get_serial(): + def _get_serial(*_): return MockSerial() def _dmi_data(): @@ -169,14 +169,14 @@ class TestSmartOSDataSource(MockerTestCase): dsrc = self._get_ds() ret = dsrc.get_data() self.assertTrue(ret) - self.assertEquals(str(mock_returns['disable_iptables_flag']), + self.assertEquals(mock_returns['disable_iptables_flag'], dsrc.metadata['iptables_disable']) def test_motd_sys_info(self): dsrc = self._get_ds() ret = dsrc.get_data() self.assertTrue(ret) - self.assertEquals(str(mock_returns['enable_motd_sys_info']), + self.assertEquals(mock_returns['enable_motd_sys_info'], dsrc.metadata['motd_sys_info']) -- cgit v1.2.3