test/embulk/input/test_mixpanel.rb in embulk-input-mixpanel-0.5.6 vs test/embulk/input/test_mixpanel.rb in embulk-input-mixpanel-0.5.7

@@ -13,11 +13,11 @@
       FROM_DATE = "2015-02-22".freeze
       TO_DATE = "2015-03-02".freeze
       DAYS = 8
       DATES = Date.parse(FROM_DATE)..(Date.parse(FROM_DATE) + DAYS - 1)
       TIMEZONE = "Asia/Tokyo".freeze
-
+      JOB_START_TIME = 1506407051000
       DURATIONS = [
         {from_date: FROM_DATE, to_date: "2015-02-28"}, # It has 7 days between 2015-02-22 and 2015-02-28
         {from_date: "2015-03-01", to_date: TO_DATE},
       ]
 
@@ -43,11 +43,17 @@
             params = params.merge(from_date: from_date, to_date: to_date)
             stub(klass).export(params) { records }
           end
         end
       end
-
+      def satisfy_task_ignore_start_time(expected_task)
+        satisfy{|input_task|
+          assert_not_nil(input_task[:job_start_time])
+          assert_equal(expected_task, input_task.merge(job_start_time: expected_task[:job_start_time]))
+          true
+        }
+      end
       def setup_logger
         stub(Embulk).logger { ::Logger.new(IO::NULL) }
       end
 
       class GuessTest < self
@@ -218,27 +224,26 @@
           def setup
           end
 
           def test_ignore_early_days
             stub(Embulk).logger { Logger.new(File::NULL) }
-
-            mock(Mixpanel).resume(task.merge(dates: target_dates), columns, 1, &control)
+            mock(Mixpanel).resume(satisfy_task_ignore_start_time(task.merge(dates: target_dates)), columns, 1, &control)
             Mixpanel.transaction(transaction_config, &control)
           end
 
           def test_info
-            stub(Mixpanel).resume(task.merge(dates: target_dates), columns, 1, &control)
+            stub(Mixpanel).resume(satisfy_task_ignore_start_time(task.merge(dates: target_dates)), columns, 1, &control)
 
             info_message_regexp = /#{Regexp.escape(target_dates.first)}.+#{Regexp.escape(target_dates.last)}/
             mock(Embulk.logger).info(info_message_regexp)
             stub(Embulk.logger).warn
 
             Mixpanel.transaction(transaction_config, &control)
           end
 
           def test_warn
-            stub(Mixpanel).resume(task.merge(dates: target_dates), columns, 1, &control)
+            stub(Mixpanel).resume(satisfy_task_ignore_start_time(task.merge(dates: target_dates)), columns, 1, &control)
             stub(Embulk.logger).info
 
             ignore_dates = dates.map{|date| date.to_s}.to_a - target_dates
             warn_message_regexp = /#{Regexp.escape(ignore_dates.first)}.+#{Regexp.escape(ignore_dates.last)}/
             mock(Embulk.logger).warn(warn_message_regexp)
@@ -268,12 +273,11 @@
         end
 
         class TimezoneTest < self
           def test_valid_timezone
             timezone = TIMEZONE
-            mock(Mixpanel).resume(transaction_task(timezone), columns, 1, &control)
-
+            mock(Mixpanel).resume(satisfy_task_ignore_start_time(transaction_task(timezone)), columns, 1, &control)
             Mixpanel.transaction(transaction_config(timezone), &control)
           end
 
           def test_invalid_timezone
             timezone = "#{TIMEZONE}ooooo"
@@ -308,11 +312,11 @@
 
         class DaysTest < self
           def test_valid_days
             days = 5
 
-            mock(Mixpanel).resume(transaction_task(days), columns, 1, &control)
+            mock(Mixpanel).resume(satisfy_task_ignore_start_time(transaction_task(days)), columns, 1, &control)
             Mixpanel.transaction(transaction_config(days), &control)
           end
 
           def test_next_to_date
             next_config_diff = Mixpanel.resume(transaction_task(1).merge(incremental: true), columns, 1) do
@@ -539,11 +543,12 @@
             fetch_unknown_columns: false,
             fetch_custom_properties: false,
             retry_initial_wait_sec: 0,
             retry_limit: 3,
             latest_fetched_time: 0,
-            slice_range: 7
+            slice_range: 7,
+            job_start_time: JOB_START_TIME
           }
         end
       end
 
       class RunTest < self
