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