How to use set_io_scheduler method in autotest

Best Python code snippet using autotest_python

test_actions.py

Source:test_actions.py

1#!.venv3/bin/python32# Copyright 2015 Canonical Ltd.3#4# This file is part of the Cassandra Charm for Juju.5#6# This program is free software: you can redistribute it and/or modify7# it under the terms of the GNU General Public License version 3, as8# published by the Free Software Foundation.9#10# This program is distributed in the hope that it will be useful, but11# WITHOUT ANY WARRANTY; without even the implied warranties of12# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR13# PURPOSE.  See the GNU General Public License for more details.14#15# You should have received a copy of the GNU General Public License16# along with this program.  If not, see <http://www.gnu.org/licenses/>.17import errno18from itertools import repeat19import os.path20import re21import shutil22import subprocess23import tempfile24from textwrap import dedent25import unittest26from unittest.mock import ANY, call, patch, sentinel27import yaml28import cassandra29from charmhelpers.core import hookenv30from tests.base import TestCaseBase31import actions32from coordinator import coordinator33import helpers34class TestActions(TestCaseBase):35    def test_set_proxy(self):36        # NB. Environment is already mocked.37        os.environ['http_proxy'] = ''38        os.environ['https_proxy'] = ''39        actions.set_proxy('')40        self.assertEqual(os.environ['http_proxy'], '')41        self.assertEqual(os.environ['https_proxy'], '')42        hookenv.config()['http_proxy'] = 'foo'43        actions.set_proxy('')44        self.assertEqual(os.environ['http_proxy'], 'foo')45        self.assertEqual(os.environ['https_proxy'], 'foo')46    @patch('subprocess.check_call')47    def test_swapoff(self, check_call):48        fstab = (49            b'UUID=abc / ext4 errors=remount-ro 0 1\n'50            b'/dev/mapper/cryptswap1 none swap sw 0 0')51        with tempfile.NamedTemporaryFile() as f:52            f.write(fstab)53            f.flush()54            actions.swapoff('', f.name)55            f.seek(0)56            self.assertTrue(b'swap' not in f.read())57        check_call.assert_called_once_with(['swapoff', '-a'])58    @patch('subprocess.check_call')59    def test_swapoff_fails(self, check_call):60        check_call.side_effect = RuntimeError()61        actions.swapoff('', '')62        # A warning is generated if swapoff fails.63        hookenv.log.assert_any_call(ANY, hookenv.WARNING)64    @patch('subprocess.check_call')65    def test_swapoff_lxc(self, check_call):66        # Under LXC, the swapoff action does nothing except log.67        helpers.is_lxc.return_value = True68        actions.swapoff('')69        self.assertFalse(check_call.called)70    @patch('charmhelpers.core.host.write_file')71    @patch('subprocess.check_call')72    def test_reset_sysctl(self, check_call, write_file):73        actions.reset_sysctl('')74        ctl_file = '/etc/sysctl.d/99-cassandra.conf'75        # Magic value per Cassandra best practice.76        write_file.assert_called_once_with(ctl_file,77                                           b"vm.max_map_count = 131072\n")78        check_call.assert_called_once_with(['sysctl', '-p',79                                            '/etc/sysctl.d/99-cassandra.conf'])80    @patch('subprocess.check_call')81    @patch('charmhelpers.core.host.write_file')82    def test_reset_sysctl_expected_fails(self, write_file, check_call):83        check_call.side_effect = repeat(OSError(errno.EACCES,84                                                'Permission Denied'))85        actions.reset_sysctl('')86        # A warning is generated if permission denied was raised.87        hookenv.log.assert_any_call(ANY, hookenv.WARNING)88    @patch('subprocess.check_call')89    @patch('charmhelpers.core.host.write_file')90    def test_reset_sysctl_fails_badly(self, write_file, check_call):91        # Other OSErrors are reraised since we don't know how to handle92        # them.93        check_call.side_effect = repeat(OSError(errno.EFAULT, 'Whoops'))94        self.assertRaises(OSError, actions.reset_sysctl, '')95    @patch('subprocess.check_call')96    def test_reset_sysctl_lxc(self, check_call):97        helpers.is_lxc.return_value = True98        actions.reset_sysctl('')99        self.assertFalse(check_call.called)100        hookenv.log.assert_any_call('In an LXC. '101                                    'Leaving sysctl unchanged.')102    @patch('helpers.get_cassandra_edition')103    @patch('helpers.ensure_cassandra_snap_installed')104    @patch('helpers.get_cassandra_packages')105    @patch('helpers.ensure_package_status')106    def test_ensure_cassandra_package_status_snap(self, ensure_package_status,107                                                  get_cassandra_packages,108                                                  ensure_snap_installed,109                                                  get_cassandra_edition):110        get_cassandra_packages.return_value = sentinel.cassandra_packages111        get_cassandra_edition.return_value = 'apache-snap'112        actions.ensure_cassandra_package_status('')113        ensure_package_status.assert_called_once_with(114            sentinel.cassandra_packages)115        ensure_snap_installed.assert_called_once_with()116    @patch('subprocess.check_call')117    @patch('helpers.get_jre')118    @patch('helpers.get_cassandra_packages')119    @patch('helpers.install_packages')120    def test_install_cassandra_packages(self, install_packages,121                                        get_cassandra_packages,122                                        get_jre, check_call):123        get_cassandra_packages.return_value = sentinel.cassandra_packages124        get_jre.return_value = 'openjdk'125        actions.install_cassandra_packages('')126        install_packages.assert_called_once_with(sentinel.cassandra_packages)127        check_call.assert_called_once_with(['update-java-alternatives',128                                            '--jre-headless', '--set',129                                            'java-1.8.0-openjdk-amd64'])130    @patch('subprocess.check_call')131    @patch('helpers.get_jre')132    @patch('helpers.get_cassandra_packages')133    @patch('helpers.install_packages')134    def test_install_cassandra_packages_oracle(self, install_packages,135                                               get_cassandra_packages,136                                               get_jre, check_call):137        get_cassandra_packages.return_value = sentinel.cassandra_packages138        get_jre.return_value = 'oracle'139        actions.install_cassandra_packages('')140        install_packages.assert_called_once_with(sentinel.cassandra_packages)141        # No alternatives selected, as the Oracle JRE installer method142        # handles this.143        self.assertFalse(check_call.called)144    @patch('actions._install_oracle_jre_tarball')145    @patch('actions._fetch_oracle_jre')146    def test_install_oracle_jre(self, fetch, install_tarball):147        fetch.return_value = sentinel.tarball148        actions.install_oracle_jre('')149        self.assertFalse(fetch.called)150        self.assertFalse(install_tarball.called)151        hookenv.config()['jre'] = 'oracle'152        actions.install_oracle_jre('')153        fetch.assert_called_once_with()154        install_tarball.assert_called_once_with(sentinel.tarball)155    @patch('helpers.status_set')156    @patch('urllib.request')157    def test_fetch_oracle_jre(self, req, status_set):158        config = hookenv.config()159        url = 'https://foo.example.com/server-jre-7u42-linux-x64.tar.gz'160        expected_tarball = os.path.join(hookenv.charm_dir(), 'lib',161                                        'server-jre-7u42-linux-x64.tar.gz')162        config['private_jre_url'] = url163        # Create a dummy tarball, since the mock urlretrieve won't.164        os.makedirs(os.path.dirname(expected_tarball))165        with open(expected_tarball, 'w'):166            pass  # Empty file167        self.assertEqual(actions._fetch_oracle_jre(), expected_tarball)168        req.urlretrieve.assert_called_once_with(url, expected_tarball)169    def test_fetch_oracle_jre_local(self):170        # Create an existing tarball. If it is found, it will be used171        # without needing to specify a remote url or actually download172        # anything.173        expected_tarball = os.path.join(hookenv.charm_dir(), 'lib',174                                        'server-jre-7u42-linux-x64.tar.gz')175        os.makedirs(os.path.dirname(expected_tarball))176        with open(expected_tarball, 'w'):177            pass  # Empty file178        self.assertEqual(actions._fetch_oracle_jre(), expected_tarball)179    @patch('helpers.status_set')180    def test_fetch_oracle_jre_notfound(self, status_set):181        with self.assertRaises(SystemExit) as x:182            actions._fetch_oracle_jre()183            self.assertEqual(x.code, 0)184            status_set.assert_called_once_with('blocked', ANY)185    @patch('subprocess.check_call')186    @patch('charmhelpers.core.host.mkdir')187    @patch('os.path.isdir')188    def test_install_oracle_jre_tarball(self, isdir, mkdir, check_call):189        isdir.return_value = False190        dest = '/usr/lib/jvm/java-8-oracle'191        actions._install_oracle_jre_tarball(sentinel.tarball)192        mkdir.assert_called_once_with(dest)193        check_call.assert_has_calls([194            call(['tar', '-xz', '-C', dest,195                  '--strip-components=1', '-f', sentinel.tarball]),196            call(['update-alternatives', '--install',197                  '/usr/bin/java', 'java',198                  os.path.join(dest, 'bin', 'java'), '1']),199            call(['update-alternatives', '--set', 'java',200                  os.path.join(dest, 'bin', 'java')]),201            call(['update-alternatives', '--install',202                  '/usr/bin/javac', 'javac',203                  os.path.join(dest, 'bin', 'javac'), '1']),204            call(['update-alternatives', '--set', 'javac',205                  os.path.join(dest, 'bin', 'javac')])])206    @patch('os.path.exists')207    @patch('subprocess.check_call')208    @patch('charmhelpers.core.host.mkdir')209    @patch('os.path.isdir')210    def test_install_oracle_jre_tarball_already(self, isdir,211                                                mkdir, check_call, exists):212        isdir.return_value = True213        exists.return_value = True  # jre already installed214        # Store the version previously installed.215        hookenv.config()['oracle_jre_tarball'] = sentinel.tarball216        dest = '/usr/lib/jvm/java-8-oracle'217        actions._install_oracle_jre_tarball(sentinel.tarball)218        self.assertFalse(mkdir.called)  # The jvm dir already existed.219        exists.assert_called_once_with('/usr/lib/jvm/java-8-oracle/bin/java')220        # update-alternatives done, but tarball not extracted.221        check_call.assert_has_calls([222            call(['update-alternatives', '--install',223                  '/usr/bin/java', 'java',224                  os.path.join(dest, 'bin', 'java'), '1']),225            call(['update-alternatives', '--set', 'java',226                  os.path.join(dest, 'bin', 'java')]),227            call(['update-alternatives', '--install',228                  '/usr/bin/javac', 'javac',229                  os.path.join(dest, 'bin', 'javac'), '1']),230            call(['update-alternatives', '--set', 'javac',231                  os.path.join(dest, 'bin', 'javac')])])232    @patch('helpers.configure_cassandra_yaml')233    def test_configure_cassandra_yaml(self, configure_cassandra_yaml):234        # actions.configure_cassandra_yaml is just a wrapper around the235        # helper.236        actions.configure_cassandra_yaml('')237        configure_cassandra_yaml.assert_called_once_with()238    @patch('helpers.get_cassandra_env_file')239    @patch('charmhelpers.core.host.write_file')240    def test_configure_cassandra_env(self, write_file, env_file):241        def _wf(path, contents, perms=None):242            with open(path, 'wb') as f:243                f.write(contents)244        write_file.side_effect = _wf245        # cassandra-env.sh is a shell script that unfortunately246        # embeds configuration we need to change.247        existing_config = dedent('''\248                                 Everything is ignored249                                 unless a regexp matches250                                 #MAX_HEAP_SIZE="1G"251                                 #HEAP_NEWSIZE="800M"252                                 #JMX_PORT="1234"253                                 And done254                                 ''')255        with tempfile.TemporaryDirectory() as tempdir:256            cassandra_env = os.path.join(tempdir, 'c.sh')257            env_file.return_value = cassandra_env258            with open(cassandra_env, 'w', encoding='UTF-8') as f:259                f.write(existing_config)260            overrides = dict(261                max_heap_size=re.compile('^MAX_HEAP_SIZE=(.*)$', re.M),262                heap_newsize=re.compile('^HEAP_NEWSIZE=(.*)$', re.M))263            for key in overrides:264                hookenv.config()[key] = ''265            # By default, the settings will be commented out.266            actions.configure_cassandra_env('')267            with open(cassandra_env, 'r', encoding='UTF-8') as f:268                generated_env = f.read()269            for config_key, regexp in overrides.items():270                with self.subTest(override=config_key):271                    self.assertIsNone(regexp.search(generated_env))272            # Settings can be overridden.273            for config_key, regexp in overrides.items():274                hookenv.config()[config_key] = '{} val'.format(config_key)275            actions.configure_cassandra_env('')276            with open(cassandra_env, 'r') as f:277                generated_env = f.read()278            for config_key, regexp in overrides.items():279                with self.subTest(override=config_key):280                    match = regexp.search(generated_env)281                    self.assertIsNotNone(match)282                    # Note the value has been shell quoted.283                    self.assertTrue(284                        match.group(1).startswith(285                            "'{} val'".format(config_key)))286            # Settings can be returned to the defaults.287            for config_key, regexp in overrides.items():288                hookenv.config()[config_key] = ''289            actions.configure_cassandra_env('')290            with open(cassandra_env, 'r', encoding='UTF-8') as f:291                generated_env = f.read()292            for config_key, regexp in overrides.items():293                with self.subTest(override=config_key):294                    self.assertIsNone(regexp.search(generated_env))295    @patch('helpers.get_cassandra_rackdc_file')296    def test_configure_cassandra_rackdc(self, rackdc_file):297        hookenv.config()['datacenter'] = 'test_dc'298        hookenv.config()['rack'] = 'test_rack'299        with tempfile.NamedTemporaryFile() as rackdc:300            rackdc_file.return_value = rackdc.name301            actions.configure_cassandra_rackdc('')302            with open(rackdc.name, 'r') as f:303                self.assertEqual(f.read().strip(),304                                 'dc=test_dc\nrack=test_rack')305    @patch('helpers.connect')306    @patch('helpers.get_auth_keyspace_replication')307    @patch('helpers.num_nodes')308    def test_needs_reset_auth_keyspace_replication(self, num_nodes,309                                                   get_auth_ks_rep,310                                                   connect):311        num_nodes.return_value = 4312        connect().__enter__.return_value = sentinel.session313        connect().__exit__.return_value = False314        get_auth_ks_rep.return_value = {'another': '8'}315        self.assertTrue(actions.needs_reset_auth_keyspace_replication())316    @patch('helpers.connect')317    @patch('helpers.get_auth_keyspace_replication')318    @patch('helpers.num_nodes')319    def test_needs_reset_auth_keyspace_replication_false(self, num_nodes,320                                                         get_auth_ks_rep,321                                                         connect):322        config = hookenv.config()323        config['datacenter'] = 'mydc'324        connect().__enter__.return_value = sentinel.session325        connect().__exit__.return_value = False326        num_nodes.return_value = 3327        get_auth_ks_rep.return_value = {'another': '8',328                                        'mydc': '3'}329        self.assertFalse(actions.needs_reset_auth_keyspace_replication())330    @patch('helpers.set_active')331    @patch('helpers.repair_auth_keyspace')332    @patch('helpers.connect')333    @patch('helpers.set_auth_keyspace_replication')334    @patch('helpers.get_auth_keyspace_replication')335    @patch('helpers.num_nodes')336    @patch('charmhelpers.core.hookenv.is_leader')337    def test_reset_auth_keyspace_replication(self, is_leader, num_nodes,338                                             get_auth_ks_rep,339                                             set_auth_ks_rep,340                                             connect, repair, set_active):341        is_leader.return_value = True342        num_nodes.return_value = 4343        coordinator.grants = {}344        coordinator.requests = {hookenv.local_unit(): {}}345        coordinator.grant('repair', hookenv.local_unit())346        config = hookenv.config()347        config['datacenter'] = 'mydc'348        connect().__enter__.return_value = sentinel.session349        connect().__exit__.return_value = False350        get_auth_ks_rep.return_value = {'another': '8'}351        self.assertTrue(actions.needs_reset_auth_keyspace_replication())352        actions.reset_auth_keyspace_replication('')353        set_auth_ks_rep.assert_called_once_with(354            sentinel.session,355            {'class': 'NetworkTopologyStrategy', 'another': '8', 'mydc': 4})356        repair.assert_called_once_with()357        set_active.assert_called_once_with()358    @patch('helpers.stop_cassandra')359    def test_stop_cassandra(self, helpers_stop_cassandra):360        actions.stop_cassandra('ignored')361        helpers_stop_cassandra.assert_called_once_with()362    @patch('helpers.start_cassandra')363    def test_start_cassandra(self, helpers_start_cassandra):364        actions.start_cassandra('ignored')365        helpers_start_cassandra.assert_called_once_with()366    @patch('os.path.isdir')367    @patch('helpers.get_all_database_directories')368    @patch('helpers.set_io_scheduler')369    def test_reset_all_io_schedulers(self, set_io_scheduler, dbdirs, isdir):370        hookenv.config()['io_scheduler'] = sentinel.io_scheduler371        dbdirs.return_value = dict(372            data_file_directories=[sentinel.d1, sentinel.d2],373            commitlog_directory=sentinel.cl,374            saved_caches_directory=sentinel.sc)375        isdir.return_value = True376        actions.reset_all_io_schedulers('')377        set_io_scheduler.assert_has_calls([378            call(sentinel.io_scheduler, sentinel.d1),379            call(sentinel.io_scheduler, sentinel.d2),380            call(sentinel.io_scheduler, sentinel.cl),381            call(sentinel.io_scheduler, sentinel.sc)],382            any_order=True)383        # If directories don't exist yet, nothing happens.384        set_io_scheduler.reset_mock()385        isdir.return_value = False386        actions.reset_all_io_schedulers('')387        self.assertFalse(set_io_scheduler.called)388    def test_config_key_lists_complete(self):389        # Ensure that we have listed all keys in either390        # RESTART_REQUIRED_KEYS, RESTART_NOT_REQUIRED_KEYS or391        # UNCHANGEABLE_KEYS. This is to ensure that RESTART_REQUIRED_KEYS392        # is maintained as new config items are added over time.393        config_path = os.path.join(os.path.dirname(__file__), os.pardir,394                                   'config.yaml')395        with open(config_path, 'r') as f:396            config = yaml.safe_load(f)397        combined = actions.RESTART_REQUIRED_KEYS.union(398            actions.RESTART_NOT_REQUIRED_KEYS).union(399                actions.UNCHANGEABLE_KEYS)400        for key in config['options']:401            with self.subTest(key=key):402                self.assertIn(key, combined)403    @patch('charmhelpers.core.host.write_file')404    def test_install_maintenance_crontab(self, write_file):405        # First 7 units get distributed, one job per day.406        hookenv.local_unit.return_value = 'foo/0'407        actions.install_maintenance_crontab('')408        write_file.assert_called_once_with('/etc/cron.d/cassandra-maintenance',409                                           ANY)410        contents = write_file.call_args[0][1]411        # Not the complete command, but includes all the expanded412        # variables.413        expected = (b'\n0 0 * * 0 cassandra run-one-until-success '414                    b'nodetool repair -pr')415        self.assertIn(expected, contents)416        # Next 7 units distributed 12 hours out of sync with the first417        # batch.418        hookenv.local_unit.return_value = 'foo/8'419        actions.install_maintenance_crontab('')420        contents = write_file.call_args[0][1]421        expected = (b'\n0 12 * * 1 cassandra run-one-until-success '422                    b'nodetool repair -pr')423        self.assertIn(expected, contents)424        # Later units per helpers.week_spread()425        hookenv.local_unit.return_value = 'foo/411'426        actions.install_maintenance_crontab('')427        contents = write_file.call_args[0][1]428        expected = (b'\n37 8 * * 5 cassandra run-one-until-success '429                    b'nodetool repair -pr')430        self.assertIn(expected, contents)431    @patch('helpers.mountpoint')432    @patch('helpers.get_cassandra_version')433    @patch('charmhelpers.core.host.write_file')434    @patch('charmhelpers.contrib.charmsupport.nrpe.NRPE')435    @patch('helpers.local_plugins_dir')436    def test_nrpe_external_master_relation(self, local_plugins_dir, nrpe,437                                           write_file, cassandra_version,438                                           mountpoint):439        mountpoint.side_effect = os.path.dirname440        cassandra_version.return_value = '2.2'441        # The fake charm_dir() needs populating.442        plugin_src_dir = os.path.join(os.path.dirname(__file__),443                                      os.pardir, 'files')444        shutil.copytree(plugin_src_dir,445                        os.path.join(hookenv.charm_dir(), 'files'))446        with tempfile.TemporaryDirectory() as d:447            local_plugins_dir.return_value = d448            actions.nrpe_external_master_relation('')449            # The expected file was written to the expected filename450            # with required perms.451            with open(os.path.join(plugin_src_dir, 'check_cassandra_heap.sh'),452                      'rb') as f:453                write_file.assert_called_once_with(454                    os.path.join(d, 'check_cassandra_heap.sh'), f.read(),455                    perms=0o555)456            nrpe().add_check.assert_has_calls([457                call(shortname='cassandra_heap',458                     description='Check Cassandra Heap',459                     check_cmd='check_cassandra_heap.sh localhost 80 90'),460                call(description=('Check Cassandra Disk '461                                  '/var/lib/cassandra'),462                     shortname='cassandra_disk_var_lib_cassandra',463                     check_cmd=('check_disk -u GB -w 50% -c 25% -K 5% '464                                '-p /var/lib/cassandra'))],465                any_order=True)466            nrpe().write.assert_called_once_with()467    @patch('helpers.get_cassandra_version')468    @patch('charmhelpers.core.host.write_file')469    @patch('os.path.exists')470    @patch('charmhelpers.contrib.charmsupport.nrpe.NRPE')471    def test_nrpe_external_master_relation_no_local(self, nrpe, exists,472                                                    write_file, ver):473        ver.return_value = '2.2'474        # If the local plugins directory doesn't exist, we don't attempt475        # to write files to it. Wait until the subordinate has set it476        # up.477        exists.return_value = False478        actions.nrpe_external_master_relation('')479        self.assertFalse(write_file.called)480    @patch('helpers.mountpoint')481    @patch('helpers.get_cassandra_version')482    @patch('os.path.exists')483    @patch('charmhelpers.contrib.charmsupport.nrpe.NRPE')484    def test_nrpe_external_master_relation_disable_heapchk(self, nrpe, exists,485                                                           ver, mountpoint):486        ver.return_value = '2.2'487        exists.return_value = False488        mountpoint.side_effect = os.path.dirname489        # Disable our checks490        config = hookenv.config()491        config['nagios_heapchk_warn_pct'] = 0  # Only one needs to be disabled.492        config['nagios_heapchk_crit_pct'] = 90493        actions.nrpe_external_master_relation('')494        exists.assert_called_once_with(helpers.local_plugins_dir())495        nrpe().add_check.assert_has_calls([496            call(shortname='cassandra_disk_var_lib_cassandra',497                 description=ANY, check_cmd=ANY)], any_order=True)498    @patch('helpers.get_cassandra_version')499    @patch('os.path.exists')500    @patch('charmhelpers.contrib.charmsupport.nrpe.NRPE')501    def test_nrpe_external_master_relation_disable_diskchk(self, nrpe,502                                                           exists, ver):503        ver.return_value = '2.2'504        exists.return_value = False505        # Disable our checks506        config = hookenv.config()507        config['nagios_disk_warn_pct'] = 0  # Only one needs to be disabled.508        config['magios_disk_crit_pct'] = 50509        actions.nrpe_external_master_relation('')510        exists.assert_called_once_with(helpers.local_plugins_dir())511        nrpe().add_check.assert_called_once_with(shortname='cassandra_heap',512                                                 description=ANY,513                                                 check_cmd=ANY)514    @patch('helpers.get_bootstrapped_ips')515    @patch('helpers.get_seed_ips')516    @patch('charmhelpers.core.hookenv.leader_set')517    @patch('charmhelpers.core.hookenv.is_leader')518    def test_maintain_seeds(self, is_leader, leader_set,519                            seed_ips, bootstrapped_ips):520        is_leader.return_value = True521        seed_ips.return_value = set(['1.2.3.4'])522        bootstrapped_ips.return_value = set(['2.2.3.4', '3.2.3.4',523                                             '4.2.3.4', '5.2.3.4'])524        actions.maintain_seeds('')525        leader_set.assert_called_once_with(seeds='2.2.3.4,3.2.3.4,4.2.3.4')526    @patch('helpers.get_bootstrapped_ips')527    @patch('helpers.get_seed_ips')528    @patch('charmhelpers.core.hookenv.leader_set')529    @patch('charmhelpers.core.hookenv.is_leader')530    def test_maintain_seeds_start(self, is_leader, leader_set,531                                  seed_ips, bootstrapped_ips):532        seed_ips.return_value = set()533        bootstrapped_ips.return_value = set()534        actions.maintain_seeds('')535        # First seed is the first leader, which lets is get everything536        # started.537        leader_set.assert_called_once_with(seeds=hookenv.unit_private_ip())538    @patch('charmhelpers.core.host.pwgen')539    @patch('helpers.query')540    @patch('helpers.set_unit_superusers')541    @patch('helpers.ensure_user')542    @patch('helpers.encrypt_password')543    @patch('helpers.superuser_credentials')544    @patch('helpers.connect')545    @patch('charmhelpers.core.hookenv.is_leader')546    @patch('charmhelpers.core.hookenv.leader_set')547    @patch('charmhelpers.core.hookenv.leader_get')548    def test_reset_default_password(self, leader_get, leader_set, is_leader,549                                    connect, sup_creds, encrypt_password,550                                    ensure_user, set_sups, query, pwgen):551        is_leader.return_value = True552        leader_get.return_value = None553        connect().__enter__.return_value = sentinel.session554        connect().__exit__.return_value = False555        connect.reset_mock()556        sup_creds.return_value = (sentinel.username, sentinel.password)557        encrypt_password.return_value = sentinel.pwhash558        pwgen.return_value = sentinel.random_password559        actions.reset_default_password('')560        # First, a superuser account for the unit was created.561        connect.assert_called_once_with('cassandra', 'cassandra',562                                        timeout=120, auth_timeout=120)563        encrypt_password.assert_called_once_with(sentinel.password)564        ensure_user.assert_called_once_with(sentinel.session,565                                            sentinel.username,566                                            sentinel.pwhash,567                                            superuser=True)568        set_sups.assert_called_once_with([hookenv.local_unit()])569        # After that, the default password is reset.570        query.assert_called_once_with(sentinel.session,571                                      'ALTER USER cassandra WITH PASSWORD %s',572                                      cassandra.ConsistencyLevel.ALL,573                                      (sentinel.random_password,))574        # Flag stored to avoid attempting this again.575        leader_set.assert_called_once_with(default_admin_password_changed=True)576    @patch('helpers.connect')577    @patch('charmhelpers.core.hookenv.is_leader')578    @patch('charmhelpers.core.hookenv.leader_get')579    def test_reset_default_password_noop(self, leader_get, is_leader, connect):580        leader_get.return_value = True581        is_leader.return_value = True582        actions.reset_default_password('')  # noop583        self.assertFalse(connect.called)584    @patch('helpers.update_hosts_file')585    @patch('socket.gethostname')586    def test_update_etc_hosts(self, gethostname, update_hosts_file):587        gethostname.return_value = sentinel.hostname588        actions.update_etc_hosts('')589        update_hosts_file.assert_called_once_with(590            '/etc/hosts', {'10.20.0.1': sentinel.hostname})591if __name__ == '__main__':...

