<commit_before># -*- coding: utf-8 -*-

from unittest import TestCase
from os import path
import json

from rdflib import ConjunctiveGraph, Graph, URIRef, Literal, RDF, XSD
from rdflib.plugins.stores.sparqlstore import SPARQLUpdateStore
from rdflib.namespace import FOAF

from oldman import create_resource_manager, parse_graph_safely
from oldman.attribute import OMAttributeTypeCheckError, OMRequiredPropertyError
from oldman.exception import OMClassInstanceError, OMAttributeAccessError, OMUniquenessError
from oldman.exception import OMWrongResourceError, OMObjectNotFoundError, OMHashIriError, OMEditError
from oldman.exception import OMDifferentBaseIRIError, OMForbiddenSkolemizedIRIError, OMUnauthorizedTypeChangeError
from oldman.rest.crud import CRUDController


default_graph = ConjunctiveGraph()
#default_graph = ConjunctiveGraph(SPARQLUpdateStore(queryEndpoint="http://localhost:3030/test/query",
#                                                   update_endpoint="http://localhost:3030/test/update"))
#schema_graph = default_graph.get_context(URIRef("http://localhost/schema"))
schema_graph = Graph()
data_graph = default_graph.get_context(URIRef("http://localhost/data"))

BIO = "http://purl.org/vocab/bio/0.1/"
REL = "http://purl.org/vocab/relationship/"
CERT = "http://www.w3.org/ns/auth/cert#"
RDFS = "http://www.w3.org/2000/01/rdf-schema#"
WOT = "http://xmlns.com/wot/0.1/"

MY_VOC = "http://example.com/vocab#"
local_person_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "foaf": FOAF,
            "bio": BIO,
            "rel": REL,
            "cert": CERT,
            "wot": WOT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalPerson",
    "@type": "hydra:Class",
    "subClassOf": "foaf:Person",
    "supportedProperty": [
        {
            "property": "foaf:name",
            "required": True
        },
        {
            "property": "foaf:mbox",
            "required": True
        },
        {
            "property": "foaf:weblog"
        },
        {
            "property": "bio:olb",
            "required": True
        },
        {
            "property": "foaf:knows"
        },
        {
            "property": "rel:parentOf"
        },
        {
            "property": "cert:key"
        },
        {
            "property": "wot:hasKey"
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_person_def), format="json-ld")

local_rsa_key_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "rdfs": RDFS,
            "cert": CERT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalRSAPublicKey",
    "@type": "hydra:Class",
    "subClassOf": "cert:RSAPublicKey",
    "supportedProperty": [
        {
            "property": "cert:exponent",
            "required": True
        },
        {
            "property": "cert:modulus",
            "required": True
        },
        {
            "property": "rdfs:label"
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_rsa_key_def), format="json-ld")

local_gpg_key_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "wot": WOT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalGPGPublicKey",
    "@type": "hydra:Class",
    "subClassOf": "wot:PubKey",
    "supportedProperty": [
        {
            "property": "wot:fingerprint",
            "required": True,
        },
        {
            "property": "wot:hex_id",
            "required": True
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_gpg_key_def), format="json-ld")
#print schema_graph.serialize(format="turtle")

context = {
    "@context": {
        "myvoc": MY_VOC,
        "rdfs": RDFS,
        "foaf": "http://xmlns.com/foaf/0.1/",
        "bio": "http://purl.org/vocab/bio/0.1/",
        "xsd": "http://www.w3.org/2001/XMLSchema#",
        "rel": "http://purl.org/vocab/relationship/",
        "cert": CERT,
        "wot": WOT,
        "id": "@id",
        "types": "@type",
        "LocalPerson": "myvoc:LocalPerson",
        "Person": "foaf:Person",
        "LocalRSAPublicKey": "myvoc:LocalRSAPublicKey",
        "LocalGPGPublicKey": "myvoc:LocalGPGPublicKey",
        "name": {
            "@id": "foaf:name",
            "@type": "xsd:string"
        },
        "mboxes": {
            "@id": "foaf:mbox",
            "@type": "xsd:string",
            "@container": "@set"
        },
        "blog": {
            "@id": "foaf:weblog",
            "@type": "@id"
        },
        "short_bio_fr": {
            "@id": "bio:olb",
            "@language": "fr"
        },
        "short_bio_en": {
            "@id": "bio:olb",
            "@language": "en"
        },
        "friends": {
            "@id": "foaf:knows",
            "@type": "@id",
            "@container": "@set"
        },
        "children": {
            "@id": "rel:parentOf",
            "@type": "@id",
            "@container": "@list"
        },
        "gpg_key": {
            "@id": "wot:hasKey",
            "@type": "@id"
        },
        "keys": {
            "@id": "cert:key",
            "@type": "@id",
            "@container": "@set"
        },
        # For keys
        "modulus": {
            "@id": "cert:modulus",
            "@type": "xsd:hexBinary"
        },
        "exponent": {
            "@id": "cert:exponent",
            "@type": "xsd:integer"
        },
        "label": {
            "@id": "rdfs:label",
            "@type": "xsd:string"
        },
        # For GPG
        "fingerprint": {
            "@id": "wot:fingerprint",
            "@type": "xsd:hexBinary"
        },
        "hex_id": {
            "@id": "wot:hex_id",
            "@type": "xsd:hexBinary"
        }
    }
}

# Retrieve namespaces when the two graphs are independent
for prefix, namespace in schema_graph.namespace_manager.namespaces():
    data_graph.namespace_manager.bind(prefix, namespace)

manager = create_resource_manager(schema_graph, data_graph)
# Model classes are generated here!
lp_model = manager.create_model("LocalPerson", context, iri_prefix="http://localhost/persons/",
                               iri_fragment="me")
rsa_model = manager.create_model("LocalRSAPublicKey", context)
gpg_model = manager.create_model("LocalGPGPublicKey", context)

crud_controller = CRUDController(manager)

bob_name = "Bob"
bob_blog = "http://blog.example.com/"
bob_email1 = "bob@localhost"
bob_email2 = "<EMAIL>"
bob_emails = {bob_email1, bob_email2}
bob_bio_en = "I grow up in ... ."
bob_bio_fr = u"J'ai grandi en ... ."
alice_name = "Alice"
alice_mail = "<EMAIL>"
alice_bio_en = "I am an expert on this and that"
john_name = "John"
john_mail = "<EMAIL>"
john_bio_en = "Supporter of Linked Data"
key_modulus = "b42dbf23ee820be938bee7298893e434f8f74d4be2bbe39408776d695168d09262da2a849962"
key_exponent = 65537
key_label = "Key 1"
gpg_fingerprint = "<KEY>"
gpg_hex_id = "002640e2"

prof_type = MY_VOC + "Professor"
researcher_type = MY_VOC + "Researcher"


