Add tests cases for copying Docker units.

This commit adds several test cases for copying Docker units.
Additionally, it adds a handy utils module that contains several
convenient functions that are useful for these test cases, as well
as for other future test cases.

Fixes #98

pulp_smash/tests/docker/cli/test_copy.py

@@ -0,0 +1,244 @@
+# coding=utf-8
+"""Tests for copying docker units between repositories."""
+from __future__ import unicode_literals
+import re
+
+import unittest2
+from packaging.version import Version
+
+from pulp_smash import utils
+from pulp_smash.tests.docker.cli import utils as docker_utils
+
+
+class CopyAllImagesTestCase(docker_utils.BaseTestCase,
+                            docker_utils.SuccessMixin):
+    """Test copying all Images from one repository to another."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create and sync a docker repository with a v1 registry."""
+        super(CopyAllImagesTestCase, cls).setUpClass()
+        if cls.cfg.version < Version('2.8'):
+            raise unittest2.SkipTest('These tests require Pulp 2.8 or above.')
+        # Create and sync a repository with a v1 feed to bring some Images in.
+        docker_utils.create_repo(
+            cls.cfg, cls.repo_id, upstream_name='library/busybox',
+            sync_v1=True, sync_v2=False)
+        docker_utils.sync_repo(cls.cfg, cls.repo_id)
+
+        # Now create a feedless repository and copy the images from the first
+        # repository over.
+        cls.copy_target = utils.uuid4()
+        docker_utils.create_repo(cls.cfg, cls.copy_target)
+        cls.completed_proc = docker_utils.copy(cls.cfg, 'image', cls.repo_id,
+                                               cls.copy_target)
+
+    @classmethod
+    def tearDownClass(cls):
+        """Delete the copy_target repo."""
+        super(CopyAllImagesTestCase, cls).tearDownClass()
+        docker_utils.delete_repo(cls.cfg, cls.copy_target)
+
+    def test_positive_copy_output(self):
+        """Assert that the list of image_ids in the copy output are correct."""
+        # We need to limit the fields to image_id as a hack to work around
+        # https://pulp.plan.io/issues/1696. Doing this has a side effect of
+        # showing the image_ids without splitting them with a line break
+        # (sigh).
+        units_in_src = docker_utils.search(self.cfg, 'image', self.repo_id,
+                                           fields=['image_id']).stdout
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the image_ids. Hello regex my old
+        # friend, I've come to talk with you again…
+        image_ids_in_src = set(re.findall(r'(?:Image Id:)\s*(.*)',
+                                          units_in_src))
+
+        image_ids_printed = set(re.findall(r'(?: {2})(\w*)',
+                                           self.completed_proc.stdout))
+
+        self.assertEqual(image_ids_printed, image_ids_in_src)
+
+    def test_positive_units_copied(self):
+        """Assert that all units were copied."""
+        units_in_src = docker_utils.search(self.cfg, 'image',
+                                           self.repo_id).stdout
+        units_in_dest = docker_utils.search(self.cfg, 'image',
+                                            self.copy_target).stdout
+
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the unit_ids. Hello regex my old
+        # friend, I've come to talk with you again…
+        units_in_src = set(re.findall(r'(?:Unit Id:)\s*(.*)', units_in_src))
+        units_in_dest = set(re.findall(r'(?:Unit Id:)\s*(.*)', units_in_dest))
+
+        self.assertEqual(units_in_src, units_in_dest)
+
+    def test_return_code(self):
+        """Assert that the copy command has a return code of 0."""
+        self.assertEqual(self.completed_proc.returncode, 0)
+
+    def test_task_succeeded(self):
+        """Assert that "Copied" appears in stdout."""
+        self.assertIn('Copied', self.completed_proc.stdout)
+
+
+class CopyAllManifestsTestCase(docker_utils.BaseTestCase,
+                               docker_utils.SuccessMixin):
+    """Test copying all Manifests from one repository to another."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create and sync a docker repository with a v2 registry."""
+        super(CopyAllManifestsTestCase, cls).setUpClass()
+        if cls.cfg.version < Version('2.8'):
+            raise unittest2.SkipTest('These tests require Pulp 2.8 or above.')
+        # Create and sync a repository with a v2 feed to bring some Manifests
+        # in.
+        docker_utils.create_repo(
+            cls.cfg, cls.repo_id, upstream_name='library/busybox',
+            sync_v1=False, sync_v2=True)
+        docker_utils.sync_repo(cls.cfg, cls.repo_id)
+
+        # Now create a feedless repository and copy the manifests from the
+        # first repository over.
+        cls.copy_target = utils.uuid4()
+        docker_utils.create_repo(cls.cfg, cls.copy_target)
+        cls.completed_proc = docker_utils.copy(cls.cfg, 'manifest',
+                                               cls.repo_id, cls.copy_target)
+
+    @classmethod
+    def tearDownClass(cls):
+        """Delete the copy_target repo."""
+        super(CopyAllManifestsTestCase, cls).tearDownClass()
+        docker_utils.delete_repo(cls.cfg, cls.copy_target)
+
+    def test_positive_copy_output(self):
+        """Assert that the list of manifests in the copy output are correct."""
+        units_in_src = docker_utils.search(self.cfg, 'manifest', self.repo_id,
+                                           fields=['digest']).stdout
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the digests. Hello regex my old
+        # friend, I've come to talk with you again…
+        manifest_ids_in_src = set(re.findall(r'(?:Digest:)\s*(.*)',
+                                             units_in_src))
+
+        # The manifest digests are printed after "docker_manifest:" in the copy
+        # output.
+        manifest_ids = self.completed_proc.stdout.split('docker_manifest:')[1]
+        manifest_ids_printed = re.findall(r'(?: {2})(\w*:\w*)', manifest_ids)
+        # Due to https://pulp.plan.io/issues/1696 pulp-admin has split the last
+        # character of each of the manifest_ids_in_src items onto the next line
+        # which wasn't matched by the regex. That's stupid, but we're going to
+        # work around it by chopping the last character off of these as well.
+        manifest_ids_printed = set([d[:-1] for d in manifest_ids_printed])
+
+        self.assertEqual(manifest_ids_printed, manifest_ids_in_src)
+
+    def test_positive_units_copied(self):
+        """Assert that all units were copied."""
+        units_in_src = docker_utils.search(self.cfg, 'manifest',
+                                           self.repo_id).stdout
+        units_in_dest = docker_utils.search(self.cfg, 'manifest',
+                                            self.copy_target).stdout
+
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the unit_ids. Hello regex my old
+        # friend, I've come to talk with you again…
+        units_in_src = set(re.findall(r'(?:Unit Id:)\s*(.*)', units_in_src))
+        units_in_dest = set(re.findall(r'(?:Unit Id:)\s*(.*)', units_in_dest))
+
+        self.assertEqual(units_in_src, units_in_dest)
+
+    def test_return_code(self):
+        """Assert that the copy command has a return code of 0."""
+        self.assertEqual(self.completed_proc.returncode, 0)
+
+    def test_task_succeeded(self):
+        """Assert that "Copied" appears in stdout."""
+        self.assertIn('Copied', self.completed_proc.stdout)
+
+
+class CopyAllTagsTestCase(docker_utils.BaseTestCase,
+                          docker_utils.SuccessMixin):
+    """Test copying all Tags from one repository to another."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create and sync a docker repository with a v2 registry."""
+        super(CopyAllTagsTestCase, cls).setUpClass()
+        if cls.cfg.version < Version('2.8'):
+            raise unittest2.SkipTest('These tests require Pulp 2.8 or above.')
+        # Create and sync a repository with a v2 feed to bring some Tags in.
+        docker_utils.create_repo(
+            cls.cfg, cls.repo_id, upstream_name='library/busybox',
+            sync_v1=False, sync_v2=True)
+        docker_utils.sync_repo(cls.cfg, cls.repo_id)
+
+        # Now create a feedless repository and copy the tags from the first
+        # repository over.
+        cls.copy_target = utils.uuid4()
+        docker_utils.create_repo(cls.cfg, cls.copy_target)
+        cls.completed_proc = docker_utils.copy(cls.cfg, 'tag', cls.repo_id,
+                                               cls.copy_target)
+
+    @classmethod
+    def tearDownClass(cls):
+        """Delete the copy_target repo."""
+        super(CopyAllTagsTestCase, cls).tearDownClass()
+        docker_utils.delete_repo(cls.cfg, cls.copy_target)
+
+    def test_positive_copy_output(self):
+        """Assert that the list of tags in the copy output are correct."""
+        units_in_src = docker_utils.search(self.cfg, 'tag', self.repo_id,
+                                           fields=['name']).stdout
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the digests. Hello regex my old
+        # friend, I've come to talk with you again…
+        tag_ids_in_src = set(re.findall(r'(?:Digest:)\s*(.*)', units_in_src))
+
+        # The tag digests are printed after "docker_tag:" in the copy output.
+        tag_ids = self.completed_proc.stdout.split('docker_tag:')[1]
+        tag_ids_printed = re.findall(r'(?: {2})(\w*:\w*)', tag_ids)
+        # Due to https://pulp.plan.io/issues/1696 pulp-admin has split the last
+        # character of each of the tag_ids_in_src items onto the next line
+        # which wasn't matched by the regex. That's stupid, but we're going to
+        # work around it by chopping the last character off of these as well.
+        tag_ids_printed = set([d[:-1] for d in tag_ids_printed])
+
+        self.assertEqual(tag_ids_printed, tag_ids_in_src)
+
+    def test_positive_units_copied(self):
+        """Assert that all units were copied."""
+        tags_in_src = docker_utils.search(self.cfg, 'tag', self.repo_id).stdout
+        tags_in_dest = docker_utils.search(self.cfg, 'tag',
+                                           self.copy_target).stdout
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the names. Hello regex my old
+        # friend, I've come to talk with you again…
+        tags_in_src = set(re.findall(r'(?:Name:)\s*(.*)', tags_in_src))
+        tags_in_dest = set(re.findall(r'(?:Name:)\s*(.*)', tags_in_dest))
+
+        # The Manifests should have been pulled over as well since they are
+        # recursively copied.
+        manifests_in_src = docker_utils.search(self.cfg, 'manifest',
+                                               self.repo_id).stdout
+        manifests_in_dest = docker_utils.search(self.cfg, 'manifest',
+                                                self.copy_target).stdout
+        # Due to https://pulp.plan.io/issues/1693 it is not possible to get
+        # pulp-admin to limit the output to the unit_ids. Hello regex my old
+        # friend, I've come to talk with you again…
+        manifests_in_src = set(re.findall(r'(?:Name:)\s*(.*)',
+                                          manifests_in_src))
+        manifests_in_dest = set(re.findall(r'(?:Name:)\s*(.*)',
+                                           manifests_in_dest))
+
+        self.assertEqual(tags_in_src, tags_in_dest)
+        self.assertEqual(manifests_in_src, manifests_in_dest)
+
+    def test_return_code(self):
+        """Assert that the copy command has a return code of 0."""
+        self.assertEqual(self.completed_proc.returncode, 0)
+
+    def test_task_succeeded(self):
+        """Assert that "Copied" appears in stdout."""
+        self.assertIn('Copied', self.completed_proc.stdout)

