Fix database connection using a socket on CI

.github/workflows/ci.yaml

@@ -15,34 +15,38 @@ jobs:
         - '3.3'
     services:
       postgres:
-        image: manageiq/postgresql:13
+        image: fryguy9/postgresql-with-config:latest
         env:
           POSTGRESQL_USER: root
           POSTGRESQL_PASSWORD: smartvm
-          POSTGRESQL_DATABASE: vmdb_production
-        options: "--name postgres --volume /tmp/postgresql-cfg/:/opt/app-root/src/postgresql-cfg/
-          --health-cmd pg_isready --health-interval 2s --health-timeout 5s --health-retries
-          5"
+          POSTGRESQL_DATABASE: temp
+        options: "--name postgres --health-cmd pg_isready --health-interval 2s --health-timeout 5s
+          --health-retries 5"
         ports:
         - 5432:5432
     steps:
     - uses: actions/checkout@v4
-    - name: Override postgres settings
-      run: |
-        sudo chown -R $(id -u):$(id -g) /tmp/postgresql-cfg
-        cp -rf .github/ci/postgresql-cfg/* /tmp/postgresql-cfg
-        /usr/bin/docker run --rm -v "/var/run/docker.sock":"/var/run/docker.sock" docker restart postgres
-        until [ "`/usr/bin/docker inspect -f {{.State.Health.Status}} postgres`" == "healthy" ]; do sleep 0.5; done
+    - name: Verify libpq
+      run: ls -al /usr/lib
     - name: Set up Ruby
       uses: ruby/setup-ruby@v1
       with:
         ruby-version: "${{ matrix.ruby-version }}"
         bundler-cache: true
       timeout-minutes: 30
+    - name: Set up tests
+      run: bundle exec rake spec:setup
+      env:
+        POSTGRESQL_HOST: localhost
+        POSTGRESQL_USER: root
+        POSTGRESQL_PASSWORD: smartvm
     - name: Run tests
       run: bundle exec rake
       env:
-        CC_TEST_REPORTER_ID: "${{ secrets.CC_TEST_REPORTER_ID }}"
+        POSTGRESQL_HOST: localhost
+        POSTGRESQL_USER: root
+        POSTGRESQL_PASSWORD: smartvm
+        CC_TEST_REPORTER_ID: ${{ secrets.CC_TEST_REPORTER_ID }}
     - if: ${{ github.ref == 'refs/heads/master' && matrix.ruby-version == '3.3' }}
       name: Report code coverage
       continue-on-error: true

Rakefile

@@ -1,17 +1,31 @@
 require "bundler/gem_tasks"
 require "rspec/core/rake_task"
 
+require_relative "spec/support/database_helper"
+require_relative "spec/support/connection_helper"
+
 def create_database(dbname)
-  require "pg"
-  c = PG::Connection.new(:dbname => "postgres")
+  c = ConnectionHelper.connection_for("postgres")
   c.async_exec("CREATE DATABASE #{dbname}")
 rescue PG::DuplicateDatabase => err
   raise unless err.message =~ /already exists/
 end
 
+def create_tables(dbname)
+  c = ConnectionHelper.connection_for(dbname)
+
+  DatabaseHelper.tables.each do |t|
+    c.async_exec(<<-SQL)
+      CREATE TABLE IF NOT EXISTS #{t} (
+        id   SERIAL PRIMARY KEY,
+        data VARCHAR(50)
+      )
+    SQL
+  end
+end
+
 def drop_database(dbname)
-  require "pg"
-  c = PG::Connection.new(:dbname => "postgres")
+  c = ConnectionHelper.connection_for("postgres")
   c.async_exec("DROP DATABASE #{dbname}")
 rescue PG::InvalidCatalogName => err
   raise unless err.message =~ /does not exist/
@@ -22,6 +36,9 @@ namespace :spec do
   task :setup => :teardown do
     create_database("logical_test")
     create_database("logical_test_target")
+
+    create_tables("logical_test")
+    create_tables("logical_test_target")
   end
 
   desc "Teardown the test databases"

lib/pg/logical_replication/client.rb

@@ -353,6 +353,9 @@ def safe_list(list)
       end
 
       def typed_exec(sql, *params)
+        puts "sql: #{sql}"
+        puts "params: #{params.inspect}"
+        puts "type_map_for_queries: #{self.class.type_map_for_queries(connection).inspect}"
         result = connection.async_exec(sql, params, nil, self.class.type_map_for_queries(connection))
         result.map_types!(self.class.type_map_for_results(connection))
       end

spec/client_spec.rb

@@ -8,8 +8,6 @@
   let(:pub_name) { "test_pub" }
   let(:sub_name) { "test_sub" }
 
-  before(:all) { DatabaseHelper.create_tables }
-
   around do |example|
     DatabaseHelper.with_clean_environment { example.call }
   end

spec/spec_helper.rb

@@ -4,80 +4,4 @@
 $LOAD_PATH.unshift File.expand_path("../../lib", __FILE__)
 require "pg-logical_replication"
 
-module ConnectionHelper
-  def self.source_database_connection
-    @source_database_connection ||= PG::Connection.new(:dbname => "logical_test").tap do |c|
-      c.set_notice_receiver { |r| nil }
-    end
-  end
-
-  def self.target_database_connection
-    @target_database_connection ||= PG::Connection.new(:dbname => "logical_test_target").tap do |c|
-      c.set_notice_receiver { |r| nil }
-    end
-  end
-
-  def self.with_each_connection
-    [source_database_connection, target_database_connection].each do |conn|
-      yield conn
-    end
-  end
-end
-
-module DatabaseHelper
-  def self.tables
-    %w(table1 table2 table3 table4)
-  end
-
-  def self.create_tables
-    ConnectionHelper.with_each_connection do |conn|
-      tables.each do |t|
-        conn.async_exec(<<-SQL)
-          CREATE TABLE IF NOT EXISTS #{t} (
-            id   SERIAL PRIMARY KEY,
-            data VARCHAR(50)
-          )
-        SQL
-      end
-    end
-  end
-
-  def self.drop_tables
-    ConnectionHelper.with_each_connection do |conn|
-      tables.each { |t| conn.async_exec("DROP TABLE IF EXISTS #{t}") }
-    end
-  end
-
-  def self.drop_subscriptions
-    conn = ConnectionHelper.target_database_connection
-    # Subscriptions are visible from all databases in the cluster so we need to specify only the subs from the target database.
-    conn.async_exec("SELECT subname::TEXT FROM pg_subscription AS sub JOIN pg_database ON sub.subdbid = pg_database.oid WHERE pg_database.datname = current_database()").values.flatten.each do |s|
-      conn.async_exec("ALTER subscription #{s} DISABLE")
-      conn.async_exec("ALTER subscription #{s} SET (slot_name = NONE)")
-      conn.async_exec("DROP SUBSCRIPTION #{s}")
-    end
-  end
-
-  def self.drop_publications
-    conn = ConnectionHelper.source_database_connection
-    conn.async_exec("SELECT pubname::TEXT from pg_publication").values.flatten.each do |p|
-      conn.async_exec("DROP PUBLICATION #{p}")
-    end
-  end
-
-  def self.drop_replication_slots
-    conn = ConnectionHelper.source_database_connection
-    # replication_slots are visible from all databases in the cluster so we need to specify only the slots from the source database.
-    conn.async_exec("SELECT slot_name::TEXT FROM pg_replication_slots WHERE slot_type = 'logical' AND NOT active AND database = current_database()").values.flatten.each do |slot|
-      conn.async_exec("SELECT pg_drop_replication_slot('#{slot}')")
-    end
-  end
-
-  def self.with_clean_environment
-    yield
-  ensure
-    drop_subscriptions
-    drop_publications
-    drop_replication_slots
-  end
-end
+Dir[File.join(__dir__, "support/**/*.rb")].each { |f| require f }