class ModelTest(TestCase):

    def tearDown(self):
        """ Clears the data graph """
        data_graph.update("CLEAR DEFAULT")
        manager.clear_resource_cache()

    def create_bob(self):
        return lp_model.create(name=bob_name, blog=bob_blog, mboxes=bob_emails,
                               short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)

    def create_alice(self):
        return lp_model.create(name=alice_name, mboxes={alice_mail}, short_bio_en=alice_bio_en)

    def create_john(self):
        return lp_model.create(name=john_name, mboxes={john_mail}, short_bio_en=john_bio_en)

    def create_rsa_key(self):
        return rsa_model.create(exponent=key_exponent, modulus=key_modulus, label=key_label)

    def create_gpg_key(self):
        return gpg_model.create(fingerprint=gpg_fingerprint, hex_id=gpg_hex_id)

    def test_bio_requirement(self):
        bob = lp_model.new()
        bob.name = bob_name
        bob.blog = bob_blog
        bob.mboxes = {bob_email1}

        self.assertFalse(bob.is_valid())
        self.assertRaises(OMRequiredPropertyError, bob.save)

        # Bio is required
        bob.short_bio_en = bob_bio_en
        self.assertTrue(bob.is_valid())
        bob.save()

    def test_person_types(self):
        bob = self.create_bob()
        expected_types = ["http://example.com/vocab#LocalPerson",
                          "http://xmlns.com/foaf/0.1/Person"]
        self.assertEquals(bob.types, expected_types)

        # Check the triplestore
        type_request = """SELECT ?t WHERE {?x a ?t }"""
        retrieved_types = {str(r) for r, in data_graph.query(type_request, initBindings={'x': URIRef(bob.id)})}
        self.assertEquals(set(expected_types), retrieved_types)

    def test_bob_in_triplestore(self):
        request = """ASK {?x foaf:name "%s"^^xsd:string }""" % bob_name
        self.assertFalse(bool(data_graph.query(request)))
        self.create_bob()
        self.assertTrue(bool(data_graph.query(request)))

    def test_bob_attributes(self):
        bob = self.create_bob()
        self.assertEquals(bob_name, bob.name)
        self.assertEquals(bob_blog, bob.blog.id)
        self.assertEquals(bob_emails, bob.mboxes)
        self.assertEquals(bob_bio_en, bob.short_bio_en)
        self.assertEquals(bob_bio_fr, bob.short_bio_fr)

    def test_bob_loading(self):
        bob = self.create_bob()
        bob_uri = bob.id

        # Not saved
        bob.name = "You should not retrieve this string"

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)

        self.assertEquals(bob_name, bob.name)
        self.assertEquals(bob_blog, bob.blog.id)
        self.assertEquals(bob_emails, bob.mboxes)
        self.assertEquals(bob_bio_en, bob.short_bio_en)
        self.assertEquals(bob_bio_fr, bob.short_bio_fr)

    def test_string_validation(self):
        bob = self.create_bob()
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.name = 2

    def test_not_saved(self):
        bob = self.create_bob()
        new_name = "<NAME>"
        bob.name = new_name
        # Not saved
        self.assertFalse(bool(data_graph.query("""ASK {?x foaf:name "%s"^^xsd:string }""" % new_name)))

    def test_objects_access(self):
        """ Object manager is only accessible at the class level """
        bob = self.create_bob()
        self.assertRaises(AttributeError, getattr, bob, "objects")

    def test_multiple_mboxes(self):
        bob = self.create_bob()
        email3 = "<EMAIL>"
        bob.mboxes = {bob_email2, email3}
        bob.save()

        mbox_query = """ASK {?x foaf:mbox "%s"^^xsd:string }"""
        self.assertTrue(bool(data_graph.query(mbox_query % bob_email2)))
        self.assertTrue(bool(data_graph.query(mbox_query % email3)))
        # Has been removed
        self.assertFalse(bool(data_graph.query(mbox_query % bob_email1)))

    def test_list_assignment_instead_of_set(self):
        bob = lp_model.new()
        bob.name = bob_name
        bob.short_bio_en = bob_bio_en

        # List assignment instead of a set
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.mboxes = [bob_email1, bob_email2]

    def test_reset(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        bob.save()
        bob_uri = bob.id
        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)

        self.assertEquals(bob.short_bio_en, None)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

    def test_reset_and_requirement(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        self.assertTrue(bob.is_valid())
        bob.short_bio_fr = None
        self.assertFalse(bob.is_valid())

    def test_language(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        bob.save()
        bob_id = bob.id

        # To make sure this object won't be retrieved in the cache
        forbidden_string = "You should not retrieve this string"
        bob.short_bio_en = forbidden_string
        self.assertEquals(bob.short_bio_en, forbidden_string)

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.short_bio_en, None)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

        bob_bio_en_2 = "Test-driven developer."
        bob.short_bio_en = bob_bio_en_2
        bob.save()
        bob.short_bio_en = "You should not retrieve this string (again)"

        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.short_bio_en, bob_bio_en_2)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

    def test_rsa_key(self):
        rsa_key = self.create_rsa_key()
        rsa_skolemized_iri = rsa_key.id
        del rsa_key
        manager.clear_resource_cache()

        rsa_key = rsa_model.get(id=rsa_skolemized_iri)
        self.assertEquals(rsa_key.exponent, key_exponent)
        self.assertEquals(rsa_key.modulus, key_modulus)
        self.assertEquals(rsa_key.label, key_label)
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.exponent = "String not a int"
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.modulus = "not an hexa value"
        # Values should already be encoded in hexadecimal strings
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.modulus = 235
        rsa_key.modulus = format(235, "x")
        with self.assertRaises(OMRequiredPropertyError):
            rsa_model.create(exponent=key_exponent)

    def test_filter_two_bobs(self):
        #Bob 1
        self.create_bob()

        bob2_mail = "<EMAIL>"
        bob2_bio_en = "I am a double."
        # Bob 2
        lp_model.create(name=bob_name, mboxes={bob2_mail}, short_bio_en=bob2_bio_en)

        bobs = list(lp_model.filter(name=bob_name))
        self.assertEquals(len(bobs), 2)
        self.assertEquals(bobs[0].name, bobs[1].name)
        self.assertEquals(bobs[0].name, bob_name)
        self.assertNotEquals(bobs[0].mboxes, bobs[1].mboxes)

        bobs2 = set(lp_model.filter(name=bob_name,
                                               # mboxes is NOT REQUIRED to be exhaustive
                                               mboxes={bob_email2}))
        self.assertEquals(len(bobs2), 1)
        bobs3 = set(lp_model.filter(name=bob_name,
                                               mboxes={bob_email1, bob_email2}))
        self.assertEquals(bobs2, bobs3)

        # Nothing
        bobs4 = list(lp_model.filter(name=bob_name,
                                                mboxes={bob_email1, bob_email2, bob2_mail}))
        self.assertEquals(len(bobs4), 0)

    def test_wrong_filter(self):
        with self.assertRaises(OMAttributeAccessError):
            lp_model.filter(undeclared_attr="not in datastore")

    def test_set_validation(self):
        with self.assertRaises(OMAttributeTypeCheckError):
            # Mboxes should be a set
            lp_model.create(name="Lola", mboxes="<EMAIL>",
                                       short_bio_en="Will not exist.")
        with self.assertRaises(OMAttributeTypeCheckError):
            # Mboxes should be a set not a list
            lp_model.create(name="Lola", mboxes=["<EMAIL>"],
                                       short_bio_en="Will not exist.")

    def test_children_object_assignment(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        # Children
        bob_children = [alice, john]
        bob_children_ids = [c.id for c in bob_children]
        bob.children = bob_children
        bob_uri = bob.id
        bob.save()

        # Force reload from the triplestore
        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)
        self.assertEquals(bob_children_ids, [c.id for c in bob.children])

    def test_children_uri_assignment(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        bob_uri = bob.id
        bob_children_uris = [alice.id, john.id]
        bob.children = bob_children_uris
        bob.save()

        # Force reload from the triplestore
        del bob
        manager.clear_resource_cache()

        bob = lp_model.get(id=bob_uri)
        self.assertEquals(bob.id, bob_uri)
        self.assertEquals(bob.name, bob_name)
        self.assertEquals(bob_children_uris, [c.id for c in bob.children])

    def test_set_assignment_instead_of_list(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        #Set assignment instead of a list
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.children = {alice.id, john.id}

    def test_children_list(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        # Children
        bob_children = [alice, john]
        bob.children = bob_children
        bob.save()

        children_request = """SELECT ?child
                              WHERE
                              { <%s> rel:parentOf ?children.
                                ?children rdf:rest*/rdf:first ?child
                              } """ % bob.id
        children_found = [str(r) for r, in data_graph.query(children_request)]
        #print default_graph.serialize(format="turtle")
        # No guarantee about the order
        self.assertEquals(set(children_found), set([c.id for c in bob_children]))

    def test_bob_json(self):
        bob = self.create_bob()
        bob_json = json.loads(bob.to_json())
        self.assertEquals(bob_json["name"], bob_name)
        self.assertEquals(bob_json["blog"], bob_blog)
        self.assertEquals(set(bob_json["mboxes"]), bob_emails)
        self.assertEquals(bob_json["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_json["short_bio_fr"], bob_bio_fr)
        self.assertEquals(bob_json["types"], lp_model.ancestry_iris)

    def test_bob_jsonld(self):
        bob = self.create_bob()
        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["blog"], bob_blog)
        self.assertEquals(set(bob_jsonld["mboxes"]), bob_emails)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_jsonld["short_bio_fr"], bob_bio_fr)
        self.assertTrue("@context" in bob_jsonld)
        self.assertEquals(bob_jsonld["@context"], context["@context"])
        self.assertEquals(bob_jsonld["types"], lp_model.ancestry_iris)

    def test_rsa_jsonld(self):
        rsa_key = self.create_rsa_key()
        key_jsonld = json.loads(rsa_key.to_jsonld())
        self.assertEquals(key_jsonld["modulus"], key_modulus)
        self.assertEquals(key_jsonld["exponent"], key_exponent)
        self.assertEquals(key_jsonld["label"], key_label)
        # Blank node so IRI must not appear
        self.assertFalse("id" in key_jsonld)

    def test_rdf(self):
        bob = self.create_bob()
        bob_uri = URIRef(bob.id)
        g = Graph()
        g.parse(data=bob.to_rdf("turtle"), format="turtle")
        self.assertEquals(g.value(bob_uri, URIRef(FOAF + "name")).toPython(), bob_name)
        self.assertEquals(g.value(bob_uri, URIRef(FOAF + "weblog")).toPython(), bob_blog)
        self.assertEquals({mbox.toPython() for mbox in g.objects(bob_uri, URIRef(FOAF + "mbox"))},
                          bob_emails)
        self.assertEquals({bio.toPython() for bio in g.objects(bob_uri, URIRef(BIO + "olb"))},
                          {bob_bio_en, bob_bio_fr})

    def test_is_blank_node(self):
        bob = self.create_bob()
        self.assertFalse(bob.is_blank_node())
        alice = lp_model.new()
        self.assertFalse(alice.is_blank_node())

        raoul = lp_model.new(id="http://localhost/.well-known/genid/2387335")
        self.assertTrue(raoul.is_blank_node())

    def test_same_document(self):
        bob = self.create_bob()
        alice = self.create_alice()
        self.assertFalse(bob.in_same_document(alice))

        partial_uri = u"http://localhost/persons"
        bob_uri = partial_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = partial_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        self.assertTrue(bob.in_same_document(alice))

    def test_children_jsonld(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()
        bob_children = [alice, john]
        bob.children = bob_children
        bob.save()

        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["blog"], bob_blog)
        self.assertEquals(set(bob_jsonld["mboxes"]), bob_emails)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_jsonld["short_bio_fr"], bob_bio_fr)
        self.assertEquals(bob_jsonld["@context"], context["@context"])
        self.assertEquals(bob_jsonld["children"], [c.id for c in bob_children])

    def test_friendship_jsonld(self):
        friendship_uri = u"http://localhost/friendship"
        bob_uri = friendship_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = friendship_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        bob_friends = {alice}
        bob.friends = bob_friends
        bob.save()
        alice_friends = {bob}
        alice.friends = alice_friends
        alice.save()

        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals([c["id"] for c in bob_jsonld["friends"]],
                          [c.id for c in bob_friends])
        self.assertEquals(["@context" in c for c in bob_jsonld["friends"]],
                          [False])
        self.assertEquals(bob_jsonld["friends"][0]["friends"][0], bob_uri)

    def test_friendship_rdf(self):
        friendship_uri = u"http://localhost/friendship"
        bob_uri = friendship_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = friendship_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        bob_friends = {alice}
        bob.friends = bob_friends
        bob.save()
        alice_friends = {bob}
        alice.friends = alice_friends
        alice.save()

        g = Graph()
        g.parse(data=bob.to_rdf("turtle"), format="turtle")
        self.assertEquals(g.value(URIRef(bob_uri), URIRef(FOAF + "knows")).toPython(), alice_uri)
        self.assertEquals(g.value(URIRef(bob_uri), URIRef(FOAF + "name")).toPython(), bob_name)
        self.assertEquals(g.value(URIRef(alice_uri), URIRef(FOAF + "name")).toPython(), alice_name)

    def test_bob_key_jsonld(self):
        bob = self.create_bob()
        bob_iri = bob.id
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.save()
        del bob
        del rsa_key
        manager.clear_resource_cache()

        bob = lp_model.get(id=bob_iri)
        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)

        key_jsonld = bob_jsonld["keys"][0]
        self.assertEquals(key_jsonld["modulus"], key_modulus)
        self.assertEquals(key_jsonld["exponent"], key_exponent)
        self.assertEquals(key_jsonld["label"], key_label)
        self.assertFalse("id" in key_jsonld)
        self.assertFalse("@context" in key_jsonld)

    def test_out_of_band_update(self):
        jason_uri = URIRef("https://example.com/jason#me")
        data_graph.add((jason_uri, URIRef(FOAF + "name"), Literal("Jason")))
        data_graph.add((jason_uri, URIRef(BIO + "olb"), Literal("Jason was a warrior", lang="en")))

        # LocalPerson and Person types are missing
        with self.assertRaises(OMClassInstanceError):
            lp_model.get(id=str(jason_uri))

        for class_iri in lp_model.ancestry_iris:
            data_graph.add((jason_uri, RDF.type, URIRef(class_iri)))

        # Mboxes is still missing
        manager.clear_resource_cache()
        jason = lp_model.get(id=str(jason_uri))
        self.assertFalse(jason.is_valid())

        mboxes = {"<EMAIL>", "<EMAIL>"}
        data_graph.parse(data=json.dumps({"@id": jason_uri,
                                          "@type": ["LocalPerson", "Person"],
                                          # Required
                                          "mboxes": list(mboxes)}),
                         context=context, format="json-ld")

        # Clear the cache (out-of-band update)
        manager.clear_resource_cache()
        jason = lp_model.get(id=jason_uri)
        self.assertEquals(jason.mboxes, mboxes)
        self.assertTrue(jason.is_valid())

    def test_iri_uniqueness(self):
        bob = self.create_bob()
        bob_iri = bob.id

        with self.assertRaises(OMUniquenessError):
            lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist")

        with self.assertRaises(OMUniquenessError):
            lp_model.create(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist")

        with self.assertRaises(OMUniquenessError):
            lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist", create=True)

        # Forces the creation (by claiming your are not)
        # Dangerous!
        short_bio_en = u"Is forced to exist"
        bob2 = lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=short_bio_en, create=False)
        self.assertEquals(bob2.short_bio_en, short_bio_en)

    def test_gpg_key(self):
        bob = self.create_bob()
        bob_id = bob.id
        bob.gpg_key = self.create_gpg_key()
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

        bob.save()
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

    def test_delete_bob(self):
        bob = self.create_bob()
        request = """ASK {?x foaf:name "%s"^^xsd:string }""" % bob_name
        self.assertTrue(bool(data_graph.query(request)))

        bob.delete()
        self.assertFalse(bool(data_graph.query(request)))

    def test_delete_rsa_but_no_alice(self):
        ask_modulus = """ASK {?x cert:modulus "%s"^^xsd:hexBinary }""" % key_modulus
        ask_alice = """ASK {?x foaf:name "%s"^^xsd:string }""" % alice_name
        self.assertFalse(bool(data_graph.query(ask_modulus)))
        self.assertFalse(bool(data_graph.query(ask_alice)))

        bob = self.create_bob()
        alice = self.create_alice()
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.children = [alice]
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_modulus)))
        self.assertTrue(bool(data_graph.query(ask_alice)))

        bob.delete()
        # Blank node is deleted
        self.assertFalse(bool(data_graph.query(ask_modulus)))
        # Alice is not (non-blank)
        self.assertTrue(bool(data_graph.query(ask_alice)))

    def test_rsa_key_removal(self):
        ask_modulus = """ASK {?x cert:modulus "%s"^^xsd:hexBinary }""" % key_modulus
        self.assertFalse(bool(data_graph.query(ask_modulus)))

        bob = self.create_bob()
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_modulus)))

        bob.keys = None
        bob.save()
        self.assertFalse(bool(data_graph.query(ask_modulus)))

    def test_gpg_key_removal(self):
        ask_fingerprint = """ASK {?x wot:fingerprint "%s"^^xsd:hexBinary }""" % gpg_fingerprint
        bob = self.create_bob()
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))
        bob.gpg_key = self.create_gpg_key()
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))

        bob.gpg_key = None
        bob.save()
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_delete_gpg(self):
        ask_fingerprint = """ASK {?x wot:fingerprint "%s"^^xsd:hexBinary }""" % gpg_fingerprint
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

        bob = self.create_bob()
        gpg_key = self.create_gpg_key()
        self.assertEquals(gpg_key.fingerprint, gpg_fingerprint)
        bob.gpg_key = gpg_key
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))

        bob.delete()
        # Blank node is deleted
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_basic_bob_full_update(self):
        bob = self.create_bob()
        bob_dict = bob.to_dict()
        boby_name = "Boby"
        bob_dict["name"] = boby_name
        bob.full_update(bob_dict)
        self.assertEquals(bob.name, boby_name)

        bob_dict.pop("short_bio_en")
        bob.full_update(bob_dict)
        self.assertEquals(bob.short_bio_en, None)

    def test_bob_gpg_update(self):
        bob = self.create_bob()
        ask_fingerprint = """ASK {?x wot:fingerprint "%s"^^xsd:hexBinary }""" % gpg_fingerprint
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))
        bob.gpg_key = self.create_gpg_key()
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))
        bob_dict = bob.to_dict()

        # GPG key blank-node is included as a dict
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.full_update(bob_dict)

        # Replace the dict by an IRI
        bob_dict["gpg_key"] = bob.gpg_key.id
        bob.full_update(bob_dict)
        bob.gpg_key.fingerprint = gpg_fingerprint

        bob_dict["gpg_key"] = None
        bob.full_update(bob_dict)
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_wrong_update(self):<commit_msg>xsd:hexBinary removed from ASK queries (not supported by Fuseki)<commit_after># -*- coding: utf-8 -*-

