x
/
packetary
Code Issues Proposed changes

Implemented input data validation 

Change-Id: I8407cf4cbb69e60b7def891625522f3ca3c822fe
Implements:  blueprint unify-input-data
This commit is contained in:
Bulat Gaifullin 2016-01-26 11:12:40 +03:00 committed by Uladzimir_Niakhai
parent d99d1228ef
commit 9e107cde9a
18 changed files with 683 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
packetary/api.py
View File

@ -19,6 +19,7 @@
from collections import defaultdict
import logging
import jsonschema
import six
from packetary.controllers import RepositoryController
@ -28,7 +29,8 @@ from packetary.objects import PackageRelation
from packetary.objects import PackagesForest
from packetary.objects import PackagesTree
from packetary.objects.statistics import CopyStatistics
from packetary.schemas import PACKAGE_FILES_SCHEMA
from packetary.schemas import PACKAGES_SCHEMA
logger = logging.getLogger(__package__)
@ -122,6 +124,7 @@ class RepositoryApi(object):
:param package_files: The list of URLs of packages
"""
self._validate_repo_data(repo_data)
self._validate_package_files(package_files)
return self.controller.create_repository(repo_data, package_files)
def get_packages(self, repos_data, requirements_data=None,
@ -215,7 +218,6 @@ class RepositoryApi(object):
self._validate_requirements_data(requirements_data)
result = []
for r in requirements_data:
self._validate_requirements_data(r)
versions = r.get('versions', None)
if versions is None:
result.append(PackageRelation.from_args((r['name'],)))
@ -227,9 +229,34 @@ class RepositoryApi(object):
return result
def _validate_repo_data(self, repo_data):
# TODO(bgaifullin) implement me
pass
schema = self.controller.get_repository_data_schema()
self._validate_data(repo_data, schema)
def _validate_requirements_data(self, requirements_data):
# TODO(bgaifullin) implement me
pass
self._validate_data(requirements_data, PACKAGES_SCHEMA)
def _validate_package_files(self, package_files):
self._validate_data(package_files, PACKAGE_FILES_SCHEMA)
def _validate_data(self, data, schema):
"""Validate the input data using jsonschema validation.
:param data: a data to validate represented as a dict
:param schema: a schema to validate represented as a dict;
must be in JSON Schema Draft 4 format.
"""
try:
jsonschema.validate(data, schema)
except jsonschema.ValidationError as e:
self._raise_validation_error("data", e.message, e.absolute_path)
except jsonschema.SchemaError as e:
self._raise_validation_error(
"schema", e.message, e.absolute_schema_path
)
@staticmethod
def _raise_validation_error(what, details, path):
message = "Invalid {0}: {1}.".format(what, details)
if path:
message = "\n".join((message, "Field: {0}".format(".".join(path))))
raise ValueError(message)

7
packetary/controllers/repository.py
View File

@ -137,6 +137,13 @@ class RepositoryController(object):
self.assign_packages(repo, packages)
return repo
def get_repository_data_schema(self):
"""Return jsonschema to validate data for required driver.
:return : Return a jsonschema represented as a dict
"""
return self.driver.repository_data_schema()
def _copy_packages(self, target, packages, observer):
with self.context.async_section() as section:
for package in packages:

4
packetary/drivers/base.py
View File

@ -111,3 +111,7 @@ class RepositoryDriverBase(object):
:return: the integer value that is relevant repository`s priority
less number means greater priority
"""
@abc.abstractmethod
def get_repository_data_schema(self):
"""Gets the json scheme for repository data validation."""

8
packetary/drivers/deb_driver.py
View File