pulp_smash/tests/docker/cli/utils.py

@@ -0,0 +1,118 @@
+# coding=utf-8
+"""Common utilities that assist in testing the CLI with docker repositories."""
+from __future__ import unicode_literals
+
+import unittest2
+
+from pulp_smash import cli, config, utils
+
+
+FEED = 'https://index.docker.io'
+
+
+def copy(server_config, unit_type, src_repo_id, dest_repo_id):
+    """Use pulp-admin to copy all instances of unit_type from src to dest."""
+    command = ('pulp-admin docker repo copy {unit_type} --from-repo-id {src} '
+               '--to-repo-id {dest}')
+    command = command.format(unit_type=unit_type, src=src_repo_id,
+                             dest=dest_repo_id)
+    command = command.split()
+
+    return cli.Client(server_config, cli.echo_handler).run(command)
+
+
+def create_repo(server_config, repo_id, upstream_name=None, sync_v1=False,
+                sync_v2=False):
+    """Use pulp-admin to create a repo with the given parameters."""
+    extra_flags = ''
+    if upstream_name:
+        extra_flags += ' --upstream-name {n}'.format(n=upstream_name)
+
+    # Handle whether we are syncing, and if so which APIs
+    if sync_v1 or sync_v2:
+        extra_flags += ' --feed {f}'.format(f=FEED)
+    if sync_v1:
+        extra_flags += ' --enable-v1 true'
+    else:
+        extra_flags += ' --enable-v1 false'
+    if sync_v2:
+        extra_flags += ' --enable-v2 true'
+    else:
+        extra_flags += ' --enable-v2 false'
+
+    command = ('pulp-admin docker repo create --repo-id '
+               '{repo_id}{extra_flags}').format(
+                   repo_id=repo_id, extra_flags=extra_flags).split()
+
+    return cli.Client(server_config, cli.echo_handler).run(command)
+
+
+def delete_repo(server_config, repo_id):
+    """Delete the repo given by repo_id."""
+    command = 'pulp-admin docker repo delete --repo-id {repo_id}'.format(
+        repo_id=repo_id).split()
+
+    return cli.Client(server_config, cli.echo_handler).run(command)
+
+
+def search(server_config, unit_type, repo_id, fields=None):
+    """
+    Search the given repo for all units of given unit_type.
+
+    unit_type should be a string: "image", "blob", "manifest", or "tag".
+    repo_id is the repo_id you wish to search.
+    fields is a list of strings of field names you want to retrieve.
+    """
+    extra_flags = ''
+    if fields:
+        extra_flags += ' --fields {f}'.format(f=','.join(fields))
+
+    command = ('pulp-admin docker repo search {unit_type} --repo-id '
+               '{repo_id}{eflags}')
+    command = command.format(unit_type=unit_type, repo_id=repo_id,
+                             eflags=extra_flags).split()
+
+    return cli.Client(server_config, cli.echo_handler).run(command)
+
+
+def sync_repo(server_config, repo_id):
+    """Synchronize the given repo."""
+    return cli.Client(server_config, cli.echo_handler).run(
+        'pulp-admin docker repo sync run --repo-id {}'.format(repo_id).split()
+    )
+
+
+class BaseTestCase(unittest2.TestCase):
+    """A Base class for testing Docker content. It logs in for you."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Provide a server config and a repository ID."""
+        cls.cfg = config.get_config()
+        cls.repo_id = utils.uuid4()
+        cli.Client(cls.cfg).run(
+            'pulp-admin login -u {} -p {}'
+            .format(cls.cfg.auth[0], cls.cfg.auth[1]).split()
+        )
+
+    @classmethod
+    def tearDownClass(cls):
+        """Delete the created repository."""
+        command = 'pulp-admin docker repo delete --repo-id {}'
+        cli.Client(cls.cfg).run(command.format(cls.repo_id).split())
+
+
+class SuccessMixin(object):
+    """Add some common assertion to Test cases."""
+
+    def test_return_code(self):
+        """Assert the "sync" command has a return code of 0."""
+        self.assertEqual(self.completed_proc.returncode, 0)
+
+    def test_task_succeeded(self):
+        """Assert the phrase "Task Succeeded" is in stdout."""
+        self.assertIn('Task Succeeded', self.completed_proc.stdout)
+
+    def test_task_failed(self):
+        """Assert the phrase "Task Failed" is not in stdout."""
+        self.assertNotIn('Task Failed', self.completed_proc.stdout)