   test "emitted monads include an error for rows with variable length" do
     stream = Stream.map(["a,\"be\"", ",c,d", "e,f", "g,,h", "i,j", "k,l"], &(&1))
 
-    assert stream
-    |> Decoder.decode
-    |> Enum.filter(fn
-      { :error, message } -> message |> String.contains?("row with length")
-    _ -> false
-    end)
-    |> Enum.count == 2
+    errors = stream |> Decoder.decode |> filter_errors |> Enum.into([])
+    assert errors == [
+      error: "Encountered a row with length 3 instead of 2",
+      error: "Encountered a row with length 3 instead of 2"
+    ]
   end
 
   test "delivers correctly ordered rows" do


  @moduletag timeout: 1000

  test "parses strings into a list of token tuples and emits them" do
    stream = Stream.map(["a,be", "c,d"], &(&1))
    result = Decoder.decode!(stream) |> Enum.into([])

  test "emitted monads include an error for non-unicode files" do
    stream = "./fixtures/broken-encoding.csv" |> Path.expand(__DIR__) |> File.stream!

    errors = stream |> Decoder.decode |> filter_errors |> Enum.into([])
    assert errors == [
      error: "Invalid encoding",
      error: "Encountered a row with length 1 instead of 0" #TODO halt the stream on invalid encoding?
    ]
  end

  test "discards any state in the current message queues when halted" do
