Allow passing through RSpec arguments

bin/rspecq

@@ -103,6 +103,13 @@ OptionParser.new do |o|
     opts[:fail_fast] = v
   end
 
+  o.on("--junit-output filepath", String, "Output junit formatted xml "    \
+       "for CI suites to the defined file path. Substitution parameters "  \
+       "{{TEST_ENV_NUMBER}} - parallel gem proc number. "                  \
+       "{{JOB_INDEX}} - increments with each suite that is run.") do |v|
+    opts[:junit_output] = v
+  end
+
   o.on("--reproduction", "Enable reproduction mode: run rspec on the given files " \
        "and examples in the exact order they are given. Incompatible with " \
        "--timings.") do |v|
@@ -140,6 +147,7 @@ opts[:queue_wait_timeout] ||= Integer(ENV["RSPECQ_QUEUE_WAIT_TIMEOUT"] || DEFAUL
 opts[:redis_url] ||= ENV["RSPECQ_REDIS_URL"]
 opts[:fail_fast] ||= Integer(ENV["RSPECQ_FAIL_FAST"] || DEFAULT_FAIL_FAST)
 opts[:reproduction] ||= env_set?("RSPECQ_REPRODUCTION")
+opts[:junit_output] ||= ENV["RSPECQ_JUNIT_OUTPUT"]
 
 # rubocop:disable Style/RaiseArgs
 raise OptionParser::MissingArgument.new(:build) if opts[:build].nil?
@@ -156,6 +164,19 @@ end
 
 Sentry.init if ENV["SENTRY_DSN"]
 
+# Use the RSpec parser to parse any command line args intended for rspec such
+# as `-- --format JUnit -o foo.xml` so that we can pass these args to rspec
+# while removing the files_or_dirs_to_run since we want to pull those from the
+# queue. OptionParser above mutates ARGV, so only options after `--`  or
+# non-flag arguments (such as files) will make it to this point.
+files_or_dirs_to_run = RSpec::Core::Parser.new(ARGV).parse[:files_or_directories_to_run]
+if files_or_dirs_to_run.length.zero?
+  opts[:rspec_args] = ARGV
+else
+  opts[:rspec_args] = ARGV[0...-files_or_dirs_to_run.length]
+  opts[:files_or_dirs_to_run] = files_or_dirs_to_run
+end
+
 if opts[:report]
   reporter = RSpecQ::Reporter.new(
     build_id: opts[:build],
@@ -172,7 +193,8 @@ else
     redis_opts: redis_opts
   )
 
-  worker.files_or_dirs_to_run = ARGV if ARGV.any?
+  worker.rspec_args = opts[:rspec_args]
+  worker.files_or_dirs_to_run = opts[:files_or_dirs_to_run] if opts[:files_or_dirs_to_run]
   worker.populate_timings = opts[:timings]
   worker.file_split_threshold = opts[:file_split_threshold]
   worker.max_requeues = opts[:max_requeues]
@@ -181,5 +203,6 @@ else
   worker.seed = Integer(opts[:seed]) if opts[:seed]
   worker.reproduction = opts[:reproduction]
   worker.tags = opts[:tags]
+  worker.junit_output = opts[:junit_output]
   worker.work
 end

lib/rspecq.rb

@@ -11,6 +11,7 @@ module RSpecQ
 require_relative "rspecq/formatters/example_count_recorder"
 require_relative "rspecq/formatters/failure_recorder"
 require_relative "rspecq/formatters/job_timing_recorder"
+require_relative "rspecq/formatters/junit_formatter"
 require_relative "rspecq/formatters/worker_heartbeat_recorder"
 
 require_relative "rspecq/queue"

lib/rspecq/formatters/junit_formatter.rb

@@ -0,0 +1,44 @@
+require "rspec_junit_formatter"
+
+module RSpecQ
+  module Formatters
+    # Junit output formatter that handles outputting of requeued examples,
+    # parallel gem, and multiple suites per rspecq run.
+    class JUnitFormatter < RSpecJUnitFormatter
+      def initialize(queue, job, max_requeues, job_index, path)
+        @queue = queue
+        @job = job
+        @max_requeues = max_requeues
+        @requeued_examples = []
+        path = path.gsub(/{{TEST_ENV_NUMBER}}/,ENV["TEST_ENV_NUMBER"].to_s)
+        path = path.gsub(/{{JOB_INDEX}}/, job_index.to_s)
+        RSpec::Support::DirectoryMaker.mkdir_p(File.dirname(path))
+        output_file = File.new(path, "w")
+        super(output_file)
+      end
+
+      def example_failed(notification)
+        # if it is requeued, store the notification
+        if @queue.requeueable_job?(notification.example.id, @max_requeues)
+          @requeued_examples << notification.example
+        end
+      end
+
+      private
+
+      def example_count
+        @summary_notification.example_count - @requeued_examples.size
+      end
+
+      def failure_count
+        @summary_notification.failure_count - @requeued_examples.size
+      end
+
+      def examples
+        @examples_notification.notifications.reject do |example_notification|
+          @requeued_examples.map(&:id).include?(example_notification.example.id)
+        end
+      end
+    end
+  end
+end

lib/rspecq/queue.rb

@@ -71,6 +71,19 @@ class Queue
       return true
     LUA
 
+    REQUEUEABLE_JOB = <<~LUA.freeze
+      local key_requeues = KEYS[1]
+      local job = ARGV[1]
+      local max_requeues = ARGV[2]
+
+      local requeued_times = redis.call('hget', key_requeues, job)
+      if requeued_times and requeued_times >= max_requeues then
+        return nil
+      end
+      redis.call('hincrby', key_requeues, job, 1)
+      return true
+    LUA
+
     STATUS_INITIALIZING = "initializing".freeze
     STATUS_READY = "ready".freeze
 
@@ -145,6 +158,15 @@ def requeue_job(example, max_requeues, original_worker_id)
       )
     end
 
