test_schemas.py

from openapi_spec_validator import openapi_v30_spec_validator
from openapi_spec_validator.readers import read_from_filename
from openapi_schema_validator import validate
from jsonschema import FormatChecker
from tests.fixtures.schemas import (
    acknowledgement_schema,
    message_header,
    observation_schema,
    planning_configuration_schema,
    platform_status_schema,
    mission_plan_schema,
)

import unittest
import json
import os
from jsonschema.validators import RefResolver
from openapi_schema_validator import validate

# class ValidateResponse:
#     def __init__(self, valid, error):
#         self.valid = valid
#         self.error = error

SCHEMA_DIR = "projects/soar/swagger.json"
MOCK_DATA_DIR = "examples/"

class SchemaError(Exception):
    """
    """
    def __init__(self, exception_type, message):
        self.type = exception_type
        self.message = message
        super().__init__(self.message)


class TestSpecs(unittest.TestCase):
    def test_swagger_specs(self):
        """
        Test specs (swagger.json generated from generate_schema_config.py)
        """
        schema, spec_url = read_from_filename("tests/fixtures/swagger.json")
        self.assertIsNone(openapi_v30_spec_validator.validate(schema))

class TestAllMessageExamples(unittest.TestCase):
    def test_schema_specs(self):
        """
        Test specs (swagger.json generated from generate_schema_config.py)
        """
        schema_ref = open(SCHEMA_DIR)
        schema = json.load(schema_ref)
        schema_ref.close()


        partner_dir_list = os.listdir(MOCK_DATA_DIR)
        

        for partner in partner_dir_list:
            message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner))
            for message_type in message_list:
                print("Testing message %s by %s now..." % (message_type, partner))

                f = open(os.path.join(MOCK_DATA_DIR, partner, message_type))
                mock_message = json.load(f)
                ref_resolver = RefResolver.from_schema(schema)


                self.assertIsNone(
                    validate(
                        mock_message,
                        schema["components"]["schemas"]["MESSAGE"],
                        resolver=ref_resolver,
                    )
                )
                f.close()
        
        schema_ref.close()
        

class TestSchema(unittest.TestCase):
    def test_sample_hydrosurv_messages(self):
        """
        Test schema - Hydrosurv Messages
        """
        message_types = ["acknowledgement", "mission_plan", "platform_status"]
        for item in message_types:
            f = open("examples/hydrosurv_adapter/%s.json" % item)
            mock_data = json.load(f)
            header_data = mock_data["header"]
            payload_data = mock_data["payload"]
            self.assertIsNone(
                validate(
                    header_data,
                    message_header,
                    format_checker=FormatChecker(),
                )
            )
            if item == "acknowledgement":
                self.assertIsNone(
                    validate(
                        payload_data,
                        acknowledgement_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "mission_plan":
                self.assertIsNone(
                    validate(
                        payload_data,
                        mission_plan_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "platform_status":
                self.assertIsNone(
                    validate(
                        payload_data,
                        platform_status_schema,
                        format_checker=FormatChecker(),
                    )
                )

            f.close()

    def test_sample_gui_messages(self):
        """
        Test schema - GUI Messages
        """
        message_types = ["planning_configuration"]
        for item in message_types:
            f = open("examples/gui_adapter/%s.json" % item)
            mock_data = json.load(f)
            header_data = mock_data["header"]
            payload_data = mock_data["payload"]
            self.assertIsNone(
                validate(
                    header_data,
                    message_header,
                    format_checker=FormatChecker(),
                )
            )
            if item == "planning_configuration":
                self.assertIsNone(
                    validate(
                        payload_data,
                        planning_configuration_schema,
                        format_checker=FormatChecker(),
                    )
                )

            f.close()

    def test_sample_ecosub_messages(self):
        """
        Test schema - Ecosub Messages
        """
        message_types = [
            "observation",
            "mission_plan",
            "platform_status",
            "platform_status-from_usbl_example",
        ]
        for item in message_types:
            f = open("examples/ecosub_adapter/%s.json" % item)
            mock_data = json.load(f)
            header_data = mock_data["header"]
            payload_data = mock_data["payload"]
            self.assertIsNone(
                validate(
                    header_data,
                    message_header,
                    format_checker=FormatChecker(),
                )
            )
            if item == "platform_status-from_usbl_example.json":
                self.assertIsNone(
                    validate(
                        payload_data,
                        platform_status_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "mission_plan":
                self.assertIsNone(
                    validate(
                        payload_data,
                        mission_plan_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "platform_status":
                self.assertIsNone(
                    validate(
                        payload_data,
                        platform_status_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "observation":
                self.assertIsNone(
                    validate(
                        payload_data,
                        observation_schema,
                        format_checker=FormatChecker(),
                    )
                )

            f.close()

    def test_sample_autonomy_engine_messages(self):
        """
        Test schema - Autonomy Engine Messages
        """
        message_types = [
            "planning_configuration",
            "mission_plan_AH1",
            "mission_plan_ECOSUB",
            "mission_plan_HYDROSURV",
            "platform_status",
            "platform_status-from_usbl_example",
            "acknowledgement",
        ]
        for item in message_types:
            f = open("examples/autonomy_engine_adapter/%s.json" % item)
            mock_data = json.load(f)
            header_data = mock_data["header"]
            payload_data = mock_data["payload"]
            self.assertIsNone(
                validate(
                    header_data,
                    message_header,
                    format_checker=FormatChecker(),
                )
            )
            if item == "platform_status-from_usbl_example.json":
                self.assertIsNone(
                    validate(
                        payload_data,
                        platform_status_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "mission_plan_AH1":
                self.assertIsNone(
                    validate(
                        payload_data,
                        mission_plan_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "mission_plan_ECOSUB":
                self.assertIsNone(
                    validate(
                        payload_data,
                        mission_plan_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "mission_plan_HYDROSURV":
                self.assertIsNone(
                    validate(
                        payload_data,
                        mission_plan_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "platform_status":
                self.assertIsNone(
                    validate(
                        payload_data,
                        platform_status_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "observation":
                self.assertIsNone(
                    validate(
                        payload_data,
                        observation_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "planning_configuration":
                self.assertIsNone(
                    validate(
                        payload_data,
                        planning_configuration_schema,
                        format_checker=FormatChecker(),
                    )
                )
            elif item == "acknowledgement":
                self.assertIsNone(
                    validate(
                        payload_data,
                        acknowledgement_schema,
                        format_checker=FormatChecker(),
                    )
                )

            f.close()


if __name__ == "__main__":
    unittest.main()