Best Python code snippet using slash
plugin.py
Source:plugin.py  
...144    :return py:class:`DAG`: a tuple that contains the DAG145    """146    dag_default_args_fix = _get_fixture("dag_default_args", session)147    dag_fix = _get_fixture("dag", session)148    dag_default_args = _compute_fixture_value(dag_default_args_fix, session)149    dag = _compute_fixture_value(dag_fix, session)150    # we clean up any resources created during dag initialization151    # in an attempt to not mess up with pytest normals operation152    session._setupstate._callfinalizers(session)153    session.items[0]._fixture_defs = {}154    return dag155def _get_fixture(argname, session):156    """ Get the fixture named after argname from the session.157    The fixture must be located at a nodeid that must be the parent of all the158    collected test items.  This is so, because the plugin will only construct a159    single DAG, if the user specifies a fixture that covers only part of the160    requested test items, the desired DAG becomes ambiguous since we do not161    know if a given fixture applies to items it was not meant to cover.162    :return: returns an instance of :py:class:`FixtureDef` if fixture found163    otherwise `None`.164    """165    # look for the requested fixturedefs in the fixturemanager166    fixs = session._fixturemanager._arg2fixturedefs.get(argname, [])167    # we first want to make sure that the fixtures registered in this plugin168    # are read only at the end in case the user or no other plugin registered169    # fixtures with the same name meant for use instead of the default fixtures170    # defined here.171    for i, fix in enumerate(fixs):172        # is the fixture function defined in the same file as this one?173        # then it must be the default fixture.174        # pytest does not record the location of fixtures registered through175        # plugin, that is the reason why we have to inspect the obejct176        if fix.func.__globals__["__name__"] != __name__:177            continue178        # if the fixture is not at the top of the list, let's put it there.  we179        # modify the list just before we break the loop, so there should be no180        # risk to modifying the list in the loop itself.181        del fixs[i]182        fixs.insert(0, fix)183        break184    # loop through the reversed list of fixtures (so, from the more narrow185    # location to the more broad), the first fixture that encompasses all the186    # colleted items found and whose scope is "session" is returned187    parent_fix = None188    for fix in fixs[::-1]:189        for item in session.items:190            if not nodes.ischildnode(fix.baseid, item._nodeid):191                parent_fix = None192                break193            parent_fix = fix194        if parent_fix and parent_fix.scope == "session":195            break196    return parent_fix197def _compute_fixture_value(fixturedef, session):198    """ Computes the fixturedef evaluated in the scope of the first item in the199    list of collected items. """200    # since the fixturedef by definition is located above all of the collected201    # items, its evaluation should be the same regardless of the item chosen202    # for scoping.203    request = session.items[0]._request204    request._compute_fixture_value(fixturedef)205    res = fixturedef.cached_result[0]206    # we add the fixturedef to force this fixturedef to be found207    # this so that for instance our fixture dag will use dag_default_args that208    # we compute here209    request._fixture_defs[fixturedef.argname] = fixturedef210    return res211def _pytest_branch_callable(items):212    """ Generates the callable for the MultiBranchPythonOperator taking into213    account the list of collected items. """214    def _callable(**kwargs):215        # we copy the items, as the list is modified by `pytest` throughout the216        # script life cycle.217        _items = items.copy()218        markerexprs = []...fixture_store.py
Source:fixture_store.py  
...243        return fixtures_set244    def get_fixture_value(self, fixture, name=None):245        if name is None:246            name = fixture.info.name247        value = self._compute_fixture_value(name, fixture)248        return value249    def get_value(self, variation, parameter_or_fixture):250        fixture_id = parameter_or_fixture.info.id251        fixtureobj = self.get_fixture_by_id(parameter_or_fixture.info.id)252        if isinstance(fixtureobj, Parametrization):253            value = parameter_or_fixture.get_value_by_index(variation.param_value_indices[fixture_id])254        else:255            value = self.get_fixture_value(parameter_or_fixture)256        return value257    def iter_parametrization_variations(self, fixture_ids=(), funcs=(), methods=()):258        if self._unresolved_fixture_ids:259            raise UnresolvedFixtureStore()260        variation_factory = VariationFactory(self)261        for fixture_id in fixture_ids:262            variation_factory.add_needed_fixture_id(fixture_id)263        for func in funcs:264            variation_factory.add_needed_fixtures_from_function(func)265        for method in methods:266            variation_factory.add_needed_fixtures_from_method(method)267        return variation_factory.iter_variations()268    def _compute_fixture_value(self, name, fixture, relative_name=None):269        if relative_name is None:270            relative_name = name271        assert not fixture.is_parameter()272        if fixture.info.id in self._computing:273            raise CyclicFixtureDependency(274                'Fixture {!r} is a part of a dependency cycle!'.format(name))275        active_fixture = self.get_active_fixture(fixture)276        if active_fixture is not None:277            if self._is_active_fixture_valid(fixture):278                _logger.trace("Fixture {} did not change", fixture)279                return active_fixture.value280            else:281                _logger.trace("Fixture {} no longer valid. Recomputing", fixture)282                self._deactivate_fixture(active_fixture.fixture)283        self._computing.add(fixture.info.id)284        try:285            fixture_value = self._call_fixture(fixture, relative_name=relative_name)286        except:287            exc_info = sys.exc_info()288            self._deactivate_fixture(fixture)289            reraise(*exc_info)290        finally:291            self._computing.discard(fixture.info.id)292        return fixture_value293    def _is_active_fixture_valid(self, fixture):294        assert fixture.info.id in self._active_fixture_dependencies, "Fixture dependencies not updated"295        new_dependencies = self._compute_fixture_dependencies(fixture)296        return new_dependencies.issubset(self._active_fixture_dependencies[fixture.info.id])297    def _compute_fixture_dependencies(self, fixture):298        param_indices = self._compute_all_needed_parametrization_ids(fixture)299        if not param_indices:300            return frozenset()301        assert ctx.session is not None, "Dependency computation requires an active session"302        variation = ctx.session.variations.get_current_variation()303        assert variation is not None, "Dependency computation requires current variation"304        return frozenset((param_id, variation.param_value_indices[param_id])305                         for param_id in self._compute_all_needed_parametrization_ids(fixture))306    def _call_fixture(self, fixture, relative_name):307        assert relative_name308        active_fixture = ActiveFixture(fixture)309        kwargs = {}310        if fixture.keyword_arguments is None:311            raise UnresolvedFixtureStore('Fixture {} is unresolved!'.format(fixture.info.name))312        for required_name, needed_fixture in fixture.keyword_arguments.items():313            if needed_fixture.is_parameter():314                continue315            kwargs[required_name] = self._compute_fixture_value(316                required_name, needed_fixture,317                relative_name='{} -> {}'.format(relative_name, required_name))318        assert fixture.info.id not in self._active_fixtures_by_scope[fixture.info.scope]319        _logger.trace("Activating fixture {}...", fixture)320        self._active_fixtures_by_scope[fixture.info.scope][fixture.info.id] = active_fixture321        self._active_fixture_dependencies[fixture.info.id] = self._compute_fixture_dependencies(fixture)322        prev_context_fixture = slash_context.fixture323        slash_context.fixture = active_fixture324        try:325            returned = active_fixture.value = fixture.get_value(kwargs, active_fixture)326        finally:327            slash_context.fixture = prev_context_fixture328        _logger.trace(' -- {} = {!r}', relative_name, returned)329        return returned...fixtures.py
Source:fixtures.py  
...72    def cached_store_for_module(self):73        return self.node.module74    def cached_store_for_session(self):75        return self.node.session76    def _compute_fixture_value(self, fixturedef):77        """78        Creates a SubRequest based on "self" and calls the execute method of the given79        fixturedef object. This will force the FixtureDef object to throw away any previous results80        and compute a new fixture value, which will be stored into the FixtureDef object itself.81        :param FixtureDef fixturedef:82        """83        # prepare a subrequest object before calling fixture function84        # (latter managed by fixturedef)85        argname = fixturedef.argname86        funcitem = self._pyfuncitem87        scope = fixturedef.scope88        try:89            param = funcitem.callspec.getparam(argname)90        except (AttributeError, ValueError):...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
