diff --git a/docs/admin/reference/settings.md b/docs/admin/reference/settings.md index c7c974d83b8..8e0e958c744 100644 --- a/docs/admin/reference/settings.md +++ b/docs/admin/reference/settings.md @@ -421,6 +421,26 @@ A JSON object representing the security scheme advertised for the `PulpRemoteUse Defaults to `{"type": "mutualTLS"}`, which represents x509 certificate based authentication. +### STAGING\_AUTOCOMMIT\_DELAY\_SECONDS + +Time in seconds from the first staged content until a repository with autocommit enabled becomes +eligible for a commit. +The scheduler's scan interval is separate; this setting does not control how often repositories +are checked. + +Defaults to `300` seconds. + +### STAGING\_AUTOCOMMIT\_SCAN\_INTERVAL\_SECONDS + +How often, in seconds, the scheduled task checks for repositories whose autocommit deadline has +passed and dispatches their commit tasks. +This is separate from `STAGING_AUTOCOMMIT_DELAY_SECONDS`, which determines when staged content first +becomes eligible for a commit. +A repository may wait almost one scan interval after its deadline before the scanner sees it. +Worker scheduling, task pickup, and waiting for the repository lock can add more time. + +Defaults to `60` seconds. + ### TASK\_DIAGNOSTICS When enabled, users are allowed to request various diagnostics for analysis by a developer. diff --git a/docs/user/guides/stage-repository-content.md b/docs/user/guides/stage-repository-content.md new file mode 100644 index 00000000000..3f93e66ba0d --- /dev/null +++ b/docs/user/guides/stage-repository-content.md @@ -0,0 +1,102 @@ +# Manage Repository Staging + +A repository staging area holds content associated with a repository before it belongs to a repository version. + +Staged content is not served by a distribution and does not trigger checks and operations that creating a repository version would. +Because it doesn't have those checks, it can be used to avoid repository version creation overhead. + +You can inspect the staging area, commit its content into a new repository version, or discard it. + +These examples use the file plugin, but staging works with repositories from any plugin. +Replace `/repositories/file/file/` with your plugin's repository endpoint. + +The commands use HTTPie and the `admin` user. +Set `PULP_API_URL` to the API base URL, including its port and version, without a trailing slash. +Set `PULP_PASSWORD` to the `admin` password and `REPOSITORY_NAME` to the name of the new repository. + +## Enable Staging + +Set `staging_enabled` when creating a repository. +It defaults to `false`. + +```bash +http --auth "admin:$PULP_PASSWORD" POST "$PULP_API_URL/repositories/file/file/" \ + name="$REPOSITORY_NAME" staging_enabled:=true +``` + +The staging endpoints return `404` while staging is disabled for the repository. +Set `REPOSITORY_ID` to the ID in the returned `pulp_href` before using the following commands. + +## Commit Staged Content Automatically + +Set `staging_autocommit` on a repository with staging enabled to commit staged content on a schedule. +It defaults to `false`. + +```bash +http --auth "admin:$PULP_PASSWORD" PATCH \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/" \ + staging_autocommit:=true +``` + +The first content staged after autocommit is enabled starts the delay set by +[`STAGING_AUTOCOMMIT_DELAY_SECONDS`](site:pulpcore/docs/admin/reference/settings/#staging_autocommit_delay_seconds). +Staging more content does not postpone that deadline. +A separate scheduled task checks for expired deadlines at the interval set by +[`STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS`](site:pulpcore/docs/admin/reference/settings/#staging_autocommit_scan_interval_seconds) +and dispatches a commit task for each ready repository. +The commit task may wait for other tasks holding the repository lock. + +Autocommit only affects content already placed in staging; enabling it does not change the file +upload endpoint. +A manual commit or restore clears the pending deadline. +If a commit fails, the staged content and deadline remain available for another attempt. + +## Inspect Staged Content + +Use the repository's `pulp_href` followed by `staging/` to list staged content URLs. +The list is paginated with `limit` and `offset`. +An empty list returns `{"count": 0, "next": null, "previous": null, "results": []}`. +`HEAD` on `staging/` returns `200` when staging is enabled. + +To retrieve one staged content unit by path, append its `relative_path` to `staging/`. +Set `RELATIVE_PATH` to the staged content unit's relative path before using the path commands. + +```bash +# List staged content URLs. +http --auth "admin:$PULP_PASSWORD" GET \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/" \ + limit==20 offset==0 + +# Check whether staging is enabled. +http --auth "admin:$PULP_PASSWORD" HEAD \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/" + +# Retrieve content at a relative path, or check whether it exists. +http --auth "admin:$PULP_PASSWORD" GET \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/${RELATIVE_PATH}/" +http --auth "admin:$PULP_PASSWORD" HEAD \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/${RELATIVE_PATH}/" +``` + +The response contains the content unit's `pulp_href` and plugin-specific fields. +Nested paths remain part of `relative_path`; URL-encode reserved characters. +`GET` and `HEAD` return `404` when the path is not staged. + +## Commit or Discard Staged Content + +Commit staged content to create a new repository version when the content changes, or discard all staged content without creating a version: + +```bash +# Commit staged content. +http --auth "admin:$PULP_PASSWORD" POST \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/?action=commit" + +# Discard staged content. +http --auth "admin:$PULP_PASSWORD" POST \ + "$PULP_API_URL/repositories/file/file/${REPOSITORY_ID}/staging/?action=restore" +``` + +If staging contains content, either action returns `202 Accepted` with a task URL in the `task` field. + +Wait for that task to complete before checking the repository version or staging area. +If staging is empty, either action returns `204 No Content` and creates no task. diff --git a/docs/user/guides/upload-publish.md b/docs/user/guides/upload-publish.md index 2b8042e9c81..fddb2c581cc 100644 --- a/docs/user/guides/upload-publish.md +++ b/docs/user/guides/upload-publish.md @@ -16,6 +16,8 @@ the content from. You can automatically associate newly uploaded content with a Repository when using the `/pulp/api/v3/content///` API by passing the `repository reference`. +For the repository staging API, see [Manage Repository Staging](site:pulpcore/docs/user/guides/stage-repository-content/). + ## Chunked Uploads For large file uploads, Pulp provides an [Uploads API](site:pulpcore/restapi/#tag/Uploads). To begin diff --git a/pulp_file/app/viewsets.py b/pulp_file/app/viewsets.py index 96c8bb47b30..3ad851fe8a8 100644 --- a/pulp_file/app/viewsets.py +++ b/pulp_file/app/viewsets.py @@ -207,7 +207,7 @@ class FileRepositoryViewSet(RepositoryViewSet, ModifyRepositoryActionMixin, Role ], }, { - "action": ["modify"], + "action": ["modify", "staging_post"], "principal": "authenticated", "effect": "allow", "condition": [ @@ -215,6 +215,12 @@ class FileRepositoryViewSet(RepositoryViewSet, ModifyRepositoryActionMixin, Role "has_model_or_domain_or_obj_perms:file.view_filerepository", ], }, + { + "action": ["staging", "staging_head", "staging_path", "staging_path_head"], + "principal": "authenticated", + "effect": "allow", + "condition": ["has_model_or_domain_or_obj_perms:file.view_filerepository"], + }, { "action": ["list_roles", "add_role", "remove_role"], "principal": "authenticated", diff --git a/pulp_file/tests/unit/test_repository_api.py b/pulp_file/tests/unit/test_repository_api.py new file mode 100644 index 00000000000..c1d3f9a957a --- /dev/null +++ b/pulp_file/tests/unit/test_repository_api.py @@ -0,0 +1,484 @@ +import json +from datetime import datetime, timedelta +from secrets import token_hex +from urllib.parse import urlencode, urljoin, urlsplit + +import pytest +from django.test import override_settings +from django.utils import timezone +from django_guid import clear_guid, set_guid +from rest_framework import status +from rest_framework.test import APIClient + +from pulpcore.app.contexts import with_user +from pulpcore.app.models import RepositoryVersion, StagedContent, Task, TaskSchedule +from pulpcore.app.tasks import repository as repository_tasks +from pulpcore.app.util import configure_staging_autocommit, get_prn, resolve, reverse + +from pulp_file.app.models import FileContent, FileRepository +from pulp_file.app.viewsets import FileContentViewSet, FileRepositoryViewSet + +AUTOCOMMIT_DELAY_SECONDS = 300 + + +def update_repository(repository: FileRepository, *, autocommit: bool = False) -> None: + repository.staging_enabled = True + repository.staging_autocommit = autocommit + repository.save(update_fields=["staging_enabled", "staging_autocommit"]) + + +def create_file_content(relative_path: str, digest: str | None = None) -> FileContent: + return FileContent.objects.create( + relative_path=relative_path, + digest=digest if digest is not None else token_hex(32), + ) + + +def get_detail_url(viewset, pk, trailing_path: str = "", query_params=None) -> str: + """Return a viewset detail url with an optional trailing path and query parameters. + + Examples: + FileRepositoryViewSet -> repositories/file/file/{pk}/ + FileContentViewSet -> content/file/file/{pk}/ + """ + url_name = viewset.view_name() + detail_pattern = "{basename}-detail" + url = urljoin( + reverse(detail_pattern.format(basename=url_name), kwargs={"pk": pk}), trailing_path + ) + return urlsplit(url)._replace(query=urlencode(query_params or {})).geturl() + + +def get_list_url(viewset, trailing_path: str = "") -> str: + """Return a viewset list url with an optional trailing path. + + Examples: + FileRepositoryViewSet -> repositories/file/file/ + FileContentViewSet -> content/file/file/ + """ + url_name = viewset.view_name() + detail_pattern = "{basename}-list" + return urljoin(reverse(detail_pattern.format(basename=url_name)), trailing_path) + + +@pytest.fixture +def repository(db): + with with_user(None): + return FileRepository.objects.create(name="file-repository-test") + + +@pytest.fixture +def api_client(db, django_user_model): + user = django_user_model.objects.create_superuser( + username="file-repository-test-admin", + email="admin@example.com", + password="test-password", + ) + client = APIClient() + client.force_authenticate(user=user) + return client + + +class TestRepositoryEndpoint: + def test_detail_exposes_default_staging_fields(self, api_client, repository): + detail_url = get_detail_url(FileRepositoryViewSet, repository.pk) + response = api_client.get(detail_url) + + assert response.status_code == status.HTTP_200_OK + assert response.data["staging_enabled"] is False + assert response.data["staging_autocommit"] is False + assert "next_commit" not in response.data + + def test_list_exposes_staging_fields(self, api_client, repository): + update_repository(repository, autocommit=True) + + list_url = get_list_url(FileRepositoryViewSet) + response = api_client.get(list_url) + + assert response.status_code == status.HTTP_200_OK + assert len(response.data["results"]) == 1 + assert response.data["results"][0]["staging_enabled"] is True + assert response.data["results"][0]["staging_autocommit"] is True + assert "next_commit" not in response.data["results"][0] + + +class TestRepositoryNextCommit: + @override_settings(STAGING_AUTOCOMMIT_DELAY_SECONDS=AUTOCOMMIT_DELAY_SECONDS) + def test_staging_sets_deadline_only_after_autocommit_enabled(self, repository): + update_repository(repository) + assert repository.next_commit is None + + # Staging alone leaves the deadline unset + repository.stage_file(create_file_content("without-autocommit.txt")) + repository.refresh_from_db() + assert repository.next_commit is None + + # The first staged file after enabling autocommit starts the delay + update_repository(repository, autocommit=True) + + t_before = timezone.now() + timedelta(seconds=AUTOCOMMIT_DELAY_SECONDS) + repository.stage_file(create_file_content("with-autocommit.txt")) + t_after = timezone.now() + timedelta(seconds=AUTOCOMMIT_DELAY_SECONDS) + + repository.refresh_from_db() + assert t_before <= repository.next_commit <= t_after + + def test_later_staged_file_keeps_first_deadline(self, repository): + update_repository(repository, autocommit=True) + + repository.stage_file(create_file_content("first.txt")) + repository.refresh_from_db() + first_deadline = repository.next_commit + assert first_deadline is not None + + # Later files do not postpone the pending commit + repository.stage_file(create_file_content("second.txt")) + repository.refresh_from_db() + + assert repository.next_commit == first_deadline + + +class TestRepositoryStagingEndpoint: + @pytest.mark.parametrize( + ("method", "query_params"), + [ + ("get", {"relative_path": ""}), + ("head", {"relative_path": ""}), + ("get", {"relative_path": "nested/file.txt"}), + ("head", {"relative_path": "nested/file.txt"}), + ("get", {"path": "old-name.txt"}), + ("head", {"path": "old-name.txt"}), + ("post", {}), + ("post", {"action": "invalid"}), + ("post", {"action": "clear"}), + ], + ) + def test_invalid_query_params_returns_400(self, api_client, repository, method, query_params): + """The methods should only accept valid query params.""" + update_repository(repository) + url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/", query_params) + request = getattr(api_client, method) + + response = request(url) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + + @pytest.mark.parametrize( + ("method", "trailing_path", "query_params"), + [ + ("get", "staging/", {}), + ("head", "staging/", {}), + ("get", "staging/missing.txt/", {}), + ("head", "staging/missing.txt/", {}), + ("post", "staging/", {"action": "commit"}), + ("post", "staging/", {"action": "restore"}), + ], + ) + def test_staging_disabled_returns_404( + self, api_client, repository, method, trailing_path, query_params + ): + """If repository has staging disabled, no action can be taken.""" + staging_url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/") + + # sanity check the staging_url exists + assert resolve(staging_url) + + url = get_detail_url(FileRepositoryViewSet, repository.pk, trailing_path, query_params) + request = getattr(api_client, method) + response = request(url) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_staging_list_returns_paginated_content_urls(self, api_client, repository): + update_repository(repository) + staging_url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/") + + # Empty staging retains the pagination envelope + empty_response = api_client.get(staging_url) + assert empty_response.status_code == status.HTTP_200_OK + assert empty_response.data == { + "count": 0, + "next": None, + "previous": None, + "results": [], + } + + first = create_file_content("first.txt") + second = create_file_content("nested/second.txt") + repository.stage_file(first) + repository.stage_file(second) + + response = api_client.get(staging_url) + + assert response.status_code == status.HTTP_200_OK + assert set(response.data) == {"count", "next", "previous", "results"} + assert response.data["count"] == 2 + + # Results are ordered by content ID + expected_urls = [ + get_detail_url(FileContentViewSet, content.pk) + for content in sorted((first, second), key=lambda content: content.pk) + ] + assert response.data["results"] == expected_urls + assert api_client.head(staging_url).status_code == status.HTTP_200_OK + + second_page_url = get_detail_url( + FileRepositoryViewSet, repository.pk, "staging/", {"limit": "1", "offset": "1"} + ) + second_page = api_client.get(second_page_url) + assert second_page.status_code == status.HTTP_200_OK + assert set(second_page.data) == {"count", "next", "previous", "results"} + + # Count covers the whole staging area, even on the second page + assert second_page.data["count"] == 2 + assert second_page.data["results"] == expected_urls[1:] + assert api_client.head(second_page_url).status_code == status.HTTP_200_OK + + def test_staging_path_get_and_head(self, api_client, repository): + update_repository(repository) + content = create_file_content("nested/file.txt") + repository.stage_file(content) + found_url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/nested/file.txt/") + + response = api_client.get(found_url) + + # Nested path segments are part of relative_path + assert response.status_code == status.HTTP_200_OK + assert response.data["pulp_href"] == get_detail_url(FileContentViewSet, content.pk) + assert response.data["relative_path"] == content.relative_path + assert api_client.head(found_url).status_code == status.HTTP_200_OK + + # Both methods report an unstaged path as missing + missing_url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/missing.txt/") + assert api_client.get(missing_url).status_code == status.HTTP_404_NOT_FOUND + assert api_client.head(missing_url).status_code == status.HTTP_404_NOT_FOUND + + @pytest.mark.parametrize("action", ["commit", "restore"]) + def test_commit_and_restore_return_204_for_empty_staging(self, api_client, repository, action): + update_repository(repository) + url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/", {"action": action}) + task_count = Task.objects.count() + + response = api_client.post(url) + + assert response.status_code == status.HTTP_204_NO_CONTENT + assert response.data is None + assert Task.objects.count() == task_count + + @pytest.mark.parametrize("action", ["commit", "restore"]) + def test_commit_and_restore_return_tasks_for_staged_content( + self, api_client, repository, action + ): + task_names = { + "commit": "pulpcore.app.tasks.repository.commit_staging", + "restore": "pulpcore.app.tasks.repository.restore_staging", + } + update_repository(repository) + content = create_file_content("file.txt") + repository.stage_file(content) + url = get_detail_url(FileRepositoryViewSet, repository.pk, "staging/", {"action": action}) + + response = api_client.post(url) + + # Staging actions return tasks + assert response.status_code == status.HTTP_202_ACCEPTED + task_pk = resolve(response.data["task"]).kwargs["pk"] + task = Task.objects.get(pk=task_pk) + assert response.data == {"task": reverse("tasks-detail", kwargs={"pk": task.pk})} + assert task.name == task_names[action] + assert task.enc_kwargs == {"repository_pk": str(repository.pk)} + + # Staging actions lock on the repository + repository_prn = get_prn(repository) + assert repository_prn in task.reserved_resources_record + assert f"shared:{repository_prn}" not in task.reserved_resources_record + + +class TestRepositoryStagingTasks: + def test_commit_creates_version_and_clears_staging_and_deadline(self, repository): + update_repository(repository, autocommit=True) + + content = create_file_content("file.txt") + repository.stage_file(content) + repository.refresh_from_db() + first_deadline = repository.next_commit + assert first_deadline is not None + + repository_tasks.commit_staging(repository.pk) + + # Commit clears staging and the deadline, then creates a version + repository.refresh_from_db() + assert repository.latest_version().content.filter(pk=content.pk).exists() + assert not StagedContent.objects.filter(repository=repository).exists() + assert repository.next_commit is None + + # New staged content starts a new deadline + repository.stage_file(create_file_content("later.txt")) + repository.refresh_from_db() + assert repository.next_commit > first_deadline + + def test_restore_clears_staging_and_deadline_without_new_version(self, repository): + update_repository(repository, autocommit=True) + + initial_version = repository.latest_version() + content = create_file_content("file.txt") + repository.stage_file(content) + repository.refresh_from_db() + first_deadline = repository.next_commit + assert first_deadline is not None + + repository_tasks.restore_staging(repository.pk) + + # Restore clears staging and the deadline without creating a version + repository.refresh_from_db() + assert repository.latest_version().pk == initial_version.pk + assert not StagedContent.objects.filter(repository=repository).exists() + assert repository.next_commit is None + + # New staged content starts a new deadline + repository.stage_file(create_file_content("later.txt")) + repository.refresh_from_db() + assert repository.next_commit > first_deadline + + def test_failed_commit_preserves_staging_and_deadline(self, repository, monkeypatch): + update_repository(repository, autocommit=True) + + content = create_file_content("file.txt") + repository.stage_file(content) + repository.refresh_from_db() + deadline = repository.next_commit + assert deadline is not None + + def fail_add_content(*args, **kwargs): + raise RuntimeError("simulated commit failure") + + monkeypatch.setattr(RepositoryVersion, "add_content", fail_add_content) + + with pytest.raises(RuntimeError, match="simulated commit failure"): + repository_tasks.commit_staging(repository.pk) + + # Failed commits leave the staged content and deadline available for retry + repository.refresh_from_db() + assert repository.next_commit == deadline + assert StagedContent.objects.filter(repository=repository, content=content).exists() + + +class TestRepositoryAutocommitScheduler: + @staticmethod + def create_repo_and_stage_content( + next_commit: datetime | None, *, autocommit: bool = True + ) -> FileRepository: + with with_user(None): + repository = FileRepository.objects.create(name=f"autocommit-{token_hex(8)}") + update_repository(repository, autocommit=autocommit) + repository.stage_file(create_file_content(f"{repository.name}.txt")) + repository.next_commit = next_commit + repository.save(update_fields=["next_commit"]) + return repository + + def test_dispatches_due_repositories_with_serializable_task_ids(self, db): + now = timezone.now() + t_past = now - timedelta(seconds=1) + t_far_future = now + timedelta(hours=1) + + # These will trigger autocommit tasks + due_repositories = ( + self.create_repo_and_stage_content(next_commit=t_past), + self.create_repo_and_stage_content(next_commit=t_past), + ) + # These won't trigger autocommit tasks + self.create_repo_and_stage_content(next_commit=t_far_future) + self.create_repo_and_stage_content(next_commit=t_past, autocommit=False) + self.create_repo_and_stage_content(next_commit=None) + + with with_user(None): + set_guid(token_hex(16)) + try: + task_ids = repository_tasks.dispatch_staging_autocommits() + repeated_task_ids = repository_tasks.dispatch_staging_autocommits() + finally: + clear_guid() + + # Only due repositories with autocommit enabled receive commit tasks + tasks = list(Task.objects.filter(name="pulpcore.app.tasks.repository.commit_staging")) + assert len(tasks) == 2 + assert repeated_task_ids == [] + assert {task.enc_kwargs["repository_pk"] for task in tasks} == { + str(due.pk) for due in due_repositories + } + + # Task results contain serializable identifiers for the dispatched tasks + assert isinstance(task_ids, list) + assert len(task_ids) == len(tasks) + assert set(task_ids) == {str(task.pk) for task in tasks} + assert json.loads(json.dumps(task_ids)) == task_ids + + # Each commit task reserves its repository exclusively + for task in tasks: + due = next( + due for due in due_repositories if str(due.pk) == task.enc_kwargs["repository_pk"] + ) + repository_prn = get_prn(due) + assert repository_prn in task.reserved_resources_record + assert f"shared:{repository_prn}" not in task.reserved_resources_record + + @override_settings(STAGING_AUTOCOMMIT_DELAY_SECONDS=AUTOCOMMIT_DELAY_SECONDS) + def test_dispatches_without_deadline_when_oldest_staged_content_is_due(self, db): + now = timezone.now() + repository = self.create_repo_and_stage_content(next_commit=None) + oldest = StagedContent.objects.get(repository=repository) + oldest.pulp_created = now - timedelta(seconds=30) + oldest.save(update_fields=["pulp_created"]) + + newer = repository.stage_file(create_file_content("newer.txt")) + newer.pulp_created = now - timedelta(seconds=10) + newer.save(update_fields=["pulp_created"]) + repository.next_commit = None + repository.save(update_fields=["next_commit"]) + + with with_user(None): + empty = FileRepository.objects.create(name=f"empty-autocommit-{token_hex(8)}") + update_repository(empty, autocommit=True) + disabled = self.create_repo_and_stage_content(next_commit=None, autocommit=False) + expired = self.create_repo_and_stage_content(next_commit=None) + expired_staged = StagedContent.objects.get(repository=expired) + expired_staged.pulp_created = now - timedelta(seconds=AUTOCOMMIT_DELAY_SECONDS + 1) + expired_staged.save(update_fields=["pulp_created"]) + expired.stage_file(create_file_content("expired-newer.txt")) + expired.next_commit = None + expired.save(update_fields=["next_commit"]) + + with with_user(None): + set_guid(token_hex(16)) + try: + task_ids = repository_tasks.dispatch_staging_autocommits() + finally: + clear_guid() + + # Only an old-enough staged row dispatches without a stored deadline + repository.refresh_from_db() + empty.refresh_from_db() + disabled.refresh_from_db() + expired.refresh_from_db() + assert repository.next_commit is None + assert empty.next_commit is None + assert disabled.next_commit is None + assert expired.next_commit is None + assert len(task_ids) == 1 + assert Task.objects.get(pk=task_ids[0]).enc_kwargs == {"repository_pk": str(expired.pk)} + + def test_scan_schedule_uses_configured_interval(self, db): + name = "Scan repositories for staging autocommit" + with override_settings(STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS=17): + configure_staging_autocommit() + + schedule = TaskSchedule.objects.get(name=name) + assert schedule.task_name == "pulpcore.app.tasks.repository.dispatch_staging_autocommits" + assert schedule.dispatch_interval == timedelta(seconds=17) + + with override_settings(STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS=29): + configure_staging_autocommit() + + schedule.refresh_from_db() + assert TaskSchedule.objects.filter(name=name).count() == 1 + assert schedule.dispatch_interval == timedelta(seconds=29) diff --git a/pulpcore/app/migrations/0162_repository_staging_autocommit_and_more.py b/pulpcore/app/migrations/0162_repository_staging_autocommit_and_more.py new file mode 100644 index 00000000000..4fe87d3f21c --- /dev/null +++ b/pulpcore/app/migrations/0162_repository_staging_autocommit_and_more.py @@ -0,0 +1,40 @@ +# Generated by Django 5.2.18 on 2026-10-22 19:16 + +import django.db.models.deletion +import django_lifecycle.mixins +import pulpcore.app.models.base +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0161_upstreampulp_remote_policy'), + ] + + operations = [ + migrations.AddField( + model_name='repository', + name='staging_autocommit', + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name='repository', + name='staging_enabled', + field=models.BooleanField(default=False), + ), + migrations.CreateModel( + name='StagedContent', + fields=[ + ('pulp_id', models.UUIDField(default=pulpcore.app.models.base.pulp_uuid, editable=False, primary_key=True, serialize=False)), + ('pulp_created', models.DateTimeField(auto_now_add=True)), + ('pulp_last_updated', models.DateTimeField(auto_now=True, null=True)), + ('content', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='staged_memberships', to='core.content')), + ('repository', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='core.repository')), + ], + options={ + 'unique_together': {('repository', 'content')}, + }, + bases=(django_lifecycle.mixins.LifecycleModelMixin, models.Model), + ), + ] diff --git a/pulpcore/app/migrations/0163_repository_next_commit.py b/pulpcore/app/migrations/0163_repository_next_commit.py new file mode 100644 index 00000000000..c2cc33ec652 --- /dev/null +++ b/pulpcore/app/migrations/0163_repository_next_commit.py @@ -0,0 +1,15 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0162_repository_staging_autocommit_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="repository", + name="next_commit", + field=models.DateTimeField(db_index=True, null=True), + ), + ] diff --git a/pulpcore/app/models/__init__.py b/pulpcore/app/models/__init__.py index e800a5c5762..4d5786e33d4 100644 --- a/pulpcore/app/models/__init__.py +++ b/pulpcore/app/models/__init__.py @@ -69,6 +69,7 @@ Remote, Repository, RepositoryContent, + StagedContent, RepositoryVersion, RepositoryVersionContentDetails, ) @@ -155,6 +156,7 @@ "Remote", "Repository", "RepositoryContent", + "StagedContent", "RepositoryVersion", "RepositoryVersionContentDetails", "CreatedResource", diff --git a/pulpcore/app/models/content.py b/pulpcore/app/models/content.py index dc9ef3fecd0..2b30e0f9ab0 100644 --- a/pulpcore/app/models/content.py +++ b/pulpcore/app/models/content.py @@ -508,6 +508,7 @@ def orphaned(self, orphan_protection_time, content_pks=None): if content_pks: return self.filter( version_memberships__isnull=True, + staged_memberships__isnull=True, timestamp_of_interest__lt=expiration, pk__in=content_pks, ) @@ -516,6 +517,7 @@ def orphaned(self, orphan_protection_time, content_pks=None): return self.filter( pulp_domain=domain_pk, version_memberships__isnull=True, + staged_memberships__isnull=True, timestamp_of_interest__lt=expiration, ) diff --git a/pulpcore/app/models/domain.py b/pulpcore/app/models/domain.py index e8e22eb76d1..2c1d07688e8 100644 --- a/pulpcore/app/models/domain.py +++ b/pulpcore/app/models/domain.py @@ -75,10 +75,14 @@ def prevent_default_deletion(self): @hook(BEFORE_DELETE, when="name", is_not="default") def _cleanup_orphans_pre_delete(self): - protected_content_set = self.content_set.exclude(version_memberships__isnull=True) + protected_content_set = self.content_set.exclude( + version_memberships__isnull=True, staged_memberships__isnull=True + ) if protected_content_set.exists(): raise DomainProtectedError() - self.content_set.filter(version_memberships__isnull=True).delete() + self.content_set.filter( + version_memberships__isnull=True, staged_memberships__isnull=True + ).delete() for artifact in self.artifact_set.all().iterator(): # Delete on by one to properly cleanup the storage. artifact.delete() diff --git a/pulpcore/app/models/repository.py b/pulpcore/app/models/repository.py index de093fa6ebc..476a3303ccf 100644 --- a/pulpcore/app/models/repository.py +++ b/pulpcore/app/models/repository.py @@ -5,6 +5,7 @@ import logging from collections import defaultdict from contextlib import suppress +from datetime import timedelta from gettext import gettext as _ from os import path @@ -15,6 +16,7 @@ from django.core.validators import MinValueValidator from django.db import models, transaction from django.db.models import F, Func, Q, Value +from django.utils import timezone from django_lifecycle import AFTER_UPDATE, BEFORE_CREATE, BEFORE_DELETE, hook from rest_framework.exceptions import APIException @@ -55,6 +57,9 @@ class Repository(MasterModel): retain_checkpoints (models.PositiveIntegerField): Number of checkpoint publications to keep as checkpoints. user_hidden (models.BooleanField): Whether to expose this repo to users via the API + staging_enabled (models.BooleanField): Whether repository staging is enabled. + staging_autocommit (models.BooleanField): Whether staged content is committed on a schedule. + next_commit (models.DateTimeField): When staged content is eligible for automatic commit. Relations: @@ -75,6 +80,9 @@ class Repository(MasterModel): retain_repo_versions = models.PositiveIntegerField(default=None, null=True) retain_checkpoints = models.PositiveIntegerField(default=None, null=True) user_hidden = models.BooleanField(default=False) + staging_enabled = models.BooleanField(default=False) + staging_autocommit = models.BooleanField(default=False) + next_commit = models.DateTimeField(null=True, db_index=True) content = models.ManyToManyField( "Content", through="RepositoryContent", related_name="repositories" ) @@ -455,6 +463,25 @@ def pull_through_add_content(self, content_artifact): body = {"repository_pk": self.pk, "add_content_units": [cpk], "remove_content_units": []} return dispatch(aadd_and_remove, kwargs=body, exclusive_resources=[self], immediate=True) + def stage_file(self, content): + """Associate a file content unit with this repository's staging area.""" + staged_content, _ = StagedContent.objects.get_or_create(repository=self, content=content) + if self.staging_autocommit: + with transaction.atomic(): + # no_key=True allows concurrent staging to create FK reference to the repository even + repository = ( + Repository.objects.select_for_update(skip_locked=True, no_key=True) + .filter(pk=self.pk, staging_autocommit=True, next_commit__isnull=True) + .first() + ) + if repository is not None: + repository.next_commit = timezone.now() + timedelta( + seconds=settings.STAGING_AUTOCOMMIT_DELAY_SECONDS + ) + repository.save(update_fields=["next_commit"]) + self.next_commit = repository.next_commit + return staged_content + async def async_pull_through_add_content(self, content_artifact): cpk = content_artifact.content_id already_present = RepositoryContent.objects.filter( @@ -881,6 +908,18 @@ class Meta: ) +class StagedContent(BaseModel): + """Content associated with a repository outside its versions.""" + + content = models.ForeignKey( + "Content", on_delete=models.PROTECT, related_name="staged_memberships" + ) + repository = models.ForeignKey(Repository, on_delete=models.CASCADE) + + class Meta: + unique_together = ("repository", "content") + + class RepositoryVersionQuerySet(models.QuerySet): """A queryset that provides repository version filtering methods.""" diff --git a/pulpcore/app/serializers/repository.py b/pulpcore/app/serializers/repository.py index 922d741c978..dd8034a6e80 100644 --- a/pulpcore/app/serializers/repository.py +++ b/pulpcore/app/serializers/repository.py @@ -53,6 +53,12 @@ class RepositorySerializer(ModelSerializer): required=False, min_value=1, ) + staging_enabled = serializers.BooleanField( + help_text=_("Whether content can be staged in this repository."), required=False + ) + staging_autocommit = serializers.BooleanField( + help_text=_("Whether staged content is committed on a schedule."), required=False + ) remote = DetailRelatedField( help_text=_("An optional remote to use by default when syncing."), view_name_pattern=r"remotes(-.*/.*)-detail", @@ -79,6 +85,8 @@ class Meta: "description", "retain_repo_versions", "retain_checkpoints", + "staging_enabled", + "staging_autocommit", "remote", ) diff --git a/pulpcore/app/settings.py b/pulpcore/app/settings.py index 8a4f1bf1413..ca61bf440fa 100644 --- a/pulpcore/app/settings.py +++ b/pulpcore/app/settings.py @@ -301,6 +301,10 @@ # On SIGINT, this value represents the time before the worker will attempt to kill the subprocess. TASK_GRACE_INTERVAL = 600 +# Seconds from the first staged content until it is eligible for automatic commit +STAGING_AUTOCOMMIT_DELAY_SECONDS = 300 +STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS = 60 + # how long to protect ephemeral items in minutes ORPHAN_PROTECTION_TIME = 24 * 60 @@ -593,6 +597,16 @@ }, ) +staging_autocommit_scan_interval_validator = Validator( + "STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS", + is_type_of=(int, float), + gt=0, + messages={ + "is_type_of": "{name} must be a number.", + "operations": "{name} must be greater than zero.", + }, +) + distributed_publication_retention_period_validator = Validator( "DISTRIBUTED_PUBLICATION_RETENTION_PERIOD", is_type_of=int, @@ -699,6 +713,7 @@ def validate_db_encryption_key_hook(settings): authentication_json_header_openapi_security_scheme_validator, otel_pulp_api_histogram_buckets_validator, otel_metrics_dispatch_interval_validator, + staging_autocommit_scan_interval_validator, distributed_publication_retention_period_validator, ], post_hooks=( diff --git a/pulpcore/app/tasks/repository.py b/pulpcore/app/tasks/repository.py index 5441d80f1bd..c7befcf4264 100644 --- a/pulpcore/app/tasks/repository.py +++ b/pulpcore/app/tasks/repository.py @@ -2,17 +2,25 @@ import hashlib from concurrent.futures import ThreadPoolExecutor from contextvars import copy_context +from datetime import timedelta from gettext import gettext as _ +from itertools import chain from logging import getLogger from asgiref.sync import sync_to_async +from django.conf import settings from django.db import transaction +from django.db.models import Exists, OuterRef +from django.utils import timezone from pulpcore.app import models +from pulpcore.app.contexts import with_domain from pulpcore.app.models import ProgressReport from pulpcore.app.serializers import RepositoryVersionSerializer -from pulpcore.app.util import get_domain +from pulpcore.app.util import get_domain, get_prn +from pulpcore.constants import TASK_INCOMPLETE_STATES from pulpcore.exceptions.base import RepositoryVersionDeleteError +from pulpcore.tasking.tasks import dispatch log = getLogger(__name__) @@ -207,6 +215,77 @@ def _serialize_version(version): return RepositoryVersionSerializer(version, context={"request": None}).data +def commit_staging(repository_pk, **kwargs): + """Commit staged content to a repository version and clear those staged entries.""" + with transaction.atomic(): + repository = models.Repository.objects.get(pk=repository_pk).cast() + staged = models.StagedContent.objects.filter(repository=repository) + staged_rows = list(staged.values_list("pk", "content_id")) + result = None + if staged_rows: + with repository.new_version() as version: + version.add_content( + models.Content.objects.filter( + pk__in=[content_id for _, content_id in staged_rows] + ) + ) + + staged.filter(pk__in=[pk for pk, _ in staged_rows]).delete() + result = _serialize_version(version) + + models.Repository.objects.filter(pk=repository_pk).update(next_commit=None) + return result + + +def restore_staging(repository_pk, **kwargs): + """Discard staged content without changing repository versions.""" + with transaction.atomic(): + models.StagedContent.objects.filter(repository_id=repository_pk).delete() + models.Repository.objects.filter(pk=repository_pk).update(next_commit=None) + + +def dispatch_staging_autocommits() -> list[str]: + """Dispatch commit tasks for repositories whose staging deadline has passed.""" + now = timezone.now() + # A concurrent commit can clear a deadline while newer content remains staged + # Treat old-enough staged content as due when that deadline is missing + old_enough_staged = models.StagedContent.objects.filter( + repository_id=OuterRef("pk"), + pulp_created__lte=now - timedelta(seconds=settings.STAGING_AUTOCOMMIT_DELAY_SECONDS), + ) + due_without_deadline = ( + models.Repository.objects.filter(staging_autocommit=True, next_commit__isnull=True) + .filter(Exists(old_enough_staged)) + .select_related("pulp_domain") + .order_by("pk") + ) + + task_ids = [] + commit_task_name = f"{commit_staging.__module__}.{commit_staging.__name__}" + # Stored deadlines remain authoritative when present + due_repositories = ( + models.Repository.objects.filter(staging_autocommit=True, next_commit__lte=now) + .select_related("pulp_domain") + .order_by("pk") + ) + for repository in chain(due_repositories.iterator(), due_without_deadline.iterator()): + # Avoid queuing another commit while one is waiting or running + if models.Task.objects.filter( + name=commit_task_name, + state__in=TASK_INCOMPLETE_STATES, + reserved_resources_record__contains=[get_prn(repository)], + ).exists(): + continue + with with_domain(repository.pulp_domain): + task = dispatch( + commit_staging, + kwargs={"repository_pk": repository.pk}, + exclusive_resources=[repository], + ) + task_ids.append(str(task.pk)) + return task_ids + + def add_and_remove( repository_pk, add_content_units, diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 24f50528ac4..8363408154a 100644 --- a/pulpcore/app/util.py +++ b/pulpcore/app/util.py @@ -567,6 +567,19 @@ def configure_periodic_telemetry(): models.TaskSchedule.objects.filter(task_name=task_name).delete() +def configure_staging_autocommit(): + task_name = "pulpcore.app.tasks.repository.dispatch_staging_autocommits" + models.TaskSchedule.objects.update_or_create( + name="Scan repositories for staging autocommit", + defaults={ + "task_name": task_name, + "dispatch_interval": timedelta( + seconds=settings.STAGING_AUTOCOMMIT_SCAN_INTERVAL_SECONDS + ), + }, + ) + + @lru_cache(maxsize=1) def _artifact_serving_distribution(): return models.ArtifactDistribution.objects.get() diff --git a/pulpcore/app/viewsets/repository.py b/pulpcore/app/viewsets/repository.py index 8900582b9e9..300df7e9aee 100644 --- a/pulpcore/app/viewsets/repository.py +++ b/pulpcore/app/viewsets/repository.py @@ -4,17 +4,21 @@ from django.db.models import Max, Q from django_filters import Filter from drf_spectacular.utils import extend_schema -from rest_framework import mixins, serializers +from rest_framework import mixins, serializers, status from rest_framework.decorators import action +from rest_framework.exceptions import NotFound +from rest_framework.response import Response from rest_framework.viewsets import GenericViewSet from pulpcore.app import tasks from pulpcore.app.models import ( Content, + ContentArtifact, Remote, Repository, RepositoryContent, RepositoryVersion, + StagedContent, ) from pulpcore.app.response import OperationPostponedResponse from pulpcore.app.serializers import ( @@ -25,7 +29,12 @@ RepositorySerializer, RepositoryVersionSerializer, ) -from pulpcore.app.util import resolve_prn, resolve_repository_version_href +from pulpcore.app.util import ( + get_url, + get_viewset_for_model, + resolve_prn, + resolve_repository_version_href, +) from pulpcore.app.viewsets import ( AsyncRemoveMixin, AsyncUpdateMixin, @@ -173,11 +182,131 @@ class ImmutableRepositoryViewSet( """ +class StagingBaseQuerySerializer(serializers.Serializer): + def validate(self, attrs): + unknown = set(self.initial_data) - set(self.fields) + if unknown: + raise serializers.ValidationError( + {field: _("Unknown query parameter.") for field in sorted(unknown)} + ) + return attrs + + +class StagingGetQuerySerializer(StagingBaseQuerySerializer): + limit = serializers.IntegerField(required=False, min_value=1) + offset = serializers.IntegerField(required=False, min_value=0) + + +class StagingPostQuerySerializer(StagingBaseQuerySerializer): + action = serializers.ChoiceField(choices=("commit", "restore")) + + class RepositoryViewSet(ImmutableRepositoryViewSet, AsyncUpdateMixin, LabelsMixin): """ A ViewSet for an ordinary repository. """ + def filter_queryset(self, queryset): + if self.action in { + "staging", + "staging_head", + "staging_post", + "staging_path", + "staging_path_head", + }: + return queryset + return super().filter_queryset(queryset) + + def _check_staging_enabled(self): + repository = self.get_object() + if not repository.staging_enabled: + raise NotFound() + return repository + + @action(detail=True, methods=["get"]) + def staging(self, request, pk=None): + repository = self._check_staging_enabled() + query = StagingGetQuerySerializer(data=request.query_params.dict()) + query.is_valid(raise_exception=True) + staged = ( + StagedContent.objects.filter(repository=repository) + .select_related("content") + .order_by("content_id") + ) + page = self.paginate_queryset(staged) + if page is not None: + contents = [get_url(membership.content, request=request) for membership in page] + return self.get_paginated_response(contents) + contents = [get_url(membership.content, request=request) for membership in staged] + return Response(contents) + + @staging.mapping.head + def staging_head(self, request, pk=None): + self._check_staging_enabled() + query = StagingGetQuerySerializer(data=request.query_params.dict()) + query.is_valid(raise_exception=True) + return Response(status=status.HTTP_200_OK) + + def _get_staged_content_by_path(self, repository, relative_path): + content_ids = ContentArtifact.objects.filter(relative_path=relative_path).values( + "content_id" + ) + path_filter = Q(content_id__in=content_ids) + for content_type in repository.CONTENT_TYPES: + if any(field.name == "relative_path" for field in content_type._meta.get_fields()): + matching_ids = content_type.objects.filter(relative_path=relative_path).values("pk") + path_filter |= Q(content_id__in=matching_ids) + + membership = ( + StagedContent.objects.filter(repository=repository) + .filter(path_filter) + .select_related("content") + .order_by("content_id") + .first() + ) + if membership is None: + raise NotFound() + return membership.content.cast() + + @action(detail=True, methods=["get"], url_path=r"staging/(?P.+)") + def staging_path(self, request, pk=None, relative_path=None): + repository = self._check_staging_enabled() + query = StagingBaseQuerySerializer(data=request.query_params.dict()) + query.is_valid(raise_exception=True) + content = self._get_staged_content_by_path(repository, relative_path) + serializer = get_viewset_for_model(content).serializer_class( + content, context={"request": request} + ) + return Response(serializer.data) + + @staging_path.mapping.head + def staging_path_head(self, request, pk=None, relative_path=None): + repository = self._check_staging_enabled() + query = StagingBaseQuerySerializer(data=request.query_params.dict()) + query.is_valid(raise_exception=True) + self._get_staged_content_by_path(repository, relative_path) + return Response(status=status.HTTP_200_OK) + + @staging.mapping.post + def staging_post(self, request, pk=None): + repository = self._check_staging_enabled() + query = StagingPostQuerySerializer(data=request.query_params.dict()) + query.is_valid(raise_exception=True) + if not StagedContent.objects.filter(repository=repository).exists(): + return Response(status=status.HTTP_204_NO_CONTENT) + + task_function = ( + tasks.repository.commit_staging + if query.validated_data["action"] == "commit" + else tasks.repository.restore_staging + ) + task = dispatch( + task_function, + kwargs={"repository_pk": str(repository.pk)}, + exclusive_resources=[repository], + ) + return OperationPostponedResponse(task, request) + class RepositoryVersionFilter(BaseFilterSet): # e.g. diff --git a/pulpcore/openapi/__init__.py b/pulpcore/openapi/__init__.py index 1ec2f5b6e8b..9d020d1f70b 100644 --- a/pulpcore/openapi/__init__.py +++ b/pulpcore/openapi/__init__.py @@ -376,6 +376,19 @@ def convert_endpoint_path_params(self, path, view, schema): resource_model, prefix, pulp_model_alias ) resource_path = "%s}/" % path.rsplit(sep="}", maxsplit=1)[0] + # TODO: review this implementation + lookup_names = { + "pulp_id", + getattr(view, "lookup_url_kwarg", None) or getattr(view, "lookup_field", None), + } + if parent_viewset: + parent_lookup = getattr(parent_viewset, "router_lookup", None) + if parent_lookup: + lookup_names.add(f"{parent_lookup}_pk") + # Keep action parameters after the resource href. + for match in re.finditer(r"\{([^{}]+)\}/", path): + if match.group(1) in lookup_names: + resource_path = path[: match.end()] # This check prevents /pulp/{pulp_domain}/ from being converted for non-detail endpoints # Possibly affects plugin-specific urls, depends on their url structure if not resource_path.endswith("/{pulp_domain}/"): diff --git a/pulpcore/plugin/models/__init__.py b/pulpcore/plugin/models/__init__.py index ac699d1cb7d..f0a346a2a92 100644 --- a/pulpcore/plugin/models/__init__.py +++ b/pulpcore/plugin/models/__init__.py @@ -36,6 +36,7 @@ Remote, RemoteArtifact, RepositoryContent, + StagedContent, RepositoryVersion, SigningService, Task, @@ -83,6 +84,7 @@ "Remote", "RemoteArtifact", "RepositoryContent", + "StagedContent", "RepositoryVersion", "SigningService", "Task", diff --git a/pulpcore/tasking/_util.py b/pulpcore/tasking/_util.py index 092a42ed801..aae052882a9 100644 --- a/pulpcore/tasking/_util.py +++ b/pulpcore/tasking/_util.py @@ -23,6 +23,7 @@ configure_analytics, configure_cleanup, configure_periodic_telemetry, + configure_staging_autocommit, ) from pulpcore.constants import TASK_FINAL_STATES, TASK_STATES from pulpcore.tasking.tasks import dispatch @@ -41,6 +42,7 @@ def startup_hook(): configure_analytics() configure_cleanup() configure_periodic_telemetry() + configure_staging_autocommit() def delete_incomplete_resources(task): diff --git a/pulpcore/tests/unit/test_openapi.py b/pulpcore/tests/unit/test_openapi.py new file mode 100644 index 00000000000..4cfc3ca0373 --- /dev/null +++ b/pulpcore/tests/unit/test_openapi.py @@ -0,0 +1,56 @@ +import pytest + +from pulpcore.app.viewsets.repository import RepositoryVersionViewSet, RepositoryViewSet +from pulpcore.openapi import PulpSchemaGenerator + + +@pytest.mark.parametrize( + ("path", "viewset_class", "action", "expected"), + [ + pytest.param( + "/repositories/", RepositoryViewSet, "list", "/repositories/", id="collection" + ), + pytest.param( + "/repositories/{pulp_id}/", + RepositoryViewSet, + "retrieve", + "{repository_href}", + id="detail", + ), + pytest.param( + "/repositories/{pulp_id}/staging/", + RepositoryViewSet, + "staging", + "{repository_href}staging/", + id="detail-action", + ), + pytest.param( + "/repositories/{pulp_id}/staging/{relative_path}/", + RepositoryViewSet, + "staging_path", + "{repository_href}staging/{relative_path}/", + id="detail-action-with-path-parameter", + ), + pytest.param( + "/repositories/{repository_pk}/versions/", + RepositoryVersionViewSet, + "list", + "{repository_href}versions/", + id="nested-collection", + ), + pytest.param( + "/repositories/{repository_pk}/versions/{number}/", + RepositoryVersionViewSet, + "retrieve", + "{repositories_repository_version_href}", + id="nested-detail", + ), + ], +) +def test_convert_endpoint_path_params(path, viewset_class, action, expected): + viewset = viewset_class() + viewset.action = action + + converted = PulpSchemaGenerator().convert_endpoint_path_params(path, viewset, None) + + assert converted == expected