Adding mirror true check for rpm/file plugin

This commit will test cross plugin scenario. If the user syncs multiple
plugin contents into the same repo with the option mirror=True, the
existing content should be overwritten.

closes #4448

pulpcore/tests/functional/api/using_plugin/constants.py

@@ -3,38 +3,32 @@
 from urllib.parse import urljoin
 
 from pulp_smash.constants import PULP_FIXTURES_BASE_URL
-from pulp_smash.pulp3.constants import (
-    BASE_PUBLISHER_PATH,
-    BASE_REMOTE_PATH,
-    CONTENT_PATH
-)
-
-
-FILE_CONTENT_NAME = 'file.file'
-
-FILE_CONTENT_PATH = urljoin(CONTENT_PATH, 'file/files/')
-
-FILE_REMOTE_PATH = urljoin(BASE_REMOTE_PATH, 'file/file/')
-
-FILE_PUBLISHER_PATH = urljoin(BASE_PUBLISHER_PATH, 'file/file/')
-
-FILE_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'file/')
-"""The URL to a file repository."""
-
-FILE_FIXTURE_MANIFEST_URL = urljoin(FILE_FIXTURE_URL, 'PULP_MANIFEST')
-"""The URL to a file repository manifest."""
-
-FILE_FIXTURE_COUNT = 3
-"""The number of packages available at :data:`FILE_FIXTURE_URL`."""
-
-FILE_FIXTURE_SUMMARY = {FILE_CONTENT_NAME: FILE_FIXTURE_COUNT}
-"""The desired content summary after syncing :data:`FILE_FIXTURE_URL`."""
-
-FILE2_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'file2/')
-"""The URL to a file repository."""
 
-FILE2_FIXTURE_MANIFEST_URL = urljoin(FILE2_FIXTURE_URL, 'PULP_MANIFEST')
-"""The URL to a file repository manifest"""
+from pulp_file.tests.functional.constants import ( # noqa: F401
+    FILE2_FIXTURE_MANIFEST_URL,
+    FILE2_FIXTURE_URL,
+    FILE2_URL,
+    FILE_CONTENT_NAME,
+    FILE_CONTENT_PATH,
+    FILE_FIXTURE_COUNT,
+    FILE_FIXTURE_MANIFEST_URL,
+    FILE_FIXTURE_SUMMARY,
+    FILE_FIXTURE_URL,
+    FILE_LARGE_FIXTURE_MANIFEST_URL,
+    FILE_LARGE_FIXTURE_URL,
+    FILE_PUBLISHER_PATH,
+    FILE_REMOTE_PATH,
+    FILE_URL,
+)
+from pulp_rpm.tests.functional.constants import ( # noqa: F401
+    RPM_UNSIGNED_FIXTURE_URL,
+    RPM_REMOTE_PATH,
+    RPM_PACKAGE_CONTENT_NAME,
+    RPM_UPDATE_CONTENT_NAME,
+    RPM_PACKAGES_COUNT,
+    RPM_UPDATE_COUNT,
+    RPM_FIXTURE_SUMMARY,
+)
 
 FILE_MANY_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'file-many/')
 """The URL to a file repository containing many files."""
@@ -48,20 +42,6 @@
 FILE_MANY_FIXTURE_COUNT = 250
 """The number of packages available at :data:`FILE_MANY_FIXTURE_URL`."""
 
-FILE_LARGE_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, 'file-large/')
-"""The URL to a file repository containing a large number of files."""
 
 FILE_LARGE_FIXTURE_COUNT = 10
 """The number of packages available at :data:`FILE_LARGE_FIXTURE_URL`."""
-
-FILE_LARGE_FIXTURE_MANIFEST_URL = urljoin(
-    FILE_LARGE_FIXTURE_URL,
-    'PULP_MANIFEST'
-)
-"""The URL to a file repository manifest."""
-
-FILE_URL = urljoin(FILE_FIXTURE_URL, '1.iso')
-"""The URL to an ISO file at :data:`FILE_FIXTURE_URL`."""
-
-FILE2_URL = urljoin(FILE2_FIXTURE_URL, '1.iso')
-"""The URL to an ISO file at :data:`FILE2_FIXTURE_URL`."""

