How to use object_from_name method in Nose2

Best Python code snippet using nose2

universe.py

Source:universe.py

1from collections import defaultdict, deque2from itertools import product3from ss.model.functions import Object, Function, Predicate, initialize, process_domain, Atom, Predicate, NegatedAtom4from ss.model.problem import reset_derived, apply_axioms, dump_evaluations5from ss.model.operators import applicable, apply, Goal6from ss.model.streams import Stream7from ss.to_pddl import pddl_domain, pddl_problem8def get_mapping(atoms1, atoms2, initial={}):9    assert len(atoms1) == len(atoms2)10    mapping = initial.copy()11    for t, a in zip(atoms1, atoms2):12        assert t.head.function is a.head.function13        for p, v in zip(t.head.args, a.head.args):14            if mapping.get(p, v) == v:15                mapping[p] = v16            else:17                return None18    return mapping19class Universe(object):20    _domain_name = 'stripstream'21    _problem_name = _domain_name22    def __init__(self, problem, initial, use_bounds, only_eager, evaluate=True):23        self.problem = problem24        self.use_bounds = use_bounds25        self.only_eager = only_eager26        self.evaluate = evaluate27        self.evaluations = set()28        self.value_from_head = {}29        self.atoms_from_predicate = defaultdict(set)30        self.fluents = self.problem.fluents()31        self.computed = set()32        self.name_from_object = {}33        self.object_from_name = {}34        self.action_from_name = {}35        for action in problem.actions:36            assert action.name not in self.action_from_name37            self.action_from_name[action.name] = action38        self.axioms_from_derived = defaultdict(list)39        for axiom in problem.axioms:40            self.axioms_from_derived[axiom.effect.head.function].append(axiom)41        self.functions = set()42        for action in (problem.actions + problem.axioms):43            self.functions.update(action.functions())44        for literal in problem.goal:45            self.functions.add(literal.head.function)46        self.streams_from_predicate = defaultdict(list)47        self.stream_queue = deque()48        self.stream_instances = set()49        for stream in problem.streams:50            if only_eager and not stream.eager:51                continue52            for i, atom in enumerate(stream.domain):53                self.streams_from_predicate[54                    atom.head.function].append((stream, i))55        self.defined_functions = {f for f in self.functions if f.is_defined()}56        for func in self.defined_functions:57            if (use_bounds and (not func.is_bound_defined())) or (only_eager and (not func.eager)):58                continue59            if use_bounds and isinstance(func, Predicate) and (func.bound is False):60                continue61            for i, atom in enumerate(func.domain):62                self.streams_from_predicate[63                    atom.head.function].append((func, i))64        for atom in initial:65            self.add_eval(atom)66        for action in problem.actions:67            for obj in action.constants():68                self.add_eval(Object(obj))69        for literal in problem.goal:70            for obj in literal.head.args:71                self.add_eval(Object(obj))72        for stream in problem.streams:73            if only_eager and not stream.eager:74                continue75            if not stream.domain:76                self._add_instance(stream, {})77    def _add_object(self, obj):78        if obj in self.name_from_object:79            return80        name = 'x{}'.format(len(self.name_from_object))81        self.name_from_object[obj] = name82        assert name not in self.object_from_name83        self.object_from_name[name] = obj84    def _add_instance(self, relation, mapping):85        inputs = tuple(mapping[p] for p in relation.inputs)86        if isinstance(relation, Stream):87            instance = relation.get_instance(inputs)88            if not instance.enumerated and (instance not in self.stream_instances):89                self.stream_instances.add(instance)90                self.stream_queue.append(instance)91        elif isinstance(relation, Function):92            head = relation.get_head(inputs)93            if not head.computed() and (head not in self.computed):94                if self.use_bounds:95                    self.add_eval(head.get_bound())96                else:97                    self.add_eval(head.get_eval())98        else:99            raise ValueError(relation)100    def _update_stream_instances(self, atom):101        if not self.evaluate:102            return103        for relation, i in self.streams_from_predicate[atom.head.function]:104            values = [self.atoms_from_predicate.get(a.head.function, {}) if i != j else {atom}105                      for j, a in enumerate(relation.domain)]106            for combo in product(*values):107                mapping = get_mapping(relation.domain, combo)108                if mapping is not None:109                    self._add_instance(relation, mapping)110    def is_fluent(self, e):111        return e.head.function in self.fluents112    def is_derived(self, e):113        return e.head.function in self.axioms_from_derived114    def is_static(self, e):115        return not self.is_derived(e) and not self.is_fluent(e)116    def _operator_instances(self, operator):117        static_atoms = process_domain({a for a in operator.preconditions if isinstance(a, Atom) and self.is_static(a)}118                                      | {Object(p) for p in operator.parameters})119        values = [self.atoms_from_predicate.get(120            a.head.function, {}) for a in static_atoms]121        for combo in product(*values):122            mapping = get_mapping(static_atoms, combo)123            if mapping is not None:124                yield operator, tuple(mapping[p] for p in operator.parameters)125    def action_instances(self):126        for action in self.action_from_name.values():127            for instance in self._operator_instances(action):128                yield instance129    def axiom_instances(self):130        for axioms in self.axioms_from_derived.values():131            for axiom in axioms:132                for instance in self._operator_instances(axiom):133                    yield instance134    def add_eval(self, eval):135        if eval in self.evaluations:136            return137        if self.value_from_head.get(eval.head, eval.value) != eval.value:138            raise ValueError('{}: {} != {}'.format(139                eval.head, self.value_from_head[eval.head], eval.value))140        self.value_from_head[eval.head] = eval.value141        self.functions.add(eval.head.function)142        self.evaluations.add(eval)143        for obj in eval.head.args:144            self._add_object(obj)145        if isinstance(eval, Atom) and (eval not in self.atoms_from_predicate[eval.head.function]):146            self.atoms_from_predicate[eval.head.function].add(eval)147            self._update_stream_instances(eval)148        for implied in eval.head.implied():149            self.add_eval(implied)150    def pddl(self):151        predicates = set(152            filter(lambda f: isinstance(f, Predicate), self.functions))153        functions = self.functions - predicates154        initial_str = [e.substitute(self.name_from_object)155                       for e in self.evaluations if not isinstance(e, NegatedAtom)]156        goal_str = [l.substitute(self.name_from_object)157                    for l in self.problem.goal]158        return pddl_domain(self._domain_name,159                           self.object_from_name.keys(),160                           predicates, functions,161                           [a.substitute_constants(self.name_from_object)162                            for a in self.action_from_name.values()],163                           [a.substitute_constants(self.name_from_object)164                            for axioms in self.axioms_from_derived.values() for a in axioms]),               pddl_problem(self._domain_name, self._problem_name,165                                                                                                                          [],166                                                                                                                          initial_str, goal_str,167                                                                                                                          self.problem.objective)168    def state_fluents(self, state):169        return frozenset(filter(lambda e: not self.is_static(e),170                                (initialize(h, v) for h, v in state.iteritems())))171    def print_plan(self, plan):172        plan_instances = [self.action_from_name[173            name].instantiate(args) for name, args in plan]174        print plan_instances175        axioms = [axiom.instantiate(args)176                  for axiom, args in self.axiom_instances()]177        state = apply(self.evaluations, defaultdict(bool))178        reset_derived(self.axioms_from_derived, state)179        print 0, self.state_fluents(state)180        for i, instance in enumerate(plan_instances):181            apply_axioms(axioms, state)182            assert instance.applicable(state)183            state = instance.apply(state)184            reset_derived(self.axioms_from_derived, state)185            print i + 1, self.state_fluents(state)186        apply_axioms(axioms, state)187        assert applicable(self.problem.goal, state)188    def is_solution(self, plan):189        plan_instances = [action.instantiate(190            args) for action, args in plan] + [Goal(self.problem.goal)]191        axiom_instances = [axiom.instantiate(192            args) for axiom, args in self.axiom_instances()]193        state = apply(self.evaluations, defaultdict(bool))194        for instance in plan_instances:195            reset_derived(self.axioms_from_derived, state)196            apply_axioms(axiom_instances, state)197            if not instance.applicable(state):198                return False199            state = instance.apply(state)200        return True201    def convert_plan(self, plan):202        if plan is None:203            return None204        new_plan = []205        for action_name, arg_names in plan:206            action = self.action_from_name[action_name]207            args = tuple(self.object_from_name[a] for a in arg_names)208            new_plan.append((action, args))209        return new_plan210    def dump(self):211        print 'Evaluations'...