@@ -569,20 +574,20 @@
           @plugin.run
         end
 
         def test_run
           stub(@plugin).preview? { false }
-          mock(@page_builder).add(anything).times(records.length)
+          mock(@page_builder).add(anything).times(records.length * 2)
           mock(@page_builder).finish
 
           @plugin.run
         end
 
         def test_timezone
           stub(@plugin).preview? { false }
           adjusted = record_epoch - timezone_offset_seconds
-          mock(@page_builder).add(["FOO", adjusted, "event"]).times(records.length)
+          mock(@page_builder).add(["FOO", adjusted, "event"]).times(records.length * 2)
           mock(@page_builder).finish
 
           @plugin.run
         end
 
@@ -590,13 +595,13 @@
 
           def test_default_slice_range
             plugin = Mixpanel.new(task.merge(slice_range: 2), nil, nil, @page_builder)
             stub(plugin).preview? {false}
             stub(plugin).fetch(["2015-02-22", "2015-02-23"],0){[]}
-            stub(plugin).fetch(["2015-02-22", "2015-02-25"],0){[]}
-            stub(plugin).fetch(["2015-02-22", "2015-02-27"],0){[]}
-            stub(plugin).fetch(["2015-02-22", "2015-03-01"],0){[]}
+            stub(plugin).fetch(["2015-02-24", "2015-02-25"],0){[]}
+            stub(plugin).fetch(["2015-02-26", "2015-02-27"],0){[]}
+            stub(plugin).fetch(["2015-02-28", "2015-03-01"],0){[]}
             mock(@page_builder).finish
             plugin.run
           end
         end
 
@@ -642,20 +647,14 @@
             stub(plugin).preview? {false}
             adjusted = record_epoch - timezone_offset_seconds
             mock(page_builder).add(["FOO", adjusted, "event"]).times(records.length * 2)
             mock(page_builder).finish
             any_instance_of(MixpanelApi::Client) do |klass|
-              stub(klass).export(satisfy{|params|
-                '(abc==def) and properties["mp_processing_time_ms"] > 1'==params["where"]
-              }).once do |params, block|
+              stub(klass).export() do |params, block|
+                assert_equal("(abc==def) and properties[\"mp_processing_time_ms\"] > 1 and properties[\"mp_processing_time_ms\"] < #{JOB_START_TIME}",params["where"])
                 records.each{|record| block.call(record) }
               end
-              stub(klass).export(satisfy{|params|
-                '(abc==def) and properties["mp_processing_time_ms"] > 1234567919'==params["where"]
-              }).once do |params, block|
-                records.each{|record| block.call(record) }
-              end
             end
             task_report = plugin.run
             assert_equal(1234567919, task_report[:latest_fetched_time])
           end
 
@@ -665,20 +664,14 @@
             stub(plugin).preview? {false}
             adjusted = record_epoch - timezone_offset_seconds
             mock(page_builder).add(["FOO", adjusted, "event"]).times(records.length * 2)
             mock(page_builder).finish
             any_instance_of(MixpanelApi::Client) do |klass|
-              stub(klass).export(satisfy {|params|
-                'properties["mp_processing_time_ms"] > 1'==params["where"]
-              }).once() do |params,block|
+              stub(klass).export() do |params, block|
+                assert_equal("properties[\"mp_processing_time_ms\"] > 1 and properties[\"mp_processing_time_ms\"] < #{JOB_START_TIME}",params["where"])
                 records.each{|record| block.call(record) }
               end
-              stub(klass).export(satisfy {|params|
-                'properties["mp_processing_time_ms"] > 1234567919' == params["where"]
-              }).once() do |params,block|
-                records.each{|record| block.call(record) }
-              end
             end
             task_report = plugin.run
             assert_equal(1234567919, task_report[:latest_fetched_time])
           end
 
@@ -780,11 +773,11 @@
               properties["foo"],
               adjusted_time,
               {"int" => properties["int"], "event" => record["event"]}.to_json
             ]
 
-            mock(@page_builder).add(added).times(records.length)
+            mock(@page_builder).add(added).times(records.length * 2)
             mock(@page_builder).finish
 
             @plugin.run
           end
 
@@ -832,10 +825,11 @@
           fetch_unknown_columns: false,
           fetch_custom_properties: false,
           retry_initial_wait_sec: 2,
           retry_limit: 3,
           latest_fetched_time: 0,
-          slice_range: 7
+          slice_range: 7,
+          job_start_time: JOB_START_TIME
         }
       end
 
       def records
         [