import pytest from core.builtin_concepts import BuiltinConcepts from core.concept import Concept from core.global_symbols import EVENT_CONCEPT_PRECEDENCE_MODIFIED, CONCEPT_COMPARISON_CONTEXT, \ EVENT_RULE_PRECEDENCE_MODIFIED, RULE_COMPARISON_CONTEXT from core.sheerka.services.SheerkaComparisonManager import SheerkaComparisonManager, ComparisonObj from core.sheerka.services.SheerkaConceptManager import ChickenAndEggError from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka class TestSheerkaGreaterThanManager(TestUsingMemoryBasedSheerka): @staticmethod def get_comparison_objs(lst, prop_name="prop_name", context="#"): for item in lst: if ">>" in item: a = item.split(">>")[0] yield ComparisonObj("id", prop_name, a.strip(), None, ">>", context) elif "<<" in item: a = item.split("<<")[0] yield ComparisonObj("id", prop_name, a.strip(), None, "<<", context) elif ">" in item: a, b = item.split(">") yield ComparisonObj("id", prop_name, a.strip(), b.strip(), ">", context) elif "<" in item: a, b = item.split("<") yield ComparisonObj("id", prop_name, a.strip(), b.strip(), "<", context) else: a, b = item.split("=") yield ComparisonObj("id", prop_name, a.strip(), b.strip(), "=", context) @staticmethod def execution_definition(context, service, concepts_map, definition): if ">>" in definition: a = concepts_map[definition.split(">>")[0].strip()] return service.set_is_greatest(context, "prop_name", a) if "<<" in definition: a = concepts_map[definition.split("<<")[0].strip()] return service.set_is_lesser(context, "prop_name", a) if ">" in definition: a, b = [concepts_map[e.strip()] for e in definition.split(">")] return service.set_is_greater_than(context, "prop_name", a, b) if "<" in definition: a, b = [concepts_map[e.strip()] for e in definition.split("<")] return service.set_is_less_than(context, "prop_name", a, b) def test_i_can_add_a_is_greater_than(self): sheerka, context, one, two = self.init_concepts("one", "two", cache_only=False) service = sheerka.services[SheerkaComparisonManager.NAME] res = service.set_is_greater_than(context, "prop_name", two, one) assert res.status assert sheerka.isinstance(res.body, BuiltinConcepts.SUCCESS) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#")] weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2} # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#")] def test_i_can_add_is_greater_than_for_rules(self): sheerka, context, r1, r2 = self.init_test(cache_only=False).with_format_rules(("True", "true"), ("False", "false"), compile_rule=False).unpack() service = sheerka.services[SheerkaComparisonManager.NAME] res = service.set_is_greater_than(context, "prop_name", r2, r1) assert res.status assert sheerka.isinstance(res.body, BuiltinConcepts.SUCCESS) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ComparisonObj(context.event.get_digest(), "prop_name", r2.str_id, r1.str_id, ">", "#")] weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {r1.str_id: 1, r2.str_id: 2} # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ComparisonObj(context.event.get_digest(), "prop_name", r2.str_id, r1.str_id, ">", "#")] def test_i_can_add_a_is_less_than(self): sheerka, context, one, two = self.init_concepts("one", "two", cache_only=False) service = sheerka.services[SheerkaComparisonManager.NAME] res = service.set_is_less_than(context, "prop_name", one, two) assert res.status assert sheerka.isinstance(res.body, BuiltinConcepts.SUCCESS) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ComparisonObj(context.event.get_digest(), "prop_name", one.str_id, two.str_id, "<", "#")] weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2} # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ComparisonObj(context.event.get_digest(), "prop_name", one.str_id, two.str_id, "<", "#")] def test_i_can_add_is_less_than_for_rules(self): sheerka, context, r1, r2 = self.init_test(cache_only=False).with_format_rules(("True", "true"), ("False", "false"), compile_rule=False, create_new=True).unpack() service = sheerka.services[SheerkaComparisonManager.NAME] res = service.set_is_less_than(context, "prop_name", r1, r2) assert res.status assert sheerka.isinstance(res.body, BuiltinConcepts.SUCCESS) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ComparisonObj(context.event.get_digest(), "prop_name", r1.str_id, r2.str_id, "<", "#")] weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {r1.str_id: 1, r2.str_id: 2} # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ComparisonObj(context.event.get_digest(), "prop_name", r1.str_id, r2.str_id, "<", "#")] def test_i_can_add_multiples_constraints(self): sheerka, context, one, two, three, four = self.init_concepts("one", "two", "three", "four", cache_only=False) service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_greater_than(context, "prop_name", three, two) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#"), ComparisonObj(context.event.get_digest(), "prop_name", three.str_id, two.str_id, ">", "#") ] # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#"), ComparisonObj(context.event.get_digest(), "prop_name", three.str_id, two.str_id, ">", "#") ] sheerka.om.clear(SheerkaComparisonManager.COMPARISON_ENTRY) # reset the cache service.set_is_greater_than(context, "prop_name", four, three) in_cache = sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_cache == [ ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#"), ComparisonObj(context.event.get_digest(), "prop_name", three.str_id, two.str_id, ">", "#"), ComparisonObj(context.event.get_digest(), "prop_name", four.str_id, three.str_id, ">", "#"), ] weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2, "c:three|1003:": 3, "c:four|1004:": 4} def test_i_can_add_multiple_constraints_2(self): sheerka, context, one, two, three = self.init_concepts("one", "two", "three") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_greater_than(context, "prop_name", three, two) weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2, "c:three|1003:": 3} service.set_is_greater_than(context, "prop_name", three, one) # should not change order weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2, "c:three|1003:": 3} def test_i_can_add_lesser_than_and_opposite_greater_than(self): sheerka, context, one, two = self.init_concepts("one", "two") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_less_than(context, "prop_name", one, two) weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2} def test_i_can_support_multiple_ontology_layers(self): sheerka, context, one, two, three = self.init_concepts("one", "two", "three", cache_only=False) service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) # sanity check expected_in_cache = [ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#")] assert sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") == expected_in_cache expected_weights = {"c:one|1001:": 1, "c:two|1002:": 2} assert sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") == expected_weights # I still can access to the previous values sheerka.push_ontology(context, "new ontology") assert sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") == expected_in_cache assert sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") == expected_weights # I can modify service.set_is_greater_than(context, "prop_name", three, two) expected_in_cache2 = [ ComparisonObj(context.event.get_digest(), "prop_name", two.str_id, one.str_id, ">", "#"), ComparisonObj(context.event.get_digest(), "prop_name", three.str_id, two.str_id, ">", "#")] assert sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") == expected_in_cache2 expected_weights2 = {"c:one|1001:": 1, "c:two|1002:": 2, "c:three|1003:": 3} assert sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") == expected_weights2 # I can retrieve the previous values sheerka.pop_ontology(context) assert sheerka.om.get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") == expected_in_cache assert sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") == expected_weights @pytest.mark.parametrize("entries, expected", [ (["two > one"], {'c:one|1001:': 1, 'c:two|1002:': 2}), (["one < two"], {'c:one|1001:': 1, 'c:two|1002:': 2}), (["three > two", "one < two"], {'c:one|1001:': 1, 'c:two|1002:': 2, 'c:three|1003:': 3}), (["three > one", "one < two"], {'c:one|1001:': 1, 'c:two|1002:': 2, 'c:three|1003:': 2}), (["three >>"], {'c:three|1003:': 2}), (["one <<"], {'c:one|1001:': 0}), ]) def test_i_can_get_concept_weight(self, entries, expected): sheerka, context, *concepts = self.init_concepts("one", "two", "three") service = sheerka.services[SheerkaComparisonManager.NAME] concepts_map = dict(zip(["one", "two", "three"], concepts)) for entry in entries: self.execution_definition(context, service, concepts_map, entry) assert service.get_weights("prop_name") == expected def test_i_can_get_concept_weight_when_no_comparison_is_defined(self): sheerka, context = self.init_concepts() service = sheerka.services[SheerkaComparisonManager.NAME] assert service.get_weights("prop_name") == {} def test_i_can_recover_from_deleted_weight(self): sheerka, context, one = self.init_concepts("one") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_lesser(context, "prop_name", one) sheerka.om.clear(service.RESOLVED_COMPARISON_ENTRY) assert service.get_weights("prop_name") == {"c:one|1001:": 0} def test_i_can_get_partition(self): sheerka, context, one, two, three = self.init_concepts("one", "two", "three") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_less_than(context, "prop_name", two, three) res = service.get_partition("prop_name") assert res == { 1: ["c:one|1001:"], 2: ["c:two|1002:"], 3: ["c:three|1003:"], } def test_i_can_toposort(self): # using sample test from https://code.activestate.com/recipes/578272-topological-sort/ comparison_objs = self.get_comparison_objs([ "2 < 11", "11 > 9", "9 < 8", "10 < 11", "3 > 10", "11 < 7", "11 < 5", "7 > 8", "3 > 8"]) assert list(SheerkaComparisonManager.toposort(comparison_objs)) == [{'2', '9', '10'}, {'8', '11'}, {'3', '5', '7'}] def test_i_can_toposort_when_no_data(self): assert list(SheerkaComparisonManager.toposort([])) == [] def test_i_cannot_toposort_when_cycle(self): comparison_objs = self.get_comparison_objs([ "1 < 2", "2 < 3", "3 < 1", "1 < 4", # no issue with this "2 < 5", # no issue with this ]) with pytest.raises(ChickenAndEggError) as ex: list(SheerkaComparisonManager.toposort(comparison_objs)) assert ex.value.concepts == {"1", "2", "3"} def test_i_can_detect_chicken_and_egg(self): sheerka, context, one, two = self.init_concepts("one", "two") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) res = service.set_is_greater_than(context, "prop_name", one, two) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.CHICKEN_AND_EGG) assert set(res.body.body) == {one.str_id, two.str_id} def test_i_can_detect_more_complex_chicken_and_egg(self): sheerka, context, one, two, three, four, five = self.init_concepts("one", "two", "three", "four", "five") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_greater_than(context, "prop_name", five, four) service.set_is_greater_than(context, "prop_name", four, three) service.set_is_greater_than(context, "prop_name", five, two) res = service.set_is_greater_than(context, "prop_name", one, five) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.CHICKEN_AND_EGG) assert set(res.body.body) == {one.str_id, two.str_id, five.str_id} def test_i_can_give_the_same_information_in_many_ways(self): sheerka, context, one, two = self.init_concepts("one", "two") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", two, one) service.set_is_less_than(context, "prop_name", one, two) weighted = sheerka.om.get(SheerkaComparisonManager.RESOLVED_COMPARISON_ENTRY, "prop_name|#") assert weighted == {"c:one|1001:": 1, "c:two|1002:": 2} def test_methods_are_correctly_bound(self): sheerka, context, one, two = self.init_concepts("one", "two") res = sheerka.set_is_greater_than(context, "prop_name", two, one) assert res.status def test_a_lesser_concept_has_the_lowest_weight(self): sheerka, context, one, two, three = self.init_concepts("one", "two", "three", cache_only=False) service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_lesser(context, "prop_name", one) assert service.get_weights("prop_name") == {"c:one|1001:": 0} # DEFAULT_COMPARISON_VALUE - 1 sheerka.set_is_greater_than(context, "prop_name", three, two) assert service.get_weights("prop_name") == {"c:one|1001:": 0, "c:two|1002:": 1, "c:three|1003:": 2} # I can commit sheerka.om.commit(context) in_db = sheerka.om.current_sdp().get(SheerkaComparisonManager.COMPARISON_ENTRY, "prop_name|#") assert in_db == [ ComparisonObj(context.event.get_digest(), "prop_name", one.str_id, None, "<<", "#"), ComparisonObj(context.event.get_digest(), "prop_name", three.str_id, two.str_id, ">", "#") ] def test_i_can_define_an_order_for_lesser_concepts(self): sheerka, context, lesser, less_l, even_more_l = self.init_concepts("lesser", "less_l", "even_less_l") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_lesser(context, "prop_name", lesser) service.set_is_lesser(context, "prop_name", less_l) service.set_is_lesser(context, "prop_name", even_more_l) sheerka.set_is_less_than(context, "prop_name", less_l, lesser) sheerka.set_is_greater_than(context, "prop_name", less_l, even_more_l) assert service.get_weights("prop_name") == {"c:lesser|1001:": 0, "c:less_l|1002:": -1, "c:even_less_l|1003:": -2} def test_i_cannot_define_less_than_a_lesser_if_not_a_lesser_itself(self): """ minus_one cannot be defined as less that one is one is defined as lesser unless minus_one is defined as lesser itself :return: """ sheerka, context, lesser, foo, bar = self.init_concepts("lesser", "foo", "bar") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_lesser(context, "prop_name", lesser) # sanity check res = sheerka.set_is_greater_than(context, "prop_name", foo, lesser) assert res.status res = sheerka.set_is_less_than(context, "prop_name", bar, lesser) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.IS_LESSER_CONSTRAINT_ERROR) def test_a_greatest_concept_has_the_highest_weight(self): sheerka, context, one, two, three = self.init_concepts("one", "two", "three") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greatest(context, "prop_name", three) assert service.get_weights("prop_name") == {"c:three|1003:": 2} # DEFAULT_COMPARISON_VALUE + 1 sheerka.set_is_greater_than(context, "prop_name", two, one) assert service.get_weights("prop_name") == {"c:one|1001:": 1, "c:two|1002:": 2, "c:three|1003:": 3} def test_i_cannot_define_greater_than_a_greatest_if_not_a_greater_itself(self): sheerka, context, greatest, foo, bar = self.init_concepts("greatest", "foo", "bar") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greatest(context, "prop_name", greatest) # sanity check res = sheerka.set_is_less_than(context, "prop_name", foo, greatest) assert res.status res = sheerka.set_is_greater_than(context, "prop_name", bar, greatest) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.IS_GREATEST_CONSTRAINT_ERROR) @pytest.mark.parametrize("definitions, expected", [ (["foo >>", "foo <<"], BuiltinConcepts.IS_GREATEST_CONSTRAINT_ERROR), (["foo <<", "foo >>"], BuiltinConcepts.IS_LESSER_CONSTRAINT_ERROR), ]) def test_i_cannot_define_a_concept_as_lesser_and_greatest_at_the_same_time(self, definitions, expected): sheerka, context, foo = self.init_concepts("foo") service = sheerka.services[SheerkaComparisonManager.NAME] concepts_map = {"foo": foo} self.execution_definition(context, service, concepts_map, definitions[0]) res = self.execution_definition(context, service, concepts_map, definitions[1]) assert not res.status assert sheerka.isinstance(res.body, expected) def test_i_can_define_an_order_for_greatest_concepts(self): sheerka, context, greatest, more_g, even_more_g = self.init_concepts("greatest", "more_g", "even_more_g") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greatest(context, "prop_name", greatest) service.set_is_greatest(context, "prop_name", more_g) service.set_is_greatest(context, "prop_name", even_more_g) sheerka.set_is_less_than(context, "prop_name", greatest, more_g) sheerka.set_is_greater_than(context, "prop_name", even_more_g, more_g) assert service.get_weights("prop_name") == {"c:greatest|1001:": 2, "c:more_g|1002:": 3, "c:even_more_g|1003:": 4} def test_i_can_mix_all_comparisons(self): sheerka, context, one, two, three, four, five, six, three_and_half = self.init_concepts( "one", "two", "three", "four", "five", "six", "two_and_half") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_lesser(context, "prop_name", one) service.set_is_lesser(context, "prop_name", two) sheerka.set_is_less_than(context, "prop_name", one, two) service.set_is_greatest(context, "prop_name", five) service.set_is_greatest(context, "prop_name", six) sheerka.set_is_greater_than(context, "prop_name", six, five) sheerka.set_is_less_than(context, "prop_name", three, four) assert service.get_weights("prop_name") == { "c:one|1001:": -1, "c:two|1002:": 0, "c:three|1003:": 1, "c:four|1004:": 2, "c:five|1005:": 3, "c:six|1006:": 4 } sheerka.set_is_less_than(context, "prop_name", three_and_half, four) sheerka.set_is_greater_than(context, "prop_name", three_and_half, three) assert service.get_weights("prop_name") == { "c:one|1001:": -1, "c:two|1002:": 0, "c:three|1003:": 1, "c:two_and_half|1007:": 2, "c:four|1004:": 3, "c:five|1005:": 4, "c:six|1006:": 5 } def test_i_can_mix_all_comparisons_2(self): sheerka, context, zero, one, two, three, four, five, thousand = self.init_concepts( "zero", "one", "two", "three", "four", "five", "thousand") service = sheerka.services[SheerkaComparisonManager.NAME] service.set_is_greater_than(context, "prop_name", four, one) service.set_is_less_than(context, "prop_name", one, five) service.set_is_less_than(context, "prop_name", one, three) service.set_is_less_than(context, "prop_name", two, three) service.set_is_greater_than(context, "prop_name", two, one) service.set_is_greater_than(context, "prop_name", five, four) service.set_is_less_than(context, "prop_name", three, four) service.set_is_greatest(context, "prop_name", thousand) service.set_is_lesser(context, "prop_name", zero) service.set_is_less_than(context, "prop_name", two, five) service.set_is_less_than(context, "prop_name", two, four) service.set_is_greater_than(context, "prop_name", five, one) service.set_is_greater_than(context, "prop_name", five, three) assert service.get_weights("prop_name") == { "c:zero|1001:": 0, "c:one|1002:": 1, "c:two|1003:": 2, "c:three|1004:": 3, "c:four|1005:": 4, "c:five|1006:": 5, "c:thousand|1007:": 6 } @pytest.mark.parametrize("definition", [ "one >>", "one <<", "one > two", "one < two", ]) def test_i_cannot_add_the_same_definition_twice(self, definition): sheerka, context, *concepts = self.init_concepts("one", "two") service = sheerka.services[SheerkaComparisonManager.NAME] concepts_map = dict(zip(["one", "two"], concepts)) self.execution_definition(context, service, concepts_map, definition) res = self.execution_definition(context, service, concepts_map, definition) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.COMPARISON_ALREADY_DEFINED) def test_an_event_is_fired_when_modifying_concepts_precedence(self): sheerka, context, one, two, foo = self.init_concepts("one", "two", "foo") event_received = False def receive_event(c): nonlocal event_received event_received = True sheerka.subscribe(EVENT_CONCEPT_PRECEDENCE_MODIFIED, receive_event) sheerka.set_is_greater_than(context, foo, one, two) assert not event_received sheerka.set_is_greater_than(context, BuiltinConcepts.PRECEDENCE, foo, one, two) assert not event_received sheerka.set_is_greater_than(context, BuiltinConcepts.PRECEDENCE, one, two, CONCEPT_COMPARISON_CONTEXT) assert event_received def test_an_event_is_fired_when_modifying_rule_precedence(self): sheerka, context, r1, r2 = self.init_test(cache_only=False).with_format_rules(("True", "true"), ("False", "false"), compile_rule=False).unpack() foo = Concept("foo") event_received = False sheerka.om.clear(SheerkaComparisonManager.COMPARISON_ENTRY) def receive_event(c): nonlocal event_received event_received = True sheerka.subscribe(EVENT_RULE_PRECEDENCE_MODIFIED, receive_event) sheerka.set_is_greater_than(context, foo, r1, r2) assert not event_received sheerka.set_is_greater_than(context, BuiltinConcepts.PRECEDENCE, foo, r1, r2) assert not event_received sheerka.set_is_greater_than(context, BuiltinConcepts.PRECEDENCE, r1, r2, RULE_COMPARISON_CONTEXT) assert event_received @pytest.mark.parametrize("entries, expected", [ ([], (set(), set(), set())), (["a<<", "a<<", "b<<", "a>", "a>>", "b>>", "a>", "c=d", "eh"], ({"b"}, {"a"}, {"a", "b", "c", "d", "e", "f", "g", "h"})), ]) def test_i_can_get_objs_groups(self, entries, expected): comparison_objs = self.get_comparison_objs(entries) assert SheerkaComparisonManager.get_objs_groups(comparison_objs) == expected @pytest.mark.parametrize("entries, greatest, lowest, expected_tuple", [ # greatest (["a < b"], {"a", "b"}, set(), (["a < b"], [], [], [])), (["a > b"], {"a", "b"}, set(), (["a > b"], [], [], [])), # lowest (["a < b"], set(), {"a", "b"}, ([], ["a < b"], [], [])), (["a > b"], set(), {"a", "b"}, ([], ["a > b"], [], [])), # equiv (["a = b"], set(), set(), ([], [], ["a = b"], [])), # neither (["a < b"], set(), set(), ([], [], [], ["a < b"])), (["a > b"], set(), set(), ([], [], [], ["a > b"])), # irrelevant information (["a > b"], {"a"}, set(), ([], [], [], [])), (["a > b"], set(), {"b"}, ([], [], [], [])), (["a < b"], {"b"}, set(), ([], [], [], [])), (["a < b"], set(), {"a"}, ([], [], [], [])), # mixed (["a b"], {"a": 2, "b": 1}), (["a < b", "b > a"], {"a": 1, "b": 2}), (["a < b", "b < c"], {"a": 1, "b": 2, "c": 3}), (["a < b", "a < c"], {"a": 1, "b": 2, "c": 2}), # greatest between themselves (["a < b", "a>>", "b>>"], {"a": 2, "b": 3}), (["a > b", "a>>", "b>>"], {"a": 3, "b": 2}), (["a < b", "b > a", "a>>", "b>>"], {"a": 2, "b": 3}), (["a < b", "b < c", "a>>", "b>>", "c>>"], {"a": 2, "b": 3, "c": 4}), (["a < b", "a < c", "a>>", "b>>", "c>>"], {"a": 2, "b": 3, "c": 3}), # lowest between themselves (["a < b", "a<<", "b<<"], {"a": -1, "b": 0}), (["a > b", "a<<", "b<<"], {"a": 0, "b": -1}), (["a < b", "b > a", "a<<", "b<<"], {"a": -1, "b": 0}), (["a < b", "b < c", "a<<", "b<<", "c<<"], {"a": -2, "b": -1, "c": 0}), (["a < b", "a < c", "a<<", "b<<", "c<<"], {"a": -1, "b": 0, "c": 0}), # greatest that does not appear in other relations (["a >>", "b < c"], {"a": 3, "b": 1, "c": 2}), (["a >>", "a > b"], {"a": 2, "b": 1}), (["a >>", "a > b", "c > d", "d > e"], {"a": 4, "b": 1, "c": 3, "d": 2, "e": 1}), # lowest that does not appear in other relations (["a <<", "b < c"], {"a": 0, "b": 1, "c": 2}), (["a <<", "a < b"], {"a": 0, "b": 1}), (["a <<", "a < b", "c < d", "d < e"], {"a": 0, "b": 1, "c": 1, "d": 2, "e": 3}), (["z <<", "a<<", "b<<", "c<<", "a < b", "b < c"], {"a": -2, "b": -1, "c": 0, "z": -3}), # eq (["a = b"], {"a": 1, "b": 1}), (["a = b", "b > c"], {"a": 2, "b": 2, "c": 1}), (["a = z", "z>>", "b d"], {"a": 2, "b": 2, "c": 2, "d": 1}), (["a = b", "b = c", "a > d"], {"a": 2, "b": 2, "c": 2, "d": 1}), #(["a = b", "b = c", "c > d"], {"a": 2, "b": 2, "c": 2, "d": 1}), # not working # mix greatest and lesser (["a >>", "b<<", "a > b"], {"a": 2, "b": 0}), ]) def test_i_can_compute_weight_new(self, entries, expected): comparison_objs = list(self.get_comparison_objs(entries)) assert SheerkaComparisonManager.compute_weights(comparison_objs) == expected @pytest.mark.parametrize("previous_entries, new_entry, items_in_cycle", [ (["a > b"], "b > a", {"a", "b"}), (["a > b", "c > d"], "b > a", {"a", "b"}), (["a < b", "b < c"], "c < a", {"a", "b", "c"}), ]) def test_validate_new_entry_i_can_detect_cycle(self, previous_entries, new_entry, items_in_cycle): sheerka, context = self.init_test().unpack() new_co = list(self.get_comparison_objs([new_entry]))[0] previous_comparison_objs = list(self.get_comparison_objs(previous_entries)) res = SheerkaComparisonManager.validate_new_entry(context, new_co, previous_comparison_objs) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.CHICKEN_AND_EGG) assert set(res.body.body) == items_in_cycle @pytest.mark.parametrize("previous_entries, new_entry", [ (["a > b"], "a > b"), (["a < b"], "a < b"), (["a = b"], "a = b"), (["a <<"], "a <<"), (["a >>"], "a >>"), ]) def test_validate_new_entry_i_can_detect_duplicate_entries(self, previous_entries, new_entry): sheerka, context = self.init_test().unpack() new_co = list(self.get_comparison_objs([new_entry]))[0] previous_comparison_objs = list(self.get_comparison_objs(previous_entries)) res = SheerkaComparisonManager.validate_new_entry(context, new_co, previous_comparison_objs) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.COMPARISON_ALREADY_DEFINED) @pytest.mark.parametrize("previous_entries, new_entry", [ (["a>>"], "a<<"), (["a>>"], "a < b"), (["a>>"], "b > a"), ]) def test_validate_new_entry_i_can_detect_is_greatest_constraint_error(self, previous_entries, new_entry): sheerka, context = self.init_test().unpack() new_co = list(self.get_comparison_objs([new_entry]))[0] previous_comparison_objs = list(self.get_comparison_objs(previous_entries)) res = SheerkaComparisonManager.validate_new_entry(context, new_co, previous_comparison_objs) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.IS_GREATEST_CONSTRAINT_ERROR) @pytest.mark.parametrize("previous_entries, new_entry", [ (["a<<"], "a>>"), (["a<<"], "a > b"), (["a<<"], "b < a"), ]) def test_validate_new_entry_i_can_detect_is_lesser_constraint_error(self, previous_entries, new_entry): sheerka, context = self.init_test().unpack() new_co = list(self.get_comparison_objs([new_entry]))[0] previous_comparison_objs = list(self.get_comparison_objs(previous_entries)) res = SheerkaComparisonManager.validate_new_entry(context, new_co, previous_comparison_objs) assert not res.status assert sheerka.isinstance(res.body, BuiltinConcepts.IS_LESSER_CONSTRAINT_ERROR)