Update threaded listener

Handle denied and approved requests
Add specs around the listener

app/models/approval_request.rb

@@ -3,4 +3,11 @@ class ApprovalRequest < ApplicationRecord
   enum :state => [:undecided, :approved, :denied]
 
   belongs_to :order_item
+
+  def update_message(level, message)
+    self.updated_at = Time.zone.now
+    ProgressMessage.create(:level         => level,
+                           :message       => message,
+                           :order_item_id => order_item_id)
+  end
 end

db/migrate/20190322191656_add_approval_request_reason.rb

@@ -0,0 +1,5 @@
+class AddApprovalRequestReason < ActiveRecord::Migration[5.2]
+  def change
+    add_column :approval_requests, :reason, :string
+  end
+end

db/schema.rb

@@ -21,6 +21,7 @@
     t.integer "order_item_id"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.string "reason"
     t.integer "state", default: 0
   end
 

lib/approval_request_listener.rb

@@ -2,9 +2,8 @@
 
 class ApprovalRequestListener
   SERVICE_NAME = "platform.approval".freeze
-  CLIENT_AND_GROUP_REF = "catalog-api-worker".freeze
-
-  class ApprovalRequestNotFound < StandardError; end
+  CLIENT_AND_GROUP_REF = "approval-catalog-api-worker".freeze
+  EVENT_REQUEST_FINISHED = 'request_finished'.freeze
 
   attr_accessor :messaging_client_options, :client
 
@@ -19,7 +18,7 @@ def run
   def subscribe_to_approval_updates
     self.client = ManageIQ::Messaging::Client.open(messaging_client_options)
 
-    client.subscribe_messages(
+    client.subscribe_topic(
       :service   => SERVICE_NAME,
       :max_bytes => 500_000
     ) do |messages|
@@ -39,21 +38,27 @@ def process_message(msg)
       :level   => "info",
       :message => "Task update message received with payload: #{msg.payload}"
     )
-    if msg.payload["decision"] == "approved"
-      approval = ApprovalRequest.find_by(:approval_request_ref => msg.payload["request_id"])
-      raise ApprovalRequestNotFound if approval.nil?
-      approval.status = msg.payload["reason"]
-      # TODO Make ProgressMessage polymorphic to support multiple model types
-      approval.update_message('info', 'Approval Complete')
-      approval.save!
+    if msg.message == EVENT_REQUEST_FINISHED && (msg.payload["decision"] == "approved" || msg.payload["decision"] == "denied")
+      approval = ApprovalRequest.find_by!(:approval_request_ref => msg.payload["request_id"])
+      update_and_log_state(approval, msg.payload)
     end
-  rescue ApprovalRequestNotFound
+  rescue ActiveRecord::RecordNotFound
     ProgressMessage.create(
       :level   => "error",
       :message => "Could not find Approval Request with request_id of #{msg.payload["request_id"]}"
     )
   end
 
