charm-nova-cloud-controller/unit_tests/test_nova_cc_utils.py

from mock import patch, MagicMock
from copy import deepcopy
from unit_tests.test_utils import CharmTestCase, patch_open

from charmhelpers.core import hookenv

_conf = hookenv.config
hookenv.config = MagicMock()

import hooks.nova_cc_utils as utils

hookenv.config = _conf

TO_PATCH = [
    'config',
    'log',
    'network_manager',
    'os_release',
    'relation_ids',
    'remote_unit',
    '_save_script_rc',
]

SCRIPTRC_ENV_VARS = {
    'OPENSTACK_PORT_MCASTPORT': 5404,
    'OPENSTACK_SERVICE_API_EC2': 'nova-api-ec2',
    'OPENSTACK_SERVICE_API_OS_COMPUTE': 'nova-api-os-compute',
    'OPENSTACK_SERVICE_CERT': 'nova-cert',
    'OPENSTACK_SERVICE_CONDUCTOR': 'nova-conductor',
    'OPENSTACK_SERVICE_OBJECTSTORE': 'nova-objectstore',
    'OPENSTACK_SERVICE_SCHEDULER': 'nova-scheduler',
}


AUTHORIZED_KEYS = """
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC27Us7lSjCpa7bumXAgc nova-compute-1
ssh-rsa BBBBB3NzaC1yc2EBBBBDBQBBBBBBBQC27Us7lSjCpa7bumXBgc nova-compute-2
ssh-rsa CCCCB3NzaC1yc2ECCCCDCQCBCCCBCQC27Us7lSjCpa7bumXCgc nova-compute-3
"""

BASE_ENDPOINTS = {
    'ec2_admin_url': 'http://foohost.com:8773/services/Cloud',
    'ec2_internal_url': 'http://foohost.com:8773/services/Cloud',
    'ec2_public_url': 'http://foohost.com:8773/services/Cloud',
    'ec2_region': 'RegionOne',
    'ec2_service': 'ec2',
    'nova_admin_url': 'http://foohost.com:8774/v1.1/\$tenant_id)s',
    'nova_internal_url': 'http://foohost.com:8774/v1.1/\$tenant_id)s',
    'nova_public_url': 'http://foohost.com:8774/v1.1/\$tenant_id)s',
    'nova_region': 'RegionOne',
    'nova_service': 'nova',
    's3_admin_url': 'http://foohost.com:3333',
    's3_internal_url': 'http://foohost.com:3333',
    's3_public_url': 'http://foohost.com:3333',
    's3_region': 'RegionOne',
    's3_service': 's3'
}


