<s> from unittest . mock import mock_open <EOL> import collections <EOL> import json <EOL> from test . framework . base_unit_test_case import BaseUnitTestCase <EOL> from app . util . event_log import EventLog , LOG_CACHE_EXPIRE_TIME_IN_HOURS <EOL> class TestEventLog ( BaseUnitTestCase ) : <EOL> _FAKE_EVENTLOG_CONTENTS = '''<STR_LIT>''' <EOL> def test_get_events_should_return_all_events_when_no_params_specified ( self ) : <EOL> open_mock = mock_open ( read_data = self . _FAKE_EVENTLOG_CONTENTS ) <EOL> self . patch ( '<STR_LIT>' , new = open_mock , create = True ) <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> events = event_log . get_events ( ) <EOL> returned_event_ids = [ event [ '<STR_LIT>' ] for event in events ] <EOL> self . assertListEqual ( returned_event_ids , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_events_should_return_all_events_when_no_params_specified_with_cache ( self ) : <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> cache_events = [ json . loads ( line ) <EOL> for line in self . _FAKE_EVENTLOG_CONTENTS . strip ( ) . splitlines ( ) ] <EOL> event_log . _log_cache = cache_events <EOL> events = event_log . get_events ( ) <EOL> returned_event_ids = [ event [ '<STR_LIT>' ] for event in events ] <EOL> self . assertListEqual ( returned_event_ids , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_events_should_return_all_events_since_a_specified_timestamp ( self ) : <EOL> open_mock = mock_open ( read_data = self . _FAKE_EVENTLOG_CONTENTS ) <EOL> self . patch ( '<STR_LIT>' , new = open_mock , create = True ) <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> since_timestamp = <NUM_LIT> <EOL> events = event_log . get_events ( since_timestamp = since_timestamp ) <EOL> returned_event_ids = [ event [ '<STR_LIT>' ] for event in events ] <EOL> self . assertListEqual ( returned_event_ids , [ <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_events_should_return_all_events_since_a_specified_event_id ( self ) : <EOL> open_mock = mock_open ( read_data = self . _FAKE_EVENTLOG_CONTENTS ) <EOL> self . patch ( '<STR_LIT>' , new = open_mock , create = True ) <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> since_id = <NUM_LIT:2> <EOL> events = event_log . get_events ( since_id = since_id ) <EOL> returned_event_ids = [ event [ '<STR_LIT>' ] for event in events ] <EOL> self . assertListEqual ( returned_event_ids , [ <NUM_LIT:3> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_events_raises_exception_if_both_timestamp_and_id_are_specified ( self ) : <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError , <EOL> msg = '<STR_LIT>' <EOL> '<STR_LIT>' ) : <EOL> event_log . get_events ( since_timestamp = <NUM_LIT:0.0> , since_id = <NUM_LIT:0> ) <EOL> def test_get_events_should_return_events_from_file_if_timestamp_not_in_range ( self ) : <EOL> log_lines = self . _FAKE_EVENTLOG_CONTENTS . strip ( ) . splitlines ( ) <EOL> open_mock = mock_open ( read_data = self . _FAKE_EVENTLOG_CONTENTS ) <EOL> self . patch ( '<STR_LIT>' , new = open_mock , create = True ) <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> event_log . _log_cache = [ json . loads ( line ) for line in log_lines [ <NUM_LIT:2> : ] ] <EOL> since_timestamp = <NUM_LIT> <EOL> events = event_log . get_events ( since_timestamp = since_timestamp ) <EOL> returned_event_ids = [ event [ '<STR_LIT>' ] for event in events ] <EOL> self . assertListEqual ( returned_event_ids , [ <NUM_LIT:2> , <NUM_LIT:3> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_events_should_write_to_cache ( self ) : <EOL> self . patch ( '<STR_LIT>' ) <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> event_log . record_event ( '<STR_LIT>' , message = '<STR_LIT:foo>' ) <EOL> event = event_log . _log_cache [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event [ '<STR_LIT:message>' ] , '<STR_LIT:foo>' ) <EOL> def test_oldest_cache_event_is_stale_should_be_true_if_event_timestamp_is_greater_than_expire_time ( self ) : <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> cache_events = [ json . loads ( line ) <EOL> for line in self . _FAKE_EVENTLOG_CONTENTS . strip ( ) . splitlines ( ) ] <EOL> event_log . _log_cache = cache_events <EOL> events = event_log . get_events ( ) <EOL> new_timestamp = event_log . _log_cache [ <NUM_LIT:0> ] [ '<STR_LIT>' ] + LOG_CACHE_EXPIRE_TIME_IN_HOURS * <NUM_LIT> * <NUM_LIT> + <NUM_LIT:1> <EOL> new_event = { '<STR_LIT>' : <NUM_LIT:4> , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : new_timestamp } <EOL> event_log . _log_cache . append ( new_event ) <EOL> self . assertTrue ( event_log . _oldest_cache_event_is_stale ( ) ) <EOL> def test_oldest_cache_event_is_stale_should_be_false_if_event_timestamp_is_less_than_expire_time ( self ) : <EOL> event_log = EventLog ( filename = '<STR_LIT>' ) <EOL> cache_events = [ json . loads ( line ) <EOL> for line in self . _FAKE_EVENTLOG_CONTENTS . strip ( ) . splitlines ( ) ] <EOL> event_log . _log_cache = cache_events <EOL> new_timestamp = event_log . _log_cache [ <NUM_LIT:0> ] [ '<STR_LIT>' ] + <NUM_LIT> <EOL> time_mock = self . patch ( '<STR_LIT>' ) <EOL> time_mock . time . return_value = new_timestamp <EOL> self . assertFalse ( event_log . _oldest_cache_event_is_stale ( ) ) </s>
