How to use stop_fixture method in gabbi

Best Python code snippet using gabbi_python

fixtures.py

Source:fixtures.py

...57            'oslo_utils.uuidutils.generate_uuid',58            return_value=FAKE_UUID)59        patcher.start()60        self.patcher = patcher61    def stop_fixture(self):62        self.patcher.stop()63class BaseExtensionFixture(fixture.GabbiFixture, metaclass=abc.ABCMeta):64    klass = None65    namespace = None66    stevedore_mgr = None67    assert_args = {}68    @abc.abstractmethod69    def setup_fake_modules(self):70        pass71    def start_fixture(self):72        fake_extensions = self.setup_fake_modules()73        self.mock = mock.patch(self.klass)74        fake_mgr = self.stevedore_mgr.make_test_instance(75            fake_extensions,76            self.namespace)77        self.patch = self.mock.start()78        self.patch.return_value = fake_mgr79    def stop_fixture(self):80        self.patch.assert_called_with(81            self.namespace,82            **self.assert_args)83        self.mock.stop()84class CollectorExtensionsFixture(BaseExtensionFixture):85    klass = 'stevedore.driver.DriverManager'86    namespace = 'cloudkitty.collector.backends'87    stevedore_mgr = driver.DriverManager88    assert_args = {89        'invoke_kwds': {'period': 3600},90        'invoke_on_load': True}91    def setup_fake_modules(self):92        def fake_metric(start,93                        end=None,94                        project_id=None,95                        q_filter=None):96            return None97        fake_module1 = tests.FakeCollectorModule()98        fake_module1.collector_name = 'fake1'99        fake_module1.get_compute = fake_metric100        fake_module2 = tests.FakeCollectorModule()101        fake_module2.collector_name = 'fake2'102        fake_module2.get_volume = fake_metric103        fake_module3 = tests.FakeCollectorModule()104        fake_module3.collector_name = 'fake3'105        fake_module3.get_compute = fake_metric106        fake_extensions = [107            extension.Extension(108                'fake1',109                'cloudkitty.tests.FakeCollectorModule1',110                None,111                fake_module1),112            extension.Extension(113                'fake2',114                'cloudkitty.tests.FakeCollectorModule2',115                None,116                fake_module2),117            extension.Extension(118                'fake3',119                'cloudkitty.tests.FakeCollectorModule3',120                None,121                fake_module3)]122        return fake_extensions[0]123class RatingModulesFixture(BaseExtensionFixture):124    klass = 'stevedore.extension.ExtensionManager'125    namespace = 'cloudkitty.rating.processors'126    stevedore_mgr = extension.ExtensionManager127    assert_args = {128        'invoke_on_load': True}129    def setup_fake_modules(self):130        class FakeConfigController(rating.RatingRestControllerBase):131            _custom_actions = {132                'test': ['GET']133            }134            @wsme_pecan.wsexpose(wtypes.text)135            def get_test(self):136                """Return the list of every mapping type available.137                """138                return 'OK'139        fake_module1 = tests.FakeRatingModule()140        fake_module1.module_name = 'fake1'141        fake_module1.set_priority(3)142        fake_module2 = tests.FakeRatingModule()143        fake_module2.module_name = 'fake2'144        fake_module2.config_controller = FakeConfigController145        fake_module2.set_priority(1)146        fake_module3 = tests.FakeRatingModule()147        fake_module3.module_name = 'fake3'148        fake_module3.set_priority(2)149        fake_extensions = [150            extension.Extension(151                'fake1',152                'cloudkitty.tests.FakeRatingModule1',153                None,154                fake_module1),155            extension.Extension(156                'fake2',157                'cloudkitty.tests.FakeRatingModule2',158                None,159                fake_module2),160            extension.Extension(161                'fake3',162                'cloudkitty.tests.FakeRatingModule3',163                None,164                fake_module3)]165        return fake_extensions166class ConfigFixture(fixture.GabbiFixture):167    auth_strategy = 'noauth'168    def start_fixture(self):169        self.conf = None170        conf = conf_fixture.Config().conf171        policy_opts.set_defaults(conf)172        msg_conf = conffixture.ConfFixture(conf)173        msg_conf.transport_url = 'fake:/'174        conf.import_group('api', 'cloudkitty.api.app')175        conf.set_override('auth_strategy', self.auth_strategy)176        conf.set_override('connection', 'sqlite:///', 'database')177        conf.set_override('policy_file',178                          os.path.abspath('etc/cloudkitty/policy.yaml'),179                          group='oslo_policy')180        conf.set_override('api_paste_config',181                          os.path.abspath(182                              'cloudkitty/tests/gabbi/gabbi_paste.ini')183                          )184        conf.import_group('storage', 'cloudkitty.storage')185        conf.set_override('backend', 'sqlalchemy', 'storage')186        conf.set_override('version', '1', 'storage')187        self.conf = conf188        self.conn = ck_db_api.get_instance()189        migration = self.conn.get_migration()190        migration.upgrade('head')191    def stop_fixture(self):192        if self.conf:193            self.conf.reset()194        db.get_engine().dispose()195class ConfigFixtureStorageV2(ConfigFixture):196    def start_fixture(self):197        super(ConfigFixtureStorageV2, self).start_fixture()198        self.conf.set_override('backend', 'influxdb', 'storage')199        self.conf.set_override('version', '2', 'storage')200class ConfigFixtureKeystoneAuth(ConfigFixture):201    auth_strategy = 'keystone'202    def start_fixture(self):203        # Mocking the middleware process_request which check for credentials204        # here, the only check done is that the hardcoded token is the one205        # send by the query. If not, 401, else 200.206        def _mock_proc_request(self, request):207            token = 'c93e3e31342e4e32ba201fd3d70878b5'208            http_code = 401209            if 'X-Auth-Token' in request.headers and \210               request.headers['X-Auth-Token'] == token:211                http_code = 200212            return webob.Response(213                status_code=http_code,214                content_type='application/json'215            )216        self._orig_func = middleware.auth_token.AuthProtocol.process_request217        middleware.auth_token.AuthProtocol.process_request = _mock_proc_request218        super(ConfigFixtureKeystoneAuth, self).start_fixture()219    def stop_fixture(self):220        super(ConfigFixtureKeystoneAuth, self).stop_fixture()221        middleware.auth_token.AuthProtocol.process_request = self._orig_func222class BaseFakeRPC(fixture.GabbiFixture):223    endpoint = None224    def start_fixture(self):225        messaging.setup()226        target = oslo_messaging.Target(topic='cloudkitty',227                                       server=cfg.CONF.host,228                                       version='1.0')229        endpoints = [230            self.endpoint()231        ]232        self.server = messaging.get_server(target, endpoints)233        self.server.start()234    def stop_fixture(self):235        self.server.stop()236class ScopeStateResetFakeRPC(BaseFakeRPC):237    class FakeRPCEndpoint(object):238        target = oslo_messaging.Target(version='1.0')239        def reset_state(self, ctxt, res_data):240            pass241    endpoint = FakeRPCEndpoint242class QuoteFakeRPC(BaseFakeRPC):243    class FakeRPCEndpoint(object):244        target = oslo_messaging.Target(namespace='rating',245                                       version='1.0')246        def quote(self, ctxt, res_data):247            return str(1.0)248    endpoint = FakeRPCEndpoint249class BaseStorageDataFixture(fixture.GabbiFixture):250    def create_fake_data(self, begin, end, project_id):251        cpu_point = dataframe.DataPoint(252            unit="nothing",253            qty=1,254            groupby={"fake_meta": 1.0, "project_id": project_id},255            metadata={"dummy": True},256            price=decimal.Decimal('1.337'),257        )258        image_point = dataframe.DataPoint(259            unit="nothing",260            qty=1,261            groupby={"fake_meta": 1.0, "project_id": project_id},262            metadata={"dummy": True},263            price=decimal.Decimal('0.121'),264        )265        data = [266            dataframe.DataFrame(267                start=begin, end=end,268                usage=collections.OrderedDict({"cpu": [cpu_point, cpu_point]}),269            ),270            dataframe.DataFrame(271                start=begin, end=end,272                usage=collections.OrderedDict(273                    {"image.size": [image_point, image_point]}),274            ),275        ]276        return data277    def start_fixture(self):278        auth = mock.patch(279            'keystoneauth1.loading.load_auth_from_conf_options',280            return_value=dict())281        session = mock.patch(282            'keystoneauth1.loading.load_session_from_conf_options',283            return_value=dict())284        with auth:285            with session:286                self.storage = storage.get_storage(conf=test_utils.load_conf())287        self.storage.init()288        self.initialize_data()289    def stop_fixture(self):290        model = models.RatedDataFrame291        session = db.get_session()292        q = utils.model_query(293            model,294            session)295        q.delete()296class StorageDataFixture(BaseStorageDataFixture):297    def initialize_data(self):298        nodata_duration = (24 * 3 + 12) * 3600299        hour_delta = datetime.timedelta(seconds=3600)300        tenant_list = ['8f82cc70-e50c-466e-8624-24bdea811375',301                       '7606a24a-b8ad-4ae0-be6c-3d7a41334a2e']302        data_dt = INITIAL_DT + datetime.timedelta(303            seconds=nodata_duration + 3600)304        data_duration = datetime.timedelta(seconds=(24 * 2 + 8) * 3600)305        iter_dt = data_dt306        while iter_dt < data_dt + data_duration:307            data = self.create_fake_data(308                iter_dt, iter_dt + hour_delta, tenant_list[0])309            self.storage.push(data, tenant_list[0])310            iter_dt += hour_delta311        iter_dt = data_dt312        while iter_dt < data_dt + data_duration / 2:313            data = self.create_fake_data(314                iter_dt, iter_dt + hour_delta, tenant_list[1])315            self.storage.push(data, tenant_list[1])316            iter_dt += hour_delta317class NowStorageDataFixture(BaseStorageDataFixture):318    def initialize_data(self):319        dt = tzutils.get_month_start(naive=True).replace(tzinfo=tz.tzutc())320        hour_delta = datetime.timedelta(seconds=3600)321        limit = dt + hour_delta * 12322        while dt < limit:323            project_id = '3d9a1b33-482f-42fd-aef9-b575a3da9369'324            data = self.create_fake_data(dt, dt + hour_delta, project_id)325            self.storage.push(data, project_id)326            dt += hour_delta327class ScopeStateFixture(fixture.GabbiFixture):328    def start_fixture(self):329        self.sm = storage_state.StateManager()330        self.sm.init()331        data = [332            ('aaaa', datetime.datetime(2019, 1, 1), 'fet1', 'col1', 'key1'),333            ('bbbb', datetime.datetime(2019, 2, 2), 'fet1', 'col1', 'key2'),334            ('cccc', datetime.datetime(2019, 3, 3), 'fet1', 'col2', 'key1'),335            ('dddd', datetime.datetime(2019, 4, 4), 'fet1', 'col2', 'key2'),336            ('eeee', datetime.datetime(2019, 5, 5), 'fet2', 'col1', 'key1'),337            ('ffff', datetime.datetime(2019, 6, 6), 'fet2', 'col1', 'key2'),338            ('gggg', datetime.datetime(2019, 6, 6), 'fet2', 'col2', 'key1'),339            ('hhhh', datetime.datetime(2019, 6, 6), 'fet2', 'col2', 'key2'),340        ]341        for d in data:342            self.sm.set_state(343                d[0], d[1], fetcher=d[2], collector=d[3], scope_key=d[4])344    def stop_fixture(self):345        session = db.get_session()346        q = utils.model_query(347            self.sm.model,348            session)349        q.delete()350class CORSConfigFixture(fixture.GabbiFixture):351    """Inject mock configuration for the CORS middleware."""352    def start_fixture(self):353        # Here we monkeypatch GroupAttr.__getattr__, necessary because the354        # paste.ini method of initializing this middleware creates its own355        # ConfigOpts instance, bypassing the regular config fixture.356        def _mock_getattr(instance, key):357            if key != 'allowed_origin':358                return self._original_call_method(instance, key)359            return "http://valid.example.com"360        self._original_call_method = cfg.ConfigOpts.GroupAttr.__getattr__361        cfg.ConfigOpts.GroupAttr.__getattr__ = _mock_getattr362    def stop_fixture(self):363        """Remove the monkeypatch."""364        cfg.ConfigOpts.GroupAttr.__getattr__ = self._original_call_method365class MetricsConfFixture(fixture.GabbiFixture):366    """Inject Metrics configuration mock to the get_metrics_conf() function"""367    def start_fixture(self):368        self._original_function = ck_utils.load_conf369        ck_utils.load_conf = mock.Mock(370            return_value=tests.samples.METRICS_CONF,371        )372    def stop_fixture(self):373        """Remove the get_metrics_conf() monkeypatch."""374        ck_utils.load_conf = self._original_function375class NowInfluxStorageDataFixture(NowStorageDataFixture):376    def start_fixture(self):377        cli = influx_utils.FakeInfluxClient()378        st = storage.get_storage()379        st._conn = cli380        self._get_storage_patch = mock.patch(381            'cloudkitty.storage.get_storage',382            new=lambda **kw: st,383        )384        self._get_storage_patch.start()385        v2_api_summary.Summary.reload()386        v2_api_dataframes.DataFrameList.reload()387        super(NowInfluxStorageDataFixture, self).start_fixture()388    def initialize_data(self):389        data = test_utils.generate_v2_storage_data(390            start=tzutils.get_month_start(),391            end=tzutils.localized_now().replace(hour=0),392        )393        self.storage.push([data])394    def stop_fixture(self):395        self._get_storage_patch.stop()396class InfluxStorageDataFixture(StorageDataFixture):397    def start_fixture(self):398        cli = influx_utils.FakeInfluxClient()399        st = storage.get_storage()400        st._conn = cli401        self._get_storage_patch = mock.patch(402            'cloudkitty.storage.get_storage',403            new=lambda **kw: st,404        )405        self._get_storage_patch.start()406        v2_api_summary.Summary.reload()407        v2_api_dataframes.DataFrameList.reload()408        super(InfluxStorageDataFixture, self).start_fixture()409    def stop_fixture(self):410        self._get_storage_patch.stop()411class UTCFixture(fixture.GabbiFixture):412    """Set the local timezone to UTC"""413    def start_fixture(self):414        self._tzmock = mock.patch('cloudkitty.utils.tz._LOCAL_TZ', tz.tzutc())415        self._tzmock.start()416    def stop_fixture(self):417        self._tzmock.stop()418def setup_app():419    messaging.setup()420    # FIXME(sheeprine): Extension fixtures are interacting with transformers421    # loading, since collectors are not needed here we shunt them422    no_collector = mock.patch(423        'cloudkitty.collector.get_collector',424        return_value=None)425    with no_collector:...

