Remove customized operation_id from OrphansView

CHANGES/plugin_api/7446.bugfix

@@ -0,0 +1 @@
+Remove customized operation_id from OrphansView

pulpcore/app/views/orphans.py

@@ -12,7 +12,6 @@ class OrphansView(APIView):
         description="Trigger an asynchronous task that deletes all"
         "orphaned content and artifacts.",
         summary="Delete orphans",
-        operation_id="orphans_delete",
         responses={202: AsyncOperationResponseSerializer},
     )
     def delete(self, request, format=None):

pulpcore/openapi/__init__.py

@@ -55,6 +55,9 @@ def _tokenize_path(self):
             if not tokenized_path and getattr(self.view, "get_view_name", None):
                 tokenized_path.extend(self.view.get_view_name().split())
 
+        path = "/".join(tokenized_path).replace("pulp/api/v3/", "")
+        tokenized_path = path.split("/")
+
         return tokenized_path
 
     def get_tags(self):

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

@@ -4,28 +4,29 @@
 import unittest
 from django.conf import settings
 from random import choice
-from requests.exceptions import HTTPError
 
-from pulp_smash import api, cli, config, utils
+from pulp_smash import cli, config
 from pulp_smash.exceptions import CalledProcessError
-from pulp_smash.pulp3.constants import ARTIFACTS_PATH, MEDIA_PATH
+from pulp_smash.pulp3.bindings import monitor_task
+from pulp_smash.pulp3.constants import MEDIA_PATH
 from pulp_smash.pulp3.utils import (
-    delete_orphans,
     delete_version,
     gen_repo,
     get_content,
     get_versions,
-    modify_repo,
-    sync,
 )
 
