Moved _get_datatyoe_schema into MS05Utils

nmostesting/MS05Utils.py

@@ -40,6 +40,7 @@ def __init__(self, apis, protocol_api_key):
     def reset(self):
         self.device_model = None
         self.class_manager = None
+        self.datatype_schemas = None
         self.load_reference_resources()
 
     # Overridden functions specialized for IS-12 and IS-14
@@ -162,7 +163,7 @@ def load_reference_resources(self):
         # Load MS-05 datatype descriptors
         self.reference_datatype_descriptors = self._load_model_descriptors(datatype_paths)
 
-        # Generate MS-05 datatype schemas from MS-05 datatype descriptors
+        # Generate reference MS-05 datatype schemas from MS-05 datatype descriptors
         self.reference_datatype_schemas = self.generate_json_schemas(
             datatype_descriptors=self.reference_datatype_descriptors,
             schema_path=os.path.join(self.apis[self.protocol_api_key]["spec_path"], 'APIs/schemas/'))
@@ -186,7 +187,7 @@ def generate_json_schemas(self, datatype_descriptors, schema_path):
             os.makedirs(base_schema_path)
 
         for name, descriptor in datatype_descriptors.items():
-            json_schema = self.descriptor_to_schema(descriptor)
+            json_schema = self._descriptor_to_schema(descriptor)
             with open(os.path.join(base_schema_path, name + '.json'), 'w') as output_file:
                 json.dump(json_schema, output_file, indent=4)
                 datatype_schema_names.append(name)
@@ -198,7 +199,7 @@ def generate_json_schemas(self, datatype_descriptors, schema_path):
 
         return datatype_schemas
 
-    def descriptor_to_schema(self, descriptor):
+    def _descriptor_to_schema(self, descriptor):
         variant_type = ['number', 'string', 'boolean', 'object', 'array', 'null']
 
         json_schema = {}
@@ -264,7 +265,7 @@ def descriptor_to_schema(self, descriptor):
 
         return json_schema
 
-    def generate_device_model_datatype_schemas(self, test):
+    def _generate_device_model_datatype_schemas(self, test):
         # Generate datatype schemas based on the datatype decriptors
         # queried from the Node under test's Device Model.
         # This will include any Non-standard data types
@@ -278,6 +279,13 @@ def generate_device_model_datatype_schemas(self, test):
         except Exception as e:
             raise NMOSTestException(test.FAIL(f"Unable to create Device Model schemas: {e.message}"))
 
+    def get_datatype_schema(self, test, type_name):
+        """Get generated JSON schema for datatype specified, based on descriptor queried from the Node under Test"""
+        if not self.datatype_schemas:
+            self.datatype_schemas = self._generate_device_model_datatype_schemas(test)
+
+        return self.datatype_schemas.get(type_name)
+
     def validate_reference_datatype_schema(self, test, payload, datatype_name, context=""):
         """Validate payload against reference datatype schema"""
         context += f"{datatype_name}: "
@@ -424,18 +432,7 @@ def _nc_object_factory(self, test, class_id, oid, role, _role_path=None):
         except NMOSTestException as e:
             raise NMOSTestException(test.FAIL("Error in Device Model " + role + ": " + str(e.args[0].detail)))
 
-    def get_class_manager(self, test):
-        """Get the Class Manager queried from the Node under test's Device Model"""
-        if not self.class_manager:
-            self.class_manager = self._get_manager(test, StandardClassIds.NCCLASSMANAGER.value)
-
-        return self.class_manager
-
-    def get_device_manager(self, test):
-        """Get the Device Manager queried from the Node under test's Device Model"""
-        return self._get_manager(test, StandardClassIds.NCDEVICEMANAGER.value)
-
-    def _get_manager(self, test, class_id):
+    def _get_object_by_class_id(self, test, class_id):
         device_model = self.query_device_model(test)
         members = device_model.find_members_by_class_id(class_id, include_derived=True)
 
@@ -450,6 +447,17 @@ def _get_manager(self, test, class_id):
 
         return members[0]
 
