feature: Add reset triplestore content method

knora/create_ontology.py

@@ -43,7 +43,8 @@ def main(args):
         exit(0)
 
     # create the knora connection object
-    con = Knora(args.server, args.user, args.password, ontology.get("prefixes"))
+    con = Knora(args.server, ontology.get("prefixes"))
+    con.login(args.user, args.password)
 
     # bulk_templ = con.create_schema(ontology["project"]["shortcode"], ontology["project"]["ontology"]["name"])
 

knora/knora.py

@@ -92,17 +92,21 @@ class Knora:
     This is the main class which holds all the methods for communication with the Knora backend.
     """
 
-    def __init__(self, server: str, email: str, password: str, prefixes: Dict[str, str] = None):
+    def __init__(self, server: str, prefixes: Dict[str, str] = None):
         """
         Constructor requiring the server address, the user and password of KNORA
         :param server: Address of the server, e.g https://api.dasch.swiss
-        :param email: Email of user, e.g., root@example.com
-        :param password: Password of the user, e.g. test
         :param prefixes: Ontology prefixes used
         """
         self.server = server
         self.prefixes = prefixes
 
+    def login(self, email: str, password: str):
+        """
+        Method to login into KNORA which creates a session token.
+        :param email: Email of user, e.g., root@example.com
+        :param password: Password of the user, e.g. test
+        """
         credentials = {
             "email": email,
             "password": password
@@ -114,24 +118,23 @@ def __init__(self, server: str, email: str, password: str, prefixes: Dict[str, s
             headers={'Content-Type': 'application/json; charset=UTF-8'},
             data=jsondata
         )
-
         self.on_api_error(req)
-
         result = req.json()
         self.token = result["token"]
 
     def get_token(self):
         return self.token
 
-    def __del__(self):
+    def logout(self):
         req = requests.delete(
             self.server + '/v2/authentication',
             headers={'Authorization': 'Bearer ' + self.token}
         )
-
-        result = req.json()
+        self.on_api_error(req)
         self.token = None
 
+    def __del__(self):
+        self.logout()
 
     def on_api_error(self, res):
         """
@@ -484,7 +487,7 @@ def delete_ontology(self, onto_iri: str, last_onto_date = None):
                               headers={'Authorization': 'Bearer ' + self.token})
         self.on_api_error(req)
         res = req.json()
-        return req
+        return res
 
     def get_ontology_graph(self,
                            shortcode: str,
@@ -1224,13 +1227,45 @@ def create_schema(self, shortcode: str, shortname: str):
         return schema
 
     def reset_triplestore_content(self):
-        rdfdata = []
+        rdfdata = [
+            {
+                "path": "knora-ontologies/knora-admin.ttl",
+                "name": "http://www.knora.org/ontology/knora-admin"
+            },
+            {
+                "path": "knora-ontologies/knora-base.ttl",
+                "name": "http://www.knora.org/ontology/knora-base"
+            },
+            {
+                "path": "knora-ontologies/standoff-onto.ttl",
+                "name": "http://www.knora.org/ontology/standoff"
+            },
+            {
+                "path": "knora-ontologies/standoff-data.ttl",
+                "name": "http://www.knora.org/data/standoff"
+            },
+            {
+                "path": "knora-ontologies/salsah-gui.ttl",
+                "name": "http://www.knora.org/ontology/salsah-gui"
+            },
+            {
+                "path": "_test_data/all_data/admin-data.ttl",
+                "name": "http://www.knora.org/data/admin"
+            },
+            {
+                "path": "_test_data/all_data/permissions-data.ttl",
+                "name": "http://www.knora.org/data/permissions"
+            },
+            {
+                "path": "_test_data/all_data/system-data.ttl",
+                "name": "http://www.knora.org/data/0000/SystemProject"
+            }
+        ]
         jsondata = json.dumps(rdfdata)
-        url = self.server + '/admin/store/ResetTriplestoreContent'
+        url = self.server + '/admin/store/ResetTriplestoreContent?prependdefaults=false'
 
         req = requests.post(url,
-                            headers={'Content-Type': 'application/json; charset=UTF-8',
-                                     'Authorization': 'Bearer ' + self.token},
+                            headers={'Content-Type': 'application/json; charset=UTF-8'},
                             data=jsondata)
         self.on_api_error(req)
         res = req.json()

tests/test_knora.py

@@ -3,24 +3,34 @@
 
 @pytest.fixture()
 def con():
-    server = "http://0.0.0.0:3333"
-    email = "root@example.com"
-    password = "test"
-    # projectcode = "00FE"
-    # ontoname = "KPT"
+    def _con(login: bool = True):
+        server = "http://0.0.0.0:3333"
+        email = "root@example.com"
+        password = "test"
+        # projectcode = "00FE"
+        # ontoname = "KPT"
+        con = Knora(server)
+        if (login):
+            con.login(email, password)
+        return con
 
-    con = Knora(server, email, password)
-    return con
+    return _con
 
+
+# resets the content of the triplestore
 def test_reset_triplestore_content(con):
-    res = con.reset_triplestore_content
-    assert(res == "gaga")
+    res = con(False).reset_triplestore_content()
+    assert(res)
+
 
+# retrieves user information
 def test_get_user(con):
-    res = con.get_user("http://rdfh.ch/users/root")
+    res = con(True).get_user("http://rdfh.ch/users/root")
     assert(res["username"] == "root")
 
+# creates a user
 def test_create_user(con):
+    connection = con(True)
     user = {
         "username": "testtest",
         "email": "testtest@example.com",
@@ -30,15 +40,21 @@ def test_create_user(con):
         "lang": "en"
     }
 
-    user_iri = con.create_user(
+    user_iri = connection.create_user(
         username=user["username"],
         email=user["email"],
         givenName=user["givenName"],
         familyName=user["familyName"],
         password="password",
         lang=user["lang"] if user.get("lang") is not None else "en")
 
-    res = con.get_user(user_iri)
+    print(user_iri)
+
+    res = connection.get_user(user_iri)
 
     assert(res["username"] == "testtest")
     assert(res["email"] == "testtest@example.com")
+
+    # try to login
+    connection.logout()
+    connection.login(res["email"], "test")