from __future__ import absolute_import
from __future__ import print_function

import datetime

from freezegun import freeze_time

from corehq.celery_monitoring.heartbeat import Heartbeat, HeartbeatNeverRecorded, \
    HEARTBEAT_FREQUENCY
from testil import assert_raises, eq


def test_heartbeat():
    hb = Heartbeat('celery_periodic')
    hb.clear_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_blockage_duration()

    seen_time = datetime.datetime.utcnow()

    with freeze_time(seen_time):
        hb.mark_seen()
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(seconds=0))

    with freeze_time(seen_time + datetime.timedelta(minutes=10)):
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(minutes=10) - HEARTBEAT_FREQUENCY)

from __future__ import absolute_import
from __future__ import print_function

import datetime

from freezegun import freeze_time

from corehq.celery_monitoring.heartbeat import Heartbeat, HeartbeatNeverRecorded, \
    HEARTBEAT_FREQUENCY
from testil import assert_raises, eq

from corehq.celery_monitoring.signals import TimeToStartTimer


def test_heartbeat():
    hb = Heartbeat('celery_periodic')
    hb.clear_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_blockage_duration()

    seen_time = datetime.datetime.utcnow()

    with freeze_time(seen_time):
        hb.mark_seen()
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(seconds=0))

    with freeze_time(seen_time + datetime.timedelta(minutes=10)):
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(minutes=10) - HEARTBEAT_FREQUENCY)


def test_time_to_start_timer():
    task_id = 'abc123'
    delay = datetime.timedelta(seconds=6)

    start_time = datetime.datetime.utcnow()

    # starts empty
    eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)

    with freeze_time(start_time):
        TimeToStartTimer(task_id).start_timing()

    with freeze_time(start_time + delay):
        time_to_start = TimeToStartTimer(task_id).stop_and_pop_timing()

    eq(time_to_start, delay)
    # can only pop once, second time empty
    eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)

Add simple test for celery time to start timer
Add simple test for celery time to start timer

  from __future__ import absolute_import
  from __future__ import print_function
  
  import datetime
  
  from freezegun import freeze_time
  
  from corehq.celery_monitoring.heartbeat import Heartbeat, HeartbeatNeverRecorded, \
      HEARTBEAT_FREQUENCY
  from testil import assert_raises, eq
+ 
+ from corehq.celery_monitoring.signals import TimeToStartTimer
  
  
  def test_heartbeat():
      hb = Heartbeat('celery_periodic')
      hb.clear_last_seen()
  
      with assert_raises(HeartbeatNeverRecorded):
          hb.get_last_seen()
  
      with assert_raises(HeartbeatNeverRecorded):
          hb.get_blockage_duration()
  
      seen_time = datetime.datetime.utcnow()
  
      with freeze_time(seen_time):
          hb.mark_seen()
          eq(hb.get_last_seen(), seen_time)
          eq(hb.get_blockage_duration(), datetime.timedelta(seconds=0))
  
      with freeze_time(seen_time + datetime.timedelta(minutes=10)):
          eq(hb.get_last_seen(), seen_time)
          eq(hb.get_blockage_duration(), datetime.timedelta(minutes=10) - HEARTBEAT_FREQUENCY)
  
+ 
+ def test_time_to_start_timer():
+     task_id = 'abc123'
+     delay = datetime.timedelta(seconds=6)
+ 
+     start_time = datetime.datetime.utcnow()
+ 
+     # starts empty
+     eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)
+ 
+     with freeze_time(start_time):
+         TimeToStartTimer(task_id).start_timing()
+ 
+     with freeze_time(start_time + delay):
+         time_to_start = TimeToStartTimer(task_id).stop_and_pop_timing()
+ 
+     eq(time_to_start, delay)
+     # can only pop once, second time empty
+     eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)
+ 
Add simple test for celery time to start timer
## Code Before:
from __future__ import absolute_import
from __future__ import print_function

import datetime

from freezegun import freeze_time

from corehq.celery_monitoring.heartbeat import Heartbeat, HeartbeatNeverRecorded, \
    HEARTBEAT_FREQUENCY
from testil import assert_raises, eq


def test_heartbeat():
    hb = Heartbeat('celery_periodic')
    hb.clear_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_blockage_duration()

    seen_time = datetime.datetime.utcnow()

    with freeze_time(seen_time):
        hb.mark_seen()
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(seconds=0))

    with freeze_time(seen_time + datetime.timedelta(minutes=10)):
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(minutes=10) - HEARTBEAT_FREQUENCY)

## Instruction:
Add simple test for celery time to start timer
## Code After:
from __future__ import absolute_import
from __future__ import print_function

import datetime

from freezegun import freeze_time

from corehq.celery_monitoring.heartbeat import Heartbeat, HeartbeatNeverRecorded, \
    HEARTBEAT_FREQUENCY
from testil import assert_raises, eq

from corehq.celery_monitoring.signals import TimeToStartTimer


def test_heartbeat():
    hb = Heartbeat('celery_periodic')
    hb.clear_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_last_seen()

    with assert_raises(HeartbeatNeverRecorded):
        hb.get_blockage_duration()

    seen_time = datetime.datetime.utcnow()

    with freeze_time(seen_time):
        hb.mark_seen()
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(seconds=0))

    with freeze_time(seen_time + datetime.timedelta(minutes=10)):
        eq(hb.get_last_seen(), seen_time)
        eq(hb.get_blockage_duration(), datetime.timedelta(minutes=10) - HEARTBEAT_FREQUENCY)


def test_time_to_start_timer():
    task_id = 'abc123'
    delay = datetime.timedelta(seconds=6)

    start_time = datetime.datetime.utcnow()

    # starts empty
    eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)

    with freeze_time(start_time):
        TimeToStartTimer(task_id).start_timing()

    with freeze_time(start_time + delay):
        time_to_start = TimeToStartTimer(task_id).stop_and_pop_timing()

    eq(time_to_start, delay)
    # can only pop once, second time empty
    eq(TimeToStartTimer(task_id).stop_and_pop_timing(), None)
