Fixes #27517 - Pulp3 content upload via hammer (#8304)

* Fixes #27517 - Pulp3 content upload via hammer

* Fixes #27517 - Rubocop fixes

* Fixes #27517 - Add tests

* Fixes #27517 - Fix test

app/controllers/katello/api/v2/content_uploads_controller.rb

@@ -8,18 +8,20 @@ class Api::V2::ContentUploadsController < Api::V2::ApiController
 
     api :POST, "/repositories/:repository_id/content_uploads", N_("Create an upload request")
     param :repository_id, :number, :required => true, :desc => N_("repository id")
+    param :size, :number, :required => true, :desc => N_("Size of file to upload")
     def create
-      render :json => @repository.backend_content_service(::SmartProxy.pulp_master).create_upload
+      render :json => @repository.backend_content_service(::SmartProxy.pulp_master).create_upload(params[:size])
     end
 
     api :PUT, "/repositories/:repository_id/content_uploads/:id", N_("Upload a chunk of the file's content")
     param :repository_id, :number, :required => true, :desc => N_("Repository id")
     param :id, String, :required => true, :desc => N_("Upload request id")
+    param :size, :number, :required => true, :desc => N_("Size of file to upload")
     param :offset, :number, :required => true, :desc => N_("The offset in the file where the content starts")
     param :content, File, :required => true, :desc => N_("The actual file contents")
     def update
       @repository.backend_content_service(::SmartProxy.pulp_master)
-        .upload_chunk(params[:id], params[:offset], params[:content])
+        .upload_chunk(params[:id], params[:offset], params[:content], params[:size])
       head :no_content
     end
 

app/lib/actions/katello/repository/import_upload.rb

@@ -3,6 +3,7 @@ module Actions
   module Katello
     module Repository
       class ImportUpload < Actions::EntryAction
+        include Actions::Katello::PulpSelector
         def plan(repository, uploads, options = {})
           action_subject(repository)
           repo_service = repository.backend_service(::SmartProxy.pulp_master)
@@ -11,24 +12,24 @@ def plan(repository, uploads, options = {})
           unit_keys = repo_service.unit_keys(uploads)
           generate_metadata = options.fetch(:generate_metadata, true)
           sync_capsule = options.fetch(:sync_capsule, true)
-          repo_type = repository.content_type
 
-          options[:content_type] ||= ::Katello::RepositoryTypeManager.find(repo_type).default_managed_content_type.label
+          options[:content_type] ||= ::Katello::RepositoryTypeManager.find(repository.content_type).default_managed_content_type.label
           unit_type_id = SmartProxy.pulp_master.content_service(options[:content_type])::CONTENT_TYPE
 
           sequence do
             upload_results = concurrence do
               upload_ids.zip(unit_keys).collect do |upload_id, unit_key|
-                if unit_type_id == 'docker_tag'
-                  unit_metadata = unit_key
-                end
-                import_upload = plan_action(Pulp::Repository::ImportUpload,
-                                            pulp_id: repository.pulp_id,
-                                            unit_type_id: unit_type_id,
-                                            unit_key: unit_key,
-                                            upload_id: upload_id,
-                                            unit_metadata: unit_metadata)
-
+                unit_metadata = unit_key if unit_type_id == 'docker_tag'
+                import_upload_args = {
+                  pulp_id: repository.pulp_id,
+                  unit_type_id: unit_type_id,
+                  unit_key: unit_key,
+                  upload_id: upload_id,
+                  unit_metadata: unit_metadata
+                }
+                import_upload = plan_pulp_action([Actions::Pulp::Repository::ImportUpload,
+                                                  Actions::Pulp3::Orchestration::Repository::ImportUpload],
+                                  repository, SmartProxy.pulp_master, import_upload_args)
                 plan_action(FinishUpload, repository, :dependency => import_upload.output,
                             generate_metadata: false, content_type: options[:content_type])
                 import_upload.output

app/lib/actions/pulp/orchestration/repository/upload_content.rb

@@ -3,13 +3,14 @@ module Pulp
     module Orchestration
       module Repository
         class UploadContent < Pulp::Abstract
