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!!