Skip to content

Commit

Permalink
Translate tests to using bindings
Browse files Browse the repository at this point in the history 
[noissue]
  • Loading branch information
Matthias Dellweg committed Feb 18, 2020
1 parent d4fad49 commit 26f7a91
Show file tree
Hide file tree
Showing 12 changed files with 708 additions and 425 deletions.
4 changes: 2 additions & 2 deletions flake8.cfg
View file
@@ -1,6 +1,6 @@
[flake8]
exclude = ./docs/*,*/migrations/*
ignore = W503,Q000,D100,D104,D106,D200,D401,D402,D202
ignore = W503,Q000,D100,D104,D106,D200,D401,D402,E203
max-line-length = 100

# Flake8-quotes extension codes
Expand All @@ -16,4 +16,4 @@ max-line-length = 100
# D200: one-line docstring should fit on one line with quotes
# D401: first line should be imperative (nitpicky)
# D402: first line should not be the function’s “signature” (false positives)
# D202: no blank line after docstring--disabled until https://github.com/PyCQA/pydocstyle/issues/361 is fixed
# E203: no whitespace around ':'. disabled until https://github.com/PyCQA/pycodestyle/issues/373 is fixed
6 changes: 1 addition & 5 deletions pulp_deb/app/serializers/content_serializers.py
View file
Expand Up @@ -68,7 +68,6 @@ class GenericContentSerializer(SingleArtifactContentUploadSerializer, ContentChe

def deferred_validate(self, data):
"""Validate the GenericContent data."""

data = super().deferred_validate(data)

data["sha256"] = data["artifact"].sha256
Expand Down Expand Up @@ -389,13 +388,12 @@ def __init__(self, *args, **kwargs):

def deferred_validate(self, data):
"""Validate that the artifact is a package and extract it's values."""

data = super().deferred_validate(data)

try:
package_paragraph = debfile.DebFile(fileobj=data["artifact"].file).debcontrol()
except Exception: # TODO: Be more specific
raise ValidationError(_("Not a valid Deb Package"))
raise ValidationError(_("Unable to read Deb Package"))

from822_serializer = self.Meta.from822_serializer.from822(data=package_paragraph)
from822_serializer.is_valid(raise_exception=True)
Expand Down Expand Up @@ -469,7 +467,6 @@ class PackageSerializer(BasePackageSerializer):

def deferred_validate(self, data):
"""Validate for 'normal' Package (not installer)."""

data = super().deferred_validate(data)

if data.get("section") == "debian-installer":
Expand All @@ -489,7 +486,6 @@ class InstallerPackageSerializer(BasePackageSerializer):

def deferred_validate(self, data):
"""Validate for InstallerPackage."""

data = super().deferred_validate(data)

if data.get("section") != "debian-installer":
Expand Down
161 changes: 84 additions & 77 deletions pulp_deb/tests/functional/api/test_crud_content_unit.py
View file
Expand Up @@ -2,24 +2,25 @@
"""Tests that perform actions over content unit."""
import unittest

from requests.exceptions import HTTPError
from tempfile import NamedTemporaryFile

from pulp_smash import api, config, utils
from pulp_smash.exceptions import TaskReportError
from pulp_smash.pulp3.constants import ARTIFACTS_PATH
from pulp_smash import utils
from pulp_smash.pulp3.utils import delete_orphans

from pulp_deb.tests.functional.constants import (
DEB_GENERIC_CONTENT_PATH,
DEB_GENERIC_CONTENT_URL,
DEB_PACKAGE_PATH,
DEB_PACKAGE_URL,
)
from pulp_deb.tests.functional.utils import (
deb_generic_content_api,
deb_package_api,
gen_artifact,
gen_deb_content_attrs,
gen_deb_content_upload_attrs,
gen_deb_package_attrs,
gen_deb_package_upload_attrs,
monitor_task,
PulpTaskError,
skip_if,
)
from pulp_deb.tests.functional.utils import set_up_module as setUpModule # noqa:F401
Expand All @@ -32,57 +33,53 @@ class GenericContentUnitTestCase(unittest.TestCase):
* `Pulp #2872 <https://pulp.plan.io/issues/2872>`_
* `Pulp #3445 <https://pulp.plan.io/issues/3445>`_
* `Pulp Smash #870 <https://github.com/PulpQE/pulp-smash/issues/870>`_
* `Pulp Smash #870 <https://github.com/pulp/pulp-smash/issues/870>`_
"""

gen_content_attrs = staticmethod(gen_deb_content_attrs)
gen_content_verify_attrs = staticmethod(gen_deb_content_attrs)
CONTENT_PATH = DEB_GENERIC_CONTENT_PATH
content_api = deb_generic_content_api
CONTENT_URL = DEB_GENERIC_CONTENT_URL

@classmethod
def setUpClass(cls):
"""Create class-wide variable."""
cls.cfg = config.get_config()
delete_orphans(cls.cfg)
delete_orphans()
cls.content_unit = {}
cls.client = api.Client(cls.cfg, api.json_handler)
files = {"file": utils.http_get(cls.CONTENT_URL)}
cls.artifact = cls.client.post(ARTIFACTS_PATH, files=files)
cls.artifact = gen_artifact(cls.CONTENT_URL)

@classmethod
def tearDownClass(cls):
"""Clean class-wide variable."""
delete_orphans(cls.cfg)
delete_orphans()

def test_01_create_content_unit(self):
"""Create content unit."""
attrs = self.gen_content_attrs(self.artifact)
call_report = self.client.post(self.CONTENT_PATH, data=attrs)
created_resources = next(api.poll_spawned_tasks(self.cfg, call_report))["created_resources"]
self.content_unit.update(self.client.get(created_resources[0]))
for key, val in self.gen_content_verify_attrs(self.artifact).items():
response = self.content_api.create(**attrs)
created_resources = monitor_task(response.task)
content_unit = self.content_api.read(created_resources[0])
self.content_unit.update(content_unit.to_dict())
for key, val in attrs.items():
with self.subTest(key=key):
self.assertEqual(self.content_unit[key], val)

@skip_if(bool, "content_unit", False)
def test_02_read_content_unit(self):
"""Read a content unit by its href."""
content_unit = self.client.get(self.content_unit["pulp_href"])
content_unit = self.content_api.read(self.content_unit["pulp_href"]).to_dict()
for key, val in self.content_unit.items():
with self.subTest(key=key):
self.assertEqual(content_unit[key], val)

@skip_if(bool, "content_unit", False)
def test_02_read_content_units(self):
"""Read a content unit by its relative_path."""
page = self.client.get(
self.CONTENT_PATH, params={"relative_path": self.content_unit["relative_path"]}
)
self.assertEqual(len(page["results"]), 1)
page = self.content_api.list(relative_path=self.content_unit["relative_path"])
self.assertEqual(len(page.results), 1)
for key, val in self.content_unit.items():
with self.subTest(key=key):
self.assertEqual(page["results"][0][key], val)
self.assertEqual(page.results[0].to_dict()[key], val)

@skip_if(bool, "content_unit", False)
def test_03_partially_update(self):
Expand All @@ -91,9 +88,10 @@ def test_03_partially_update(self):
This HTTP method is not supported and a HTTP exception is expected.
"""
attrs = self.gen_content_attrs(self.artifact)
with self.assertRaises(HTTPError) as exc:
self.client.patch(self.content_unit["pulp_href"], attrs)
self.assertEqual(exc.exception.response.status_code, 405)
with self.assertRaises(AttributeError) as exc:
self.content_api.partial_update(self.content_unit["pulp_href"], attrs)
msg = "object has no attribute 'partial_update'"
self.assertIn(msg, exc.exception.args[0])

@skip_if(bool, "content_unit", False)
def test_03_fully_update(self):
Expand All @@ -102,27 +100,29 @@ def test_03_fully_update(self):
This HTTP method is not supported and a HTTP exception is expected.
"""
attrs = self.gen_content_attrs(self.artifact)
with self.assertRaises(HTTPError) as exc:
self.client.put(self.content_unit["pulp_href"], attrs)
self.assertEqual(exc.exception.response.status_code, 405)
with self.assertRaises(AttributeError) as exc:
self.content_api.update(self.content_unit["pulp_href"], attrs)
msg = "object has no attribute 'update'"
self.assertIn(msg, exc.exception.args[0])

@skip_if(bool, "content_unit", False)
def test_04_delete(self):
"""Attempt to delete a content unit using HTTP DELETE.
This HTTP method is not supported and a HTTP exception is expected.
"""
with self.assertRaises(HTTPError) as exc:
self.client.delete(self.content_unit["pulp_href"])
self.assertEqual(exc.exception.response.status_code, 405)
with self.assertRaises(AttributeError) as exc:
self.content_api.delete(self.content_unit["pulp_href"])
msg = "object has no attribute 'delete'"
self.assertIn(msg, exc.exception.args[0])


class PackageTestCase(GenericContentUnitTestCase):
"""CRUD content unit."""

gen_content_attrs = staticmethod(gen_deb_package_attrs)
gen_content_verify_attrs = staticmethod(gen_deb_package_attrs)
CONTENT_PATH = DEB_PACKAGE_PATH
content_api = deb_package_api
CONTENT_URL = DEB_PACKAGE_URL


Expand All @@ -136,58 +136,63 @@ class GenericContentUnitUploadTestCase(unittest.TestCase):

gen_content_upload_attrs = staticmethod(gen_deb_content_upload_attrs)
gen_content_upload_verify_attrs = staticmethod(gen_deb_content_upload_attrs)
CONTENT_PATH = DEB_GENERIC_CONTENT_PATH
content_api = deb_generic_content_api
CONTENT_URL = DEB_GENERIC_CONTENT_URL

@classmethod
def setUpClass(cls):
"""Create class-wide variable."""
cls.cfg = config.get_config()
delete_orphans(cls.cfg)
delete_orphans()
cls.content_unit = {}
cls.client = api.Client(cls.cfg, api.smart_handler)
cls.files = {"file": utils.http_get(cls.CONTENT_URL)}
cls.file = utils.http_get(cls.CONTENT_URL)
cls.attrs = cls.gen_content_upload_attrs()

@classmethod
def tearDownClass(cls):
"""Clean class-wide variable."""
delete_orphans(cls.cfg)
delete_orphans()

def test_01_create_content_unit(self):
"""Create content unit."""
content_unit = self.client.post(self.CONTENT_PATH, data=self.attrs, files=self.files)
self.content_unit.update(content_unit)
with NamedTemporaryFile() as temp_file:
temp_file.write(self.file)
temp_file.flush()
response = self.content_api.create(**self.attrs, file=temp_file.name)
created_resources = monitor_task(response.task)
content_unit = self.content_api.read(created_resources[0])
self.content_unit.update(content_unit.to_dict())
for key, val in self.attrs.items():
with self.subTest(key=key):
self.assertEqual(self.content_unit[key], val)

@skip_if(bool, "content_unit", False)
def test_02_read_content_unit(self):
"""Read a content unit by its href."""
content_unit = self.client.get(self.content_unit["pulp_href"])
content_unit = self.content_api.read(self.content_unit["pulp_href"]).to_dict()
for key, val in self.content_unit.items():
with self.subTest(key=key):
self.assertEqual(content_unit[key], val)

@skip_if(bool, "content_unit", False)
def test_02_read_content_units(self):
"""Read a content unit by its relative_path."""
page = self.client.using_handler(api.json_handler).get(
self.CONTENT_PATH, params={"relative_path": self.content_unit["relative_path"]}
)
self.assertEqual(len(page["results"]), 1)
page = self.content_api.list(relative_path=self.content_unit["relative_path"])
self.assertEqual(len(page.results), 1)
for key, val in self.content_unit.items():
with self.subTest(key=key):
self.assertEqual(page["results"][0][key], val)
self.assertEqual(page.results[0].to_dict()[key], val)

@skip_if(bool, "content_unit", False)
def test_03_fail_duplicate_content_unit(self):
"""Create content unit."""
with self.assertRaises(TaskReportError) as exc:
self.client.post(self.CONTENT_PATH, data=self.attrs, files=self.files)
self.assertEqual(exc.exception.task["state"], "failed")
error = exc.exception.task["error"]
with NamedTemporaryFile() as temp_file:
temp_file.write(self.file)
temp_file.flush()
response = self.content_api.create(**self.attrs, file=temp_file.name)
with self.assertRaises(PulpTaskError) as exc:
monitor_task(response.task)
self.assertEqual(exc.exception.task.state, "failed")
error = exc.exception.task.error
for key in ("already", "relative", "path", "sha256"):
self.assertIn(key, error["description"].lower(), error)

Expand All @@ -197,15 +202,18 @@ def test_03_duplicate_content_unit(self):
attrs = self.attrs.copy()
# Packages types only validate the filename, so we can prepend something to the path.
attrs["relative_path"] = "moved-" + self.content_unit["relative_path"]
self.client.post(self.CONTENT_PATH, data=attrs, files=self.files)
with NamedTemporaryFile() as temp_file:
temp_file.write(self.file)
temp_file.flush()
self.content_api.create(**attrs, file=temp_file.name)


class PackageUnitUploadTestCase(GenericContentUnitUploadTestCase):
"""CRUD content unit with upload feature."""

gen_content_upload_attrs = staticmethod(gen_deb_package_upload_attrs)
gen_content_upload_verify_attrs = staticmethod(gen_deb_package_upload_attrs)
CONTENT_PATH = DEB_PACKAGE_PATH
content_api = deb_package_api
CONTENT_URL = DEB_PACKAGE_URL


Expand All @@ -219,19 +227,13 @@ class DuplicateGenericContentUnit(unittest.TestCase):

gen_content_attrs = staticmethod(gen_deb_content_attrs)
gen_content_verify_attrs = staticmethod(gen_deb_content_attrs)
CONTENT_PATH = DEB_GENERIC_CONTENT_PATH
content_api = deb_generic_content_api
CONTENT_URL = DEB_GENERIC_CONTENT_URL

@classmethod
def setUpClass(cls):
"""Create class-wide variables."""
cls.cfg = config.get_config()
cls.client = api.Client(cls.cfg, api.smart_handler)

@classmethod
def tearDownClass(cls):
"""Clean created resources."""
delete_orphans(cls.cfg)
delete_orphans()

def test_raise_error(self):
"""Create a duplicate content unit using same relative_path.
Expand All @@ -241,19 +243,21 @@ def test_raise_error(self):
In order to raise an HTTP error, the same ``artifact`` and the same
``relative_path`` should be used.
"""
delete_orphans(self.cfg)
files = {"file": utils.http_get(self.CONTENT_URL)}
artifact = self.client.post(ARTIFACTS_PATH, files=files)
delete_orphans()
artifact = gen_artifact(self.CONTENT_URL)
attrs = self.gen_content_attrs(artifact)

# create first content unit.
self.client.post(self.CONTENT_PATH, attrs)
response = self.content_api.create(**attrs)
created_resources = monitor_task(response.task)
self.content_api.read(created_resources[0])

# using the same attrs used to create the first content unit.
with self.assertRaises(TaskReportError) as exc:
self.client.post(self.CONTENT_PATH, attrs)
self.assertEqual(exc.exception.task["state"], "failed")
error = exc.exception.task["error"]
response = self.content_api.create(**attrs)
with self.assertRaises(PulpTaskError) as exc:
monitor_task(response.task)
self.assertEqual(exc.exception.task.state, "failed")
error = exc.exception.task.error
for key in ("already", "relative", "path", "sha256"):
self.assertIn(key, error["description"].lower(), error)

Expand All @@ -265,24 +269,27 @@ def test_non_error(self):
In order to avoid an HTTP error, use the same ``artifact`` and
different ``relative_path``.
"""
delete_orphans(self.cfg)
files = {"file": utils.http_get(self.CONTENT_URL)}
artifact = self.client.post(ARTIFACTS_PATH, files=files)
delete_orphans()
artifact = gen_artifact(self.CONTENT_URL)
attrs = self.gen_content_attrs(artifact)

# create first content unit.
content_unit = self.client.post(self.CONTENT_PATH, attrs)
response = self.content_api.create(**attrs)
created_resources = monitor_task(response.task)
content_unit = self.content_api.read(created_resources[0])

# Packages types only validate the filename, so we can prepend something to the path.
attrs["relative_path"] = "moved-" + content_unit["relative_path"]
attrs["relative_path"] = "moved-" + content_unit.relative_path
# create second content unit.
self.client.post(self.CONTENT_PATH, attrs)
response = self.content_api.create(**attrs)
created_resources = monitor_task(response.task)
content_unit = self.content_api.read(created_resources[0])


class DuplicatePackageUnit(DuplicateGenericContentUnit):
"""Attempt to create a duplicate content unit."""

gen_content_attrs = staticmethod(gen_deb_package_attrs)
gen_content_verify_attrs = staticmethod(gen_deb_package_attrs)
CONTENT_PATH = DEB_PACKAGE_PATH
content_api = deb_package_api
CONTENT_URL = DEB_PACKAGE_URL

0 comments on commit 26f7a91

Please sign in to comment.