-          def plan(repository, _smart_proxy, file, unit_type_id)
+          def plan(repository, smart_proxy, file, unit_type_id)
             sequence do
               upload_request = plan_action(Pulp::Repository::CreateUploadRequest)
               plan_action(Pulp::Repository::UploadFile,
                           upload_id: upload_request.output[:upload_id],
                           file: file[:path])
               plan_action(Pulp::Repository::ImportUpload,
+                          repository, smart_proxy,
                           pulp_id: repository.pulp_id,
                           unit_type_id: unit_type_id,
                           unit_key: unit_key(file, repository),

app/lib/actions/pulp/repository/import_upload.rb

@@ -2,20 +2,16 @@ module Actions
   module Pulp
     module Repository
       class ImportUpload < Pulp::AbstractAsyncTask
-        input_format do
-          param :pulp_id
-          param :unit_type_id
-          param :upload_id
-          param :unit_key
-          param :unit_metadata
+        def plan(_repo, _smart_proxy, options)
+          plan_self(:options => options)
         end
 
         def invoke_external_task
-          output[:pulp_tasks] = [pulp_resources.content.import_into_repo(input[:pulp_id],
-                                                   input[:unit_type_id],
-                                                   input[:upload_id],
-                                                   input[:unit_key],
-                                                   unit_metadata: input[:unit_metadata] || {})]
+          output[:pulp_tasks] = [pulp_resources.content.import_into_repo(input[:options][:pulp_id],
+                                                   input[:options][:unit_type_id],
+                                                   input[:options][:upload_id],
+                                                   input[:options][:unit_key],
+                                                   unit_metadata: input[:options][:unit_metadata] || {})]
         end
       end
     end

app/lib/actions/pulp3/orchestration/repository/import_upload.rb

@@ -0,0 +1,26 @@
+module Actions
+  module Pulp3
+    module Orchestration
+      module Repository
+        class ImportUpload < Pulp3::AbstractAsyncTask
+          def plan(repository, smart_proxy, args)
+            file = {:filename => args.dig(:unit_key, :name)}
+            sequence do
+              action_output = plan_self(:repository_id => repository.id, :smart_proxy_id => smart_proxy.id, :upload_href => "/pulp/api/v3/uploads/" + args.dig(:upload_id) + "/", :sha256 => args.dig(:unit_key, :checksum)).output
+              artifact_action_output = plan_action(Pulp3::Repository::SaveArtifact, file, repository, action_output[:pulp_tasks], args.dig(:unit_type_id)).output
+              action_output = plan_action(Pulp3::Repository::ImportUpload, artifact_action_output[:content_unit_href], repository, smart_proxy).output
+              plan_action(Pulp3::Repository::SaveVersion, repository, action_output[:pulp_tasks]).output
+            end
+          end
+
+          def invoke_external_task
+            repo = ::Katello::Repository.find(input[:repository_id])
+            repo_backend_service = repo.backend_service(smart_proxy)
+            uploads_api = repo_backend_service.uploads_api
+            output[:pulp_tasks] = [uploads_api.commit(input[:upload_href], sha256: input[:sha256])]
+          end
+        end
+      end
+    end
+  end
+end

app/services/katello/pulp/content.rb

@@ -3,10 +3,17 @@ module Pulp
     class Content
       extend Katello::Abstract::Pulp::Content
       class << self
-        extend Forwardable
-        def_delegator :pulp_content, :create_upload_request, :create_upload
-        def_delegator :pulp_content, :delete_upload_request, :delete_upload
-        def_delegator :pulp_content, :upload_bits, :upload_chunk
+        def create_upload(_size = 0)
+          pulp_content.create_upload_request
+        end
+
+        def delete_upload(upload_id)
+          pulp_content.delete_upload_request(upload_id)
+        end
+
+        def upload_chunk(upload_id, offset, content, _size)
+          pulp_content.upload_bits(upload_id, offset, content)
+        end
 
         private def pulp_content
           SmartProxy.pulp_master.pulp_api.resources.content

app/services/katello/pulp3/content.rb

@@ -0,0 +1,53 @@
+require "pulpcore_client"
+module Katello
+  module Pulp3
+    class Content
+      extend Katello::Abstract::Pulp::Content
+      class << self
+        def create_upload(size = 0)
+          upload_href = uploads_api.create(upload_class.new(size: size))._href
+          {"upload_id" => upload_href.split("/").last}
+        end
+
+        def delete_upload(upload_href)
+          #Commit deletes upload request for pulp3. Not needed other than to implement abstract method.
+        end
+
+        def upload_chunk(upload_href, offset, content, size)
+          upload_href = "/pulp/api/v3/uploads/" + upload_href + "/"
+          offset = offset.try(:to_i)
+          size = size.try(:to_i)
+          begin
+            filechunk = Tempfile.new('filechunk', :encoding => 'ascii-8bit')
+            filechunk.write(content)
+            filechunk.flush
+            actual_chunk_size = File.size(filechunk)
+            uploads_api.update(upload_href, content_range(offset, offset + actual_chunk_size - 1, size), filechunk)
+          ensure
+            filechunk.close
+            filechunk.unlink
+          end
+        end
+
+        private
+
+        def core_api_client
+          PulpcoreClient::ApiClient.new(SmartProxy.pulp_master.pulp3_configuration(PulpcoreClient::Configuration))
+        end
+
+        def uploads_api
+          PulpcoreClient::UploadsApi.new(core_api_client)
+        end
+
+        def upload_class
+          PulpcoreClient::Upload
+        end
+
+        def content_range(start, finish, total)
+          finish = finish > total ? total : finish
+          "bytes #{start}-#{finish}/#{total}"
+        end
+      end
+    end
+  end
+end

app/services/katello/pulp3/repository.rb

@@ -50,6 +50,10 @@ def remote_options
         fail NotImplementedError
       end
 
+      def content_service
+        Katello::Pulp3::Content
+      end
+
       def self.api_client(_smart_proxy)
         fail NotImplementedError
       end
@@ -525,6 +529,12 @@ def add_content(content_unit_href)
             add_content_units: content_unit_href)
         repository_versions_api.create(repository_reference.repository_href, data)
       end