+    def requeueable_job?(job, max_requeues)
+      return false if max_requeues.zero?
+
+      @redis.eval(
+        REQUEUEABLE_JOB,
+        keys: [key_requeues_formatter_stats],
+        argv: [job, max_requeues]
+      )
+
     def save_worker_seed(worker, seed)
       @redis.hset(key("worker_seed"), worker, seed)
     end
@@ -347,6 +369,10 @@ def key_requeues
       key("requeues")
     end
 
+    def key_requeues_formatter_stats
+      key("requeues_formatter_stats")
+    end
+
     # The total number of examples, those that were requeued.
     #
     # redis: STRING<integer>

lib/rspecq/worker.rb

@@ -46,6 +46,19 @@ class Worker
     # Defaults to 0
     attr_accessor :fail_fast
 
+    # Output Junit formatted XML to a specifiedd file
+    #
+    # Example: test_results/results-{{TEST_ENV_NUMBER}}-{{JOB_INDEX}}.xml
+    # where TEST_ENV_NUMBER is substituted with the environment variable
+    # from the gem parallel test, and JOB_INDEX is incremented based
+    # on the number of test suites run in the current process.
+    attr_accessor :junit_output
+
+    # Optional arguments to pass along to rspec.
+    #
+    # Defaults to nil
+    attr_accessor :rspec_args
+
     # Time to wait for a queue to be published.
     #
     # Defaults to 30
@@ -73,6 +86,7 @@ def initialize(build_id:, worker_id:, redis_opts:)
       @file_split_threshold = 999_999
       @heartbeat_updated_at = nil
       @max_requeues = 3
+      @junit_output = nil
       @queue_wait_timeout = 30
       @seed = srand && (srand % 0xFFFF)
       @tags = []
