minor tweak to default values of APIHelper

Author: tipatterson-dev

conSys4Py/core/default_api_helpers.py

@@ -56,8 +56,6 @@ def resource_type_to_endpoint(res_type: APIResourceTypes, parent_type: APIResour
             return APITerms.DATASTREAMS.value
         case APIResourceTypes.OBSERVATION:
             return APITerms.OBSERVATIONS.value
-        case APIResourceTypes.SYSTEM:
-            return APITerms.SYSTEMS.value
         case APIResourceTypes.SYSTEM_EVENT:
             return APITerms.SYSTEM_EVENTS.value
         case APIResourceTypes.SAMPLING_FEATURE:
@@ -76,11 +74,11 @@ def resource_type_to_endpoint(res_type: APIResourceTypes, parent_type: APIResour
 
 @dataclass
 class APIHelper(ABC):
-    server_url = None
-    api_root = "/api"
-    username = None
-    password = None
-    user_auth = False
+    server_url: str = None
+    api_root: str = "api"
+    username: str = None
+    password: str = None
+    user_auth: bool = False
 
     def create_resource(self, res_type: APIResourceTypes, json_data: any, parent_res_id: str = None,
                         from_collection: bool = False, url_endpoint: str = None):
@@ -230,7 +228,7 @@ class DefaultObjectRepresentations(BaseModel):
     """
     # Part 1
     collections: str = Field(EncodingSchema.JSON.value)
-    deployments: str = Field(EncodingSchema. GEO_JSON.value)
+    deployments: str = Field(EncodingSchema.GEO_JSON.value)
     procedures: str = Field(EncodingSchema.GEO_JSON.value)
     properties: str = Field(EncodingSchema.SML_JSON.value)
     sampling_features: str = Field(EncodingSchema.GEO_JSON.value)
@@ -243,5 +241,3 @@ class DefaultObjectRepresentations(BaseModel):
     system_events: str = Field(EncodingSchema.OM_JSON.value)
     system_history: str = Field(EncodingSchema.GEO_JSON.value)
     # TODO: validate schemas for each resource to amke sure they are allowed per the spec
-
-

tests/conftest.py

@@ -5,6 +5,5 @@
 
 @pytest.fixture(autouse=True)
 def server_url():
-    print('Setting up server url')
     global server_url
     server_url = 'http://localhost:8282/sensorhub'

tests/test_default_api_helpers.py

@@ -0,0 +1,59 @@
+import conSys4Py.core.default_api_helpers as helpers
+from conSys4Py import APIResourceTypes
+
+
+def test_determine_parent_type():
+    assert helpers.determine_parent_type(APIResourceTypes.SYSTEM) == APIResourceTypes.SYSTEM
+    assert helpers.determine_parent_type(APIResourceTypes.COLLECTION) is None
+    assert helpers.determine_parent_type(APIResourceTypes.CONTROL_CHANNEL) == APIResourceTypes.SYSTEM
+    assert helpers.determine_parent_type(APIResourceTypes.COMMAND) == APIResourceTypes.CONTROL_CHANNEL
+    assert helpers.determine_parent_type(APIResourceTypes.DATASTREAM) == APIResourceTypes.SYSTEM
+    assert helpers.determine_parent_type(APIResourceTypes.OBSERVATION) == APIResourceTypes.DATASTREAM
+    assert helpers.determine_parent_type(APIResourceTypes.SYSTEM_EVENT) == APIResourceTypes.SYSTEM
+    assert helpers.determine_parent_type(APIResourceTypes.SAMPLING_FEATURE) == APIResourceTypes.SYSTEM
+    assert helpers.determine_parent_type(APIResourceTypes.PROCEDURE) is None
+    assert helpers.determine_parent_type(APIResourceTypes.PROPERTY) is None
+    assert helpers.determine_parent_type(APIResourceTypes.SYSTEM_HISTORY) is None
+    assert helpers.determine_parent_type(APIResourceTypes.DEPLOYMENT) is None
+
+
+def test_resource_type_to_ep():
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.SYSTEM) == 'systems'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.COLLECTION) == 'collections'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.CONTROL_CHANNEL) == 'controls'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.COMMAND) == 'commands'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.DATASTREAM) == 'datastreams'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.OBSERVATION) == 'observations'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.SYSTEM_EVENT) == 'systemEvents'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.SAMPLING_FEATURE) == 'samplingFeatures'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.PROCEDURE) == 'procedures'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.PROPERTY) == 'properties'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.SYSTEM_HISTORY) == 'history'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.DEPLOYMENT) == 'deployments'
+    assert helpers.resource_type_to_endpoint(APIResourceTypes.SYSTEM, APIResourceTypes.COLLECTION) == 'items'
+
+
+def test_resolve_resource_ep():
+    server_url = 'http://localhost:8181'
+    api_endpoint = 'api'
+    api_helper = helpers.APIHelper(server_url, api_endpoint)
+    assert api_helper.server_url == server_url
+    assert api_helper.api_root == api_endpoint
+    assert api_helper.resource_url_resolver(APIResourceTypes.SYSTEM) == f'{server_url}/{api_endpoint}/systems'
+    assert api_helper.resource_url_resolver(APIResourceTypes.SYSTEM,
+                                            '1234') == f'{server_url}/{api_endpoint}/systems/1234'
+    assert api_helper.resource_url_resolver(APIResourceTypes.COLLECTION, '1234',
+                                            from_collection=False) == f'{server_url}/{api_endpoint}/collections/1234'
+    assert api_helper.resource_url_resolver(APIResourceTypes.SYSTEM, '1234', '5678',
+                                            True) == f'{server_url}/{api_endpoint}/collections/5678/items/1234'
+
+
+def test_user_auth():
+    server_url = 'http://localhost:8181'
+    api_endpoint = 'api'
+    uname = 'user'
+    pword = 'pass'
+    api_helper = helpers.APIHelper(server_url, api_endpoint, uname, pword, True)
+    assert api_helper.get_helper_auth() == ('user', 'pass')
+    api_helper.user_auth = False
+    assert api_helper.get_helper_auth() is None