datapipeline/test/core/operations_test.rb

require_relative '../test_helper'

# on_error, validate, stats, hooks, watermark, dry run, limit
class OperationsTest < PipelineTest
  ROWS = (1..5).map { |i| { 'id' => i.to_s, 'shkzg' => i == 3 ? 'X' : 'S', 'betrag' => i == 4 ? 'abc' : "#{i}.50" } }.freeze
  RULES = { 'id' => { required: true, format: /\A\d+\z/ }, 'shkzg' => { in: %w[S H] }, 'betrag' => { type: :decimal } }.freeze

  def failing_map = proc { map { |r| r['id'] == '2' ? raise('kaputt') : r } }

  def test_abort_is_default
    error = assert_raises(RuntimeError) { run_pipeline(ROWS, &failing_map) }
    assert_equal 'kaputt', error.message
    assert_equal 'kaputt', DataPipeline.runs.last.error.message
  end

  def test_skip
    map = failing_map
    target, stats = run_pipeline(ROWS) { on_error :skip; instance_eval(&map) }
    assert_equal 4, target.output.size
    assert_equal 1, stats.rejected
    assert_equal 'map#1', stats.errors.first[:step]
    refute stats.ok?
  end

  def test_collect_batch_and_stream
    [false, true].each do |streamed|
      rejects = MemoryStorage.new
      map = failing_map
      target, = run_pipeline(ROWS, streamed:) { on_error :collect, to: rejects; instance_eval(&map) }
      assert_equal [ROWS[1].merge('_step' => 'map#1', '_error' => 'kaputt')], rejects.streamed_rows
      assert_equal 4, target.output.size
    end
  end

  def test_on_error_arguments
    assert_raises(ArgumentError) { DataPipeline.new { on_error :collect } }
    assert_raises(ArgumentError) { DataPipeline.new { on_error :ignore } }
  end

  def test_edge_errors_always_abort
    assert_raises(RuntimeError) { run_pipeline(ROWS) { on_error :skip; header { raise 'h' }; map { |r| r } } }
  end

  def test_skip_in_parallel_mode
    target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
    assert_equal 4, target.output.size
    assert_equal 1, stats.rejected
  end

  def test_validate_rules
    assert_raises(DataPipeline::ValidationError) { run_pipeline(ROWS) { validate RULES } }
    target, stats = run_pipeline(ROWS) { on_error :skip; validate RULES }
    assert_equal %w[1 2 5], target.output.map { |r| r['id'] }
    assert_equal 2, stats.rejected
  end

  def test_validate_block
    target, stats = run_pipeline(ROWS) { on_error :skip; validate { |r| 'zu gross' if r['id'].to_i > 3 } }
    assert_equal 3, target.output.size
    assert_equal 'zu gross', stats.errors.first[:error]
  end

  def test_validation_messages
    errors = DataPipeline::Validation.errors({ 'a' => '12345', 'd' => '2025-13-01' },
                                             'a' => { max: 3, format: /x/ }, 'd' => { type: :date }, 'b' => { required: true })
    assert_equal ['a: longer than 3', 'a: format /x/', 'd: not date', 'b: missing'], errors
    assert_empty DataPipeline::Validation.errors({ 'a' => '' }, 'a' => { max: 1, type: :integer })
  end

  def test_validate_turns_chaining_on
    target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
    assert_equal 3, target.output.size
    assert(target.output.all? { |r| r['x'] })
  end

  def test_stats
    _target, stats = run_pipeline(ROWS) { map { |r| [r, r] } }
    assert_equal [5, 10, 10], [stats.read, stats.written, stats.output]
    assert stats.ok?
    assert_kind_of Float, stats.duration
    map = failing_map
    _target, stats = run_pipeline(ROWS) { on_error :skip; instance_eval(&map) }
    assert_match(/\Aread=5 written=4 rejected=1/, stats.to_s)
  end

  def test_on_finish_after_success_and_abort
    seen = []
    run_pipeline(ROWS) { on_finish { |s| seen << s.error } }
    assert_raises(RuntimeError) { run_pipeline(ROWS) { on_finish { |s| seen << s.error.class }; map { raise 'x' } } }
    assert_equal [nil, RuntimeError], seen
  end

  def test_hook_error_is_only_logged
    log = StringIO.new
    stats = DataPipeline.new do
      logger Logger.new(log)
      read MemoryStorage.new(rows: ROWS)
      write MemoryStorage.new
      on_finish { raise 'hook' }
    end.run
    assert stats.ok?
    assert_includes log.string, 'on_finish: RuntimeError: hook'
  end

  def test_logger_messages
    log = StringIO.new
    DataPipeline.new do
      logger Logger.new(log)
      on_error :skip
      read MemoryStorage.new(rows: ROWS)
      write MemoryStorage.new
      map { |r| r['id'] == '1' ? raise('weg') : r }
    end.run
    assert_includes log.string, 'rejected by map#1: weg'
    assert_includes log.string, 'finish read=5 written=4 rejected=1'
  end

  def test_watermark
    with_tmpdir do |dir|
      file = "#{dir}/state.json"
      wm = nil
      run_pipeline(ROWS) { wm = watermark :t, column: 'id', file:, initial: 0 }
      assert_equal '5', wm.value

      run_pipeline([{ 'id' => '9' }, { 'id' => '10' }]) { watermark :t, column: 'id', file: }
      assert_equal '10', JSON.parse(File.read(file))['t'], 'numeric texts compared as numbers'

      assert_raises(RuntimeError) { run_pipeline([{ 'id' => '99' }]) { watermark :t, column: 'id', file:; map { raise 'x' } } }
      assert_equal '10', JSON.parse(File.read(file))['t'], 'not saved after abort'

      run_pipeline(ROWS, streamed: true) { watermark :s, column: 'id', file: }
      assert_equal({ 't' => '10', 's' => '5' }, JSON.parse(File.read(file)))
    end
  end

  def test_dry_run
    with_tmpdir do |dir|
      rejects = MemoryStorage.new
      target = nil
      text = capture_stdout do
        target, = run_pipeline(ROWS, dry_run: true, preview: 2) do
          watermark :d, column: 'id', file: "#{dir}/state.json"
          on_error :collect, to: rejects
          map { |r| r['id'] == '1' ? raise('x') : r }
        end
      end
      assert_nil target.written
      assert_empty target.streamed_rows
      assert_empty rejects.streamed_rows
      refute File.exist?("#{dir}/state.json")
      assert text.start_with?('DRY RUN')
      assert_equal 3, text.lines.size
      assert DataPipeline.runs.last.dry_run
    end
  end

  def test_dry_run_creates_no_csv_file
    with_tmpdir do |dir|
      capture_stdout do
        DataPipeline.new(dry_run: true) do
          logger PipelineHelper::QUIET
          read MemoryStorage.new(rows: ROWS)
          write container(:csv, file: "#{dir}/dry.csv")
        end.run
      end
      refute File.exist?("#{dir}/dry.csv")
    end
  end

  def test_limit
    target, stats = run_pipeline(ROWS, limit: 2) { map { |r| r } }
    assert_equal [2, 2], [target.output.size, stats.read]

    target, stats = run_pipeline(ROWS, streamed: true, limit: 3) { trailer { |c| { 'n' => c.read } } }
    assert_equal({ 'n' => 3 }, target.output.last)
    assert_equal 3, stats.read
  end
end