+  def update_and_log_state(approval, payload)
+    decision = payload['decision']
+    reason = payload['reason']
+    log_message = decision == "approved" ? "Approval Complete: #{reason}" : "Approval Denied: #{reason}"
+    approval.update_message("info", log_message)
+    approval.state = decision
+    approval.reason = reason
+    approval.save!
+  end
+
   def default_messaging_options
     {
       :protocol   => :Kafka,

spec/lib/approval_request_listener_spec.rb

@@ -0,0 +1,124 @@
+describe ApprovalRequestListener do
+  let(:client) { double(:client) }
+  let(:event)  { ApprovalRequestListener::EVENT_REQUEST_FINISHED }
+
+  describe "#subscribe_to_approval_updates" do
+    let(:request) do
+      { :headers => { 'x-rh-identity' => encoded_user_hash }, :original_url => 'whatever' }
+    end
+
+    around do |example|
+      ManageIQ::API::Common::Request.with_request(request) { example.call }
+    end
+
+    let(:messages) { [ManageIQ::Messaging::ReceivedMessage.new(nil, event, payload, nil)] }
+    let!(:approval_request) do
+      ApprovalRequest.create!(
+        :approval_request_ref => approval_request_ref,
+        :workflow_ref         => "2323",
+        :order_item_id        => "3434"
+      )
+    end
+
+    before do
+      allow(ManageIQ::Messaging::Client).to receive(:open).with(
+        :protocol   => :Kafka,
+        :client_ref => ApprovalRequestListener::CLIENT_AND_GROUP_REF,
+        :group_ref  => ApprovalRequestListener::CLIENT_AND_GROUP_REF
+      ).and_return(client)
+      allow(client).to receive(:subscribe_topic).with(
+        :service   => ApprovalRequestListener::SERVICE_NAME,
+        :max_bytes => 500_000
+      ).and_yield(messages)
+      allow(client).to receive(:close)
+    end
+
+    context "when the approval request is not findable" do
+      let(:approval_request_ref) { "0" }
+
+      context "when the request_id is anything else" do
+        let(:payload)    { {"request_id" => request_id } }
+        let(:request_id) { "10" }
+
+        it "creates a progress message about the payload" do
+          subject.subscribe_to_approval_updates
+          latest_progress_message = ProgressMessage.last
+          expect(latest_progress_message.level).to eq("info")
+          expect(latest_progress_message.message).to eq("Task update message received with payload: #{payload}")
+        end
+
+        it "does not update the approval request" do
+          subject.subscribe_to_approval_updates
+          approval_request.reload
+          expect(approval_request.state).to eq("undecided")
+        end
+      end
+    end
+
+    context "when the approval request is not findable " do
+      let(:approval_request_ref) { "0" }
+
+      context "and the approval is finished with an approval or denial" do
+        let(:reason)                { "System Approved" }
+        let(:decision)              { "approved" }
+        let(:request_id)            { "1" }
+        let(:payload)               { {"request_id" => request_id, "decision" => decision, "reason" => reason } }
+
+        it "creates a progress message about the approval request error" do
+          subject.subscribe_to_approval_updates
+          latest_progress_message = ProgressMessage.last
+          expect(latest_progress_message.level).to eq("error")
+          expect(latest_progress_message.message).to eq("Could not find Approval Request with request_id of 1")
+        end
+
+        it "does not update the approval request" do
+          subject.subscribe_to_approval_updates
+          approval_request.reload
+          expect(approval_request.state).to eq("undecided")
+        end
+      end
+    end
+
+    context "when the state of the approval is approved" do
+      let(:decision)              { "approved" }
+      let(:reason)                { "System Approved" }
+      let(:approval_request_ref)  { "1" }
+      let(:request_id)            { "1" }
+      let(:payload)               { {"request_id" => request_id, "decision" => decision, "reason" => reason } }
+
+      it "creates a progress message about the payload" do
+        subject.subscribe_to_approval_updates
+        latest_progress_message = ProgressMessage.second_to_last
+        expect(latest_progress_message.level).to eq("info")
+        expect(latest_progress_message.message).to eq("Task update message received with payload: #{payload}")
+      end
+
+      it "updates the approval request to be approved" do
+        subject.subscribe_to_approval_updates
+        approval_request.reload
+        expect(approval_request.state).to eq("approved")
+      end
+    end
+
+    context "when the state of the approval is denied" do
+      let(:decision)              { "denied" }
+      let(:reason)                { "System Denied" }
+      let(:approval_request_ref)  { "1" }
+      let(:request_id)            { "1" }
+      let(:payload)               { {"request_id" => request_id, "decision" => decision, "reason" => reason } }
+
+      it "creates a progress message about the payload" do
+        subject.subscribe_to_approval_updates
+        latest_progress_message = ProgressMessage.second_to_last
+        expect(latest_progress_message.level).to eq("info")
+        expect(latest_progress_message.message).to eq("Task update message received with payload: #{payload}")
+      end
+
+      it "updates the approval request to be denied" do
+        subject.subscribe_to_approval_updates
+        approval_request.reload
+        expect(approval_request.state).to eq("denied")
+      end
+    end
+  end
+end