@ -36,6 +36,7 @@ from packetary.objects import FileChecksum
from packetary.objects import Package
from packetary.objects import PackageRelation
from packetary.objects import Repository
from packetary.schemas import DEB_REPO_SCHEMA
_OPERATORS_MAPPING = {
@ -83,6 +84,9 @@ _checksum_collector = checksum_composite('md5', 'sha1', 'sha256')
class DebRepositoryDriver(RepositoryDriverBase):
def get_repository_data_schema(self):
return DEB_REPO_SCHEMA
def priority_sort(self, repo_data):
# DEB repository expects general values from 0 to 1000. 0
# to have lowest priority and 1000 -- the highest. Note that a
@ -94,7 +98,7 @@ class DebRepositoryDriver(RepositoryDriverBase):
return -priority
def get_repository(self, connection, repository_data, arch, consumer):
url = utils.normalize_repository_url(repository_data['url'])
url = utils.normalize_repository_url(repository_data['uri'])
suite = repository_data['suite']
components = repository_data.get('section')
path = repository_data.get('path')
@ -202,7 +206,7 @@ class DebRepositoryDriver(RepositoryDriverBase):
return new_repo
def create_repository(self, repository_data, arch):
url = utils.normalize_repository_url(repository_data['url'])
url = utils.normalize_repository_url(repository_data['uri'])
suite = repository_data['suite']
component = repository_data.get('section')
path = repository_data.get('path')

8
packetary/drivers/rpm_driver.py
View File

@ -36,6 +36,7 @@ from packetary.objects import PackageRelation
from packetary.objects import PackageVersion
from packetary.objects import Repository
from packetary.objects import VersionRange
from packetary.schemas import RPM_REPO_SCHEMA
urljoin = six.moves.urllib.parse.urljoin
@ -86,6 +87,9 @@ class CreaterepoCallBack(object):
class RpmRepositoryDriver(RepositoryDriverBase):
def get_repository_data_schema(self):
return RPM_REPO_SCHEMA
def priority_sort(self, repo_data):
# DEB repository expects general values from 0 to 1000. 0
# to have lowest priority and 1000 -- the highest. Note that a
@ -99,7 +103,7 @@ class RpmRepositoryDriver(RepositoryDriverBase):
def get_repository(self, connection, repository_data, arch, consumer):
consumer(Repository(
name=repository_data['name'],
url=utils.normalize_repository_url(repository_data["url"]),
url=utils.normalize_repository_url(repository_data["uri"]),
architecture=arch,
origin=""
))
@ -195,7 +199,7 @@ class RpmRepositoryDriver(RepositoryDriverBase):
def create_repository(self, repository_data, arch):
repository = Repository(
name=repository_data['name'],
url=utils.normalize_repository_url(repository_data["url"]),
url=utils.normalize_repository_url(repository_data["uri"]),
architecture=arch,
origin=repository_data.get('origin')
)

29
packetary/schemas/__init__.py Normal file
View File

@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
from packetary.schemas.deb_repo_schema import DEB_REPO_SCHEMA
from packetary.schemas.package_files_schema import PACKAGE_FILES_SCHEMA
from packetary.schemas.packages_schema import PACKAGES_SCHEMA
from packetary.schemas.rpm_repo_schema import RPM_REPO_SCHEMA
__all__ = [
"DEB_REPO_SCHEMA",
"PACKAGES_SCHEMA",
"RPM_REPO_SCHEMA",
"PACKAGE_FILES_SCHEMA"
]

56
packetary/schemas/deb_repo_schema.py Normal file
View File

@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
DEB_REPO_SCHEMA = {
"$schema": "http://json-schema.org/draft-04/schema#",
"type": "object",
"required": [
"name",
"uri",
"suite",
],
"properties": {
"name": {
"type": "string"
},
"uri": {
"type": "string"
},
"path": {
"type": "string"
},
"priority": {
"anyOf": [
{
"type": "number",
"minimum": 0,
},
{
"type": "null"
}
]
},
"suite": {
"type": "string"
},
"section": {
"type": "array",
"items": {"type": "string"}
}
}
}

26
packetary/schemas/package_files_schema.py Normal file
View File

@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
PACKAGE_FILES_SCHEMA = {
"$schema": "http://json-schema.org/draft-04/schema#",
"type": "array",
"items": {
"type": "string",
"pattern": "^(\/|file:\/\/|https?:\/\/).+$"
}
}

43
packetary/schemas/packages_schema.py Normal file
View File

@ -0,0 +1,43 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
PACKAGES_SCHEMA = {
"$schema": "http://json-schema.org/draft-04/schema#",
"type": "array",
"items": {
"type": "object",
"required": [
"name",
"versions"
],
"properties": {
"name": {
"type": "string"
},
"versions": {
"type": "array",
"items": [
{
"type": "string",
"pattern": "^([<>]=?|=)\s+.+$"
}
]
}
}
}
}

49
packetary/schemas/rpm_repo_schema.py Normal file
View File

@ -0,0 +1,49 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
RPM_REPO_SCHEMA = {
"$schema": "http://json-schema.org/draft-04/schema#",
"type": "object",
"required": [
"name",
"uri",
],
"properties": {
"name": {
"type": "string"
},
"uri": {
"type": "string"
},
"path": {
"type": "string"
},
"priority": {
"anyOf": [
{
"type": "number",
"minimum": 1,
"maximum": 99,
},
{
"type": "null"
}
]
}
}
}

6
packetary/tests/base.py
View File

@ -31,3 +31,9 @@ class TestCase(unittest.TestCase):
assertion(
exp, method(*value), "{0} != f({1})".format(exp, value)
)
def assertNotRaises(self, exception, method, *args, **kwargs):
try:
method(*args, **kwargs)
except exception as e:
self.fail("Unexpected error: {0}".format(e))

1
packetary/tests/stubs/generator.py
View File

@ -22,6 +22,7 @@ from packetary import objects
def gen_repository(name="test", url="file:///test",
architecture="x86_64", origin="Test", **kwargs):
"""Helper to create Repository object with default attributes."""
url = kwargs.pop("uri", url)
return objects.Repository(name, url, architecture, origin, **kwargs)

19
packetary/tests/test_deb_driver.py
View File

@ -21,6 +21,7 @@ import os.path as path
import six
from packetary.drivers import deb_driver
from packetary.schemas import DEB_REPO_SCHEMA
from packetary.tests import base
from packetary.tests.stubs.generator import gen_package
from packetary.tests.stubs.generator import gen_repository
@ -66,7 +67,7 @@ class TestDebDriver(base.TestCase):
def test_get_repository(self):
repos = []
repo_data = {
"name": "repo1", "url": "http://host", "suite": "trusty",
"name": "repo1", "uri": "http://host", "suite": "trusty",
"section": ["main", "universe"], "path": "my_path"
}
self.connection.open_stream.return_value = {"Origin": "Ubuntu"}
@ -99,7 +100,7 @@ class TestDebDriver(base.TestCase):
def test_get_repository_if_release_does_not_exist(self):
repo_data = {
"name": "repo1", "url": "http://host", "suite": "trusty",
"name": "repo1", "uri": "http://host", "suite": "trusty",
"section": ["main"], "path": "my_path"
}
repos = []
@ -115,7 +116,7 @@ class TestDebDriver(base.TestCase):
def test_get_repository_fail_if_error(self):
repo_data = {
"name": "repo1", "url": "http://host", "suite": "trusty",
"name": "repo1", "uri": "http://host", "suite": "trusty",
"section": ["main"], "path": "my_path"
}
repos = []
@ -132,7 +133,7 @@ class TestDebDriver(base.TestCase):
with self.assertRaisesRegexp(ValueError, "does not supported"):
self.driver.get_repository(
self.connection,
{"url": "http://host", "suite": "trusty"},
{"uri": "http://host", "suite": "trusty"},
"x86_64",
lambda x: None
)
@ -313,14 +314,14 @@ class TestDebDriver(base.TestCase):
@mock.patch("packetary.drivers.deb_driver.utils.ensure_dir_exist")
def test_create_repository(self, mkdir_mock, deb822, gzip, open, os):
repository_data = {
"name": "Test", "url": "file:///repo", "suite": "trusty",
"name": "Test", "uri": "file:///repo", "suite": "trusty",
"section": "main", "type": "rpm", "priority": "100",
"origin": "Origin", "path": "/repo"
}
repo = self.driver.create_repository(repository_data, "x86_64")
self.assertEqual(repository_data["name"], repo.name)
self.assertEqual("x86_64", repo.architecture)
self.assertEqual(repository_data["url"] + "/", repo.url)
self.assertEqual(repository_data["uri"] + "/", repo.url)
self.assertEqual(repository_data["origin"], repo.origin)
self.assertEqual(
(repository_data["suite"], repository_data["section"]),
@ -340,7 +341,7 @@ class TestDebDriver(base.TestCase):
def test_createrepository_fails_if_invalid_data(self):
repository_data = {
"name": "Test", "url": "file:///repo", "suite": "trusty",
"name": "Test", "uri": "file:///repo", "suite": "trusty",
"type": "rpm", "priority": "100",
"origin": "Origin", "path": "/repo"
}
@ -397,3 +398,7 @@ class TestDebDriver(base.TestCase):
)
rel_path = self.driver.get_relative_path(repo, "test.pkg")
self.assertEqual("pool/main/t/test.pkg", rel_path)
def test_get_repository_data_scheme(self):
schema = self.driver.get_repository_data_schema()
self.assertIs(DEB_REPO_SCHEMA, schema)

130
packetary/tests/test_repository_api.py
View File

@ -19,21 +19,31 @@
import copy
import mock
import jsonschema
from packetary.api import Configuration
from packetary.api import Context
from packetary.api import RepositoryApi
from packetary.schemas import PACKAGE_FILES_SCHEMA
from packetary.schemas import PACKAGES_SCHEMA
from packetary.tests import base
from packetary.tests.stubs import generator
from packetary.tests.stubs.helpers import CallbacksAdapter
@mock.patch("packetary.api.jsonschema")
class TestRepositoryApi(base.TestCase):
def setUp(self):
self.controller = CallbacksAdapter()
self.api = RepositoryApi(self.controller)
self.repo_data = {"name": "repo1", "url": "file:///repo1"}
self.repo_data = {"name": "repo1", "uri": "file:///repo1"}
self.requirements_data = [
{"name": "test1"}, {"name": "test2", "versions": ["< 3", "> 1"]}
]
self.schema = {}
self.repo = generator.gen_repository(**self.repo_data)
self.controller.load_repositories.return_value = [self.repo]
self.controller.get_repository_data_schema.return_value = self.schema
self._generate_packages()
def _generate_packages(self):
@ -56,7 +66,8 @@ class TestRepositoryApi(base.TestCase):
@mock.patch("packetary.api.RepositoryController")
@mock.patch("packetary.api.ConnectionsManager")
def test_create_with_config(self, connection_mock, controller_mock):
def test_create_with_config(self, connection_mock, controller_mock,
jsonschema_mock):
config = Configuration(
http_proxy="http://localhost", https_proxy="https://localhost",
retries_num=10, retry_interval=1, threads_num=8,
@ -75,7 +86,8 @@ class TestRepositoryApi(base.TestCase):
@mock.patch("packetary.api.RepositoryController")
@mock.patch("packetary.api.ConnectionsManager")
def test_create_with_context(self, connection_mock, controller_mock):
def test_create_with_context(self, connection_mock, controller_mock,
jsonschema_mock):
config = Configuration(
http_proxy="http://localhost", https_proxy="https://localhost",
retries_num=10, retry_interval=1, threads_num=8,
@ -93,42 +105,67 @@ class TestRepositoryApi(base.TestCase):
context, "deb", "x86_64"
)
def test_create_repository(self):
def test_create_repository(self, jsonschema_mock):
file_urls = ["file://test1.pkg"]
self.api.create_repository(self.repo_data, file_urls)
self.controller.create_repository.assert_called_once_with(
self.repo_data, file_urls
)
jsonschema_mock.validate.assert_has_calls(
[
mock.call(self.repo_data, self.schema),
mock.call(file_urls, PACKAGE_FILES_SCHEMA),
]
)
def test_get_packages_as_is(self):
def test_get_packages_as_is(self, jsonschema_mock):
packages = self.api.get_packages([self.repo_data], None)
self.assertEqual(5, len(packages))
self.assertItemsEqual(
self.packages,
packages
)
jsonschema_mock.validate.assert_called_once_with(
self.repo_data, self.schema
)
def test_get_packages_by_requirements_with_mandatory(self):
def test_get_packages_by_requirements_with_mandatory(self,
jsonschema_mock):
requirements = [{"name": "package1"}]
packages = self.api.get_packages(
[self.repo_data], [{"name": "package1"}], True
[self.repo_data], requirements, True
)
self.assertEqual(3, len(packages))
self.assertItemsEqual(
["package1", "package2", "package3"],
(x.name for x in packages)
)
jsonschema_mock.validate.assert_has_calls(
[
mock.call(self.repo_data, self.schema),
mock.call(requirements, PACKAGES_SCHEMA),
]
)
def test_get_packages_by_requirements_without_mandatory(self):
def test_get_packages_by_requirements_without_mandatory(self,
jsonschema_mock):
requirements = [{"name": "package4"}]
packages = self.api.get_packages(
[self.repo_data], [{"name": "package4"}], False
[self.repo_data], requirements, False
)
self.assertEqual(2, len(packages))
self.assertItemsEqual(
["package1", "package4"],
(x.name for x in packages)
)
jsonschema_mock.validate.assert_has_calls(
[
mock.call(self.repo_data, self.schema),
mock.call(requirements, PACKAGES_SCHEMA),
]
)
def test_clone_repositories_as_is(self):
def test_clone_repositories_as_is(self, jsonschema_mock):
# return value is used as statistics
mirror = copy.copy(self.repo)
mirror.url = "file:///mirror/repo"
@ -143,15 +180,19 @@ class TestRepositoryApi(base.TestCase):
)
self.assertEqual(6, stats.total)
self.assertEqual(4, stats.copied)
jsonschema_mock.validate.assert_called_once_with(
self.repo_data, self.schema
)
def test_clone_by_requirements_with_mandatory(self):
def test_clone_by_requirements_with_mandatory(self, jsonschema_mock):
# return value is used as statistics
mirror = copy.copy(self.repo)
mirror.url = "file:///mirror/repo"
requirements = [{"name": "package1"}]
self.controller.fork_repository.return_value = mirror
self.controller.assign_packages.return_value = [0, 1, 1]
stats = self.api.clone_repositories(
[self.repo_data], [{"name": "package1"}],
[self.repo_data], requirements,
"/mirror", include_mandatory=True
)
packages = {self.packages[0], self.packages[1], self.packages[2]}
@ -163,15 +204,23 @@ class TestRepositoryApi(base.TestCase):
)
self.assertEqual(3, stats.total)
self.assertEqual(2, stats.copied)
jsonschema_mock.validate.assert_has_calls(
[
mock.call(self.repo_data, self.schema),
mock.call(requirements, PACKAGES_SCHEMA),
]
)
def test_clone_by_requirements_without_mandatory(self):
def test_clone_by_requirements_without_mandatory(self,
jsonschema_mock):
# return value is used as statistics
mirror = copy.copy(self.repo)
mirror.url = "file:///mirror/repo"
requirements = [{"name": "package4"}]
self.controller.fork_repository.return_value = mirror
self.controller.assign_packages.return_value = [0, 4]
stats = self.api.clone_repositories(
[self.repo_data], [{"name": "package4"}],
[self.repo_data], requirements,
"/mirror", include_mandatory=False
)
packages = {self.packages[0], self.packages[3]}
@ -183,30 +232,65 @@ class TestRepositoryApi(base.TestCase):
)
self.assertEqual(2, stats.total)
self.assertEqual(1, stats.copied)
jsonschema_mock.validate.assert_has_calls(
[
mock.call(self.repo_data, self.schema),
mock.call(requirements, PACKAGES_SCHEMA),
]
)
def test_get_unresolved(self):
def test_get_unresolved(self, jsonschema_mock):
unresolved = self.api.get_unresolved_dependencies([self.repo_data])
self.assertItemsEqual(["package6"], (x.name for x in unresolved))
jsonschema_mock.validate.assert_called_once_with(
self.repo_data, self.schema
)
def test_load_requirements(self):
def test_load_requirements(self, jsonschema_mock):
expected = {
generator.gen_relation("test1"),
generator.gen_relation("test2", ["<", "3"]),
generator.gen_relation("test2", [">", "1"]),
}
actual = set(self.api._load_requirements(
[{"name": "test1"}, {"name": "test2", "versions": ["< 3", "> 1"]}]
self.requirements_data
))
self.assertEqual(expected, actual)
self.assertIsNone(self.api._load_requirements(None))
jsonschema_mock.validate.assert_called_once_with(
self.requirements_data,
PACKAGES_SCHEMA
)
def test_validate_repo_data(self):
# TODO(bgaifullin) implement me
pass
def test_validate_data(self, jsonschema_mock):
self.api._validate_data(self.repo_data, self.schema)
jsonschema_mock.validate.assert_called_once_with(
self.repo_data, self.schema
)
def test_validate_requirements_data(self):
# TODO(bgaifullin) implement me
pass
def test_validate_invalid_data(self, jschema_m):
jschema_m.ValidationError = jsonschema.ValidationError
jschema_m.SchemaError = jsonschema.SchemaError
paths = [("a", "b"), ()]
for path in paths:
msg = "Invalid data: error."
details = "\nField: {0}".format(".".join(path)) if path else ""
with self.assertRaisesRegexp(ValueError, msg + details):
jschema_m.validate.side_effect = jsonschema.ValidationError(
"error", path=path
)
self.api._validate_data([], {})
jschema_m.validate.assert_called_with([], {})
jschema_m.validate.reset_mock()
msg = "Invalid schema: error."
with self.assertRaisesRegexp(ValueError, msg + details):
jschema_m.validate.side_effect = jsonschema.SchemaError(
"error", schema_path=path
)
self.api._validate_data([], {})
jschema_m.validate.assert_called_with([], {})
class TestContext(base.TestCase):