pulpcore/tests/functional/api/using_plugin/test_sync_multiple_plugins.py

@@ -0,0 +1,115 @@
+# coding=utf-8
+"""Tests related to multiple plugins."""
+import os
+import unittest
+from unittest import SkipTest
+
+from pulp_smash import api, config
+from pulp_smash.pulp3.constants import REPO_PATH
+from pulp_smash.pulp3.utils import (
+    gen_remote,
+    gen_repo,
+    get_added_content_summary,
+    get_content_summary,
+    get_removed_content_summary,
+    require_pulp_plugins,
+    sync,
+)
+
+from pulpcore.tests.functional.api.using_plugin.constants import (
+    FILE_FIXTURE_MANIFEST_URL,
+    FILE_FIXTURE_SUMMARY,
+    FILE_REMOTE_PATH,
+    RPM_FIXTURE_SUMMARY,
+    RPM_REMOTE_PATH,
+    RPM_UNSIGNED_FIXTURE_URL,
+)
+from pulpcore.tests.functional.utils import skip_if
+from pulpcore.tests.functional.api.using_plugin.utils import set_up_module # noqa
+
+
+def setUpModule():
+    """Conditions to skip tests.
+
+    Skip tests if not testing Pulp 3, or if either pulpcore, pulp_file
+    or pulp_rpm aren't installed.
+
+    refer :meth:`pulpcore.tests.functional.api.using_plugin.utils.set_up_module`
+    """
+    set_up_module()
+    require_pulp_plugins('pulp_rpm', SkipTest)
+
+
+class SyncMultiplePlugins(unittest.TestCase):
+    """Sync repositories with the multiple plugins in the same repo."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Create class-wide variables."""
+        cls.cfg = config.get_config()
+        cls.client = api.Client(cls.cfg, api.json_handler)
+        cls.travis = 'TRAVIS' in os.environ
+
+    @skip_if(bool, 'travis', True)
+    def test_mirror_sync(self):
+        """Sync multiple plugin into the same repo with mirror as `True`.
+
+        This test targets the following issue: 4448
+
+        * `<https://pulp.plan.io/issues/4448>`_
+
+        This test does the following:
+
+        1. Create a repo.
+        2. Create two remotes
+            a. RPM remote
+            b. File remote
+        3. Sync the repo with RPM remote.
+        4. Sync the repo with File remote with ``Mirror=True``.
+        5. Verify whether the content in the latest version of the repo
+           has only File content and RPM content is deleted.
+        """
+        # Step 1
+        repo = self.client.post(REPO_PATH, gen_repo())
+        self.addCleanup(self.client.delete, repo['_href'])
+
+        # Step 2
+        rpm_remote = self.client.post(
+            RPM_REMOTE_PATH,
+            gen_remote(url=RPM_UNSIGNED_FIXTURE_URL)
+        )
+        self.addCleanup(self.client.delete, rpm_remote['_href'])
+
+        file_remote = self.client.post(
+            FILE_REMOTE_PATH,
+            gen_remote(url=FILE_FIXTURE_MANIFEST_URL)
+        )
+        self.addCleanup(self.client.delete, file_remote['_href'])
+
+        # Step 3
+        sync(self.cfg, rpm_remote, repo)
+        repo = self.client.get(repo['_href'])
+        self.assertIsNotNone(repo['_latest_version_href'])
+        self.assertDictEqual(
+            get_added_content_summary(repo),
+            RPM_FIXTURE_SUMMARY
+        )
+
+        # Step 4
+        sync(self.cfg, file_remote, repo, mirror=True)
+        repo = self.client.get(repo['_href'])
+        self.assertIsNotNone(repo['_latest_version_href'])
+        self.assertDictEqual(
+            get_added_content_summary(repo),
+            FILE_FIXTURE_SUMMARY
+        )
+
+        # Step 5
+        self.assertDictEqual(
+            get_content_summary(repo),
+            FILE_FIXTURE_SUMMARY
+        )
+        self.assertDictEqual(
+            get_removed_content_summary(repo),
+            RPM_FIXTURE_SUMMARY
+        )