<commit_before>from json import JSONDecodeError

from app.helpers.schema_helper import SchemaHelper
from app.schema_loader import schema_loader
import json
import jsonschema
import logging
import unittest
import os

from app import settings

from jsonschema import validate

logger = logging.getLogger(__name__)


class TestSchemaValidation(unittest.TestCase):

    def test_schema(self):

        errors = []

        files = schema_loader.available_schemas()

        schema_file = open(os.path.join(settings.EQ_SCHEMA_DIRECTORY, "schema/schema-v1.json"), encoding="utf8")
        schema = json.load(schema_file)

        for file in files:

            try:
                json_file = open(os.path.join(settings.EQ_SCHEMA_DIRECTORY, file), encoding="utf8")
                json_to_validate = json.load(json_file)

                errors.extend(self.validate_json_against_schema(file, json_to_validate, schema))

                errors.extend(self.validate_schema_contains_valid_routing_rules(file, json_to_validate))

                errors.extend(self.validate_routing_rules_has_default_if_not_all_answers_routed(file, json_to_validate))
            except JSONDecodeError as e:
                errors.append("JSON Decode Error! Could not decode [{}]. Error [{}]".format(file, e))

        if errors:
            for error in errors:
                logger.error(error)

            self.fail("{} Schema Validation Errors.".format(len(errors)))


    @staticmethod
    def validate_json_against_schema(file, json_to_validate, schema):
        try:
            validate(json_to_validate, schema)<commit_msg>Add test that checks for duplicate keys in schemas.<commit_after>from json import JSONDecodeError

from app.helpers.schema_helper import SchemaHelper
from app.schema_loader import schema_loader
import json
import jsonschema
import logging
import unittest
import os

from app import settings

from jsonschema import validate

logger = logging.getLogger(__name__)


class TestSchemaValidation(unittest.TestCase):

    def test_schema(self):

        errors = []

        files = schema_loader.available_schemas()

        schema_file = open(os.path.join(settings.EQ_SCHEMA_DIRECTORY, "schema/schema-v1.json"), encoding="utf8")
        schema = json.load(schema_file)

        for file in files:

            try:
                json_file = open(os.path.join(settings.EQ_SCHEMA_DIRECTORY, file), encoding="utf8")
                json_to_validate = json.load(json_file)

                errors.extend(self.validate_json_against_schema(file, json_to_validate, schema))

                errors.extend(self.validate_schema_contains_valid_routing_rules(file, json_to_validate))

                errors.extend(self.validate_routing_rules_has_default_if_not_all_answers_routed(file, json_to_validate))
            except JSONDecodeError as e:
                errors.append("JSON Decode Error! Could not decode [{}]. Error [{}]".format(file, e))

        if errors:
            for error in errors:
                logger.error(error)

            self.fail("{} Schema Validation Errors.".format(len(errors)))

    def test_no_duplicate_ids_in_schema(self):
        schema_files = schema_loader.available_schemas()
        ignored_keys = ['routing_rules', 'skip_condition']

        print('Checking schema files for duplicate keys.')
        for schema_file in schema_files:
            unique_id = []
            with open(os.path.join(settings.EQ_SCHEMA_DIRECTORY, schema_file), encoding="utf8") as file:
                schema_json = json.load(file)
                print('Checking %s...' % schema_file)
                for id_value in self._parse_id_values(schema_json, ignored_keys):
                    if id_value in unique_id:
                        self.fail('Duplicate Id found. schema: %s, id: %s' % (schema_file, id_value))
                    else:
                        unique_id.append(id_value)

    def _parse_id_values(self, schema_json, ignored_keys):
        for k, v in schema_json.items():
            if k == 'id':
                yield schema_json[k]
            elif k in ignored_keys:
                continue
            elif isinstance(v, dict):
                yield from self._parse_id_values(v, ignored_keys)
            elif isinstance(v, list):
                for schema_item in v:
                    if isinstance(schema_item, dict):
                        yield from self._parse_id_values(schema_item, ignored_keys)


    @staticmethod
    def validate_json_against_schema(file, json_to_validate, schema):
        try:
            validate(json_to_validate, schema)
{'fraction_non_alphanumeric': 0.08356848274365096, 'fraction_numerical': 0.001519426958975472, 'mean_word_length': 2.6169544740973314, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2353581', 'n_tokens_mistral': 1319, 'n_tokens_neox': 1256, 'n_words': 286}