How to use the pykwalify.errors.CoreError function in pykwalify

To help you get started, we’ve selected a few pykwalify examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Grokzen / pykwalify / tests / test_core.py View on Github external
def test_create_empty_core_object(self, tmpdir):
        """
        If createing a core object without any source or schema file an exception should be raised.
        """
        with pytest.raises(CoreError) as ex:
            Core()
        assert "No source file/data was loaded" in str(ex.value)

        # To trigger schema exception we must pass in a source file
        source_f = tmpdir.join("bar.json")
        source_f.write("3.14159")

        with pytest.raises(CoreError) as ex:
            Core(source_file=str(source_f))
        assert "No schema file/data was loaded" in str(ex.value)
github Grokzen / pykwalify / tests / test_core.py View on Github external
def test_load_unsupported_format(self, tmpdir):
        """
        Try to load some fileending that is not supported.
        Currently XML is not supported.
        """
        source_f = tmpdir.join("foo.xml")
        source_f.write("bar")

        schema_f = tmpdir.join("bar.xml")
        schema_f.write("bar")

        with pytest.raises(CoreError) as ex:
            Core(source_file=str(source_f))
        assert "Unable to load source_file. Unknown file format of specified file path" in str(ex.value)

        with pytest.raises(CoreError) as ex:
            Core(schema_files=[str(schema_f)])
        assert "Unknown file format. Supported file endings is" in str(ex.value)
github Grokzen / pykwalify / pykwalify / core.py View on Github external
res = re.match(rule.pattern, value, re.UNICODE)
            except TypeError:
                res = None

            if res is None:  # Not matching
                self.errors.append(SchemaError.SchemaErrorEntry(
                    msg=u"Value '{value}' does not match pattern '{pattern}'. Path: '{path}'",
                    path=path,
                    value=nativestr(str(value)),
                    pattern=rule._pattern))
            else:
                log.debug("Pattern matched...")

        if rule.range is not None:
            if not is_scalar(value):
                raise CoreError(u"value is not a valid scalar")

            r = rule.range

            try:
                v = len(value)
                value = v
            except Exception:
                pass

            self._validate_range(
                r.get("max"),
                r.get("min"),
                r.get("max-ex"),
                r.get("min-ex"),
                value,
                path,
github Grokzen / pykwalify / pykwalify / core.py View on Github external
schema_data = dict(schema_data, **data)

            self.schema = schema_data

        # Nothing was loaded so try the source_data variable
        if self.source is None:
            log.debug(u"No source file loaded, trying source data variable")
            self.source = source_data
        if self.schema is None:
            log.debug(u"No schema file loaded, trying schema data variable")
            self.schema = schema_data

        # Test if anything was loaded
        if self.source is None:
            raise CoreError(u"No source file/data was loaded")
        if self.schema is None:
            raise CoreError(u"No schema file/data was loaded")

        # Merge any extensions defined in the schema with the provided list of extensions from the cli
        for f in self.schema.get('extensions', []):
            self.extensions.append(f)

        if not isinstance(self.extensions, list) and all(isinstance(e, str) for e in self.extensions):
            raise CoreError(u"Specified extensions must be a list of file paths")

        self._load_extensions()

        if self.strict_rule_validation:
            log.info("Using strict rule keywords validation...")
github Grokzen / pykwalify / pykwalify / core.py View on Github external
def _validate_assert(self, rule, value, path):
        if not self.allow_assertions:
            raise CoreError('To allow usage of keyword "assert" you must use cli flag "--allow-assertions" or set the keyword "allow_assert" in Core class')

        # Small hack to make strings work as a value.
        if isinstance(value, str):
            assert_value_str = '"{0}"'.format(value)
        else:
            assert_value_str = '{0}'.format(value)

        assertion_string = "val = {0}; assert {1}".format(assert_value_str, rule.assertion)
        try:
            exec(assertion_string, {}, {})
        except AssertionError:
            self.errors.append(SchemaError.SchemaErrorEntry(
                msg=u"Value: '{0}' assertion expression failed ({1})".format(value, rule.assertion),
                path=path,
                value=value,
            ))
github Grokzen / pykwalify / pykwalify / core.py View on Github external
def _validate_range(self, max_, min_, max_ex, min_ex, value, path, prefix):
        """
        Validate that value is within range values.
        """
        if not isinstance(value, int) and not isinstance(value, float):
            raise CoreError("Value must be a integer type")

        log.debug(
            u"Validate range : %s : %s : %s : %s : %s : %s",
            max_,
            min_,
            max_ex,
            min_ex,
            value,
            path,
        )

        if max_ is not None and max_ < value:
                self.errors.append(SchemaError.SchemaErrorEntry(
                    msg=u"Type '{prefix}' has size of '{value}', greater than max limit '{max_}'. Path: '{path}'",
                    path=path,
                    value=nativestr(value) if tt['str'](value) else value,
github Grokzen / pykwalify / pykwalify / core.py View on Github external
method = getattr(extension, func, None)
            if method:
                found_method = True

                # No exception will should be caught. If one is raised it should bubble up all the way.
                ret = method(value, rule, path)
                if ret is not True and ret is not None:
                    msg = '%s. Path: {path}' % unicode(ret)
                    self.errors.append(SchemaError.SchemaErrorEntry(
                                    msg=msg,
                                    path=path,
                                    value=None))

                # If False or None or some other object that is interpreted as False
                if not ret:
                    raise CoreError(u"Error when running extension function : {0}".format(func))

                # Only run the first matched function. Sinc loading order is determined
                # it should be easy to determine which file is used before others
                break

        if not found_method:
            raise CoreError(u"Did not find method '{0}' in any loaded extension file".format(func))
github Grokzen / pykwalify / pykwalify / core.py View on Github external
if rule.length is not None:
            self._validate_length(
                rule.length,
                value,
                path,
                'scalar',
            )

        # Validate timestamp
        if rule.type == "timestamp":
            self._validate_scalar_timestamp(value, path)

        if rule.type == "date":
            if not is_scalar(value):
                raise CoreError(u'value is not a valid scalar')
            date_format = rule.format
            self._validate_scalar_date(value, date_format, path)
github Grokzen / pykwalify / pykwalify / core.py View on Github external
yml.constructor.add_constructor('tag:yaml.org,2002:python/none', Constructor.construct_yaml_null)
        yml.constructor.add_constructor('tag:yaml.org,2002:python/str', Constructor.construct_python_str)
        yml.constructor.add_constructor('tag:yaml.org,2002:python/tuple', Constructor.construct_python_tuple)
        yml.constructor.add_constructor('tag:yaml.org,2002:python/unicode', Constructor.construct_python_unicode)

        if data_file_obj:
            try:
                self.source = yml.load(data_file_obj.read())
            except Exception as e:
                raise CoreError("Unable to load data_file_obj input")

        if schema_file_obj:
            try:
                self.schema = yml.load(schema_file_obj.read())
            except Exception as e:
                raise CoreError("Unable to load schema_file_obj")

        if source_file is not None:
            if not os.path.exists(source_file):
                raise CoreError(u"Provided source_file do not exists on disk: {0}".format(source_file))

            with open(source_file, "r", encoding=file_encoding) as stream:
                if source_file.endswith(".json"):
                    self.source = json.load(stream)
                elif source_file.endswith(".yaml") or source_file.endswith('.yml'):
                    self.source = yml.load(stream)
                else:
                    raise CoreError(u"Unable to load source_file. Unknown file format of specified file path: {0}".format(source_file))

        if not isinstance(schema_files, list):
            raise CoreError(u"schema_files must be of list type")