149 lines
3.6 KiB
Ruby
149 lines
3.6 KiB
Ruby
#
|
|
# rake test or ruby -Itest test/core/chained_test.rb
|
|
#
|
|
# .env is NOT loaded: no test may reach the production database.
|
|
# postgres tests use a local server and a throw-away database per test:
|
|
# ETL_TEST_PG_HOST (default localhost), ETL_TEST_PG_PORT, ETL_TEST_PG_USER (default $USER), ETL_TEST_PG_PASSWORD
|
|
# without a reachable server they are skipped.
|
|
#
|
|
require 'bundler/setup'
|
|
require 'minitest/autorun'
|
|
require 'json'
|
|
require 'logger'
|
|
require 'stringio'
|
|
require 'tmpdir'
|
|
|
|
require 'datapipeline'
|
|
|
|
ROOT = File.expand_path('..', __dir__)
|
|
|
|
# source and target in memory
|
|
class MemoryStorage < DataPipeline::StoragePlugin
|
|
attr_reader :rows, :written, :streamed_rows
|
|
|
|
def initialize rows: [], **options
|
|
@options = options
|
|
@rows = rows
|
|
@written = nil
|
|
@streamed_rows = []
|
|
@closed = false
|
|
end
|
|
|
|
def read(&block)
|
|
return @rows.map(&:dup) unless block
|
|
|
|
@rows.map(&:dup).each(&block)
|
|
nil
|
|
end
|
|
|
|
def write(rows) = (@written = rows)
|
|
def write_row(row) = (@streamed_rows << row)
|
|
def close = (@closed = true)
|
|
def closed? = @closed
|
|
|
|
# rows of the last batch write or of the streamed rows
|
|
def output = @written || @streamed_rows
|
|
end
|
|
|
|
DataPipeline.register :memory, MemoryStorage
|
|
|
|
module PipelineHelper
|
|
QUIET = Logger.new(IO::NULL)
|
|
|
|
def setup
|
|
super
|
|
DataPipeline.reset_settings!
|
|
DataPipeline.runs.clear
|
|
DataPipeline.params.clear
|
|
end
|
|
|
|
# runs a pipeline from rows into a MemoryStorage, returns [target, stats]
|
|
def run_pipeline rows = [], **settings, &block
|
|
source = MemoryStorage.new(rows:)
|
|
target = MemoryStorage.new
|
|
stats = DataPipeline.new(**settings) do
|
|
logger QUIET
|
|
read source
|
|
write target
|
|
instance_eval(&block) if block
|
|
end.run
|
|
[target, stats]
|
|
end
|
|
|
|
def output rows = [], **settings, &block
|
|
run_pipeline(rows, **settings, &block).first.output
|
|
end
|
|
|
|
def with_tmpdir(&block)
|
|
Dir.mktmpdir('etl-test', &block)
|
|
end
|
|
|
|
def capture_stdout
|
|
old = $stdout
|
|
$stdout = StringIO.new
|
|
yield
|
|
$stdout.string
|
|
ensure
|
|
$stdout = old
|
|
end
|
|
end
|
|
|
|
module PgHelper
|
|
def self.config
|
|
{
|
|
host: ENV.fetch('ETL_TEST_PG_HOST', 'localhost'),
|
|
port: ENV.fetch('ETL_TEST_PG_PORT', '5432'),
|
|
username: ENV.fetch('ETL_TEST_PG_USER', ENV['USER']),
|
|
password: ENV.fetch('ETL_TEST_PG_PASSWORD', nil)
|
|
}
|
|
end
|
|
|
|
def self.admin
|
|
require 'pg'
|
|
c = config
|
|
PG.connect(host: c[:host], port: c[:port], user: c[:username], password: c[:password], dbname: 'postgres',
|
|
connect_timeout: 3)
|
|
end
|
|
|
|
def self.available?
|
|
return @available unless @available.nil?
|
|
|
|
@available = begin
|
|
admin.close
|
|
true
|
|
rescue StandardError, LoadError
|
|
false
|
|
end
|
|
end
|
|
|
|
# yields connection options for a fresh database, dropped afterwards
|
|
def with_pg_database
|
|
skip 'no local postgres (ETL_TEST_PG_HOST)' unless PgHelper.available?
|
|
|
|
name = "etl_test_#{Process.pid}_#{rand(1_000_000)}"
|
|
admin = PgHelper.admin
|
|
admin.exec("create database #{name}")
|
|
yield PgHelper.config.merge(database: name)
|
|
ensure
|
|
if admin
|
|
admin.exec("select pg_terminate_backend(pid) from pg_stat_activity where datname = '#{name}' and pid <> pg_backend_pid()")
|
|
admin.exec("drop database if exists #{name}")
|
|
admin.close
|
|
end
|
|
end
|
|
|
|
# plain sql, several statements allowed (setup of test tables)
|
|
def pg_exec options, sql
|
|
conn = PG.connect(host: options[:host], port: options[:port], user: options[:username],
|
|
password: options[:password], dbname: options[:database])
|
|
conn.exec(sql).to_a
|
|
ensure
|
|
conn&.close
|
|
end
|
|
end
|
|
|
|
class PipelineTest < Minitest::Test
|
|
include PipelineHelper
|
|
include PgHelper
|
|
end
|