class NovaCCUtilsTests(CharmTestCase):
    def setUp(self):
        super(NovaCCUtilsTests, self).setUp(utils, TO_PATCH)
        self.config.side_effect = self.test_config.get

    def test_resource_map_quantum(self):
        self.network_manager.return_value = 'quantum'
        self.relation_ids.return_value = []
        self.test_config.set('network-manager', 'Quantum')
        _map = utils.resource_map()
        confs = [
            '/etc/quantum/quantum.conf',
            '/etc/quantum/api-paste.ini'
        ]
        [self.assertIn(q_conf, _map.keys()) for q_conf in confs]

    def test_resource_map_neutron(self):
        self.network_manager.return_value = 'neutron'
        self.relation_ids.return_value = []
        self.test_config.set('network-manager', 'Quantum')
        _map = utils.resource_map()
        confs = [
            '/etc/neutron/neutron.conf',
        ]
        [self.assertIn(q_conf, _map.keys()) for q_conf in confs]

    def test_resource_map_nova_volume(self):
        self.relation_ids.return_value = ['nova-volume-service:0']
        _map = utils.resource_map()
        self.assertIn('nova-api-os-volume',
                      _map['/etc/nova/nova.conf']['services'])

    def test_determine_packages_quantum(self):
        self.network_manager.return_value = 'quantum'
        self.relation_ids.return_value = []
        self.test_config.set('network-manager', 'Quantum')
        pkgs = utils.determine_packages()
        self.assertIn('quantum-server', pkgs)

    def test_determine_packages_neutron(self):
        self.network_manager.return_value = 'neutron'
        self.relation_ids.return_value = []
        self.test_config.set('network-manager', 'Quantum')
        pkgs = utils.determine_packages()
        self.assertIn('neutron-server', pkgs)

    def test_determine_packages_nova_volume(self):
        self.relation_ids.return_value = ['nova-volume-service:0']
        pkgs = utils.determine_packages()
        self.assertIn('nova-api-os-volume', pkgs)

    def test_determine_packages_base(self):
        self.relation_ids.return_value = []
        pkgs = utils.determine_packages()
        ex = list(set(utils.BASE_PACKAGES + utils.BASE_SERVICES))
        self.assertEquals(ex, pkgs)

    @patch.object(utils, 'restart_map')
    def test_determine_ports(self, restart_map):
        restart_map.return_value = {
            '/etc/nova/nova.conf': ['nova-api-os-compute', 'nova-api-ec2'],
            '/etc/quantum/quantum.conf': ['quantum-server'],
        }
        ports = utils.determine_ports()
        ex = [8773, 8774, 9696]
        self.assertEquals(ex, sorted(ports))

    def test_save_script_rc_base(self):
        self.relation_ids.return_value = []
        utils.save_script_rc()
        self._save_script_rc.called_with(**SCRIPTRC_ENV_VARS)

    def test_save_script_quantum(self):
        self.relation_ids.return_value = []
        self.test_config.set('network-manager', 'Quantum')
        utils.save_script_rc()
        _ex = deepcopy(SCRIPTRC_ENV_VARS)
        _ex['OPENSTACK_SERVICE_API_QUANTUM'] = 'quantum-server'
        self._save_script_rc.called_with(**_ex)

    def test_save_script_nova_volume(self):
        self.relation_ids.return_value = ['nvol:0']
        utils.save_script_rc()
        _ex = deepcopy(SCRIPTRC_ENV_VARS)
        _ex['OPENSTACK_SERVICE_API_OS_VOL'] = 'nova-api-os-volume'
        self._save_script_rc.called_with(**_ex)

    def test_determine_volume_service_essex(self):
        self.os_release.return_value = 'essex'
        self.assertEquals('nova-volume', utils.volume_service())

    def test_determine_volume_service_folsom_cinder(self):
        self.os_release.return_value = 'folsom'
        self.relation_ids.return_value = ['cinder:0']
        self.assertEquals('cinder', utils.volume_service())

    def test_determine_volume_service_folsom_nova_vol(self):
        self.os_release.return_value = 'folsom'
        self.relation_ids.return_value = []
        self.assertEquals('nova-volume', utils.volume_service())

    def test_determine_volume_service_grizzly_and_beyond(self):
        pass

    @patch.object(utils, 'remove_known_host')
    @patch.object(utils, 'ssh_known_host_key')
    @patch('subprocess.check_output')
    def test_add_known_host_exists(self, check_output, host_key, rm):
        check_output.return_value = 'fookey'
        host_key.return_value = 'fookey'
        with patch_open() as (_open, _file):
            utils.add_known_host('foohost')
            self.assertFalse(rm.called)
            self.assertFalse(_file.write.called)

    @patch.object(utils, 'known_hosts')
    @patch.object(utils, 'remove_known_host')
    @patch.object(utils, 'ssh_known_host_key')
    @patch('subprocess.check_output')
    def test_add_known_host_exists_outdated(
            self, check_output, host_key, rm, known_hosts):
        check_output.return_value = 'fookey'
        host_key.return_value = 'fookey_old'
        with patch_open() as (_open, _file):
            utils.add_known_host('foohost')
            rm.assert_called_with('foohost')

    @patch.object(utils, 'known_hosts')
    @patch.object(utils, 'remove_known_host')
    @patch.object(utils, 'ssh_known_host_key')
    @patch('subprocess.check_output')
    def test_add_known_host_exists_added(
            self, check_output, host_key, rm, known_hosts):
        check_output.return_value = 'fookey'
        host_key.return_value = None
        with patch_open() as (_open, _file):
            _file.write = MagicMock()
            utils.add_known_host('foohost')
            self.assertFalse(rm.called)
            _file.write.assert_called_with('fookey\n')

    @patch('os.mkdir')
    @patch('os.path.isdir')
    def test_ssh_directory_for_unit(self, isdir, mkdir):
        self.remote_unit.return_value = 'nova-compute/0'
        isdir.return_value = False
        with patch_open() as (_open, _file):
            self.assertEquals(utils.ssh_directory_for_unit(),
                              '/etc/nova/compute_ssh/nova-compute')

    @patch.object(utils, 'ssh_directory_for_unit')
    def test_known_hosts(self, ssh_dir):
        ssh_dir.return_value = '/tmp/foo'
        self.assertEquals(utils.known_hosts(), '/tmp/foo/known_hosts')

    @patch.object(utils, 'ssh_directory_for_unit')
    def test_authorized_keys(self, ssh_dir):
        ssh_dir.return_value = '/tmp/foo'
        self.assertEquals(utils.authorized_keys(), '/tmp/foo/authorized_keys')

    @patch.object(utils, 'known_hosts')
    @patch('subprocess.check_call')
    def test_remove_host_key(self, check_call, known_hosts):
        known_hosts.return_value = '/tmp/known_hosts'
        utils.remove_known_host('foo')
        check_call.assert_called_with([
            'ssh-kegen', '-f', known_hosts(), '-R', 'foo'])

    @patch.object(utils, 'authorized_keys')
    def test_ssh_authorized_key_exists(self, keys):
        key = 'BBBBB3NzaC1yc2EBBBBDBQBBBBBBBQC27Us7lSjCpa7bumXBgc'
        with patch_open() as (_open, _file):
            _file.read.return_value = AUTHORIZED_KEYS
            self.assertTrue(utils.ssh_authorized_key_exists(key))

    @patch.object(utils, 'authorized_keys')
    def test_ssh_authorized_key_doesnt_exist(self, keys):
        key = ('xxxx')
        with patch_open() as (_open, _file):
            _file.read = MagicMock()
            _file.readreturn_value = AUTHORIZED_KEYS
            self.assertFalse(utils.ssh_authorized_key_exists(key))

    @patch.object(utils, 'known_hosts')
    @patch.object(utils, 'authorized_keys')
    @patch('os.path.isfile')
    def test_ssh_compute_remove(self, isfile, auth_key, known_host):
        isfile.return_value = False
        utils.ssh_compute_remove()
        self.assertFalse(self.remote_unit.called)

        keys_removed = (
            "\nssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC27Us7lSjCpa7bumXAgc "
            "nova-compute-1\n"
            "ssh-rsa CCCCB3NzaC1yc2ECCCCDCQCBCCCBCQC27Us7lSjCpa7bumXCgc "
            "nova-compute-3\n"
        )
        isfile.return_value = True
        self.remote_unit.return_value = 'nova-compute/2'
        with patch_open() as (_open, _file):
            _file.readlines = MagicMock()
            _file.write = MagicMock()
            _file.readlines.return_value = AUTHORIZED_KEYS.split('\n')
            utils.ssh_compute_remove()
            _file.write.assert_called_with(keys_removed)

    def test_network_manager_untranslated(self):
        self.test_config.set('network-manager', 'foo')
        self.os_release.return_value = 'folsom'

    def test_determine_endpoints_base(self):
        self.relation_ids.return_value = []
        self.assertEquals(
            BASE_ENDPOINTS, utils.determine_endpoints('http://foohost.com'))

    def test_determine_endpoints_nova_volume(self):
        self.relation_ids.return_value = ['nova-volume-service/0']
        endpoints = deepcopy(BASE_ENDPOINTS)
        endpoints.update({
            'nova-volume_admin_url':
            'http://foohost.com:8774/v1/\$(tenant_id)s',
            'nova-volume_internal_url':
            'http://foohost.com:8774/v1/\$(tenant_id)s',
            'nova-volume_public_url':
            'http://foohost.com:8774/v1/\$(tenant_id)s',
            'nova-volume_region': 'RegionOne',
            'nova-volume_service': 'nova-volume'})
        self.assertEquals(
            endpoints, utils.determine_endpoints('http://foohost.com'))

    def test_determine_endpoints_quantum_neutron(self):
        self.relation_ids.return_value = []
        self.network_manager.return_value = 'quantum'
        endpoints = deepcopy(BASE_ENDPOINTS)
        endpoints.update({
            'quantum_admin_url': 'http://foohost.com:9696',
            'quantum_internal_url': 'http://foohost.com:9696',
            'quantum_public_url': 'http://foohost.com:9696',
            'quantum_region': 'RegionOne',
            'quantum_service': 'quantum'})
        self.assertEquals(
            endpoints, utils.determine_endpoints('http://foohost.com'))