276 lines
12 KiB
Python
276 lines
12 KiB
Python
import pytest
|
|
|
|
from base import BaseTest
|
|
from common.global_symbols import NotFound, NotInit
|
|
from conftest import NewOntology
|
|
from core.BuiltinConcepts import BuiltinConcepts
|
|
from core.concept import ConceptMetadata
|
|
from core.error import ErrorContext
|
|
from helpers import get_concepts, get_metadata
|
|
from services.SheerkaConceptManager import ConceptAlreadyDefined, ConceptManager
|
|
|
|
|
|
class TestConceptManager(BaseTest):
|
|
|
|
@pytest.fixture()
|
|
def service(self, sheerka):
|
|
return sheerka.services[ConceptManager.NAME]
|
|
|
|
def test_i_can_compute_concept_digest(self, service):
|
|
"""
|
|
Two concepts with the same definition share the same digest
|
|
:return:
|
|
:rtype:
|
|
"""
|
|
metadata = get_metadata("foo", "body")
|
|
digest = service.compute_metadata_digest(metadata)
|
|
assert digest == "7c0f1708968e0312be622950d3f21d588f718f7ba568054ece64d077052a6476"
|
|
|
|
another_metadata = get_metadata("foo", "body")
|
|
other_digest = service.compute_metadata_digest(another_metadata)
|
|
|
|
assert digest == other_digest
|
|
|
|
def test_id_is_not_part_of_the_digest(self, service):
|
|
metadata1 = get_metadata("foo", "body", id=1)
|
|
metadata2 = get_metadata("foo", "body", id=2)
|
|
|
|
assert service.compute_metadata_digest(metadata1) == service.compute_metadata_digest(metadata2)
|
|
|
|
def test_i_can_compute_concept_attributes_based_on_the_metadata(self, service):
|
|
compute_all_attrs = service.compute_all_attrs
|
|
|
|
m1 = get_metadata("foo")
|
|
assert compute_all_attrs(m1.variables) == ('#where#', '#pre#', '#post#', '#body#', '#ret#')
|
|
|
|
m2 = get_metadata("bar", variables=[("var1", None), ("var2", None)])
|
|
assert compute_all_attrs(m2.variables) == ('#where#', '#pre#', '#post#', '#body#', '#ret#', 'var1', 'var2')
|
|
|
|
@pytest.mark.parametrize("definition, variables, expected", [
|
|
("foo", [], "foo"),
|
|
("foo(bar)", [], "foo ( bar )"),
|
|
("foo a", ["a"], "foo __var__0"),
|
|
("a foo b", ["a", "b"], "__var__0 foo __var__1"),
|
|
("a foo b", ["b", "a"], "__var__1 foo __var__0"),
|
|
("foo", ["foo"], "foo"),
|
|
("foo a", ["foo"], "__var__0 a"),
|
|
("foo a b", ["a"], "foo __var__0 b"),
|
|
("'foo'", [], "'foo'"),
|
|
("my name is a", ["a"], "my name is __var__0"),
|
|
("a b c d", ["b", "c"], "a __var__0 __var__1 d"),
|
|
("a 'b c' d", ["b", "c"], "a 'b c' d"),
|
|
("a | b", ["a", "b"], "__var__0 | __var__1"),
|
|
("a b a c", ["a", "b"], "__var__0 __var__1 __var__0 c"),
|
|
("a b a c", ["b", "a"], "__var__1 __var__0 __var__1 c"),
|
|
("def concept a", ["a"], "def concept __var__0"),
|
|
])
|
|
def test_i_can_create_concept_key(self, service, definition, variables, expected):
|
|
expanded_variables = tuple((v, NotInit) for v in variables)
|
|
key = service.create_concept_key(definition, None, expanded_variables)
|
|
|
|
assert key == expected
|
|
|
|
def test_the_key_is_created_from_the_definition_if_it_is_set(self, service):
|
|
assert service.create_concept_key("from name", "from definition", None) == "from definition"
|
|
assert service.create_concept_key("from name", None, None) == "from name"
|
|
|
|
def test_i_can_define_a_new_concept(self, context, service):
|
|
with NewOntology(context, "test_i_can_define_a_new_concept"):
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
|
|
assert res.status is True
|
|
|
|
metadata = res.value.metadata
|
|
assert isinstance(metadata, ConceptMetadata)
|
|
assert metadata.id == "1001"
|
|
assert metadata.name == "name"
|
|
assert metadata.key == "name"
|
|
assert metadata.body == "body"
|
|
assert metadata.digest == "c75faa4efbc9ef9dbc5174c52786d5b066e2ece41486b81c27336e292917fecb"
|
|
assert metadata.all_attrs == ('#where#', '#pre#', '#post#', '#body#', '#ret#')
|
|
|
|
# is sorted in db
|
|
om = context.sheerka.om
|
|
assert om.get(ConceptManager.CONCEPTS_BY_ID_ENTRY, metadata.id) == metadata
|
|
assert om.get(ConceptManager.CONCEPTS_BY_NAME_ENTRY, metadata.name) == metadata
|
|
assert om.get(ConceptManager.CONCEPTS_BY_KEY_ENTRY, metadata.key) == metadata
|
|
assert om.get(ConceptManager.CONCEPTS_BY_HASH_ENTRY, metadata.digest) == metadata
|
|
|
|
def test_i_cannot_create_the_same_concept_twice(self, context, service):
|
|
with NewOntology(context, "test_i_cannot_create_the_same_concept_twice"):
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
assert res.status
|
|
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
assert not res.status
|
|
assert isinstance(res.value, ErrorContext)
|
|
assert isinstance(res.value.value, ConceptAlreadyDefined)
|
|
|
|
def test_i_can_add_the_same_concept_on_different_ontologies(self, context, service):
|
|
with NewOntology(context, "test_i_can_add_the_same_concept_on_different_ontologies"):
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
assert res.status
|
|
|
|
sheerka = context.sheerka
|
|
om = sheerka.om
|
|
om.push_ontology("my_new_ontology")
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
assert res.status is True
|
|
|
|
def test_i_cannot_get_by_if_concept_does_not_exist(self, service):
|
|
assert service.get_by_id("unresolved_id") == NotFound
|
|
assert service.get_by_name("unresolved name") == NotFound
|
|
assert service.get_by_key("unresolved_hash") == NotFound
|
|
|
|
def test_i_can_get_a_newly_created_concept(self, context, service):
|
|
with NewOntology(context, "test_i_can_get_a_newly_created_concept"):
|
|
res = service.define_new_concept(context, "name", body="body")
|
|
assert res.status
|
|
metadata = res.value.metadata
|
|
|
|
assert service.get_by_id(metadata.id).id == metadata.id
|
|
assert service.get_by_name(metadata.name).name == metadata.name
|
|
assert service.get_by_key(metadata.key).key == metadata.key
|
|
|
|
def test_i_can_instantiate_a_new_concept_by_its_name(self, context, service):
|
|
with NewOntology(context, "test_i_can_instantiate_a_new_concept_by_its_name"):
|
|
res = service.define_new_concept(context, "foo", variables=[("var1", None), ("var2", None)])
|
|
assert res.status
|
|
|
|
foo = service.newn("foo", var1="value1", var2="value2")
|
|
|
|
assert foo.id == "1001"
|
|
assert foo.key == "foo"
|
|
assert foo.name == "foo"
|
|
assert foo.str_id == "c:#1001:"
|
|
assert foo.var1 == "value1"
|
|
assert foo.var2 == "value2"
|
|
|
|
def test_i_can_manage_when_concepts_with_the_same_name(self, context, service):
|
|
with NewOntology(context, "test_i_can_manage_when_concepts_with_the_same_name"):
|
|
service.define_new_concept(context, "foo", body="body1")
|
|
service.define_new_concept(context, "foo", body="body2")
|
|
|
|
concepts = service.newn("foo")
|
|
|
|
assert len(concepts) == 2
|
|
assert concepts[0].name == "foo"
|
|
assert concepts[0].get_metadata().body == "body1"
|
|
assert concepts[1].name == "foo"
|
|
assert concepts[1].get_metadata().body == "body2"
|
|
|
|
def test_i_can_instantiate_a_new_concept_by_its_id(self, context, service):
|
|
with NewOntology(context, "test_i_can_instantiate_a_new_concept_by_its_id"):
|
|
res = service.define_new_concept(context, "foo", variables=[("var1", None), ("var2", None)])
|
|
assert res.status
|
|
|
|
foo = service.newi("1001", var1="value1", var2="value2")
|
|
|
|
assert foo.id == "1001"
|
|
assert foo.key == "foo"
|
|
assert foo.name == "foo"
|
|
assert foo.str_id == "c:#1001:"
|
|
assert foo.var1 == "value1"
|
|
assert foo.var2 == "value2"
|
|
|
|
def test_i_cannot_instantiate_a_concept_which_does_not_exist(self, context, service):
|
|
foo = service.newn("foo", var1="value1", var2="value2")
|
|
assert foo.key == BuiltinConcepts.UNKNOWN_CONCEPT
|
|
assert foo.requested_name == "foo"
|
|
|
|
foo = service.newi("1001", var1="value1", var2="value2")
|
|
assert foo.key == BuiltinConcepts.UNKNOWN_CONCEPT
|
|
assert foo.requested_id == "1001"
|
|
|
|
def test_i_can_instantiate_by_name_when_multiple_results(self, context, service):
|
|
with NewOntology(context, "test_i_can_instantiate_by_name_when_multiple_results"):
|
|
service.define_new_concept(context, "foo", body="body1")
|
|
service.define_new_concept(context, "foo", body="body2")
|
|
|
|
concepts = service.newn("foo")
|
|
|
|
assert len(concepts) == 2
|
|
assert concepts[0].id == "1001"
|
|
assert concepts[0].get_metadata().body == "body1"
|
|
assert concepts[1].id == "1002"
|
|
assert concepts[1].get_metadata().body == "body2"
|
|
|
|
def test_concepts_are_removed_when_ontology_is_popped(self, context, service):
|
|
context.sheerka.om.push_ontology("new ontology")
|
|
res = service.define_new_concept(context, "foo", body="body")
|
|
assert service.get_by_id(res.value.metadata.id) is not NotFound
|
|
|
|
context.sheerka.om.pop_ontology(context)
|
|
assert service.get_by_id(res.value.metadata.id) is NotFound
|
|
|
|
def test_i_can_new(self, context, service):
|
|
with NewOntology(context, "test_i_can_new"):
|
|
res = service.define_new_concept(context, "name", body="body", variables=[("my_var", None)])
|
|
assert res.status
|
|
metadata = res.value.metadata
|
|
|
|
# I can create a new concept
|
|
res = service.new(metadata, my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
res = service.new((metadata.name, None), my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
res = service.new((None, metadata.id), my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
res = service.new("c:name:", my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
res = service.new("c:#1001:", my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
res = service.new("c:name#1001:", my_var="my_var_value")
|
|
assert res.id == metadata.id
|
|
assert res.my_var == "my_var_value"
|
|
|
|
# cannot new using id
|
|
assert service.new(f"1001").name == BuiltinConcepts.UNKNOWN_CONCEPT
|
|
|
|
def test_id_is_used_when_name_and_id_are_provided(self, context, service):
|
|
with NewOntology(context, "test_id_is_used_when_name_and_id_are_provided"):
|
|
res = service.define_new_concept(context, "name", body="body1")
|
|
metadata = res.value.metadata
|
|
service.define_new_concept(context, "name", body="body2")
|
|
|
|
assert service.new((metadata.name, metadata.id)).id == metadata.id
|
|
|
|
def test_i_can_instantiate_a_list_of_concepts(self, context, service):
|
|
with NewOntology(context, "test_i_can_instantiate_a_list_of_concepts"):
|
|
foo, bar = get_concepts(context, "foo", "bar", use_sheerka=True)
|
|
|
|
res = service.new([foo.get_metadata(), bar.get_metadata()])
|
|
|
|
assert len(res) == 2
|
|
assert context.sheerka.isinstance(res[0], foo)
|
|
assert context.sheerka.isinstance(res[1], bar)
|
|
|
|
def test_unknown_concept_is_return_if_the_identifier_is_not_found(self, service):
|
|
assert service.new("unknown").name == BuiltinConcepts.UNKNOWN_CONCEPT
|
|
|
|
def test_can_get_all_concepts(self, context, service):
|
|
with NewOntology(context, "test_i_can_new"):
|
|
service.define_new_concept(context, "foo")
|
|
service.define_new_concept(context, "bar")
|
|
context.sheerka.om.push_ontology("another ontology")
|
|
service.define_new_concept(context, "baz")
|
|
service.define_new_concept(context, "qux")
|
|
|
|
all_concepts = service.get_all_concepts()
|
|
assert [c.name for c in all_concepts if not c.is_builtin] == ["foo", "bar", "baz", "qux"]
|
|
|
|
# sanity check. Concepts are discarded when ontology is popped
|
|
context.sheerka.om.pop_ontology(context)
|
|
all_concepts = service.get_all_concepts()
|
|
assert [c.name for c in all_concepts if not c.is_builtin] == ["foo", "bar"]
|