require_relative '../test_helper' # chained steps: map, filter, header, trailer, context class ChainedTest < PipelineTest ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze def test_chained_transform_gets_previous_output out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } } assert_equal ROWS.map { |r| r.merge('b' => 1, 'c' => 2) }, out end def test_one_to_many_then_index out = output(ROWS) { chained; transform { |r| [r, r] }; transform { |r, i| r.merge('i' => i) } } assert_equal (0...6).to_a, out.map { |r| r['i'] } end def test_map_and_filter out = output(ROWS) do map { |r| r.merge('d' => r['a'] * 2) } filter { |r| r['d'] > 2 } map { |r, ctx| r.merge('i' => ctx.index) } end assert_equal [{ 'a' => 2, 'd' => 4, 'i' => 0 }, { 'a' => 3, 'd' => 6, 'i' => 1 }], out end def test_map_turns_chaining_on_for_transform out = output(ROWS) { transform { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 1) } } assert_equal 3, out.size assert(out.all? { |r| r['b'] && r['c'] }) end def test_header_and_trailer_with_context out = output(ROWS) do header { |c| { 'H' => c.read, 'W' => c.written } } filter { |r| r['a'] > 1 } map { |r| [r, r] } trailer { |c| { 'T' => c.written } } end assert_equal({ 'H' => 3, 'W' => 4 }, out.first) assert_equal({ 'T' => 4 }, out.last) assert_equal 6, out.size end def test_edge_order_follows_registration out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } } assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(2) end def test_header_and_trailer_nil_and_arrays out = output(ROWS) { header { nil }; trailer { [{ 'x' => 1 }, nil, { 'y' => 2 }] }; map { |r| r } } assert_equal ROWS + [{ 'x' => 1 }, { 'y' => 2 }], out end def test_condition_on_map assert_equal [{ 'a' => 1 }], output(ROWS) { map(condition: ->(r) { r['a'] == 1 }) { |r| r } } end def test_chained_false assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size end def test_csv_end_to_end with_tmpdir do |dir| DataPipeline.new do logger PipelineHelper::QUIET read MemoryStorage.new(rows: ROWS) filter { |r| r['a'].odd? } map { |r| { 'id' => r['a'], 'x' => 'y' } } trailer { |c| { 'id' => 'TOTAL', 'x' => c.written } } write container(:csv, file: "#{dir}/c.csv", columns: %w[id x]) end.run assert_equal "id,x\n1,y\n3,y\nTOTAL,2\n", File.read("#{dir}/c.csv") end end end