<s> from datetime import datetime , timedelta <EOL> from django . test import TestCase <EOL> from corehq . apps . repeaters . dbaccessors import ( <EOL> get_pending_repeat_record_count , <EOL> get_success_repeat_record_count , <EOL> get_failure_repeat_record_count , <EOL> get_repeat_record_count , <EOL> get_repeaters_by_domain , <EOL> get_paged_repeat_records , <EOL> iterate_repeat_records , <EOL> ) <EOL> from corehq . apps . repeaters . models import RepeatRecord , CaseRepeater <EOL> from corehq . apps . repeaters . const import RECORD_PENDING_STATE <EOL> class TestRepeatRecordDBAccessors ( TestCase ) : <EOL> dependent_apps = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> repeater_id = '<STR_LIT>' <EOL> other_id = '<STR_LIT>' <EOL> domain = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> before = datetime . utcnow ( ) - timedelta ( minutes = <NUM_LIT:5> ) <EOL> failed = RepeatRecord ( <EOL> domain = cls . domain , <EOL> failure_reason = '<STR_LIT>' , <EOL> repeater_id = cls . repeater_id , <EOL> next_event = before , <EOL> ) <EOL> success = RepeatRecord ( <EOL> domain = cls . domain , <EOL> succeeded = True , <EOL> repeater_id = cls . repeater_id , <EOL> next_event = before , <EOL> ) <EOL> pending = RepeatRecord ( <EOL> domain = cls . domain , <EOL> succeeded = False , <EOL> repeater_id = cls . repeater_id , <EOL> next_event = before , <EOL> ) <EOL> other_id = RepeatRecord ( <EOL> domain = cls . domain , <EOL> succeeded = False , <EOL> repeater_id = cls . other_id , <EOL> next_event = before , <EOL> ) <EOL> cls . records = [ <EOL> failed , <EOL> success , <EOL> pending , <EOL> other_id , <EOL> ] <EOL> for record in cls . records : <EOL> record . save ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> for record in cls . records : <EOL> record . delete ( ) <EOL> def test_get_pending_repeat_record_count ( self ) : <EOL> count = get_pending_repeat_record_count ( self . domain , self . repeater_id ) <EOL> self . assertEqual ( count , <NUM_LIT:1> ) <EOL> def test_get_success_repeat_record_count ( self ) : <EOL> count = get_success_repeat_record_count ( self . domain , self . repeater_id ) <EOL> self . assertEqual ( count , <NUM_LIT:1> ) <EOL> def test_get_failure_repeat_record_count ( self ) : <EOL> count = get_failure_repeat_record_count ( self . domain , self . repeater_id ) <EOL> self . assertEqual ( count , <NUM_LIT:1> ) <EOL> def test_get_paged_repeat_records_with_state_and_no_records ( self ) : <EOL> count = get_repeat_record_count ( '<STR_LIT>' , state = RECORD_PENDING_STATE ) <EOL> self . assertEqual ( count , <NUM_LIT:0> ) <EOL> def test_get_paged_repeat_records ( self ) : <EOL> records = get_paged_repeat_records ( self . domain , <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:2> ) <EOL> def test_get_paged_repeat_records_with_repeater_id ( self ) : <EOL> records = get_paged_repeat_records ( self . domain , <NUM_LIT:0> , <NUM_LIT:2> , repeater_id = self . other_id ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:1> ) <EOL> def test_get_paged_repeat_records_with_state ( self ) : <EOL> records = get_paged_repeat_records ( self . domain , <NUM_LIT:0> , <NUM_LIT:10> , state = RECORD_PENDING_STATE ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:2> ) <EOL> def test_get_paged_repeat_records_wrong_domain ( self ) : <EOL> records = get_paged_repeat_records ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:0> ) <EOL> def test_iterate_repeat_records ( self ) : <EOL> records = list ( iterate_repeat_records ( datetime . utcnow ( ) , chunk_size = <NUM_LIT:2> ) ) <EOL> self . assertEqual ( len ( records ) , <NUM_LIT:3> ) <EOL> class TestRepeatersDBAccessors ( TestCase ) : <EOL> dependent_apps = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> domain = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> repeater = CaseRepeater ( <EOL> domain = cls . domain , <EOL> ) <EOL> cls . repeaters = [ <EOL> repeater <EOL> ] <EOL> for repeater in cls . repeaters : <EOL> repeater . save ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> for repeater in cls . repeaters : <EOL> repeater . delete ( ) <EOL> def test_get_repeaters_by_domain ( self ) : <EOL> repeaters = get_repeaters_by_domain ( self . domain ) <EOL> self . assertEqual ( len ( repeaters ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( repeaters [ <NUM_LIT:0> ] . __class__ , CaseRepeater ) </s>
