Add synchronous awesome spawn runner

Author: kbrock

examples/awesome.asl

@@ -0,0 +1,27 @@
+{
+  "Comment": "Directory Listing",
+  "StartAt": "a",
+  "States": {
+    "a":{
+      "Type": "Pass",
+      "Next": "b"
+    },
+    "b": {
+      "Type": "Wait",
+      "Seconds": 1,
+      "Next": "ls"
+    },
+    "ls": {
+      "Type": "Task",
+      "Resource": "awesome://ls -l Gemfile",
+      "Comment": "awesome://ls -l $FILENAME",
+      "Next": "c",
+      "Parameters": {
+        "FILENAME" : "Gemfile"
+      }
+    },
+      "c": {
+        "Type": "Succeed"
+    }
+  }
+}

exe/floe

@@ -6,6 +6,7 @@ $LOAD_PATH.unshift(File.expand_path("../lib", __dir__))
 require "optimist"
 require "floe"
 require "floe/container_runner"
+require "floe/awesome_runner"
 
 opts = Optimist.options do
   version("v#{Floe::VERSION}\n")

lib/floe/awesome_runner.rb

@@ -0,0 +1,69 @@
+# frozen_string_literal: true
+
+module Floe
+  class AwesomeRunner < Floe::Runner
+    SCHEME        = "awesome"
+    SCHEME_PREFIX = "#{SCHEME}://"
+    SCHEME_OFFSET = SCHEME.length + 3
+
+    def initialize(_options = {})
+      require "awesome_spawn"
+
+      super
+    end
+
+    # @return [Hash] runner_context
+    def run_async!(resource, params = {}, _secrets = {}, _context = {})
+      raise ArgumentError, "Invalid resource" unless resource&.start_with?(SCHEME_PREFIX)
+
+      args = resource[SCHEME_OFFSET..].split
+      method = args.shift
+
+      runner_context = {}
+
+      # TODO: fix sanitization preventing params in args (e.g.: $PARAM1 => \$PARAM1)
+      result = AwesomeSpawn.run(method, :env => params, :params => args)
+      self.class.populate_results!(runner_context, :result => result)
+      runner_context
+    end
+
+    def status!(runner_context)
+    end
+
+    def running?(runner_context)
+      !runner_context["Output"]
+    end
+
+    def success?(runner_context)
+      !runner_context["Error"]
+    end
+
+    def output(runner_context)
+      runner_context["Output"]
+    end
+
+    def cleanup(runner_context)
+    end
+
+    # internal methods
+
+    def self.command_error_cause(command_result)
+      command_result.error.nil? || command_result.error.empty? ? command_result.output.to_s : command_result.error.to_s
+    end
+
+    def self.populate_results!(runner_context, result: nil, error: nil)
+      error ||= command_error_cause(result) if result&.failure?
+
+      if error
+        runner_context["Output"] = {"Error" => "States.TaskFailed", "Cause" => error}
+        runner_context["Error"]  = true
+      else
+        runner_context["Output"] = {"Result" => result.output.chomp.split("\n")}
+      end
+
+      runner_context
+    end
+  end
+end
+
+Floe::Runner.register_scheme(Floe::AwesomeRunner::SCHEME, Floe::AwesomeRunner.new)

spec/awesome_runner_spec.rb

@@ -0,0 +1,78 @@
+require_relative "../lib/floe/awesome_runner"
+
+RSpec.describe Floe::AwesomeRunner, :uses_awesome_spawn => true do
+  let(:subject)        { described_class.new(runner_options) }
+  let(:runner_options) { {} }
+  let(:container_id)   { SecureRandom.hex }
+
+  # let(:workflow) { make_workflow(ctx, {"State" => {"Type" => "Task", "Resource" => resource, "Parameters" => {"var1.$" => "$.foo.bar"}, "End" => true}}) }
+
+  describe "#run_async!" do
+    it "raises an exception without a resource" do
+      expect { subject.run_async!(nil) }.to raise_error(ArgumentError, "Invalid resource")
+    end
+
+    it "raises an exception for an invalid resource uri" do
+      expect { subject.run_async!("arn:abcd:efgh") }.to raise_error(ArgumentError, "Invalid resource")
+    end
+
+    it "calls command run with the command name" do
+      stub_good_run("ls", :params => [], :env => {}, :output => "file\nlisting\n")
+
+      subject.run_async!("awesome://ls")
+    end
+
+    it "passes environment variables to command run" do
+      stub_good_run("ls", :params => [], :env => {"FOO" => "BAR"}, :output => "file\nlisting\n")
+
+      subject.run_async!("awesome://ls", {"FOO" => "BAR"})
+    end
+  end
+
+  # describe "#status!" do
+  #   let(:runner_context) { {"container_ref" => container_id} }
+
+  #   it "returns the updated container_state" do
+  #     stub_good_run!("ls", :params => ["inspect", container_id], :output => "[{\"State\": {\"Running\": true}}]")
+
+  #     subject.status!(runner_context)
+
+  #     expect(runner_context).to include("container_state" => {"Running" => true})
+  #   end
+  # end
+
+  describe "#running?" do
+    # it "retuns true when running" do
+    #   runner_context = {"container_ref" => container_id, "container_state" => {"Running" => true}}
+    #   expect(subject.running?(runner_context)).to be_truthy
+    # end
+
+    # it "retuns false when not running" do
+    #   runner_context = {"container_ref" => container_id, "container_state" => {"Running" => false, "ExitCode" => 0}}
+    #   expect(subject.running?(runner_context)).to be_falsey
+    # end
+  end
+
+  describe "#success?" do
+    # it "retuns true when successful" do
+    #   runner_context = {"container_ref" => container_id, "container_state" => {"Running" => false, "ExitCode" => 0}}
+    #   expect(subject.success?(runner_context)).to be_truthy
+    # end
+
+    # it "retuns false when not successful" do
+    #   runner_context = {"container_ref" => container_id, "container_state" => {"Running" => false, "ExitCode" => 1}}
+    #   expect(subject.success?(runner_context)).to be_falsey
+    # end
+  end
+
+  describe "#output" do
+    let(:runner_context) { {"Output" => ["output1", "output2"]} }
+
+    it "returns log output" do
+      expect(subject.output(runner_context)).to eq(["output1", "output2"])
+    end
+  end
+
+  # describe "#cleanup" do
+  # end
+end