-from pulpcore.tests.functional.api.using_plugin.constants import (
-    FILE2_URL,
-    FILE_CONTENT_NAME,
-    FILE_CONTENT_PATH,
-    FILE_REMOTE_PATH,
-    FILE_REPO_PATH,
+from pulpcore.tests.functional.api.using_plugin.constants import FILE_CONTENT_NAME
+from pulpcore.client.pulpcore import ArtifactsApi
+from pulpcore.client.pulpcore.exceptions import ApiException
+from pulpcore.client.pulp_file import (
+    ApiClient,
+    ContentFilesApi,
+    RepositoriesFileApi,
+    RepositorySyncURL,
+    RemotesFileApi,
 )
+from pulpcore.tests.functional.utils import configuration, core_client, delete_orphans
 from pulpcore.tests.functional.api.using_plugin.utils import gen_file_remote
 from pulpcore.tests.functional.api.using_plugin.utils import (  # noqa:F401
     set_up_module as setUpModule,
@@ -49,7 +50,7 @@ class DeleteOrphansTestCase(unittest.TestCase):
     def setUpClass(cls):
         """Create class-wide variables."""
         cls.cfg = config.get_config()
-        cls.api_client = api.Client(cls.cfg, api.json_handler)
+        cls.api_client = ApiClient(configuration)
         cls.cli_client = cli.Client(cls.cfg)
 
     def test_clean_orphan_content_unit(self):
@@ -66,37 +67,47 @@ def test_clean_orphan_content_unit(self):
         5. Assert that the orphan content unit was cleaned up, and its artifact
            is not present on disk.
         """
-        repo = self.api_client.post(FILE_REPO_PATH, gen_repo())
-        self.addCleanup(self.api_client.delete, repo["pulp_href"])
+        repo_api = RepositoriesFileApi(self.api_client)
+        remote_api = RemotesFileApi(self.api_client)
+
+        repo = repo_api.create(gen_repo())
+        self.addCleanup(repo_api.delete, repo.pulp_href)
 
         body = gen_file_remote()
-        remote = self.api_client.post(FILE_REMOTE_PATH, body)
-        self.addCleanup(self.api_client.delete, remote["pulp_href"])
+        remote = remote_api.create(body)
+        self.addCleanup(remote_api.delete, remote.pulp_href)
 
-        sync(self.cfg, remote, repo)
-        repo = self.api_client.get(repo["pulp_href"])
-        content = choice(get_content(repo)[FILE_CONTENT_NAME])
+        # Sync the repository.
+        self.assertEqual(repo.latest_version_href, f"{repo.pulp_href}versions/0/")
+        repository_sync_data = RepositorySyncURL(remote=remote.pulp_href)
+        sync_response = repo_api.sync(repo.pulp_href, repository_sync_data)
+        monitor_task(sync_response.task)
+        repo = repo_api.read(repo.pulp_href)
+        content = choice(get_content(repo.to_dict())[FILE_CONTENT_NAME])
 
         # Create an orphan content unit.
-        modify_repo(self.cfg, repo, remove_units=[content])
+        repo_api.modify(repo.pulp_href, dict(remove_content_units=[content["pulp_href"]]))
+
+        artifacts_api = ArtifactsApi(core_client)
 
         if settings.DEFAULT_FILE_STORAGE == "pulpcore.app.models.storage.FileSystem":
             # Verify that the artifact is present on disk.
-            artifact_path = os.path.join(
-                MEDIA_PATH, self.api_client.get(content["artifact"])["file"]
-            )
+            artifact_path = os.path.join(MEDIA_PATH, artifacts_api.read(content["artifact"]).file)
             cmd = ("ls", artifact_path)
             self.cli_client.run(cmd, sudo=True)
 
+        file_contents_api = ContentFilesApi(self.api_client)
         # Delete first repo version. The previous removed content unit will be
         # an orphan.
-        delete_version(repo, get_versions(repo)[1]["pulp_href"])
-        content_units = self.api_client.get(FILE_CONTENT_PATH)["results"]
-        self.assertIn(content, content_units)
+        delete_version(repo, get_versions(repo.to_dict())[1]["pulp_href"])
+        content_units = file_contents_api.list().to_dict()["results"]
+        content_units_href = [c["pulp_href"] for c in content_units]
+        self.assertIn(content["pulp_href"], content_units_href)
 
         delete_orphans()
-        content_units = self.api_client.get(FILE_CONTENT_PATH)["results"]
-        self.assertNotIn(content, content_units)
+        content_units = file_contents_api.list().to_dict()["results"]
+        content_units_href = [c["pulp_href"] for c in content_units]
+        self.assertNotIn(content["pulp_href"], content_units_href)
 
         if settings.DEFAULT_FILE_STORAGE == "pulpcore.app.models.storage.FileSystem":
             # Verify that the artifact was removed from disk.
@@ -105,20 +116,21 @@ def test_clean_orphan_content_unit(self):
 
     def test_clean_orphan_artifact(self):
         """Test whether orphan artifacts units can be clean up."""
-        repo = self.api_client.post(FILE_REPO_PATH, gen_repo())
-        self.addCleanup(self.api_client.delete, repo["pulp_href"])
+        repo_api = RepositoriesFileApi(self.api_client)
+        repo = repo_api.create(gen_repo())
+        self.addCleanup(repo_api.delete, repo.pulp_href)
 
-        files = {"file": utils.http_get(FILE2_URL)}
-        artifact = self.api_client.post(ARTIFACTS_PATH, files=files)
+        artifacts_api = ArtifactsApi(core_client)
+        artifact = artifacts_api.create(file=__file__)
 
         if settings.DEFAULT_FILE_STORAGE == "pulpcore.app.models.storage.FileSystem":
-            cmd = ("ls", os.path.join(MEDIA_PATH, artifact["file"]))
+            cmd = ("ls", os.path.join(MEDIA_PATH, artifact.file))
             self.cli_client.run(cmd, sudo=True)
 
         delete_orphans()
 
-        with self.assertRaises(HTTPError):
-            self.api_client.get(artifact["pulp_href"])
+        with self.assertRaises(ApiException):
+            artifacts_api.read(artifact.pulp_href)
 
         if settings.DEFAULT_FILE_STORAGE == "pulpcore.app.models.storage.FileSystem":
             with self.assertRaises(CalledProcessError):

pulpcore/tests/functional/utils.py

@@ -4,19 +4,18 @@
 from time import sleep
 from unittest import SkipTest
 
-from pulp_smash import selectors
+from pulp_smash import config, selectors
+from pulp_smash.pulp3.bindings import monitor_task
 from pulp_smash.pulp3.utils import require_pulp_3, require_pulp_plugins
 from pulpcore.client.pulpcore import (
     ApiClient,
-    Configuration,
+    OrphansApi,
     TaskGroupsApi,
 )
 
 
-configuration = Configuration()
-configuration.username = "admin"
-configuration.password = "password"
-configuration.safe_chars_for_path_param = "/"
+cfg = config.get_config()
+configuration = cfg.get_bindings_config()
 
 
 def set_up_module():
@@ -55,3 +54,9 @@ def monitor_task_group(tg_href):
     else:
         print("The task group was successful.")
         return tg
+
+
+def delete_orphans():
+    """Delete orphans through bindings."""
+    response = OrphansApi(core_client).delete()
+    monitor_task(response.task)