test_helpers.py

Source:test_helpers.py

...81    @patch('charmhelpers.core.host.write_file')82    @patch('os.path.exists')83    @patch('os.path.isdir')84    @patch('subprocess.check_output')85    def test_set_io_scheduler(self, check_output, isdir, exists, write_file, is_container):86        is_container.return_value = False87        # Normal operation, the device is detected and the magic88        # file written.89        check_output.return_value = 'foo\n/dev/sdq 1 2 3 1% /foo\n'90        isdir.return_value = True91        exists.return_value = True92        helpers.set_io_scheduler('fnord', '/foo')93        write_file.assert_called_once_with('/sys/block/sdq/queue/scheduler',94                                           b'fnord', perms=0o644)95        # Some OSErrors we log warnings for, and continue.96        for e in (errno.EACCES, errno.ENOENT):97            with self.subTest(errno=e):98                write_file.side_effect = repeat(OSError(e, 'Whoops'))99                hookenv.log.reset_mock()100                helpers.set_io_scheduler('fnord', '/foo')101                hookenv.log.assert_has_calls([call(ANY),102                                              call(ANY, hookenv.WARNING)])103        # Other OSErrors just fail hard.104        write_file.side_effect = iter([OSError(errno.EFAULT, 'Whoops')])105        self.assertRaises(OSError, helpers.set_io_scheduler, 'fnord', '/foo')106        # If we are not under lxc, nothing happens at all except a log107        # message.108        is_container.return_value = True109        hookenv.log.reset_mock()110        write_file.reset_mock()111        helpers.set_io_scheduler('fnord', '/foo')112        self.assertFalse(write_file.called)113        hookenv.log.assert_called_once_with(ANY)  # A single INFO message.114    @patch('shutil.chown')115    def test_recursive_chown(self, chown):116        with tempfile.TemporaryDirectory() as tmpdir:117            os.makedirs(os.path.join(tmpdir, 'a', 'bb', 'ccc'))118            with open(os.path.join(tmpdir, 'top file'), 'w') as f:119                f.write('top file')120            with open(os.path.join(tmpdir, 'a', 'bb', 'midfile'), 'w') as f:121                f.write('midfile')122            helpers.recursive_chown(tmpdir, 'un', 'gn')123        chown.assert_has_calls(124            [call(os.path.join(tmpdir, 'a'), 'un', 'gn'),125             call(os.path.join(tmpdir, 'a', 'bb'), 'un', 'gn'),...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.