Enqueue restart message on job/restart v3 endpoint

lib/travis/api/v3/queries/job.rb

@@ -24,8 +24,15 @@ module Travis::API::V3
 
     def restart(user)
       raise JobAlreadyRunning if %w(received queued started).include? find.state
-      payload = { id: id, user_id: user.id, source: 'api' }
-      perform_async(:job_restart, payload)
+
+      if Travis::Features.owner_active?(:enqueue_to_hub, user)
+        service = Travis::Enqueue::Services::RestartModel.new(user, { job_id: id })
+        payload = { id: id, user_id: user.id }
+        service.push("job:restart", payload)
+      else
+        payload = { id: id, user_id: user.id, source: 'api' }
+        perform_async(:job_restart, payload)
+      end
       payload
     end
   end

spec/v3/services/job/restart_spec.rb

@@ -9,6 +9,7 @@ describe Travis::API::V3::Services::Job::Restart do
 
   before do
     Travis::Features.stubs(:owner_active?).returns(true)
+    Travis::Features.stubs(:owner_active?).with(:enqueue_to_hub, repo.owner).returns(false)
     @original_sidekiq = Sidekiq::Client
     Sidekiq.send(:remove_const, :Client) # to avoid a warning
     Sidekiq::Client = []
@@ -200,6 +201,130 @@ describe Travis::API::V3::Services::Job::Restart do
     end
   end
 
+  describe "existing repository, push access, job not already running, enqueues message for Hub" do
+    let(:params)  {{}}
+    let(:token)   { Travis::Api::App::AccessToken.create(user: repo.owner, app_id: 1)                          }
+    let(:headers) {{ 'HTTP_AUTHORIZATION' => "token #{token}"                                                 }}
+    before do
+      Travis::API::V3::Models::Permission.create(repository: repo, user: repo.owner, push: true)
+      Travis::Features.stubs(:owner_active?).with(:enqueue_to_hub, repo.owner).returns(true)
+    end
+
+    describe "canceled state" do
+      before        { job.update_attribute(:state, "canceled")                                                }
+      before        { post("/v3/job/#{job.id}/restart", params, headers)                                      }
+
+      example { expect(last_response.status).to be == 202 }
+      example { expect(JSON.load(body).to_s).to include(
+        "@type",
+        "pending",
+        "job",
+        "@href",
+        "@representation",
+        "minimal",
+        "restart",
+        "id",
+        "state_change")
+      }
+
+      example { expect(sidekiq_payload).to be == {
+        "id"     => "#{job.id}",
+        "user_id"=> repo.owner_id}
+      }
+
+      example { expect(Sidekiq::Client.last['queue']).to be == 'hub'                }
+      example { expect(Sidekiq::Client.last['class']).to be == 'Travis::Hub::Sidekiq::Worker' }
+    end
+    describe "errored state" do
+      before        { job.update_attribute(:state, "errored")                                                }
+      before        { post("/v3/job/#{job.id}/restart", params, headers)                                      }
+
+      example { expect(last_response.status).to be == 202 }
+      example { expect(JSON.load(body).to_s).to include(
+        "@type",
+        "pending",
+        "job",
+        "@href",
+        "@representation",
+        "minimal",
+        "restart",
+        "id",
+        "state_change")
+      }
+
+      example { expect(sidekiq_payload).to be == {
+        "id"     => "#{job.id}",
+        "user_id"=> repo.owner_id}
+      }
+
+      example { expect(Sidekiq::Client.last['queue']).to be == 'hub'                }
+      example { expect(Sidekiq::Client.last['class']).to be == 'Travis::Hub::Sidekiq::Worker' }
+    end
+    describe "failed state" do
+      before        { job.update_attribute(:state, "failed")                                                }
+      before        { post("/v3/job/#{job.id}/restart", params, headers)                                      }
+
+      example { expect(last_response.status).to be == 202 }
+      example { expect(JSON.load(body).to_s).to include(
+        "@type",
+        "pending",
+        "job",
+        "@href",
+        "@representation",
+        "minimal",
+        "restart",
+        "id",
+        "state_change")
+      }
+
+      example { expect(sidekiq_payload).to be == {
+        "id"     => "#{job.id}",
+        "user_id"=> repo.owner_id}
+      }
+
+      example { expect(Sidekiq::Client.last['queue']).to be == 'hub'                }
+      example { expect(Sidekiq::Client.last['class']).to be == 'Travis::Hub::Sidekiq::Worker' }
+    end
+    describe "passed state" do
+      before        { job.update_attribute(:state, "passed")                                                }
+      before        { post("/v3/job/#{job.id}/restart", params, headers)                                      }
+
+      example { expect(last_response.status).to be == 202 }
+      example { expect(JSON.load(body).to_s).to include(
+        "@type",
+        "pending",
+        "job",
+        "@href",
+        "@representation",
+        "minimal",
+        "restart",
+        "id",
+        "state_change")
+      }
+
+      example { expect(sidekiq_payload).to be == {
+        "id"     => "#{job.id}",
+        "user_id"=> repo.owner_id}
+      }
+
+      example { expect(Sidekiq::Client.last['queue']).to be == 'hub'                }
+      example { expect(Sidekiq::Client.last['class']).to be == 'Travis::Hub::Sidekiq::Worker' }
+    end
+
+    describe "setting id has no effect" do
+      before do
+        job.update_attribute(:state, "passed")
+        post("/v3/job/#{job.id}/restart", params, headers)
+      end
+      let(:params) {{ id: 42 }}
+
+      example { expect(sidekiq_payload).to be == {
+        "id"     => "#{job.id}",
+        "user_id"=> repo.owner_id}
+      }
+    end
+  end
+
   describe "existing repository, push access, job already running" do
     let(:params)  {{}}
     let(:token)   { Travis::Api::App::AccessToken.create(user: repo.owner, app_id: 1)                          }