How to use get_resource_dependencies method in localstack

Best Python code snippet using localstack_python

test_job.py

Source:test_job.py

...129        job = JobDefinition({130            'name': 'name',131            'plugin': 'plugin'})132        expected = set()133        observed = job.get_resource_dependencies()134        self.assertEqual(expected, observed)135    def test_resource_parsing_typical(self):136        job = JobDefinition({137            'name': 'name',138            'plugin': 'plugin',139            'requires': 'foo.bar == 10'})140        expected = set(['foo'])141        observed = job.get_resource_dependencies()142        self.assertEqual(expected, observed)143    def test_resource_parsing_many(self):144        job = JobDefinition({145            'name': 'name',146            'plugin': 'plugin',147            'requires': (148                "foo.bar == 10\n"149                "froz.bot == 10\n")})150        expected = set(['foo', 'froz'])151        observed = job.get_resource_dependencies()152        self.assertEqual(expected, observed)153    def test_resource_parsing_broken(self):154        job = JobDefinition({155            'name': 'name',156            'plugin': 'plugin',157            'requires': "foo.bar == bar"})158        self.assertRaises(Exception, job.get_resource_dependencies)159    def test_checksum_smoke(self):160        job1 = JobDefinition({161            'name': 'name',162            'plugin': 'plugin'163        })164        identical_to_job1 = JobDefinition({165            'name': 'name',...

context_eval_as_constant.py

Source:context_eval_as_constant.py

...19        20        We should also check that we can actually solve21        each one... maybe limit to WrapAMap?22    """23    dependencies = get_resource_dependencies(context, r)24    # print('This depends on %r' % dependencies)25    not_constants = [_ for _ in dependencies if context.is_new_function(_) ]26    if not_constants:27        # print('Not constant because of these deps: %s' % not_constants)28        return False29    else:30        return True31@contract(r=CResource, returns=ValueWithUnits)32def eval_constant_resource(context, r):33    assert can_resource_be_constant(context, r)34    def find_connection_to(s2, dp2):35        for c in context.connections:36            if c.s2 == s2 and c.dp2 == dp2:37                return c38        assert False39    @contract(functions=dict)40    def run_ndp(ndp, functions):41        rnames = ndp.get_rnames()42        f = []43        for fn in ndp.get_fnames():44            vu = functions[fn]45            f.append(vu.value)46        f = tuple(f)47        if len(ndp.get_fnames()) == 1:48            f = f[0]49        dp = ndp.get_dp()50        if isinstance(dp, Constant):51            res = ValueWithUnits(dp.c, dp.R)52        elif isinstance(dp, WrapAMap):53            amap = dp.amap54            if len(rnames) == 1:55                res = ValueWithUnits(amap(f), dp.get_res_space())56            else:57                res_R = ndp.get_rtypes(rnames)58                r = amap(f)59                assert isinstance(res_R, PosetProduct) and len(res_R) == len(rnames)60                assert isinstance(r, tuple) and len(r) == len(rnames)61                res = [ValueWithUnits(value=value, unit=unit)62                for value, unit in zip(r, res_R.subs)]63        else:64            raise NotImplementedError(type(dp))65        if len(rnames) == 1:66            resources = {rnames[0]: res}67        else:68            resources = {}69            for i, rn in enumerate(rnames):70                resources[rn] = res[i]71        return resources72    @contract(r=CResource, returns=ValueWithUnits)73    def evaluate(r):74        dp = r.dp75        ndp = context.names[dp]76        # we need to evaluate all functions77        functions = {}78        for f in ndp.get_fnames():79            c = find_connection_to(s2=f, dp2=dp)80            functions[f] = evaluate(CResource(c.dp1, c.s1))81        results = run_ndp(ndp, functions)82        return results[r.s]83    return evaluate(r)84@contract(r=CResource)85def get_resource_dependencies(context, r):86    G = cndp_get_digraph(context.names, context.connections)87    # print G.nodes(), G.edges()88    # XXX do we need to remove the self-loops?89    from networkx.algorithms.dag import ancestors90    all_ancestors = ancestors(G, r.dp)91    # print('ancestors: %s')92    return all_ancestors93def cndp_get_digraph(name2ndps, connections):94    import networkx as nx95    G = nx.DiGraph()96    for n in name2ndps:97        G.add_node(n)98    for c in connections:99        G.add_edge(c.dp1, c.dp2)...

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.