95 lines
3.5 KiB
Ruby
95 lines
3.5 KiB
Ruby
require_relative '../test_helper'
|
|
|
|
# dsl, settings, plugin lookup
|
|
class PipelineCoreTest < PipelineTest
|
|
ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
|
|
|
|
def settings_of(pipeline) = pipeline.instance_variable_get(:@settings)
|
|
|
|
def test_dsl_switches
|
|
assert_equal false, settings_of(DataPipeline.new { verbose false })[:verbose]
|
|
assert_equal true, settings_of(DataPipeline.new { verbose })[:verbose]
|
|
assert_equal false, settings_of(DataPipeline.new { streamed false })[:streamed]
|
|
assert_equal '1.0', settings_of(DataPipeline.new { version '1.0' })[:version]
|
|
assert_equal true, settings_of(DataPipeline.new { configure verbose: true })[:verbose]
|
|
end
|
|
|
|
def test_typo_raises
|
|
assert_raises(NoMethodError) { DataPipeline.new { verbsoe true } }
|
|
end
|
|
|
|
def test_unknown_plugin
|
|
error = assert_raises(RuntimeError) { DataPipeline.new { container :gibtsnicht } }
|
|
assert_includes error.message, 'GibtsnichtStorage'
|
|
end
|
|
|
|
def test_plugin_errors_pass_through
|
|
assert_raises(ArgumentError) { DataPipeline.new { container :csv } }
|
|
broken = Class.new(DataPipeline::StoragePlugin) { def initialize(**) = undefined_thing }
|
|
DataPipeline.register :broken, broken
|
|
assert_includes assert_raises(NameError) { DataPipeline.new { container :broken } }.message, 'undefined_thing'
|
|
end
|
|
|
|
def test_register_lazy_path
|
|
with_tmpdir do |dir|
|
|
File.write("#{dir}/lazy_storage.rb", "class LazyStorage < MemoryStorage; end\n")
|
|
DataPipeline.register :lazy, 'LazyStorage', path: "#{dir}/lazy_storage.rb"
|
|
assert_empty $LOADED_FEATURES.grep(/lazy_storage/)
|
|
storage = nil
|
|
DataPipeline.new { storage = container :lazy }
|
|
assert_equal 'LazyStorage', storage.class.name
|
|
refute_empty $LOADED_FEATURES.grep(/lazy_storage/)
|
|
end
|
|
end
|
|
|
|
def test_csv_plugin_by_convention
|
|
assert_equal DataPipeline::CsvStorage, DataPipeline.plugin(:csv)
|
|
end
|
|
|
|
def test_settings_are_frozen_and_replaced
|
|
assert DataPipeline.settings.frozen?
|
|
pipeline = DataPipeline.new { logger PipelineHelper::QUIET }
|
|
DataPipeline.settings(verbose: true)
|
|
assert_equal false, settings_of(pipeline)[:verbose]
|
|
assert_equal true, DataPipeline.settings[:verbose]
|
|
DataPipeline.reset_settings!
|
|
assert_equal DataPipeline::DEFAULT_SETTINGS, DataPipeline.settings
|
|
end
|
|
|
|
def test_settings_per_pipeline
|
|
_target, stats = run_pipeline(ROWS, limit: 1)
|
|
assert_equal 1, stats.read
|
|
assert_nil DataPipeline.settings[:limit]
|
|
end
|
|
|
|
def test_parallel_threads
|
|
reads = Array.new(8) { |i| Thread.new { run_pipeline(ROWS, limit: (i % 3) + 1).last.read } }.map(&:value)
|
|
assert_equal Array.new(8) { |i| (i % 3) + 1 }, reads
|
|
end
|
|
|
|
def test_removed_dsl_of_version_2
|
|
%i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
|
|
end
|
|
|
|
def test_copy_without_steps
|
|
assert_equal ROWS, output(ROWS)
|
|
end
|
|
|
|
def test_closes_source_and_target
|
|
source = MemoryStorage.new(rows: ROWS)
|
|
target = MemoryStorage.new
|
|
DataPipeline.new { logger PipelineHelper::QUIET; read source; write target }.run
|
|
assert source.closed?
|
|
assert target.closed?
|
|
end
|
|
|
|
def test_read_and_write_required
|
|
assert_includes assert_raises(ArgumentError) { DataPipeline.new { logger PipelineHelper::QUIET; read MemoryStorage.new }.run }.message, 'write'
|
|
assert_includes assert_raises(ArgumentError) { DataPipeline.new { logger PipelineHelper::QUIET; write MemoryStorage.new }.run }.message, 'read'
|
|
end
|
|
|
|
def test_block_required
|
|
assert_raises(ArgumentError) { DataPipeline.new { map } }
|
|
end
|
|
end
|