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


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 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()