RUBY-2481 unified runner changes

spec/runners/unified/event_subscriber.rb

@@ -28,4 +28,63 @@ def add_wanted_events(kind)
       @wanted_events[kind] = true
     end
   end
+
+  class StoringEventSubscriber
+    def initialize(&block)
+      @operations = {}
+      @handler = block
+    end
+
+    def started(event)
+      started_at = Time.now
+      @operations[event.operation_id] = [event, started_at]
+      @handler.call(
+        'name' => event.class.name.sub(/.*::/, '') + 'Event',
+        'commandName' => event.command_name,
+        'startTime' => started_at.to_f,
+        'address' => event.address.seed,
+      )
+    end
+
+    def succeeded(event)
+      started_event, started_at = @operations.delete(event.operation_id)
+      raise "Started event for #{event.operation_id} not found" unless started_event
+      @handler.call(
+        'name' => event.class.name.sub(/.*::/, '') + 'Event',
+        'commandName' => started_event.command_name,
+        'duration' => event.duration,
+        'startTime' => started_at.to_f,
+        'address' => started_event.address.seed,
+      )
+    end
+
+    def failed(event)
+      started_event, started_at = @operations.delete(event.operation_id)
+      raise "Started event for #{event.operation_id} not found" unless started_event
+      @handler.call(
+        'name' => event.class.name.sub(/.*::/, '') + 'Event',
+        'commandName' => started_event.command_name,
+        'duration' => event.duration,
+        'failure' => event.failure,
+        'startTime' => started_at.to_f,
+        'address' => started_event.address.seed,
+      )
+    end
+
+    def published(event)
+      payload = {
+        'name' => event.class.name.sub(/.*::/, '') + 'Event',
+        'time' => Time.now.to_f,
+        'address' => event.address.seed,
+      }.tap do |entry|
+        if event.respond_to?(:connection_id)
+          entry['connectionId'] = event.connection_id
+        end
+        if event.respond_to?(:reason)
+          entry['reason'] = event.reason
+        end
+      end
+      @handler.call(payload)
+    end
+  end
 end

spec/runners/unified/support_operations.rb

@@ -152,6 +152,42 @@ def assert_session_unpinned(op)
       assert_session_pinned(op, false)
     end
 
+    def _loop(op)
+      consume_test_runner(op)
+      use_arguments(op) do |args|
+        ops = args.use!('operations')
+
+        if store_errors = args.use('storeErrorsAsEntity')
+          entities.set(:error_list, store_errors, [])
+        end
+
+        store_iterations = args.use('storeIterationsAsEntity')
+        iterations = 0
+
+        loop do
+          break if stop?
+          begin
+            execute_operations(ops.map(&:dup))
+          rescue => e
+            if store_errors
+              STDERR.puts "Error: #{e.class}: #{e}"
+              entities.get(:error_list, store_errors) << {
+                error: "#{e.class}: #{e}",
+                time: Time.now.to_f,
+              }
+            else
+              raise
+            end
+          end
+          iterations += 1
+        end
+
+        if store_iterations
+          entities.set(:iteration_count, store_iterations, iterations)
+        end
+      end
+    end
+
     private
 
     def assert_no_arguments(op)

spec/runners/unified/test.rb

@@ -87,6 +87,27 @@ def create_entities
             opts = {}
           end
 
+          if store_events = spec.use('storeEventsAsEntities')
+            store_event_names = {}
+            store_events.each do |event_name, entity_name|
+              #event_name = event_name.gsub(/Event$/, '').gsub(/[A-Z]/) { |m| "_#{m}" }.upcase
+              #event_name = event_name.gsub(/Event$/, '').sub(/./) { |m| m.upcase }
+              store_event_names[event_name] = entity_name
+            end
+            store_event_names.values.uniq.each do |entity_name|
+              entities.set(:event_list, entity_name, [])
+            end
+            subscriber = StoringEventSubscriber.new do |payload|
+              if entity_name = store_event_names[payload['name']]
+                entities.get(:event_list, entity_name) << payload
+              end
+            end
+            opts[:sdam_proc] = lambda do |client|
+              client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+              client.subscribe(Mongo::Monitoring::CONNECTION_POOL, subscriber)
+            end
+          end
+
           create_client(**opts).tap do |client|
             if oe = spec.use('observeEvents')
               oe.each do |event|
@@ -176,6 +197,39 @@ def run
       disable_fail_points
     end
 
+    def stop!
+      @stop = true
+    end
+
+    def stop?
+      !!@stop
+    end
+
+    def cleanup
+      if $kill_transactions || true
+        begin
+          root_authorized_client.command(
+            killAllSessions: [],
+          )
+        rescue Mongo::Error::OperationFailure => e
+          if e.code == 11601
+            # operation was interrupted, ignore
+          elsif e.code == 59
+            # no such command (old server), ignore
+          else
+            raise
+          end
+        end
+        $kill_transactions = nil
+      end
+
+      entities[:client]&.each do |id, client|
+        client.close
+      end
+    end
+
+    private
+
     def execute_operations(ops)
       ops.each do |op|
         execute_operation(op)
@@ -185,9 +239,13 @@ def execute_operations(ops)
     def execute_operation(op)
       use_all(op, 'operation', op) do |op|
         name = Utils.underscore(op.use!('name'))
+        method_name = name
+        if name.to_s == 'loop'
+          method_name = "_#{name}"
+        end
         if expected_error = op.use('expectError')
           begin
-            send(name, op)
+            send(method_name, op)
           rescue Mongo::Error, BSON::String::IllegalKey => e
             if expected_error.use('isClientError')
               unless BSON::String::IllegalKey === e
@@ -227,7 +285,7 @@ def execute_operation(op)
             raise Error::ErrorMismatch, "Expected exception but none was raised"
           end
         else
-          result = send(name, op)
+          result = send(method_name, op)
           if expected_result = op.use('expectResult')
             if !expected_result.empty? && result.nil?
               raise Error::ResultMismatch, "Actual result nil but expected result #{expected_result}"
@@ -263,29 +321,6 @@ def use_arguments(op, &block)
       use_sub(op, 'arguments', &block)
     end
 
-    def cleanup
-      if $kill_transactions || true
-        begin
-          root_authorized_client.command(
-            killAllSessions: [],
-          )
-        rescue Mongo::Error::OperationFailure => e
-          if e.code == 11601
-            # operation was interrupted, ignore
-          elsif e.code == 59
-            # no such command (old server), ignore
-          else
-            raise
-          end
-        end
-        $kill_transactions = nil
-      end
-
-      entities[:client]&.each do |id, client|
-        client.close
-      end
-    end
-
     def disable_fail_points
       if $disable_fail_points
         $disable_fail_points.each do |(fail_point_command, address)|

spec/runners/unified/using_hash.rb

@@ -21,11 +21,7 @@ def wrap(v)
         self.class[v]
       when Array
         v.map do |subv|
-          if Hash === subv
-            self.class[subv]
-          else
-            subv
-          end
+          wrap(subv)
         end
       else
         v