partial tests for control channels

Author: tipatterson-dev

conSys/part_2/control_channels.py

@@ -34,7 +34,7 @@ def list_control_streams_of_system(server_addr: HttpUrl, system_id: str, api_roo
                    .with_api_root(api_root)
                    .for_resource_type(APITerms.SYSTEMS.value)
                    .with_resource_id(system_id)
-                   .for_resource_type(APITerms.CONTROL_STREAMS.value)
+                   .for_sub_resource_type(APITerms.CONTROL_STREAMS.value)
                    .build_url_from_base()
                    .with_headers(headers)
                    .with_request_method('GET')
@@ -83,7 +83,7 @@ def retrieve_control_stream_description_by_id(server_addr: HttpUrl, control_stre
 
 def update_control_stream_description_by_id(server_addr: HttpUrl, control_stream_id: str,
                                             request_body: Union[str, dict],
-                                            api_root: str = APITerms.API.value):
+                                            api_root: str = APITerms.API.value, headers: dict = None):
     """
     Updates a control stream by its id
     :return:
@@ -95,9 +95,10 @@ def update_control_stream_description_by_id(server_addr: HttpUrl, control_stream
                    .with_resource_id(control_stream_id)
                    .with_request_body(request_body)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('PUT')
                    .build())
-    resp = requests.put(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+    return api_request.make_request()
 
 
 def delete_control_stream_by_id(server_addr: HttpUrl, control_stream_id: str, api_root: str = APITerms.API.value,
@@ -130,7 +131,7 @@ def retrieve_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id
                    .with_api_root(api_root)
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .with_resource_id(control_stream_id)
-                   .for_resource_type(APITerms.SCHEMA.value)
+                   .for_sub_resource_type(APITerms.SCHEMA.value)
                    .build_url_from_base()
                    .with_headers(headers)
                    .with_request_method('GET')
@@ -139,7 +140,7 @@ def retrieve_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id
     return api_request.make_request()
 
 
-def update_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id: str, request_body: dict,
+def update_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id: str, request_body: Union[str, dict],
                                        api_root: str = APITerms.API.value, headers: dict = None):
     """
     Updates a control stream schema by its id
@@ -150,7 +151,7 @@ def update_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id:
                    .with_api_root(api_root)
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .with_resource_id(control_stream_id)
-                   .for_resource_type(APITerms.SCHEMA.value)
+                   # .for_sub_resource_type(APITerms.SCHEMA.value)
                    .with_request_body(request_body)
                    .build_url_from_base()
                    .with_headers(headers)

tests/test_script_full_suite.py

@@ -315,6 +315,46 @@ def test_list_control_streams():
     print(control_streams)
 
 
+def test_list_control_streams_of_system():
+    systems_list = Systems.list_all_systems(server_url)
+    system_id = systems_list["items"][0]["id"]
+    control_streams = ControlChannels.list_control_streams_of_system(server_url, system_id)
+    print(control_streams)
+
+
+def test_retrieve_control_stream_by_id():
+    control_streams = ControlChannels.list_all_control_streams(server_url).json()
+    control_stream_desc = ControlChannels.retrieve_control_stream_description_by_id(server_url,
+                                                                                    control_streams["items"][0]["id"])
+    control_stream_schema = ControlChannels.retrieve_control_stream_schema_by_id(server_url,
+                                                                                 control_streams["items"][0]["id"])
+    print(control_stream_desc.json())
+    print(control_stream_schema.json())
+
+
+def test_update_control_stream_by_id():
+    control_streams = ControlChannels.list_all_control_streams(server_url).json()
+
+    time_schema = TimeSchema(label="Test Control Channel Time (Updated)", definition="http://test.com/Time",
+                             uom=URI(href="http://test.com/TimeUOM"))
+    count_schema = CountSchema(label="Test Control Channel Count (Updated)", definition="http://test.com/Count")
+    bool_schema = BooleanSchema(label="Test Control Channel Boolean (Updated)", definition="http://test.com/Boolean")
+
+    control_schema = JSONControlChannelSchema(command_format=ObservationFormat.SWE_JSON.value,
+                                              params_schema=DataRecordSchema(
+                                                  label="Test Control Channel Record (Updated)",
+                                                  definition="http://test.com/Record",
+                                                  fields=[bool_schema]))
+    request_body = ControlStreamJSONSchema(name="Test Control Channel (Updated)", input_name="TestControlInput1",
+                                           schema=control_schema)
+    print(f'Request Body for Control Stream: {request_body.model_dump_json(exclude_none=True, by_alias=True)}')
+    resp = ControlChannels.update_control_stream_schema_by_id(server_url, control_streams["items"][0]["id"],
+                                                              request_body.model_dump_json(exclude_none=True,
+                                                                                           by_alias=True),
+                                                              headers=json_headers)
+    print(resp)
+
+
 """
 Teardown Section
 """