4
packetary/tests/test_repository_contoller.py
View File

@ -53,7 +53,7 @@ class TestRepositoryController(base.TestCase):
self.assertIs(self.driver, controller.driver)
def test_load_repositories(self):
repo_data = {"name": "test", "url": "file:///test1"}
repo_data = {"name": "test", "uri": "file:///test1"}
repo = gen_repository(**repo_data)
self.driver.get_repository = CallbacksAdapter()
self.driver.get_repository.side_effect = [repo]
@ -150,7 +150,7 @@ class TestRepositoryController(base.TestCase):
def test_create_repository(self):
repository_data = {
"name": "Test", "url": "file:///repo/",
"name": "Test", "uri": "file:///repo/",
"section": ("trusty", "main"), "origin": "Test"
}
repo = gen_repository(**repository_data)

11
packetary/tests/test_rpm_driver.py
View File

@ -23,6 +23,7 @@ import sys
import six
from packetary.objects import FileChecksum
from packetary.schemas import RPM_REPO_SCHEMA
from packetary.tests import base
from packetary.tests.stubs.generator import gen_repository
from packetary.tests.stubs.helpers import get_compressed
@ -68,7 +69,7 @@ class TestRpmDriver(base.TestCase):
def test_get_repository(self):
repos = []
repo_data = {"name": "os", "url": "http://host/centos/os/x86_64/"}
repo_data = {"name": "os", "uri": "http://host/centos/os/x86_64/"}
self.driver.get_repository(
self.connection,
repo_data,
@ -220,13 +221,13 @@ class TestRpmDriver(base.TestCase):
@mock.patch("packetary.drivers.rpm_driver.utils.ensure_dir_exist")
def test_create_repository(self, ensure_dir_exists_mock):
repository_data = {
"name": "Test", "url": "file:///repo/os/x86_64", "origin": "Test"
"name": "Test", "uri": "file:///repo/os/x86_64", "origin": "Test"
}
repo = self.driver.create_repository(repository_data, "x86_64")
ensure_dir_exists_mock.assert_called_once_with("/repo/os/x86_64/")
self.assertEqual(repository_data["name"], repo.name)
self.assertEqual("x86_64", repo.architecture)
self.assertEqual(repository_data["url"] + "/", repo.url)
self.assertEqual(repository_data["uri"] + "/", repo.url)
self.assertEqual(repository_data["origin"], repo.origin)
@mock.patch("packetary.drivers.rpm_driver.utils")
@ -278,3 +279,7 @@ class TestRpmDriver(base.TestCase):
)
rel_path = self.driver.get_relative_path(repo, "test.pkg")
self.assertEqual("packages/test.pkg", rel_path)
def test_get_repository_data_schema(self):
schema = self.driver.get_repository_data_schema()
self.assertIs(RPM_REPO_SCHEMA, schema)