from unittest import TestCase
from os import path
import json

from rdflib import ConjunctiveGraph, Graph, URIRef, Literal, RDF, XSD
from rdflib.plugins.stores.sparqlstore import SPARQLUpdateStore
from rdflib.namespace import FOAF

from oldman import create_resource_manager, parse_graph_safely
from oldman.attribute import OMAttributeTypeCheckError, OMRequiredPropertyError
from oldman.exception import OMClassInstanceError, OMAttributeAccessError, OMUniquenessError
from oldman.exception import OMWrongResourceError, OMObjectNotFoundError, OMHashIriError, OMEditError
from oldman.exception import OMDifferentBaseIRIError, OMForbiddenSkolemizedIRIError, OMUnauthorizedTypeChangeError
from oldman.rest.crud import CRUDController


default_graph = ConjunctiveGraph(SPARQLUpdateStore(queryEndpoint="http://localhost:3030/test/query",
                                                   update_endpoint="http://localhost:3030/test/update"))
#default_graph = ConjunctiveGraph()
schema_graph = default_graph.get_context(URIRef("http://localhost/schema"))
data_graph = default_graph.get_context(URIRef("http://localhost/data"))

BIO = "http://purl.org/vocab/bio/0.1/"
REL = "http://purl.org/vocab/relationship/"
CERT = "http://www.w3.org/ns/auth/cert#"
RDFS = "http://www.w3.org/2000/01/rdf-schema#"
WOT = "http://xmlns.com/wot/0.1/"