spec/support/connection_helper.rb

@@ -0,0 +1,27 @@
+module ConnectionHelper
+  def self.connection_for(dbname)
+    require "pg"
+    conn_str = "postgresql://#{ENV["POSTGRESQL_USER"]}:#{ENV["POSTGRESQL_PASSWORD"]}@#{ENV["POSTGRESQL_HOST"]}:5432/#{dbname}?sslmode=disable"
+    PG::Connection.new(conn_str).tap do |c|
+      raise "Invalid connection" unless c.exec("SELECT 1").values == [["1"]]
+    end
+  end
+
+  def self.source_database_connection
+    @source_database_connection ||= connection_for("logical_test").tap do |c|
+      c.set_notice_receiver { |r| nil }
+    end
+  end
+
+  def self.target_database_connection
+    @target_database_connection ||= connection_for("logical_test_target").tap do |c|
+      c.set_notice_receiver { |r| nil }
+    end
+  end
+
+  def self.with_each_connection
+    [source_database_connection, target_database_connection].each do |conn|
+      yield conn
+    end
+  end
+end

spec/support/database_helper.rb

@@ -0,0 +1,40 @@
+require_relative "connection_helper"
+
+module DatabaseHelper
+  def self.tables
+    %w(table1 table2 table3 table4)
+  end
+
+  def self.drop_subscriptions
+    conn = ConnectionHelper.target_database_connection
+    # Subscriptions are visible from all databases in the cluster so we need to specify only the subs from the target database.
+    conn.async_exec("SELECT subname::TEXT FROM pg_subscription AS sub JOIN pg_database ON sub.subdbid = pg_database.oid WHERE pg_database.datname = current_database()").values.flatten.each do |s|
+      conn.async_exec("ALTER subscription #{s} DISABLE")
+      conn.async_exec("ALTER subscription #{s} SET (slot_name = NONE)")
+      conn.async_exec("DROP SUBSCRIPTION #{s}")
+    end
+  end
+
+  def self.drop_publications
+    conn = ConnectionHelper.source_database_connection
+    conn.async_exec("SELECT pubname::TEXT from pg_publication").values.flatten.each do |p|
+      conn.async_exec("DROP PUBLICATION #{p}")
+    end
+  end
+
+  def self.drop_replication_slots
+    conn = ConnectionHelper.source_database_connection
+    # replication_slots are visible from all databases in the cluster so we need to specify only the slots from the source database.
+    conn.async_exec("SELECT slot_name::TEXT FROM pg_replication_slots WHERE slot_type = 'logical' AND NOT active AND database = current_database()").values.flatten.each do |slot|
+      conn.async_exec("SELECT pg_drop_replication_slot('#{slot}')")
+    end
+  end
+
+  def self.with_clean_environment
+    yield
+  ensure
+    drop_subscriptions
+    drop_publications
+    drop_replication_slots
+  end
+end