Merge pull request #20818 from jeremywadsack/use_transactional_fixtures_all_databases

Use notification to ensure that lazy-loaded model classes have transactions

activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb

@@ -849,6 +849,21 @@ def establish_connection(config)
 
         remove_connection(spec.name)
         owner_to_pool[spec.name] = ConnectionAdapters::ConnectionPool.new(spec)
+
+        message_bus = ActiveSupport::Notifications.instrumenter
+        payload = {
+          connection_id: object_id
+        }
+        if spec
+          payload[:class_name] = spec.name
+          payload[:config] = spec.config
+        end
+
+        message_bus.instrument('!connection.active_record', payload) do
+          owner_to_pool[spec.name] = ConnectionAdapters::ConnectionPool.new(spec)
+        end
+
+        owner_to_pool[spec.name]
       end
 
       # Returns true if there are any active connections among the connection

activerecord/lib/active_record/fixtures.rb

@@ -968,6 +968,7 @@ def setup_fixtures(config = ActiveRecord::Base)
       @fixture_cache = {}
       @fixture_connections = []
       @@already_loaded_fixtures ||= {}
+      @connection_subscriber = nil
 
       # Load fixtures once and begin transaction.
       if run_in_transaction?
@@ -977,10 +978,36 @@ def setup_fixtures(config = ActiveRecord::Base)
           @loaded_fixtures = load_fixtures(config)
           @@already_loaded_fixtures[self.class] = @loaded_fixtures
         end
+
+        # Begin transactions for connections already established
         @fixture_connections = enlist_fixture_connections
         @fixture_connections.each do |connection|
           connection.begin_transaction joinable: false
         end
+
+        # When connections are established in the future, begin a transaction too
+        @connection_subscriber = ActiveSupport::Notifications.subscribe('!connection.active_record') do |_, _, _, _, payload|
+          model_class = nil
+          begin
+            model_class = payload[:class_name].constantize if payload[:class_name]
+          rescue NameError
+            model_class = nil
+          end
+
+          if model_class
+            begin
+              connection = ActiveRecord::Base.connection_handler.retrieve_connection(model_class)
+            rescue ConnectionNotEstablished
+              connection = nil
+            end
+
+            if connection && !@fixture_connections.include?(connection)
+              connection.begin_transaction joinable: false
+              @fixture_connections << connection
+            end
+          end
+        end
+
       # Load fixtures for every test.
       else
         ActiveRecord::FixtureSet.reset_cache
@@ -995,6 +1022,7 @@ def setup_fixtures(config = ActiveRecord::Base)
     def teardown_fixtures
       # Rollback changes if a transaction is active.
       if run_in_transaction?
+        ActiveSupport::Notifications.unsubscribe(@connection_subscriber) if @connection_subscriber
         @fixture_connections.each do |connection|
           connection.rollback_transaction if connection.transaction_open?
         end

activerecord/test/cases/connection_pool_test.rb

@@ -341,6 +341,18 @@ def test_anonymous_class_exception
         end
       end
 
+      def test_connection_notification_is_called
+        payloads = []
+        subscription = ActiveSupport::Notifications.subscribe('!connection.active_record') do |name, started, finished, unique_id, payload|
+          payloads << payload
+        end
+        ActiveRecord::Base.establish_connection :arunit
+        assert_equal [:class_name, :config, :connection_id], payloads[0].keys.sort
+        assert_equal 'primary', payloads[0][:class_name]
+      ensure
+        ActiveSupport::Notifications.unsubscribe(subscription) if subscription
+      end
+
       def test_pool_sets_connection_schema_cache
         connection = pool.checkout
         schema_cache = SchemaCache.new connection

activerecord/test/cases/fixtures_test.rb

@@ -622,6 +622,46 @@ def test_it_twice_in_whatever_order_to_check_for_fixture_leakage
   end
 end
 
+class TransactionalFixturesOnConnectionNotification < ActiveRecord::TestCase
+  self.use_transactional_tests = true
+  self.use_instantiated_fixtures = false
+
+  def test_transaction_created_on_connection_notification
+    connection = stub(:transaction_open? => false)
+    connection.expects(:begin_transaction).with(joinable: false)
+    fire_connection_notification(connection)
+  end
+
+  def test_notification_established_transactions_are_rolled_back
+    # Mocha is not thread-safe so define our own stub to test
+    connection = Class.new do
+      attr_accessor :rollback_transaction_called
+      def transaction_open?; true; end
+      def begin_transaction(*args); end
+      def rollback_transaction(*args)
+        @rollback_transaction_called = true
+      end
+    end.new
+    fire_connection_notification(connection)
+    teardown_fixtures
+    assert(connection.rollback_transaction_called, "Expected <mock connection>#rollback_transaction to be called but was not")
+  end
+
+  private
+
+    def fire_connection_notification(connection)
+      ActiveRecord::Base.connection_handler.stubs(:retrieve_connection).with(Book).returns(connection)
+      message_bus = ActiveSupport::Notifications.instrumenter
+      payload = {
+        class_name: 'Book',
+        config: nil,
+        connection_id: connection.object_id
+      }
+
+      message_bus.instrument('!connection.active_record', payload) {}
+    end
+end
+
 class InvalidTableNameFixturesTest < ActiveRecord::TestCase
   fixtures :funny_jokes
   # Set to false to blow away fixtures cache and ensure our fixtures are loaded