fixture.py

Source:fixture.py

...34    """35    def __enter__(self):36        self.start_fixture()37    def __exit__(self, exc_type, value, traceback):38        self.stop_fixture()39    def start_fixture(self):40        """Implement the actual workings of starting the fixture here."""41        pass42    def stop_fixture(self):43        """Implement the actual workings of stopping the fixture here."""44        pass45class InterceptFixture(GabbiFixture):46    """Start up the wsgi intercept. This should not be called directly."""47    httplib2_intercept.install()48    def __init__(self, host, port, app):49        self.host = host50        self.port = port51        self.app = app52    def start_fixture(self):53        wsgi_intercept.add_wsgi_intercept(self.host, self.port, self.app)54    def stop_fixture(self):55        wsgi_intercept.remove_wsgi_intercept(self.host, self.port)56class SkipAllFixture(GabbiFixture):57    """A fixture that skips all the tests in the current suite."""58    def start_fixture(self):59        raise case.SkipTest('entire suite skipped')60@contextlib.contextmanager61def nest(fixtures):62    """Nest a series of fixtures.63    This is duplicated from ``nested`` in the stdlib, which has been64    deprecated because of issues with how exceptions are difficult to65    handle during ``__init__``. Gabbi needs to nest an unknown number66    of fixtures dynamically, so the ``with`` syntax that replaces67    ``nested`` will not work.68    """...

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.