From b1e18ee1c8ced5c05658db34a1927f46b39de93d Mon Sep 17 00:00:00 2001
From: jrybicki-jsc <j.rybicki@fz-juelich.de>
Date: Tue, 1 Jun 2021 10:49:17 +0200
Subject: [PATCH] removing type (reserverd keyword)
---
apiserver/storage/JsonFileStorageAdapter.py | 49 ++++++++++-----------
apiserver/storage/LocationStorage.py | 43 +++++++++---------
tests/storage_tests/test_jsonbackend.py | 18 ++++++--
3 files changed, 60 insertions(+), 50 deletions(-)
diff --git a/apiserver/storage/JsonFileStorageAdapter.py b/apiserver/storage/JsonFileStorageAdapter.py
index a136aea..40162d5 100644
--- a/apiserver/storage/JsonFileStorageAdapter.py
+++ b/apiserver/storage/JsonFileStorageAdapter.py
@@ -1,12 +1,12 @@
-import os
import json
+import os
import uuid
-
-from .LocationStorage import AbstractLocationDataStorageAdapter, LocationData, LocationDataType
+from typing import List
from apiserver.config import ApiserverSettings
-from typing import List
+from .LocationStorage import (AbstractLocationDataStorageAdapter, LocationData,
+ LocationDataType)
class StoredData:
@@ -35,31 +35,30 @@ class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
if not (os.path.exists(self.data_dir) and os.path.isdir(self.data_dir)):
raise Exception(f"Data Directory {self.data_dir} does not exist.")
- def get_list(self, type: LocationDataType) -> List:
- localpath = os.path.join(self.data_dir, type.value)
+ def __setup_path(self, value):
+ localpath = os.path.join(self.data_dir, value)
if not (os.path.isdir(localpath)):
# This type has apparently not yet been used at all,
# create its directory and return an empty json file
os.mkdir(localpath)
- return []
- else:
- allFiles = [f for f in os.listdir(
- localpath) if os.path.isfile(os.path.join(localpath, f))]
- # now each file has to be checked for its filename (= oid)
- # and the LocationData name (which is inside the json)
- retList = []
- for f in allFiles:
- with open(os.path.join(localpath, f)) as file:
- data = json.load(file)
- retList.append({data['actualData']['name']: f})
- return retList
-
- def add_new(self, type: LocationDataType, data: LocationData, usr: str):
- localpath = os.path.join(self.data_dir, type.value)
- if not (os.path.isdir(localpath)):
- # This type has apparently not yet been used at all,
- # therefore we need to create its directory
- os.mkdir(localpath)
+ return localpath
+
+
+ def get_list(self, n_type: LocationDataType) -> List:
+ local_path = self.__setup_path(n_type.value)
+ allFiles = [f for f in os.listdir(
+ local_path) if os.path.isfile(os.path.join(local_path, f))]
+ # now each file has to be checked for its filename (= oid)
+ # and the LocationData name (which is inside the json)
+ retList = []
+ for f in allFiles:
+ with open(os.path.join(local_path, f)) as file:
+ data = json.load(file)
+ retList.append({data['actualData']['name']: f})
+ return retList
+
+ def add_new(self, n_type: LocationDataType, data: LocationData, usr: str):
+ localpath = self.__setup_path(value=n_type.value)
# create a unique oid, by randomly generating one,
# and re-choosing if it is already taken
oid = str(uuid.uuid4())
diff --git a/apiserver/storage/LocationStorage.py b/apiserver/storage/LocationStorage.py
index 929a629..1ffb220 100644
--- a/apiserver/storage/LocationStorage.py
+++ b/apiserver/storage/LocationStorage.py
@@ -1,9 +1,8 @@
-from pydantic import BaseModel
-
-from typing import Optional
-from typing import Dict
from enum import Enum
+from typing import Dict, Optional, List
+
+from pydantic import BaseModel
class LocationDataType(Enum):
@@ -36,38 +35,40 @@ class AbstractLocationDataStorageAdapter:
done by the caller, this adapter assumes that the user id fulfills the criteria.
Permissions are stored as a list of user ids, and every id is authorized for full access.
'''
- # get a list of all LocationData Elements with the provided type, as pairs of {name : id}
- def get_list(self, type: LocationDataType):
+ def get_list(self, n_type: LocationDataType) -> List:
+ # get a list of all LocationData Elements with the provided type, as pairs of {name : id}
raise NotImplementedError()
- # add a new element of the provided type, assign and return the id and the new data as {id : LocationData}
- def add_new(self, type: LocationDataType, data: LocationData, usr: str):
+ def add_new(self, n_type: LocationDataType, data: LocationData, usr: str):
+ # add a new element of the provided type, assign and return the id and
+ # the new data as {id : LocationData}
raise NotImplementedError()
- # return the LocationData of the requested object (identified by oid and type)
- def get_details(self, type: LocationDataType, oid: str):
+ def get_details(self, n_type: LocationDataType, oid: str):
+ # return the LocationData of the requested object (identified by oid and type)
raise NotImplementedError()
- # change the details of the requested object, return {oid : newData}
- def update_details(self, type: LocationDataType, oid: str, data: LocationData, usr: str):
+ def update_details(self, n_type: LocationDataType, oid: str, data: LocationData, usr: str):
+ # change the details of the requested object, return {oid : newData}
raise NotImplementedError()
- def delete(self, type: LocationDataType, oid: str, usr: str):
+ def delete(self, n_type: LocationDataType, oid: str, usr: str):
raise NotImplementedError()
- # return the owner of the requested object; if multiple owners are set, return them is a list
- def get_owner(self, type: LocationDataType, oid: str):
+ def get_owner(self, n_type: LocationDataType, oid: str):
+ # return the owner of the requested object; if multiple owners are set,
+ # return them is a list
raise NotImplementedError()
- # check if the given user has permission to change the given object
- def check_perm(self, type: LocationDataType, oid: str, usr: str):
+ def check_perm(self, n_type: LocationDataType, oid: str, usr: str):
+ # check if the given user has permission to change the given object
raise NotImplementedError()
- # add user to file perm
- def add_perm(self, type: LocationDataType, oid: str, usr: str):
+ def add_perm(self, n_type: LocationDataType, oid: str, usr: str):
+ # add user to file perm
raise NotImplementedError()
- # remove user from file perm
- def rm_perm(self, type: LocationDataType, oid: str, usr: str):
+ def rm_perm(self, n_type: LocationDataType, oid: str, usr: str):
+ # remove user from file perm
raise NotImplementedError()
diff --git a/tests/storage_tests/test_jsonbackend.py b/tests/storage_tests/test_jsonbackend.py
index ea2313b..c27ac1c 100644
--- a/tests/storage_tests/test_jsonbackend.py
+++ b/tests/storage_tests/test_jsonbackend.py
@@ -10,7 +10,6 @@ import shutil
class SomeTests(unittest.TestCase):
def setUp(self):
-
Settings = namedtuple('Settings',['json_storage_path'])
self.test_config = Settings('/tmp/json_test/')
pathlib.Path(self.test_config.json_storage_path).mkdir(parents=True, exist_ok=True)
@@ -19,11 +18,22 @@ class SomeTests(unittest.TestCase):
def tearDown(self):
if os.path.exists(self.test_config.json_storage_path):
- print('Paht exists. Removing')
+ print('Path exists. Removing')
shutil.rmtree(self.test_config.json_storage_path)
- def test_getList(self):
+ def test_get_emptyList(self):
test_type = LocationDataType.DATASET
- lst = self.store.get_list(type=test_type)
+ lst = self.store.get_list(n_type=test_type)
self.assertEqual(lst, [], 'Id should not be none')
+
+ def test_not_path(self):
+ Settings = namedtuple('Settings',['json_storage_path'])
+ test_config = Settings('/tmp/json_test/blah/')
+ self.assertRaises(Exception, JsonFileStorageAdapter, test_config)
+
+
+
+
+
+
\ No newline at end of file
--
GitLab