MY_VOC = "http://example.com/vocab#"
local_person_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "foaf": FOAF,
            "bio": BIO,
            "rel": REL,
            "cert": CERT,
            "wot": WOT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalPerson",
    "@type": "hydra:Class",
    "subClassOf": "foaf:Person",
    "supportedProperty": [
        {
            "property": "foaf:name",
            "required": True
        },
        {
            "property": "foaf:mbox",
            "required": True
        },
        {
            "property": "foaf:weblog"
        },
        {
            "property": "bio:olb",
            "required": True
        },
        {
            "property": "foaf:knows"
        },
        {
            "property": "rel:parentOf"
        },
        {
            "property": "cert:key"
        },
        {
            "property": "wot:hasKey"
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_person_def), format="json-ld")

local_rsa_key_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "rdfs": RDFS,
            "cert": CERT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalRSAPublicKey",
    "@type": "hydra:Class",
    "subClassOf": "cert:RSAPublicKey",
    "supportedProperty": [
        {
            "property": "cert:exponent",
            "required": True
        },
        {
            "property": "cert:modulus",
            "required": True
        },
        {
            "property": "rdfs:label"
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_rsa_key_def), format="json-ld")

local_gpg_key_def = {
    "@context": [
        {
            "myvoc": MY_VOC,
            "wot": WOT
        },
        #"http://www.w3.org/ns/hydra/core"
        json.load(open(path.join(path.dirname(__file__), "hydra_core.jsonld")))["@context"]
    ],
    "@id": "myvoc:LocalGPGPublicKey",
    "@type": "hydra:Class",
    "subClassOf": "wot:PubKey",
    "supportedProperty": [
        {
            "property": "wot:fingerprint",
            "required": True,
        },
        {
            "property": "wot:hex_id",
            "required": True
        }
    ]
}
parse_graph_safely(schema_graph, data=json.dumps(local_gpg_key_def), format="json-ld")
#print schema_graph.serialize(format="turtle")