+
+      def unit_keys(uploads)
+        uploads.map do |upload|
+          upload.except('id')
+        end
+      end
     end
   end
 end

test/actions/katello/repository/import_upload_test.rb

@@ -19,12 +19,17 @@ module Actions
       action = create_action(action_class)
       action.expects(:action_subject).with(repo)
       plan_action(action, repo, [upload])
+      import_upload_args = {
+        pulp_id: repo.pulp_id,
+        unit_type_id: repo.unit_type_id,
+        unit_key: upload.except('id', 'name'),
+        upload_id: '1',
+        unit_metadata: nil
+      }
 
-      assert_action_planed_with(action, pulp_import_class, :pulp_id => repo.pulp_id,
-                                unit_type_id: repo.unit_type_id,
-                                unit_key: upload.except('id', 'name'),
-                                upload_id: '1',
-                                unit_metadata: nil)
+      assert_action_planed_with(action, pulp_import_class,
+                                repo, SmartProxy.pulp_master,
+                                import_upload_args)
     end
   end
 end

test/actions/katello/repository_test.rb

@@ -249,6 +249,7 @@ class UploadFilesTest < TestBase
       assert_action_planed_with(action, ::Actions::Pulp::Repository::UploadFile,
                                 upload_id: 123, file: file)
       assert_action_planed_with(action, ::Actions::Pulp::Repository::ImportUpload,
+                                puppet_repository, proxy,
                   pulp_id: puppet_repository.pulp_id,
                   unit_type_id: "puppet_module",
                   unit_key: {},
@@ -324,11 +325,16 @@ class UploadDockerTest < TestBase
       plan_action action, docker_repository, uploads,
               generate_metadata: true, sync_capsule: true
 
+      import_upload_args = {
+        pulp_id: docker_repository.pulp_id,
+        unit_type_id: 'docker_manifest',
+        unit_key: {'size' => '12333', 'checksum' => 'asf23421324', 'name' => 'test'},
+        upload_id: 1,
+        unit_metadata: nil
+      }
       assert_action_planned_with(action, ::Actions::Pulp::Repository::ImportUpload,
-                                 pulp_id: docker_repository.pulp_id,
-                                 unit_type_id: 'docker_manifest',
-                                 unit_key: {'size' => '12333', 'checksum' => 'asf23421324', 'name' => 'test'},
-                                 upload_id: 1, unit_metadata: nil
+                                 docker_repository, SmartProxy.pulp_master,
+                                 import_upload_args
                                 )
     end
 
@@ -342,11 +348,16 @@ class UploadDockerTest < TestBase
       plan_action action, docker_repository, uploads,
               generate_metadata: true, sync_capsule: true, content_type: 'docker_tag'
 
+      import_upload_args = {
+        pulp_id: docker_repository.pulp_id,
+        unit_type_id: 'docker_tag',
+        unit_key: unit_keys[0],
+        upload_id: 1,
+        unit_metadata: unit_keys[0]
+      }
       assert_action_planned_with(action, ::Actions::Pulp::Repository::ImportUpload,
-                                 pulp_id: docker_repository.pulp_id,
-                                 unit_type_id: 'docker_tag',
-                                 unit_key: unit_keys[0],
-                                 upload_id: 1, unit_metadata: unit_keys[0]
+                                 docker_repository, SmartProxy.pulp_master,
+                                 import_upload_args
                                 )
     end
   end

test/actions/pulp/repository/upload_file_test.rb

@@ -5,6 +5,9 @@ module ::Actions::Pulp::Repository
   class UploadFileTest < VCRTestBase
     let(:repo) { katello_repositories(:p_forge) }
     let(:file) { File.join(Katello::Engine.root, "test/fixtures/puppet/puppetlabs-ntp-2.0.1.tar.gz") }
+    before do
+      FactoryBot.create(:smart_proxy, :default_smart_proxy)
+    end
 
     def test_upload_file
       upload_request = run_action(::Actions::Pulp::Repository::CreateUploadRequest)
@@ -15,10 +18,12 @@ def test_upload_file
       end
 
       run_action(::Actions::Pulp::Repository::ImportUpload,
-                  pulp_id: repo.pulp_id,
-                  unit_type_id: repo.unit_type_id,
-                  unit_key: {},
-                  upload_id: upload_request.output[:upload_id])
+                 repo, SmartProxy.pulp_master,
+                    pulp_id: repo.pulp_id,
+                     unit_type_id: repo.unit_type_id,
+                     unit_key: {},
+                     upload_id: upload_request.output[:upload_id]
+                  )
 
       run_action(::Actions::Pulp::Repository::DeleteUploadRequest,
                   upload_id: upload_request.output[:upload_id])