From 557ea72eb2df2e603d94f63a4506372fc7b5c959 Mon Sep 17 00:00:00 2001 From: Liam Young Date: Sat, 15 Nov 2014 09:32:24 -0600 Subject: [PATCH] Start to add unit_tests --- .coverage | 10 ++ .coveragerc | 6 + Makefile | 4 + unit_tests/__init__.py | 3 + unit_tests/test_ceph.py | 124 ++++++++++++++++ unit_tests/test_hooks.py | 303 +++++++++++++++++++++++++++++++++++++++ unit_tests/test_utils.py | 119 +++++++++++++++ 7 files changed, 569 insertions(+) create mode 100644 .coverage create mode 100644 .coveragerc create mode 100644 unit_tests/__init__.py create mode 100644 unit_tests/test_ceph.py create mode 100644 unit_tests/test_hooks.py create mode 100644 unit_tests/test_utils.py diff --git a/.coverage b/.coverage new file mode 100644 index 00000000..a257f1b6 --- /dev/null +++ b/.coverage @@ -0,0 +1,10 @@ +}q(U collectorqUcoverage v3.7.1qUlinesq}q(US/home/liam/branches/paris-train/next-org/ceph-radosgw-next/unit_tests/test_utils.pyq]q(KKKKKKK +KKKKKKKKKKKK#K$K%K&K'K*K-K/K0K1K2K3K4K5K7K8K9K:K;K=K>K?KBKDKEKGKHKJKKKOKRKSKUKXKZK[K]K`KhKnKoKqKsKtKvKweUZ/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/core/host.pyq]q (KKK K +K K K KKKKKKKKK"K'K0K6KCKMKcKnKyKKKKKKKKKKKKMM'M3MJMPM[MeMuM~eU^/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/payload/execd.pyq +]q (KKKKK K KK$K0eUY/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/__init__.pyq ]q KaU_/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/fetch/__init__.pyq]q(KKKKKKK K KKKKKKKKK K!K"K$K%K&K'K(K)K*K,K-K.K/K0K1K2K4K5K6K7K8K9K:KK?K@KAKBKCKKKNKOKPKSKTKWKXK[K\K_KaKcKhKmKpKwKKKKKKKKKKKKKKKKMMMM?McMiM|eUH/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/ceph.pyq]q(K +K K K KKKKKKKKKKKKK K!K"K#K$K%K&K'K(K)K*K+K,K-K.K/K0K1K2K3K4K5K6K7K8K9K;K=K>K?K@KAKCKFKGKHKIKKKLKMKNKOKPKQKRKSKTKUKVKWKYKZK[K\K]K^K_K`KaKbKcKdKeKgKjKmKnKoKpKrKsKvKwKxKyK{K|K}K~KKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKeUQ/home/liam/branches/paris-train/next-org/ceph-radosgw-next/unit_tests/__init__.pyq]q(KKeUI/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/hooks.pyq]q(K +K K K KKKKK!K"K)K*K+K-K0K1K2K5K8K=K>K?K@KAKCKDKGKHKIKJKLKMKNKOKPKSKUKVKWKXK[K]K^K`KaKbKcKdKeKjKlKmKnKpKqKtKuKvKxKyKzK{K|KKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKMeU]/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/core/hookenv.pyq]q(KKKK K +K K K KKKKKKKKK&K(K)K*K+K,K/K2K=K?K@KBKCKFKGKIKNK\K`KdKhKmK{KKKKKKKKKKKKKKKMMMM'M(M;MKMLMVMWM`MaMlMmMxMyMMMMMMMMMMMMMMMMMMMMMMMMMMMMM +M M M MMMeUa/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/payload/__init__.pyq]qKaU^/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/core/__init__.pyq]qKaUI/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/utils.pyq]q(K +K K KKKKKKKKKK K(K4KCeUS/home/liam/branches/paris-train/next-org/ceph-radosgw-next/unit_tests/test_hooks.pyq]q(KKKKKKK K +K K K KKKKKKKKKKKKKKKK K!K"K#K$K&K'K(K)K+K,K-K.K0K3K4K6K7K9K:K;KK?K@KAKCKDKEKFKGKHKIKKKLKMKNKOKPKQKRKSKTKUKVKWKXKYKZK\K]K^K_K`KaKcKdKeKfKgKiKjKkKlKnKoKpKrKsKuKwKxKyK{K|K~eU[/home/liam/branches/paris-train/next-org/ceph-radosgw-next/hooks/charmhelpers/core/fstab.pyq ]q!(KKK K KKKKK!K)K+K2K9KCKJKRKhKiKpKqeuu. \ No newline at end of file diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 00000000..0e6369e1 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,6 @@ +[report] +# Regexes for lines to exclude from consideration +exclude_lines = + if __name__ == .__main__.: +include= + hooks/ceph.py diff --git a/Makefile b/Makefile index 5c9ade39..02f8c75d 100644 --- a/Makefile +++ b/Makefile @@ -5,6 +5,10 @@ lint: @flake8 --exclude hooks/charmhelpers hooks tests @charm proof +unit_test: + @$(PYTHON) /usr/bin/nosetests unit_tests +# @$(PYTHON) /usr/bin/nosetests --nologcapture --with-coverage unit_tests + test: @echo Starting Amulet tests... # coreycb note: The -v should only be temporary until Amulet sends diff --git a/unit_tests/__init__.py b/unit_tests/__init__.py new file mode 100644 index 00000000..afaed60c --- /dev/null +++ b/unit_tests/__init__.py @@ -0,0 +1,3 @@ +import sys + +sys.path.append('hooks/') diff --git a/unit_tests/test_ceph.py b/unit_tests/test_ceph.py new file mode 100644 index 00000000..2e096082 --- /dev/null +++ b/unit_tests/test_ceph.py @@ -0,0 +1,124 @@ +from mock import call, patch, MagicMock +from test_utils import CharmTestCase, patch_open + +import ceph + +TO_PATCH = [ + 'get_unit_hostname', + 'os', + 'subprocess', + 'time', +] + +class CephRadosGWCephTests(CharmTestCase): + + def setUp(self): + super(CephRadosGWCephTests, self).setUp(ceph, TO_PATCH) + + def test_is_quorum_leader(self): + self.get_unit_hostname.return_value = 'myhost' + self.subprocess.check_output.return_value = '{"state": "leader"}' + self.assertEqual(ceph.is_quorum(), True) + + def test_is_quorum_notleader(self): + self.get_unit_hostname.return_value = 'myhost' + self.subprocess.check_output.return_value = '{"state": "notleader"}' + self.assertEqual(ceph.is_quorum(), False) + + def test_is_quorum_valerror(self): + self.get_unit_hostname.return_value = 'myhost' + self.subprocess.check_output.return_value = "'state': 'bob'}" + self.assertEqual(ceph.is_quorum(), False) + + def test_is_leader(self): + self.get_unit_hostname.return_value = 'myhost' + self.os.path.exists.return_value = True + self.subprocess.check_output.return_value = '{"state": "leader"}' + self.assertEqual(ceph.is_leader(), True) + + def test_is_leader_notleader(self): + self.get_unit_hostname.return_value = 'myhost' + self.os.path.exists.return_value = True + self.subprocess.check_output.return_value = '{"state": "notleader"}' + self.assertEqual(ceph.is_leader(), False) + + def test_is_leader_valerror(self): + self.get_unit_hostname.return_value = 'myhost' + self.os.path.exists.return_value = True + self.subprocess.check_output.return_value = "'state': 'bob'}" + self.assertEqual(ceph.is_leader(), False) + + def test_is_leader_noasok(self): + self.get_unit_hostname.return_value = 'myhost' + self.os.path.exists.return_value = False + self.assertEqual(ceph.is_leader(), False) + +# def test_wait_for_quorum_yes(self): +# _is_quorum = self.patch('is_quorum') +# _is_quorum.return_value = False +# self.time.return_value = None +# ceph.wait_for_quorum() +# self.time.sleep.assert_called_with(3) + +# def test_wait_for_quorum_no(self): +# _is_quorum = self.patch('is_quorum') +# _is_quorum.return_value = True +# ceph.wait_for_quorum() +# self.assertFalse(self.time.sleep.called) + + def test_add_bootstrap_hint(self): + self.get_unit_hostname.return_value = 'myhost' + cmd = [ + "ceph", + "--admin-daemon", + '/var/run/ceph/ceph-mon.myhost.asok', + "add_bootstrap_peer_hint", + 'mypeer' + ] + self.os.path.exists.return_value = True + ceph.add_bootstrap_hint('mypeer') + self.subprocess.call.assert_called_with(cmd) + + def test_add_bootstrap_hint_noasok(self): + self.get_unit_hostname.return_value = 'myhost' + self.os.path.exists.return_value = False + ceph.add_bootstrap_hint('mypeer') + self.assertFalse(self.subprocess.call.called) + + def test_is_osd_disk(self): + # XXX Insert real sgdisk output + self.subprocess.check_output.return_value = 'Partition GUID code: 4FBD7E29-9D25-41B8-AFD0-062C0CEFF05D' + self.assertEqual(ceph.is_osd_disk('/dev/fmd0'), True) + + def test_is_osd_disk_no(self): + # XXX Insert real sgdisk output + self.subprocess.check_output.return_value = 'Partition GUID code: 5FBD7E29-9D25-41B8-AFD0-062C0CEFF05D' + self.assertEqual(ceph.is_osd_disk('/dev/fmd0'), False) + + def test_rescan_osd_devices(self): + cmd = [ + 'udevadm', 'trigger', + '--subsystem-match=block', '--action=add' + ] + ceph.rescan_osd_devices() + self.subprocess.call.assert_called_with(cmd) + + def test_zap_disk(self): + cmd = [ + 'sgdisk', '--zap-all', '/dev/fmd0', + ] + ceph.zap_disk('/dev/fmd0') + self.subprocess.check_call.assert_called_with(cmd) + + def test_import_osd_bootstrap_key(self): + self.os.path.exists.return_value = False + cmd = [ + 'ceph-authtool', + '/var/lib/ceph/bootstrap-osd/ceph.keyring', + '--create-keyring', + '--name=client.bootstrap-osd', + '--add-key=mykey', + ] + ceph.import_osd_bootstrap_key('mykey') + self.subprocess.check_call.assert_called_with(cmd) + diff --git a/unit_tests/test_hooks.py b/unit_tests/test_hooks.py new file mode 100644 index 00000000..a49f2b90 --- /dev/null +++ b/unit_tests/test_hooks.py @@ -0,0 +1,303 @@ + +from mock import call, patch, MagicMock +from test_utils import CharmTestCase, patch_open + +import hooks as ceph_hooks + +TO_PATCH = [ + 'add_source', + 'apt_update', + 'apt_install', + 'config', + 'cmp_pkgrevno', + 'execd_preinstall', + 'enable_pocket', + 'get_host_ip', + 'get_unit_hostname', + 'glob', + 'is_apache_24', + 'log', + 'lsb_release', + 'open_port', + 'os', + 'related_units', + 'relation_ids', + 'relation_set', + 'relation_get', + 'render_template', + 'shutil', + 'subprocess', + 'sys', + 'unit_get', +] + +class CephRadosGWTests(CharmTestCase): + + def setUp(self): + super(CephRadosGWTests, self).setUp(ceph_hooks, TO_PATCH) + self.config.side_effect = self.test_config.get + self.test_config.set('source', 'distro') + self.test_config.set('key', 'secretkey') + self.test_config.set('use-syslog', False) + + def test_install_www_scripts(self): + self.glob.glob.return_value = ['files/www/bob'] + ceph_hooks.install_www_scripts() + self.shutil.copy.assert_called_with('files/www/bob', '/var/www/') + + def test_install_ceph_optimised_packages(self): + self.lsb_release.return_value = {'DISTRIB_CODENAME': 'vivid'} + git_url = 'http://gitbuilder.ceph.com' + fastcgi_source = ('http://gitbuilder.ceph.com/' + 'libapache-mod-fastcgi-deb-vivid-x86_64-basic/ref/master') + apache_source = ('http://gitbuilder.ceph.com/' + 'apache2-deb-vivid-x86_64-basic/ref/master') + calls = [ + call(fastcgi_source, key='6EAEAE2203C3951A'), + call(apache_source, key='6EAEAE2203C3951A'), + ] + ceph_hooks.install_ceph_optimised_packages() + self.add_source.assert_has_calls(calls) + + def test_install_packages(self): + self.test_config.set('use-ceph-optimised-packages', '') + ceph_hooks.install_packages() + self.add_source.assert_called_with('distro', 'secretkey') + self.apt_update.assert_called() + self.apt_install.assert_called_with(['radosgw', + 'libapache2-mod-fastcgi', + 'apache2', + 'ntp'], fatal=True) + + def test_install_optimised_packages(self): + self.test_config.set('use-ceph-optimised-packages', True) + _install_packages = self.patch('install_ceph_optimised_packages') + ceph_hooks.install_packages() + self.add_source.assert_called_with('distro', 'secretkey') + self.apt_update.assert_called() + _install_packages.assert_called() + self.apt_install.assert_called_with(['radosgw', + 'libapache2-mod-fastcgi', + 'apache2', + 'ntp'], fatal=True) + + def test_install(self): + _install_packages = self.patch('install_packages') + ceph_hooks.install() + self.execd_preinstall.assert_called() + _install_packages.assert_called() + self.enable_pocket.assert_called_with('multiverse') + self.os.makedirs.called_with('/var/lib/ceph/nss') + + def test_emit_cephconf(self): + _get_keystone_conf = self.patch('get_keystone_conf') + _get_auth = self.patch('get_auth') + _get_mon_hosts = self.patch('get_mon_hosts') + _get_auth.return_value = 'cephx' + _get_keystone_conf.return_value = {'keystone_key': 'keystone_value'} + _get_mon_hosts.return_value = ['10.0.0.1:6789', '10.0.0.2:6789'] + self.get_unit_hostname.return_value = 'bob' + self.os.path.exists.return_value = False + cephcontext = { + 'auth_supported': 'cephx', + 'mon_hosts': '10.0.0.1:6789 10.0.0.2:6789', + 'hostname': 'bob', + 'old_auth': False, + 'use_syslog': 'false', + 'keystone_key': 'keystone_value', + } + self.cmp_pkgrevno.return_value = 1 + with patch_open() as (_open, _file): + ceph_hooks.emit_cephconf() + self.os.makedirs.assert_called_with('/etc/ceph') + _open.assert_called_with('/etc/ceph/ceph.conf', 'w') + self.render_template.assert_called_with('ceph.conf', cephcontext) + + def test_emit_apacheconf(self): + self.is_apache_24.return_value = True + self.unit_get.return_value = '10.0.0.1' + apachecontext = { + "hostname": '10.0.0.1', + } + with patch_open() as (_open, _file): + ceph_hooks.emit_apacheconf() + _open.assert_called_with('/etc/apache2/sites-available/rgw.conf', 'w') + self.render_template.assert_called_with('rgw', apachecontext) + + def test_apache_sites24(self): + self.is_apache_24.return_value = True + ceph_hooks.apache_sites() + calls = [ + call(['a2dissite', '000-default']), + call(['a2ensite', 'rgw']), + ] + self.subprocess.check_call.assert_has_calls(calls) + + def test_apache_sites22(self): + self.is_apache_24.return_value = False + ceph_hooks.apache_sites() + calls = [ + call(['a2dissite', 'default']), + call(['a2ensite', 'rgw']), + ] + self.subprocess.check_call.assert_has_calls(calls) + + def test_apache_modules(self): + ceph_hooks.apache_modules() + calls = [ + call(['a2enmod', 'fastcgi']), + call(['a2enmod', 'rewrite']), + ] + self.subprocess.check_call.assert_has_calls(calls) + + def test_apache_reload(self): + ceph_hooks.apache_reload() + calls = [ + call(['service', 'apache2', 'reload']), + ] + self.subprocess.call.assert_has_calls(calls) + + def test_config_changed(self): + _install_packages = self.patch('install_packages') + _emit_cephconf = self.patch('emit_cephconf') + _emit_apacheconf = self.patch('emit_apacheconf') + _install_www_scripts = self.patch('install_www_scripts') + _apache_sites = self.patch('apache_sites') + _apache_modules = self.patch('apache_modules') + _apache_reload = self.patch('apache_reload') + ceph_hooks.config_changed() + _install_packages.assert_called() + _emit_cephconf.assert_called() + _emit_apacheconf.assert_called() + _install_www_scripts.assert_called() + _apache_sites.assert_called() + _apache_modules.assert_called() + _apache_reload.assert_called() + + def test_get_mon_hosts(self): + self.relation_ids.return_value = ['monrelid'] + self.related_units.return_value = ['monunit'] + self.relation_get.return_value = '10.0.0.1' + self.get_host_ip.return_value = '10.0.0.1' + self.assertEquals(ceph_hooks.get_mon_hosts(), ['10.0.0.1:6789']) + + def test_get_conf(self): + self.relation_ids.return_value = ['monrelid'] + self.related_units.return_value = ['monunit'] + self.relation_get.return_value = 'bob' + self.assertEquals(ceph_hooks.get_conf('key'), 'bob') + + def test_get_conf_nomatch(self): + self.relation_ids.return_value = ['monrelid'] + self.related_units.return_value = ['monunit'] + self.relation_get.return_value = '' + self.assertEquals(ceph_hooks.get_conf('key'), None) + + def test_get_auth(self): + self.relation_ids.return_value = ['monrelid'] + self.related_units.return_value = ['monunit'] + self.relation_get.return_value = 'bob' + self.assertEquals(ceph_hooks.get_auth(), 'bob') + + def test_get_keystone_conf(self): + self.test_config.set('operator-roles', 'admin') + self.test_config.set('cache-size', '42') + self.test_config.set('revocation-check-interval', '21') + self.relation_ids.return_value = ['idrelid'] + self.related_units.return_value = ['idunit'] + def _relation_get(key, unit, relid): + ks_dict = { + 'auth_protocol': 'https', + 'auth_host': '10.0.0.2', + 'auth_port': '8090', + 'admin_token': 'sectocken', + } + return ks_dict[key] + self.relation_get.side_effect = _relation_get + self.assertEquals(ceph_hooks.get_keystone_conf(), + {'auth_type': 'keystone', + 'auth_protocol': 'https', + 'admin_token': 'sectocken', + 'user_roles': 'admin', + 'auth_host': '10.0.0.2', + 'cache_size': '42', + 'auth_port': '8090', + 'revocation_check_interval': '21'}) + + def test_get_keystone_conf_missinginfo(self): + self.test_config.set('operator-roles', 'admin') + self.test_config.set('cache-size', '42') + self.test_config.set('revocation-check-interval', '21') + self.relation_ids.return_value = ['idrelid'] + self.related_units.return_value = ['idunit'] + def _relation_get(key, unit, relid): + ks_dict = { + 'auth_protocol': 'https', + 'auth_host': '10.0.0.2', + 'auth_port': '8090', + } + return ks_dict[key] if key in ks_dict else None + self.relation_get.side_effect = _relation_get + self.assertEquals(ceph_hooks.get_keystone_conf(), None) + + def test_mon_relation(self): + _emit_cephconf = self.patch('emit_cephconf') + _ceph = self.patch('ceph') + _restart = self.patch('restart') + self.relation_get.return_value = 'seckey' + ceph_hooks.mon_relation() + _restart.assert_called() + _ceph.import_radosgw_key.assert_called_with('seckey') + + def test_mon_relation_nokey(self): + _emit_cephconf = self.patch('emit_cephconf') + _ceph = self.patch('ceph') + _restart = self.patch('restart') + self.relation_get.return_value = None + ceph_hooks.mon_relation() + self.assertFalse(_ceph.import_radosgw_key.called) + self.assertFalse(_restart.called) + + def test_gateway_relation(self): + self.unit_get.return_value = 'myserver' + ceph_hooks.gateway_relation() + self.relation_set.assert_called_with(hostname='myserver', port=80) + + def test_start(self): + ceph_hooks.start() + self.subprocess.call.assert_called_with(['service', 'radosgw', 'start']) + + def test_stop(self): + ceph_hooks.stop() + self.subprocess.call.assert_called_with(['service', 'radosgw', 'stop']) + + def test_start(self): + ceph_hooks.restart() + self.subprocess.call.assert_called_with(['service', 'radosgw', 'restart']) + + def test_identity_joined_early_version(self): + self.cmp_pkgrevno.return_value = -1 + ceph_hooks.identity_joined() + self.sys.exit.assert_called_with(1) + + def test_identity_joined(self): + self.cmp_pkgrevno.return_value = 1 + self.test_config.set('region', 'region1') + self.test_config.set('operator-roles', 'admin') + self.unit_get.return_value = 'myserv' + ceph_hooks.identity_joined(relid='rid') + self.relation_set.assert_called_with(service='swift', + region='region1', + public_url='http://myserv:80/swift/v1', + internal_url='http://myserv:80/swift/v1', + requested_roles='admin', + rid='rid', + admin_url='http://myserv:80/swift') + + def test_identity_changed(self): + _emit_cephconf = self.patch('emit_cephconf') + _restart = self.patch('restart') + ceph_hooks.identity_changed() + _emit_cephconf.assert_called() + _restart.assert_called() + diff --git a/unit_tests/test_utils.py b/unit_tests/test_utils.py new file mode 100644 index 00000000..526a61f7 --- /dev/null +++ b/unit_tests/test_utils.py @@ -0,0 +1,119 @@ +import logging +import os +import unittest +import yaml + +from contextlib import contextmanager +from mock import patch, MagicMock + + +def load_config(): + '''Walk backwords from __file__ looking for config.yaml, + load and return the 'options' section' + ''' + config = None + f = __file__ + while config is None: + d = os.path.dirname(f) + if os.path.isfile(os.path.join(d, 'config.yaml')): + config = os.path.join(d, 'config.yaml') + break + f = d + + if not config: + logging.error('Could not find config.yaml in any parent directory ' + 'of %s. ' % file) + raise Exception + + return yaml.safe_load(open(config).read())['options'] + + +def get_default_config(): + '''Load default charm config from config.yaml return as a dict. + If no default is set in config.yaml, its value is None. + ''' + default_config = {} + config = load_config() + for k, v in config.iteritems(): + if 'default' in v: + default_config[k] = v['default'] + else: + default_config[k] = None + return default_config + + +class CharmTestCase(unittest.TestCase): + + def setUp(self, obj, patches): + super(CharmTestCase, self).setUp() + self.patches = patches + self.obj = obj + self.test_config = TestConfig() + self.test_relation = TestRelation() + self.patch_all() + + def patch(self, method): + _m = patch.object(self.obj, method) + mock = _m.start() + self.addCleanup(_m.stop) + return mock + + def patch_all(self): + for method in self.patches: + setattr(self, method, self.patch(method)) + + +class TestConfig(object): + + def __init__(self): + self.config = get_default_config() + + def get(self, attr=None): + if not attr: + return self.get_all() + try: + return self.config[attr] + except KeyError: + return None + + def get_all(self): + return self.config + + def set(self, attr, value): + if attr not in self.config: + raise KeyError + self.config[attr] = value + + +class TestRelation(object): + + def __init__(self, relation_data={}): + self.relation_data = relation_data + + def set(self, relation_data): + self.relation_data = relation_data + + def get(self, attr=None, unit=None, rid=None): + if attr is None: + return self.relation_data + elif attr in self.relation_data: + return self.relation_data[attr] + return None + + +@contextmanager +def patch_open(): + '''Patch open() to allow mocking both open() itself and the file that is + yielded. + Yields the mock for "open" and "file", respectively. + ''' + mock_open = MagicMock(spec=open) + mock_file = MagicMock(spec=file) + + @contextmanager + def stub_open(*args, **kwargs): + mock_open(*args, **kwargs) + yield mock_file + + with patch('__builtin__.open', stub_open): + yield mock_open, mock_file