context = {
    "@context": {
        "myvoc": MY_VOC,
        "rdfs": RDFS,
        "foaf": "http://xmlns.com/foaf/0.1/",
        "bio": "http://purl.org/vocab/bio/0.1/",
        "xsd": "http://www.w3.org/2001/XMLSchema#",
        "rel": "http://purl.org/vocab/relationship/",
        "cert": CERT,
        "wot": WOT,
        "id": "@id",
        "types": "@type",
        "LocalPerson": "myvoc:LocalPerson",
        "Person": "foaf:Person",
        "LocalRSAPublicKey": "myvoc:LocalRSAPublicKey",
        "LocalGPGPublicKey": "myvoc:LocalGPGPublicKey",
        "name": {
            "@id": "foaf:name",
            "@type": "xsd:string"
        },
        "mboxes": {
            "@id": "foaf:mbox",
            "@type": "xsd:string",
            "@container": "@set"
        },
        "blog": {
            "@id": "foaf:weblog",
            "@type": "@id"
        },
        "short_bio_fr": {
            "@id": "bio:olb",
            "@language": "fr"
        },
        "short_bio_en": {
            "@id": "bio:olb",
            "@language": "en"
        },
        "friends": {
            "@id": "foaf:knows",
            "@type": "@id",
            "@container": "@set"
        },
        "children": {
            "@id": "rel:parentOf",
            "@type": "@id",
            "@container": "@list"
        },
        "gpg_key": {
            "@id": "wot:hasKey",
            "@type": "@id"
        },
        "keys": {
            "@id": "cert:key",
            "@type": "@id",
            "@container": "@set"
        },
        # For keys
        "modulus": {
            "@id": "cert:modulus",
            "@type": "xsd:hexBinary"
        },
        "exponent": {
            "@id": "cert:exponent",
            "@type": "xsd:integer"
        },
        "label": {
            "@id": "rdfs:label",
            "@type": "xsd:string"
        },
        # For GPG
        "fingerprint": {
            "@id": "wot:fingerprint",
            "@type": "xsd:hexBinary"
        },
        "hex_id": {
            "@id": "wot:hex_id",
            "@type": "xsd:hexBinary"
        }
    }
}

#default_graph.namespace_manager.bind("xsd", XSD)
default_graph.namespace_manager.bind("foaf", FOAF)
default_graph.namespace_manager.bind("wot", WOT)
default_graph.namespace_manager.bind("rel", REL)
default_graph.namespace_manager.bind("cert", CERT)

manager = create_resource_manager(schema_graph, data_graph)
# Model classes are generated here!
lp_model = manager.create_model("LocalPerson", context, iri_prefix="http://localhost/persons/",
                               iri_fragment="me")
rsa_model = manager.create_model("LocalRSAPublicKey", context)
gpg_model = manager.create_model("LocalGPGPublicKey", context)

crud_controller = CRUDController(manager)

bob_name = "Bob"
bob_blog = "http://blog.example.com/"
bob_email1 = "bob@localhost"
bob_email2 = "<EMAIL>"
bob_emails = {bob_email1, bob_email2}
bob_bio_en = "I grow up in ... ."
bob_bio_fr = u"J'ai grandi en ... ."
alice_name = "Alice"
alice_mail = "<EMAIL>"
alice_bio_en = "I am an expert on this and that"
john_name = "John"
john_mail = "<EMAIL>"
john_bio_en = "Supporter of Linked Data"
key_modulus = "b42dbf23ee820be938bee7298893e434f8f74d4be2bbe39408776d695168d09262da2a849962"
key_exponent = 65537
key_label = "Key 1"
gpg_fingerprint = "<KEY>"
gpg_hex_id = "002640e2"

prof_type = MY_VOC + "Professor"
researcher_type = MY_VOC + "Researcher"

# xsd:hexBinary behaves strangely in Fuseki. Not supported by SPARQL? http://www.w3.org/2011/rdf-wg/wiki/XSD_Datatypes
#ask_fingerprint = """ASK { ?x wot:fingerprint "%s"^^xsd:hexBinary }""" % gpg_fingerprint
ask_fingerprint = """ASK { ?x wot:fingerprint ?y }"""
#ask_modulus = """ASK {?x cert:modulus "%s"^^xsd:hexBinary }""" % key_modulus
ask_modulus = """ASK {?x cert:modulus ?y }"""