plugin.py

Source:plugin.py

...88        event.handled = True89        try:90            feature_package_name = name_from_path(91                os.path.dirname(feature_path))[0]92            feature_module = object_from_name(feature_package_name)[1]93        except Exception:94            return event.loader.failedImport(feature_path)95        return self.makeSuiteFromFeature(96            module=feature_module,97            feature_path=feature_path,98        )99    def registerInSubprocess(self, event):100        event.pluginClasses.insert(0, self.__class__)101    def loadTestsFromNames(self, event):102        is_feature = partial(FEATURE_NAME.search)103        feature_names = [test_name for test_name in event.names if104                         is_feature(test_name)]105        event.names = [test_name for test_name in event.names if106                       not is_feature(test_name)]107        test_suites = list(self._from_names(feature_names))108        if event.names:109            event.extraTests.extend(test_suites)110        else:111            event.handled = True112            return test_suites113    def loadTestsFromName(self, event):114        log.debug(event)115        if FEATURE_NAME.search(event.name) is None:116            return117        event.handled = True118        features = list(self._from_names([event.name]))119        return features[0]120    def _from_names(self, names):121        by_feature = normalize_names(names)122        for (123            feature_package_name, feature_filename,124        ), scenarios_to_run in sorted(by_feature.items()):125            feature_module = object_from_name(feature_package_name)[1]126            feature_path = os.path.join(127                os.path.dirname(feature_module.__file__), feature_filename,128            )129            suite = self.makeSuiteFromFeature(130                module=feature_module,131                feature_path=feature_path,132                scenarios_to_run=scenarios_to_run,133            )134            yield suite135def normalize_names(names):136    """Normalize a sequence of feature test names.137    Aims to:138    - Collect together separate entries referring to different scenarios in139      the same feature....

champion.py

Source:champion.py

...54    55def string_to_object(csv_string : str) -> Champion:56    champ = csv_string.split(',')57    return Champion(champ[0], float(champ[1]), float(champ[2]), float(champ[3]))58def object_from_name(name : str) -> Champion:59    for champ in get_all():60        if name == champ.name:61            return champ62        63def list_of_player_selection(player_id : int) -> list[Champion]:64    string = request.to_server(f"from_db=get_frompl={player_id}")65    return [string_to_object(champ) for champ in string.split('+')][1:]66def list_of_player_champnames(player_id : int) -> list[Champion]:67    return champlist_to_names(list_of_player_selection(player_id))68"""69Functions which interact with the databases via the server, such as adding, deleting champions. Again self-explanatory.70"""71def save_to_player_selection(champ : Champion) -> None:72    as_string = ",".join(champ.str_tuple)73    74    request.to_server(f"from_db=save_champ={as_string}")75def save_name_to_roster(champ : str) -> None:76    save_to_roster(object_from_name(champ))77def save_to_roster(champ : Champion) -> None:78    as_string = ",".join(champ.str_tuple)79    80    request.to_server(f"from_db=save_toall={as_string}")81def delete_from_roster(name : str) -> None:82    obj = object_from_name(name)83    as_string = ",".join(obj.str_tuple)84    85    print(as_string)...

test_collect.py

Source:test_collect.py

...53    include_patterns = ['pytest_nodev.collection:generate_objects_from_modules']54    objs = collect.generate_objects_from_modules(55        modules, include_patterns, module_blacklist_pattern='re')56    assert len(list(objs)) == 157def test_object_from_name():58    object_ = collect.object_from_name('pytest_nodev.collect:object_from_name')59    assert object_ is collect.object_from_name60    # instance methods compare by equality, see http://stackoverflow.com/questions/1597780861    object_ = collect.object_from_name('pytest_nodev.collect:NOMATCH_PATTERN.upper')...

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.