Add tests to check ensure_decompressed is called

test/test_event.rb

@@ -412,15 +412,23 @@ def setup
       time = Engine.now
       @times = [Fluent::EventTime.new(time.sec), Fluent::EventTime.new(time.sec + 1)]
       @records = [{ 'k' => 'v1', 'n' => 1 }, { 'k' => 'v2', 'n' => 2 }]
+      @packed_record = ''
       @entries = ''
       @times.zip(@records).each do |_time, record|
         v = ''
         v << [_time, record].to_msgpack
+        @packed_record += v
         @entries += compress(v)
       end
       @es = CompressedMessagePackEventStream.new(@entries)
     end
 
+    def ensure_data_is_decompressed
+      assert_equal @entries, @es.instance_variable_get(:@data)
+      yield
+      assert_equal @packed_record, @es.instance_variable_get(:@data)
+    end
+
     test 'dup' do
       dupped = @es.dup
       assert_kind_of CompressedMessagePackEventStream, dupped
@@ -438,21 +446,24 @@ def setup
     end
 
     test 'size' do
-      assert_equal 2, @es.size
       assert_equal 0, CompressedMessagePackEventStream.new('').size
+      ensure_data_is_decompressed { assert_equal 2, @es.size  }
     end
 
     test 'each' do
       i = 0
-      @es.each do |time, record|
-        assert_equal @times[i], time
-        assert_equal @records[i], record
-        i += 1
+      ensure_data_is_decompressed do
+        @es.each do |time, record|
+          assert_equal @times[i], time
+          assert_equal @records[i], record
+          i += 1
+        end
       end
     end
 
     test 'slice' do
-      sliced = @es.slice(1,1)
+      sliced = nil
+      ensure_data_is_decompressed { sliced = @es.slice(1,1) }
       assert_kind_of EventStream, sliced
       assert_equal 1, sliced.size
 
@@ -477,7 +488,9 @@ def setup
 
     test 'to_msgpack_stream' do
       i = 0
-      stream = @es.to_msgpack_stream
+      stream = nil
+      ensure_data_is_decompressed { stream = @es.to_msgpack_stream }
+
       Fluent::Engine.msgpack_factory.unpacker.feed_each(stream) { |time, record|
         assert_equal @times[i], time
         assert_equal @records[i], record
@@ -487,7 +500,11 @@ def setup
 
     test 'to_compressed_msgpack_stream' do
       i = 0
+      # Do not call ensure_decompressed!
+      assert_equal @entries, @es.instance_variable_get(:@data)
       compressed_stream = @es.to_compressed_msgpack_stream
+      assert_equal @entries, @es.instance_variable_get(:@data)
+
       stream = decompress(compressed_stream)
       Fluent::Engine.msgpack_factory.unpacker.feed_each(stream) { |time, record|
         assert_equal @times[i], time