@@ -82,14 +96,17 @@ def initialize(build_id:, worker_id:, redis_opts:)
       RSpec::Core::Formatters.register(Formatters::ExampleCountRecorder, :dump_summary)
       RSpec::Core::Formatters.register(Formatters::FailureRecorder, :example_failed, :message)
       RSpec::Core::Formatters.register(Formatters::WorkerHeartbeatRecorder, :example_finished)
+      RSpec::Core::Formatters.register(Formatters::JUnitFormatter, :example_failed, :start, :stop, :dump_summary)
     end
 
     def work
       puts "Working for build #{@build_id} (worker=#{@worker_id})"
 
       try_publish_queue!(queue)
+
       queue.wait_until_published(queue_wait_timeout)
       queue.save_worker_seed(@worker_id, seed)
+      idx = 0
 
       loop do
         # we have to bootstrap this so that it can be used in the first call
@@ -119,6 +136,12 @@ def work
         RSpec.configuration.detail_color = :magenta
         RSpec.configuration.seed = seed
         RSpec.configuration.backtrace_formatter.filter_gem("rspecq")
+
+        if junit_output
+          RSpec.configuration.add_formatter(Formatters::JUnitFormatter.new(queue, job, max_requeues,
+                                                                           idx, junit_output))
+        end
+
         RSpec.configuration.add_formatter(Formatters::FailureRecorder.new(queue, job, max_requeues, @worker_id))
         RSpec.configuration.add_formatter(Formatters::ExampleCountRecorder.new(queue))
         RSpec.configuration.add_formatter(Formatters::WorkerHeartbeatRecorder.new(self))
@@ -127,12 +150,13 @@ def work
           RSpec.configuration.add_formatter(Formatters::JobTimingRecorder.new(queue, job))
         end
 
-        options = ["--format", "progress", job]
+        options = [*rspec_args, "--format", "progress", job]
         tags.each { |tag| options.push("--tag", tag) }
         opts = RSpec::Core::ConfigurationOptions.new(options)
         _result = RSpec::Core::Runner.new(opts).run($stderr, $stdout)
 
         queue.acknowledge_job(job)
+        idx += 1
       end
     end
 

rspecq.gemspec

@@ -19,6 +19,7 @@ Gem::Specification.new do |s|
   end
 
   s.add_dependency "redis"
+  s.add_dependency "rspec_junit_formatter"
   s.add_dependency "sentry-ruby"
 
   s.add_development_dependency "minitest"

test/test_e2e.rb

@@ -1,6 +1,12 @@
 require "test_helpers"
 
 class TestEndToEnd < RSpecQTest
+  def after_teardown
+    Dir["./test/sample_suites/flakey_suite/test/test_results/**/*"].each do |file|
+      File.delete(file)
+    end
+  end
+
   def test_suite_with_legit_failures
     queue = exec_build("failing_suite")
 
@@ -129,4 +135,26 @@ def test_suite_with_failures_and_fail_fast
 
     assert_includes [2, 3], queue.processed_jobs.length
   end
+
+  def test_suite_with_rspec_arguments
+    queue = exec_build("tagged_suite", "-- --tag foo")
+
+    assert_equal 1, queue.example_count
+  end
+
+  def test_suite_with_junit_output
+    queue = exec_build("flakey_suite",
+                       "--junit-output test/test_results/test.{{JOB_INDEX}}.xml")
+
+    assert queue.build_successful?
+    assert_processed_jobs [
+      "./spec/foo_spec.rb",
+      "./spec/foo_spec.rb[1:1]",
+    ], queue
+
+    assert_equal({ "./spec/foo_spec.rb[1:1]" => "2" }, queue.requeued_jobs)
+    assert File.exist?("test/sample_suites/flakey_suite/test/test_results/test.0.xml")
+    assert File.exist?("test/sample_suites/flakey_suite/test/test_results/test.1.xml")
+    assert File.exist?("test/sample_suites/flakey_suite/test/test_results/test.2.xml")
+  end
 end