class ModelTest(TestCase):

    def tearDown(self):
        """ Clears the data graph """
        default_graph.update("CLEAR GRAPH <%s>" % data_graph.identifier)
        manager.clear_resource_cache()

    def create_bob(self):
        return lp_model.create(name=bob_name, blog=bob_blog, mboxes=bob_emails,
                               short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)

    def create_alice(self):
        return lp_model.create(name=alice_name, mboxes={alice_mail}, short_bio_en=alice_bio_en)

    def create_john(self):
        return lp_model.create(name=john_name, mboxes={john_mail}, short_bio_en=john_bio_en)

    def create_rsa_key(self):
        return rsa_model.create(exponent=key_exponent, modulus=key_modulus, label=key_label)

    def create_gpg_key(self):
        return gpg_model.create(fingerprint=gpg_fingerprint, hex_id=gpg_hex_id)

    def test_bio_requirement(self):
        bob = lp_model.new()
        bob.name = bob_name
        bob.blog = bob_blog
        bob.mboxes = {bob_email1}

        self.assertFalse(bob.is_valid())
        self.assertRaises(OMRequiredPropertyError, bob.save)

        # Bio is required
        bob.short_bio_en = bob_bio_en
        self.assertTrue(bob.is_valid())
        bob.save()

    def test_person_types(self):
        bob = self.create_bob()
        expected_types = ["http://example.com/vocab#LocalPerson",
                          "http://xmlns.com/foaf/0.1/Person"]
        self.assertEquals(bob.types, expected_types)

        # Check the triplestore
        type_request = """SELECT ?t WHERE {?x a ?t }"""
        retrieved_types = {str(r) for r, in data_graph.query(type_request, initBindings={'x': URIRef(bob.id)})}
        self.assertEquals(set(expected_types), retrieved_types)

    def test_bob_in_triplestore(self):
        request = """ASK { ?x foaf:name "%s"^^xsd:string }""" % bob_name
        self.assertFalse(bool(data_graph.query(request)))
        self.create_bob()
        self.assertTrue(bool(data_graph.query(request)))

    def test_bob_attributes(self):
        bob = self.create_bob()
        self.assertEquals(bob_name, bob.name)
        self.assertEquals(bob_blog, bob.blog.id)
        self.assertEquals(bob_emails, bob.mboxes)
        self.assertEquals(bob_bio_en, bob.short_bio_en)
        self.assertEquals(bob_bio_fr, bob.short_bio_fr)

    def test_bob_loading(self):
        bob = self.create_bob()
        bob_uri = bob.id

        # Not saved
        bob.name = "You should not retrieve this string"

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)

        self.assertEquals(bob_name, bob.name)
        self.assertEquals(bob_blog, bob.blog.id)
        self.assertEquals(bob_emails, bob.mboxes)
        self.assertEquals(bob_bio_en, bob.short_bio_en)
        self.assertEquals(bob_bio_fr, bob.short_bio_fr)

    def test_string_validation(self):
        bob = self.create_bob()
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.name = 2

    def test_not_saved(self):
        bob = self.create_bob()
        new_name = "<NAME>"
        bob.name = new_name
        # Not saved
        self.assertFalse(bool(data_graph.query("""ASK {?x foaf:name "%s"^^xsd:string }""" % new_name)))

    def test_objects_access(self):
        """ Object manager is only accessible at the class level """
        bob = self.create_bob()
        self.assertRaises(AttributeError, getattr, bob, "objects")

    def test_multiple_mboxes(self):
        bob = self.create_bob()
        email3 = "<EMAIL>"
        bob.mboxes = {bob_email2, email3}
        bob.save()

        mbox_query = """ASK {?x foaf:mbox "%s"^^xsd:string }"""
        self.assertTrue(bool(data_graph.query(mbox_query % bob_email2)))
        self.assertTrue(bool(data_graph.query(mbox_query % email3)))
        # Has been removed
        self.assertFalse(bool(data_graph.query(mbox_query % bob_email1)))

    def test_list_assignment_instead_of_set(self):
        bob = lp_model.new()
        bob.name = bob_name
        bob.short_bio_en = bob_bio_en

        # List assignment instead of a set
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.mboxes = [bob_email1, bob_email2]

    def test_reset(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        bob.save()
        bob_uri = bob.id
        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)

        self.assertEquals(bob.short_bio_en, None)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

    def test_reset_and_requirement(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        self.assertTrue(bob.is_valid())
        bob.short_bio_fr = None
        self.assertFalse(bob.is_valid())

    def test_language(self):
        bob = self.create_bob()
        bob.short_bio_en = None
        bob.save()
        bob_id = bob.id

        # To make sure this object won't be retrieved in the cache
        forbidden_string = "You should not retrieve this string"
        bob.short_bio_en = forbidden_string
        self.assertEquals(bob.short_bio_en, forbidden_string)

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.short_bio_en, None)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

        bob_bio_en_2 = "Test-driven developer."
        bob.short_bio_en = bob_bio_en_2
        bob.save()
        bob.short_bio_en = "You should not retrieve this string (again)"

        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.short_bio_en, bob_bio_en_2)
        self.assertEquals(bob.short_bio_fr, bob_bio_fr)

    def test_rsa_key(self):
        rsa_key = self.create_rsa_key()
        rsa_skolemized_iri = rsa_key.id
        del rsa_key
        manager.clear_resource_cache()

        rsa_key = rsa_model.get(id=rsa_skolemized_iri)
        self.assertEquals(rsa_key.exponent, key_exponent)
        self.assertEquals(rsa_key.modulus, key_modulus)
        self.assertEquals(rsa_key.label, key_label)
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.exponent = "String not a int"
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.modulus = "not an hexa value"
        # Values should already be encoded in hexadecimal strings
        with self.assertRaises(OMAttributeTypeCheckError):
            rsa_key.modulus = 235
        rsa_key.modulus = format(235, "x")
        with self.assertRaises(OMRequiredPropertyError):
            rsa_model.create(exponent=key_exponent)

    def test_filter_two_bobs(self):
        #<NAME>
        self.create_bob()

        bob2_mail = "<EMAIL>"
        bob2_bio_en = "I am a double."
        # Bob 2
        lp_model.create(name=bob_name, mboxes={bob2_mail}, short_bio_en=bob2_bio_en)

        bobs = list(lp_model.filter(name=bob_name))
        self.assertEquals(len(bobs), 2)
        self.assertEquals(bobs[0].name, bobs[1].name)
        self.assertEquals(bobs[0].name, bob_name)
        self.assertNotEquals(bobs[0].mboxes, bobs[1].mboxes)

        bobs2 = set(lp_model.filter(name=bob_name,
                                               # mboxes is NOT REQUIRED to be exhaustive
                                               mboxes={bob_email2}))
        self.assertEquals(len(bobs2), 1)
        bobs3 = set(lp_model.filter(name=bob_name,
                                               mboxes={bob_email1, bob_email2}))
        self.assertEquals(bobs2, bobs3)

        # Nothing
        bobs4 = list(lp_model.filter(name=bob_name,
                                                mboxes={bob_email1, bob_email2, bob2_mail}))
        self.assertEquals(len(bobs4), 0)

    def test_wrong_filter(self):
        with self.assertRaises(OMAttributeAccessError):
            lp_model.filter(undeclared_attr="not in datastore")

    def test_set_validation(self):
        with self.assertRaises(OMAttributeTypeCheckError):
            # Mboxes should be a set
            lp_model.create(name="Lola", mboxes="<EMAIL>",
                                       short_bio_en="Will not exist.")
        with self.assertRaises(OMAttributeTypeCheckError):
            # Mboxes should be a set not a list
            lp_model.create(name="Lola", mboxes=["<EMAIL>"],
                                       short_bio_en="Will not exist.")

    def test_children_object_assignment(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        # Children
        bob_children = [alice, john]
        bob_children_ids = [c.id for c in bob_children]
        bob.children = bob_children
        bob_uri = bob.id
        bob.save()

        # Force reload from the triplestore
        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_uri)
        self.assertEquals(bob_children_ids, [c.id for c in bob.children])

    def test_children_uri_assignment(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        bob_uri = bob.id
        bob_children_uris = [alice.id, john.id]
        bob.children = bob_children_uris
        bob.save()

        # Force reload from the triplestore
        del bob
        manager.clear_resource_cache()

        bob = lp_model.get(id=bob_uri)
        self.assertEquals(bob.id, bob_uri)
        self.assertEquals(bob.name, bob_name)
        self.assertEquals(bob_children_uris, [c.id for c in bob.children])

    def test_set_assignment_instead_of_list(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()

        #Set assignment instead of a list
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.children = {alice.id, john.id}

    def test_children_list(self):
        bob = self.create_bob()
        bob_iri = bob.id
        alice = self.create_alice()
        john = self.create_john()

        # Children
        bob_children = [alice, john]
        bob.children = bob_children
        bob.save()

        children_request = """SELECT ?child
                              WHERE
                              { <%s> rel:parentOf ?children.
                                ?children rdf:rest*/rdf:first ?child
                              } """ % bob.id
        children_found = [str(r) for r, in data_graph.query(children_request)]
        #print default_graph.serialize(format="turtle")
        # No guarantee about the order
        self.assertEquals(set(children_found), set([c.id for c in bob_children]))

        bob_children_iris = [c.id for c in bob_children]
        del bob
        manager.clear_resource_cache()
        bob = manager.get(id=bob_iri)
        self.assertEquals([c.id for c in bob.children], bob_children_iris)

    def test_bob_json(self):
        bob = self.create_bob()
        bob_json = json.loads(bob.to_json())
        self.assertEquals(bob_json["name"], bob_name)
        self.assertEquals(bob_json["blog"], bob_blog)
        self.assertEquals(set(bob_json["mboxes"]), bob_emails)
        self.assertEquals(bob_json["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_json["short_bio_fr"], bob_bio_fr)
        self.assertEquals(bob_json["types"], lp_model.ancestry_iris)

    def test_bob_jsonld(self):
        bob = self.create_bob()
        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["blog"], bob_blog)
        self.assertEquals(set(bob_jsonld["mboxes"]), bob_emails)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_jsonld["short_bio_fr"], bob_bio_fr)
        self.assertTrue("@context" in bob_jsonld)
        self.assertEquals(bob_jsonld["@context"], context["@context"])
        self.assertEquals(bob_jsonld["types"], lp_model.ancestry_iris)

    def test_rsa_jsonld(self):
        rsa_key = self.create_rsa_key()
        key_jsonld = json.loads(rsa_key.to_jsonld())
        self.assertEquals(key_jsonld["modulus"], key_modulus)
        self.assertEquals(key_jsonld["exponent"], key_exponent)
        self.assertEquals(key_jsonld["label"], key_label)
        # Blank node so IRI must not appear
        self.assertFalse("id" in key_jsonld)

    def test_rdf(self):
        bob = self.create_bob()
        bob_uri = URIRef(bob.id)
        g = Graph()
        g.parse(data=bob.to_rdf("turtle"), format="turtle")
        self.assertEquals(g.value(bob_uri, URIRef(FOAF + "name")).toPython(), bob_name)
        self.assertEquals(g.value(bob_uri, URIRef(FOAF + "weblog")).toPython(), bob_blog)
        self.assertEquals({mbox.toPython() for mbox in g.objects(bob_uri, URIRef(FOAF + "mbox"))},
                          bob_emails)
        self.assertEquals({bio.toPython() for bio in g.objects(bob_uri, URIRef(BIO + "olb"))},
                          {bob_bio_en, bob_bio_fr})

    def test_is_blank_node(self):
        bob = self.create_bob()
        self.assertFalse(bob.is_blank_node())
        alice = lp_model.new()
        self.assertFalse(alice.is_blank_node())

        raoul = lp_model.new(id="http://localhost/.well-known/genid/2387335")
        self.assertTrue(raoul.is_blank_node())

    def test_same_document(self):
        bob = self.create_bob()
        alice = self.create_alice()
        self.assertFalse(bob.in_same_document(alice))

        partial_uri = u"http://localhost/persons"
        bob_uri = partial_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = partial_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        self.assertTrue(bob.in_same_document(alice))

    def test_children_jsonld(self):
        bob = self.create_bob()
        alice = self.create_alice()
        john = self.create_john()
        bob_children = [alice, john]
        bob.children = bob_children
        bob.save()

        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["blog"], bob_blog)
        self.assertEquals(set(bob_jsonld["mboxes"]), bob_emails)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)
        self.assertEquals(bob_jsonld["short_bio_fr"], bob_bio_fr)
        self.assertEquals(bob_jsonld["@context"], context["@context"])
        self.assertEquals(bob_jsonld["children"], [c.id for c in bob_children])

    def test_friendship_jsonld(self):
        friendship_uri = u"http://localhost/friendship"
        bob_uri = friendship_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = friendship_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        bob_friends = {alice}
        bob.friends = bob_friends
        bob.save()
        alice_friends = {bob}
        alice.friends = alice_friends
        alice.save()

        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals([c["id"] for c in bob_jsonld["friends"]],
                          [c.id for c in bob_friends])
        self.assertEquals(["@context" in c for c in bob_jsonld["friends"]],
                          [False])
        self.assertEquals(bob_jsonld["friends"][0]["friends"][0], bob_uri)

    def test_friendship_rdf(self):
        friendship_uri = u"http://localhost/friendship"
        bob_uri = friendship_uri + "#bob"
        bob = lp_model.create(id=bob_uri, name=bob_name, blog=bob_blog, mboxes=bob_emails,
                                         short_bio_en=bob_bio_en, short_bio_fr=bob_bio_fr)
        alice_uri = friendship_uri + "#alice"
        alice = lp_model.create(id=alice_uri, name=alice_name, mboxes={alice_mail},
                                           short_bio_en=alice_bio_en)
        bob_friends = {alice}
        bob.friends = bob_friends
        bob.save()
        alice_friends = {bob}
        alice.friends = alice_friends
        alice.save()

        g = Graph()
        g.parse(data=bob.to_rdf("turtle"), format="turtle")
        self.assertEquals(g.value(URIRef(bob_uri), URIRef(FOAF + "knows")).toPython(), alice_uri)
        self.assertEquals(g.value(URIRef(bob_uri), URIRef(FOAF + "name")).toPython(), bob_name)
        self.assertEquals(g.value(URIRef(alice_uri), URIRef(FOAF + "name")).toPython(), alice_name)

    def test_bob_key_jsonld(self):
        bob = self.create_bob()
        bob_iri = bob.id
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.save()
        del bob
        del rsa_key
        manager.clear_resource_cache()

        bob = lp_model.get(id=bob_iri)
        bob_jsonld = json.loads(bob.to_jsonld())
        self.assertEquals(bob_jsonld["name"], bob_name)
        self.assertEquals(bob_jsonld["short_bio_en"], bob_bio_en)

        key_jsonld = bob_jsonld["keys"][0]
        self.assertEquals(key_jsonld["modulus"], key_modulus)
        self.assertEquals(key_jsonld["exponent"], key_exponent)
        self.assertEquals(key_jsonld["label"], key_label)
        self.assertFalse("id" in key_jsonld)
        self.assertFalse("@context" in key_jsonld)

    def test_out_of_band_update(self):
        jason_uri = URIRef("https://example.com/jason#me")
        data_graph.add((jason_uri, URIRef(FOAF + "name"), Literal("Jason")))
        data_graph.add((jason_uri, URIRef(BIO + "olb"), Literal("Jason was a warrior", lang="en")))

        # LocalPerson and Person types are missing
        with self.assertRaises(OMClassInstanceError):
            lp_model.get(id=str(jason_uri))

        for class_iri in lp_model.ancestry_iris:
            data_graph.add((jason_uri, RDF.type, URIRef(class_iri)))

        # Mboxes is still missing
        manager.clear_resource_cache()
        jason = lp_model.get(id=str(jason_uri))
        self.assertFalse(jason.is_valid())

        mboxes = {"<EMAIL>", "<EMAIL>"}
        data_graph.parse(data=json.dumps({"@id": jason_uri,
                                          "@type": ["LocalPerson", "Person"],
                                          # Required
                                          "mboxes": list(mboxes)}),
                         context=context, format="json-ld")

        # Clear the cache (out-of-band update)
        manager.clear_resource_cache()
        jason = lp_model.get(id=jason_uri)
        self.assertEquals(jason.mboxes, mboxes)
        self.assertTrue(jason.is_valid())

    def test_iri_uniqueness(self):
        bob = self.create_bob()
        bob_iri = bob.id

        with self.assertRaises(OMUniquenessError):
            lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist")

        with self.assertRaises(OMUniquenessError):
            lp_model.create(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist")

        with self.assertRaises(OMUniquenessError):
            lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=u"Will not exist", create=True)

        # Forces the creation (by claiming your are not)
        # Dangerous!
        short_bio_en = u"Is forced to exist"
        bob2 = lp_model.new(id=bob_iri, name=bob_name, mboxes=bob_emails, short_bio_en=short_bio_en, create=False)
        self.assertEquals(bob2.short_bio_en, short_bio_en)

    def test_gpg_key(self):
        bob = self.create_bob()
        bob_id = bob.id
        bob.gpg_key = self.create_gpg_key()
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

        bob.save()
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

        del bob
        manager.clear_resource_cache()
        bob = lp_model.get(id=bob_id)
        self.assertEquals(bob.gpg_key.fingerprint, gpg_fingerprint)
        self.assertEquals(bob.gpg_key.hex_id, gpg_hex_id)

    def test_delete_bob(self):
        bob = self.create_bob()
        request = """ASK {?x foaf:name "%s"^^xsd:string }""" % bob_name
        self.assertTrue(bool(data_graph.query(request)))

        bob.delete()
        self.assertFalse(bool(data_graph.query(request)))

    def test_delete_rsa_but_no_alice(self):
        ask_alice = """ASK {?x foaf:name "%s"^^xsd:string }""" % alice_name
        self.assertFalse(bool(data_graph.query(ask_modulus)))
        self.assertFalse(bool(data_graph.query(ask_alice)))

        bob = self.create_bob()
        alice = self.create_alice()
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.children = [alice]
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_modulus)))
        self.assertTrue(bool(data_graph.query(ask_alice)))

        bob.delete()
        # Blank node is deleted
        self.assertFalse(bool(data_graph.query(ask_modulus)))
        # Alice is not (non-blank)
        self.assertTrue(bool(data_graph.query(ask_alice)))

    def test_rsa_key_removal(self):
        self.assertFalse(bool(data_graph.query(ask_modulus)))

        bob = self.create_bob()
        rsa_key = self.create_rsa_key()
        bob.keys = {rsa_key}
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_modulus)))

        bob.keys = None
        bob.save()
        self.assertFalse(bool(data_graph.query(ask_modulus)))

    def test_gpg_key_removal(self):
        bob = self.create_bob()
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))
        bob.gpg_key = self.create_gpg_key()
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))

        bob.gpg_key = None
        bob.save()
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_delete_gpg(self):
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

        bob = self.create_bob()
        gpg_key = self.create_gpg_key()
        self.assertEquals(gpg_key.fingerprint, gpg_fingerprint)
        bob.gpg_key = gpg_key
        bob.save()
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))

        bob.delete()
        # Blank node is deleted
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_basic_bob_full_update(self):
        bob = self.create_bob()
        bob_dict = bob.to_dict()
        boby_name = "Boby"
        bob_dict["name"] = boby_name
        bob.full_update(bob_dict)
        self.assertEquals(bob.name, boby_name)

        bob_dict.pop("short_bio_en")
        bob.full_update(bob_dict)
        self.assertEquals(bob.short_bio_en, None)

    def test_bob_gpg_update(self):
        bob = self.create_bob()
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))
        bob.gpg_key = self.create_gpg_key()
        bob.save()
        #self.assertTrue(bool(default_graph.query("""ASK { GRAPH ?g {?x wot:fingerprint "%s"^^xsd:hexBinary } }""" % gpg_fingerprint)))
        self.assertTrue(bool(data_graph.query(ask_fingerprint)))
        bob_dict = bob.to_dict()

        # GPG key blank-node is included as a dict
        with self.assertRaises(OMAttributeTypeCheckError):
            bob.full_update(bob_dict)

        # Replace the dict by an IRI
        bob_dict["gpg_key"] = bob.gpg_key.id
        bob.full_update(bob_dict)
        bob.gpg_key.fingerprint = gpg_fingerprint

        bob_dict["gpg_key"] = None
        bob.full_update(bob_dict)
        self.assertFalse(bool(data_graph.query(ask_fingerprint)))

    def test_wrong_update(self):
{'fraction_non_alphanumeric': 0.12455319014329548, 'fraction_numerical': 0.005156106664979597, 'mean_word_length': 2.4665825977301385, 'pattern_counts': {'":': 220, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 2, 'lorem ipsum': 0, 'www.': 13, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 62, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7016393', 'n_tokens_mistral': 21607, 'n_tokens_neox': 20677, 'n_words': 3290}