validate.py

##############################################################################
#
#    Converter Odoo module
#    Copyright (C) 2020 XCG Consulting <https://xcg-consulting.fr>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################

import json
import math
import os
from typing import NoReturn

import odoo.addons

import jsonschema

VALIDATION_SKIP = "skip"
VALIDATION_SOFT = "soft"
VALIDATION_STRICT = "strict"


class NotInitialized(Exception):
    pass


class Validator:
    def __init__(
        self,
        repository_module_name: str,
        repository: str,
        default_url_pattern: str,
    ):
        self.repository_module_name = repository_module_name
        self.repository = repository
        # exemple "https://annonces-legales.fr/xbus/schemas/v1/{}.schema.json"
        self.default_url_pattern = default_url_pattern
        self.schemastore = {}
        self.validators = {}
        self.search_path = "not initialized, please call initialize()"
        self.initialized = False
        self.encoding = "UTF-8"

    def initialize(self) -> NoReturn:
        repo_module_basepath = os.path.dirname(
            getattr(odoo.addons, self.repository_module_name).__file__
        )
        schema_search_path = os.path.abspath(
            os.path.join(repo_module_basepath, self.repository)
        )
        self.search_path = schema_search_path

        for root, _dirs, files in os.walk(self.search_path):
            for fname in files:
                fpath = os.path.join(root, fname)
                if fpath.endswith((".json",)):
                    with open(fpath, "r", encoding=self.encoding) as schema_fd:
                        schema = json.load(schema_fd)
                        if "$id" in schema:
                            self.schemastore[schema["$id"]] = schema
        self.initialized = True

    def validate(self, schema_id, doc) -> NoReturn:
        if not self.initialized:
            raise NotInitialized("please call the initialize() method")

        uri = "file://{}/{}.schema.json".format(self.search_path, schema_id)
        validator = self.validators.get(uri)
        if validator is None:
            schema = self.schemastore[
                self.default_url_pattern.format(schema_id)
            ]

            resolver = jsonschema.RefResolver(uri, schema, self.schemastore)

            validator = jsonschema.Draft7Validator(schema, resolver=resolver)
            self.validators[uri] = validator

        validator.validate(doc)


def multiple_of(validator, db, instance, schema):
    """Fix rounding which fails when comparing e.g. 66.6 vs 0.01"""

    if not validator.is_type(instance, "number"):
        return

    if isinstance(db, float):
        quotient = instance / db
        failed = not math.isclose(round(quotient), quotient)
    else:
        failed = instance % db

    if failed:
        yield jsonschema.ValidationError(
            "%r is not a multiple of %r" % (instance, db)
        )


# Fix rounding which fails when comparing e.g. 66.6 vs 0.01
jsonschema.Draft7Validator.VALIDATORS["multipleOf"] = multiple_of