Added get_document and create_document of scope apis (#40)

* Added get_document and create_document of scope apis

* Changed test run idea

.github/workflows/tests.yaml

@@ -1,8 +1,6 @@
 name: Tests
 
-on: 
-  push:
-  pull_request:
+on:
   workflow_dispatch:
 
 jobs:

tests/test_api.py

@@ -14,7 +14,7 @@
 from upsonic_on_prem.api.urls import *
 
 from upsonic_on_prem.utils import AccessKey
-from upsonic_on_prem.utils import storage
+from upsonic_on_prem.utils import storage, storage_2, Scope
 
 
 
@@ -583,6 +583,73 @@ def test_events_get_x_api(self):
 
         storage.pop()
 
+    def test_scope_documentation(self):
+        storage_2.pop()
+        id = "test_scope_documentation"
+        accesskey = AccessKey(id)
+        accesskey.enable()
+        accesskey.set_scope_read("onur.my_function")
+
+        def my_function():
+            return "aaa"
+
+        the_scope = Scope("onur.my_function")
+        dumped_data = Fernet(base64.urlsafe_b64encode(hashlib.sha256("u".encode()).digest())).encrypt(
+            cloudpickle.dumps(my_function))
+
+        def get_document():
+            data = {"scope": "onur.my_function", }
+            response = requests.post("http://localhost:7777" + get_document_of_scope_url,
+                                     auth=HTTPBasicAuth("", id),
+                                     data=data)
+            return response.json()["result"]
+
+        the_scope.dump(dumped_data)
+        self.assertEqual(get_document(), the_scope.documentation)
+        the_scope.create_documentation()
+
+        print(the_scope.documentation)
+        self.assertEqual(get_document(), the_scope.documentation)
+
+        storage_2.pop()
+
+    def test_scope_documentation_create(self):
+        storage_2.pop()
+        id = "test_scope_documentation_create"
+        accesskey = AccessKey(id)
+        accesskey.enable()
+        accesskey.set_scope_read("onur.my_function")
+        accesskey.set_scope_write("onur.my_function")
+
+        def my_function():
+            return "aaa"
+
+        the_scope = Scope("onur.my_function")
+        dumped_data = Fernet(base64.urlsafe_b64encode(hashlib.sha256("u".encode()).digest())).encrypt(
+            cloudpickle.dumps(my_function))
+
+        def get_document():
+            data = {"scope": "onur.my_function", }
+            response = requests.post("http://localhost:7777" + get_document_of_scope_url,
+                                     auth=HTTPBasicAuth("", id),
+                                     data=data)
+            return response.json()["result"]
+
+        def create_document():
+            data = {"scope": "onur.my_function", }
+            response = requests.post("http://localhost:7777" + create_document_of_scope_url,
+                                     auth=HTTPBasicAuth("", id),
+                                     data=data)
+            return response.json()
+
+        the_scope.dump(dumped_data)
+        first = get_document()
+        self.assertEqual(first, the_scope.documentation)
+        create_document()
+
+        self.assertNotEqual(first, get_document())
+
+        storage_2.pop()
 
 
 

upsonic_on_prem/api/operations/__init__.py

@@ -1,2 +1,2 @@
 from .admin import *
-from .data import *
+from .user import *

upsonic_on_prem/api/operations/data.py → upsonic_on_prem/api/operations/user.py

@@ -35,3 +35,16 @@ def get_read_scopes_of_me():
 @app.route(get_write_scopes_of_me_url, methods=["get"])
 def get_write_scopes_of_me():
     return jsonify({"status": True, "result": AccessKey(request.authorization.password).scopes_write})
+
+
+@app.route(get_document_of_scope_url, methods=["POST"])
+def get_document_of_scope():
+    scope = request.form.get("scope")
+    return jsonify({"status": True, "result": Scope(scope).documentation})
+
+
+@app.route(create_document_of_scope_url, methods=["POST"])
+def create_document_of_scope():
+    scope = request.form.get("scope")
+    Scope(scope).create_documentation()
+    return jsonify({"status": True})

upsonic_on_prem/api/pre_process/__init__.py

@@ -50,8 +50,7 @@ def check():
 
 
     if not the_access_key.is_admin:
-        if not (
-                endpoint == dump_url or endpoint == load_url or endpoint == get_read_scopes_of_me_url or endpoint == get_write_scopes_of_me_url):
+        if not (endpoint in user_urs):
             print("endpoint", endpoint)
             print(request.endpoint)
             return Response(

upsonic_on_prem/api/pre_process/user.py

@@ -9,9 +9,9 @@ def user_pre_process(the_access_key:access_key, request):
     endpoint = "/"+request.endpoint
 
     operation_type = None
-    if endpoint == dump_url:
+    if endpoint in user_write_urls:
         operation_type = the_access_key.can_access_write
-    elif endpoint == load_url:
+    elif endpoint in user_read_urls:
         operation_type = the_access_key.can_access_read
     else:
         operation_type = free_operation

upsonic_on_prem/api/urls.py

@@ -47,3 +47,11 @@
 event_url = "/event"
 
 get_last_x_event_url = "/get_last_x_event"
+
+get_document_of_scope_url = "/get_document_of_scope"
+create_document_of_scope_url = "/create_document_of_scope"
+
+user_urs = [load_url, dump_url, get_read_scopes_of_me_url, get_write_scopes_of_me_url, get_document_of_scope_url,
+            create_document_of_scope_url]
+user_write_urls = [dump_url, create_document_of_scope_url]
+user_read_urls = [load_url, get_document_of_scope_url]