+    def get_class_manager(self, test):
+        """Get the Class Manager queried from the Node under test's Device Model"""
+        if not self.class_manager:
+            self.class_manager = self._get_object_by_class_id(test, StandardClassIds.NCCLASSMANAGER.value)
+
+        return self.class_manager
+
+    def get_device_manager(self, test):
+        """Get the Device Manager queried from the Node under test's Device Model"""
+        return self._get_object_by_class_id(test, StandardClassIds.NCDEVICEMANAGER.value)
+
     def primitive_to_python_type(self, type):
         """Convert MS-05 primitive type to corresponding Python type"""
 
@@ -493,6 +501,7 @@ def resolve_datatype(self, test, datatype):
         return datatype
 
     def create_role_path(self, base_role_path, role):
+        """Appends role to base_role_path"""
         role_path = base_role_path.copy()
         role_path.append(role)
         return role_path

nmostesting/suites/MS0501Test.py

@@ -151,13 +151,6 @@ def test_ms05_01(self, test):
 
         return test.PASS()
 
-    def get_datatype_schema(self, test, type_name):
-        """Get generated JSON schema for datatype specified"""
-        if not self.datatype_schemas:
-            self.datatype_schemas = self.ms05_utils.generate_device_model_datatype_schemas(test)
-
-        return self.datatype_schemas.get(type_name)
-
     def get_property(self, test, oid, property_id, role_path, context):
         """Get property from object. Sets self.device_model_metadata on error"""
         try:
@@ -202,7 +195,8 @@ def _validate_property_type(self, test, value, data_type, is_nullable, context="
             if not isinstance(value, self.ms05_utils.primitive_to_python_type(data_type)):
                 raise NMOSTestException(test.FAIL(context + str(value) + " is not of type " + str(data_type)))
         else:
-            self.ms05_utils.validate_schema(test, value, self.get_datatype_schema(test, data_type), context + data_type)
+            self.ms05_utils.validate_schema(test, value, self.ms05_utils.get_datatype_schema(test, data_type),
+                                            context + data_type)
 
         return
 
@@ -353,9 +347,9 @@ def check_touchpoints(self, test, oid, role_path, context):
             self.touchpoints_metadata["checked"] = True
             try:
                 for touchpoint in touchpoints:
-                    schema = self.get_datatype_schema(test, "NcTouchpointNmos") \
+                    schema = self.ms05_utils.get_datatype_schema(test, "NcTouchpointNmos") \
                         if touchpoint["contextNamespace"] == "x-nmos" \
-                        else self.get_datatype_schema(test, "NcTouchpointNmosChannelMapping")
+                        else self.ms05_utils.get_datatype_schema(test, "NcTouchpointNmosChannelMapping")
                     self.ms05_utils.validate_schema(
                         test,
                         touchpoint,
@@ -381,7 +375,7 @@ def check_block(self, test, block, class_descriptors, context=""):
             self.ms05_utils.validate_schema(
                 test,
                 descriptor,
-                self.get_datatype_schema(test, "NcBlockMemberDescriptor"),
+                self.ms05_utils.get_datatype_schema(test, "NcBlockMemberDescriptor"),
                 context=context + "NcBlockMemberDescriptor: " + str(descriptor['role']))
 
             self.check_unique_roles(descriptor['role'], role_cache)
@@ -398,7 +392,7 @@ def check_block(self, test, block, class_descriptors, context=""):
                 self.ms05_utils.validate_schema(
                     test,
                     class_descriptors[class_identifier],
-                    self.get_datatype_schema(test, "NcClassDescriptor"),
+                    self.ms05_utils.get_datatype_schema(test, "NcClassDescriptor"),
                     context=context + "NcClassDescriptor for class " + str(descriptor['classId']))
                 self.check_object_properties(test,
                                              class_descriptors[class_identifier],
@@ -973,7 +967,7 @@ def do_find_member_by_role_test(self, test, block, context=""):
 
     def check_constraint(self, test, constraint, type_name, is_sequence, test_metadata, context):
         if constraint.get("defaultValue"):
-            datatype_schema = self.get_datatype_schema(test, type_name)
+            datatype_schema = self.ms05_utils.get_datatype_schema(test, type_name)
             if isinstance(constraint.get("defaultValue"), list) is not is_sequence:
                 test_metadata["error"] = True
                 test_metadata["error_msg"] = context + (" a default value sequence was expected"