test_schemas.py

from openapi_spec_validator import OpenAPIV30SpecValidator
from openapi_spec_validator.readers import read_from_filename
from openapi_schema_validator import validate
from jsonschema.validators import RefResolver
import unittest
import json
import os
from generate_schema_config import write_schema, get_swagger_config


MOCK_DATA_DIR = "examples/"


class SchemaTestCase(unittest.TestCase):
    def load_schema(self, file_path=None):
        if file_path:
            schema_path = file_path
        else:
            schema_path = os.getenv("SCHEMA_PATH")
        schema, _ = read_from_filename(schema_path)
        return schema

    @classmethod
    def setUpClass(cls):
        test_schema_path = "tests/test_swagger.json"
        os.environ["SCHEMA_PATH"] = test_schema_path
        swagger_config = get_swagger_config()
        write_schema(swagger_config, test_schema_path)


class SchemaError(Exception):
    """
    Test config specs of swagger.json for projects
    """

    def __init__(self, exception_type, message):
        self.type = exception_type
        self.message = message
        super().__init__(message)


class TestCompiledSoarSpec(SchemaTestCase):
    """
    Test the saved version of the compiled schema has been updated
    """

    def test_compiled_spec_matches_definitions(self):
        print("TEST: definitions match project/soar/swagger.json")
        derived_schema = self.load_schema()
        saved_schema = self.load_schema("project/soar/swagger.json")
        self.assertEqual(saved_schema, derived_schema)


class TestSpecs(SchemaTestCase):
    def test_swagger_specs(self):
        """
        Test specs (swagger.json generated from generate_schema_config.py)
        """
        print("TEST: compiled schema matches openapi v3 spec")
        schema = self.load_schema()
        errors = list(OpenAPIV30SpecValidator(schema).iter_errors())
        self.assertEqual(len(errors), 0)


class TestAllMessageExamples(SchemaTestCase):
    def test_schema_specs(self):
        """
        Test specs (swagger.json generated from generate_schema_config.py)
        """
        schema = self.load_schema()

        partner_dir_list = os.listdir(MOCK_DATA_DIR)

        for partner in partner_dir_list:
            if not partner.endswith(".py"):
                print("TEST: Partner is %s" % partner.upper())
                message_list = os.listdir(os.path.join(MOCK_DATA_DIR, partner))
                for message_type in message_list:
                    if message_type.endswith(".json"):
                        print("Testing %s now..." % message_type)

                        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()
                        print("Done.")


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