287
packetary/tests/test_schemas.py Normal file
View File

@ -0,0 +1,287 @@
# -*- coding: utf-8 -*-
# Copyright 2016 Mirantis, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import jsonschema
from packetary.schemas import DEB_REPO_SCHEMA
from packetary.schemas import PACKAGE_FILES_SCHEMA
from packetary.schemas import PACKAGES_SCHEMA
from packetary.schemas import RPM_REPO_SCHEMA
from packetary.tests import base
class TestRepositorySchemaBase(base.TestCase):
def check_invalid_name(self):
self._check_invalid_type('name')
def check_invalid_uri(self):
self._check_invalid_type('uri')
def check_invalid_path(self):
self._check_invalid_type('path')
def check_required_properties(self):
repos_data = [{"name": "os"}, {"uri": "file:///repo"}]
for data in repos_data:
self.assertRaisesRegexp(
jsonschema.ValidationError,
"is a required property",
jsonschema.validate, data, self.schema
)
def check_priority(self, min_value=None, max_value=None):
if min_value is not None:
self._check_invalid_priority(min_value - 1)
self._check_valid_priority(min_value)
if max_value is not None:
self._check_invalid_priority(max_value + 1)
self._check_valid_priority(max_value)
self._check_valid_priority(None)
self._check_invalid_priority("abc")
def _check_invalid_type(self, key):
invalid_data = {key: 123}
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, invalid_data[key],
self.schema['properties'][key]
)
def _check_valid_priority(self, value):
self.assertNotRaises(
jsonschema.ValidationError, jsonschema.validate, value,
self.schema['properties']['priority']
)
def _check_invalid_priority(self, value):
self.assertRaisesRegexp(
jsonschema.ValidationError,
"is not valid under any of the given schemas",
jsonschema.validate, value, self.schema['properties']['priority']
)
class TestDebRepoSchema(TestRepositorySchemaBase):
def setUp(self):
self.schema = DEB_REPO_SCHEMA
def test_valid_repo_data(self):
repo_data = {
"name": "os", "uri": "file:///repo", "suite": "trusty",
"section": ["main", "multiverse"], "path": "/some/path",
"priority": 1001
}
self.assertNotRaises(
jsonschema.ValidationError, jsonschema.validate,
repo_data, self.schema
)
def test_priority(self):
self.check_priority(0)
def test_validation_fail_for_required_properties(self):
self.check_required_properties()
repo_data = {"name": "os", "uri": "file:///repo"}
self.assertRaisesRegexp(
jsonschema.ValidationError, "'suite' is a required property",
jsonschema.validate, repo_data, self.schema
)
def test_validation_fail_if_name_is_invalid(self):
self.check_invalid_name()
def test_validation_fail_if_uri_is_invalid(self):
self.check_invalid_uri()
def test_validation_fail_if_path_is_invalid(self):
self.check_invalid_path()
def test_validation_fail_if_suite_is_invalid(self):
repo_data = {"name": "os", "uri": "file:///repo", "suite": 123}
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, repo_data, self.schema
)
def test_validation_fail_if_section_not_array(self):
repo_data = {
"name": "os", "uri": "file:///repo", "suite": "trusty",
"section": 123
}
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'array'",
jsonschema.validate, repo_data, self.schema
)
def test_validation_fail_if_section_not_string(self):
repo_data = {
"name": "os", "uri": "file:///repo", "suite": "trusty",
"section": [123]
}
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, repo_data, self.schema
)
class TestRpmRepoSchema(TestRepositorySchemaBase):
def setUp(self):
self.schema = RPM_REPO_SCHEMA
def test_valid_repo_data(self):
repo_data = {
"name": "os", "uri": "file:///repo", "path": "/some/path",
"priority": 45
}
self.assertNotRaises(
jsonschema.ValidationError, jsonschema.validate, repo_data,
self.schema
)
def test_priority(self):
self.check_priority(1, 99)
def test_validation_fail_for_required_properties(self):
self.check_required_properties()
def test_validation_fail_if_name_is_invalid(self):
self.check_invalid_name()
def test_validation_fail_if_uri_is_invalid(self):
self.check_invalid_uri()
def test_validation_fail_if_path_is_invalid(self):
self.check_invalid_path()
class TestPackagesSchema(base.TestCase):
def setUp(self):
self.schema = PACKAGES_SCHEMA
def test_valid_requirements_data(self):
requirements_data = [
{"name": "test1", "versions": [">= 1.1.2", "<= 3"]},
{"name": "test2", "versions": ["< 3", "> 1", ">= 4"]},
{"name": "test3", "versions": ["= 3"]},
{"name": "test4", "versions": ["= 3"]}
]
self.assertNotRaises(
jsonschema.ValidationError, jsonschema.validate, requirements_data,
self.schema
)
def test_validation_fail_for_required_properties(self):
requirements_data = [
[{"name": "test1"}],
[{"versions": ["< 3", "> 1"]}]
]
for data in requirements_data:
self.assertRaisesRegexp(
jsonschema.ValidationError,
"is a required property",
jsonschema.validate, data, self.schema
)
def test_validation_fail_if_name_is_invalid(self):
requirements_data = [
{"name": 123, "versions": [">= 1.1.2", "<= 3"]},
]
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, requirements_data, self.schema
)
def test_validation_fail_if_versions_not_array(self):
requirements_data = [
{"name": "test1", "versions": 123}
]
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'array'",
jsonschema.validate, requirements_data,
self.schema
)
def test_validation_fail_if_versions_not_string(self):
requirements_data = [
{"name": "test1", "versions": [123]}
]
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, requirements_data,
self.schema
)
def test_validation_fail_if_versions_not_match(self):
versions = [
["1.1.2"], # relational operator
[">=3"], # not whitespace after ro
["== 3"] # ==
]
for version in versions:
self.assertRaisesRegexp(
jsonschema.ValidationError, "does not match",
jsonschema.validate, version,
self.schema['items']['properties']['versions']
)
class TestPackageFilesSchema(base.TestCase):
def setUp(self):
self.schema = PACKAGE_FILES_SCHEMA
def test_valid_file_urls(self):
file_urls = [
"file://test1.pkg",
"file:///test2.pkg",
"/test3.pkg",
"http://test4.pkg",
"https://test5.pkg"
]
self.assertNotRaises(
jsonschema.ValidationError, jsonschema.validate, file_urls,
self.schema
)
def test_validation_fail_if_urls_not_array(self):
file_urls = "/test1.pkg"
self.assertRaisesRegexp(
jsonschema.ValidationError, "'/test1.pkg' is not of type 'array'",
jsonschema.validate, file_urls, self.schema
)
def test_validation_fail_if_urls_not_string(self):
file_urls = [123]
self.assertRaisesRegexp(
jsonschema.ValidationError, "123 is not of type 'string'",
jsonschema.validate, file_urls, self.schema
)
def test_validation_fail_if_invalid_file_urls(self):
file_urls = [
["test1.pkg"], # does not match pattern
["./test2.pkg"], # does not match pattern
["file//test3.pkg"], # does not match pattern
["http//test4.pkg"] # does not match pattern
]
for url in file_urls[2:]:
self.assertRaisesRegexp(
jsonschema.ValidationError, "does not match",
jsonschema.validate, url, self.schema
)

1
requirements.txt
View File

@ -12,3 +12,4 @@ stevedore>=1.1.0
six>=1.5.2
python-debian>=0.1.21
lxml>=3.2
jsonschema!=2.5.0,<3.0.0,>=2.0.0 # MIT