<s> from datetime import datetime <EOL> import re <EOL> from testenv . utils import get_resource as resource <EOL> from testenv . utils import deploy_application as deploy <EOL> from testenv import TestCase <EOL> from testenv import ElasticSearchProcess <EOL> from testenv . constants import LOG_INDICES_PREFIX <EOL> class EventsTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( EventsTest , self ) . setUp ( ) <EOL> timestamp = datetime . now ( ) <EOL> from testenv import testenv_instance <EOL> index_name = '<STR_LIT>' . format ( LOG_INDICES_PREFIX , <EOL> timestamp . strftime ( '<STR_LIT>' ) ) <EOL> testenv_instance . handle_logs_with_es ( index_name ) <EOL> self . deployment_id = self . _create_deployment ( ) <EOL> def tearDown ( self ) : <EOL> ElasticSearchProcess . remove_log_indices ( ) <EOL> super ( EventsTest , self ) . tearDown ( ) <EOL> def test_events_without_logs ( self ) : <EOL> events = self . client . events . list ( include_logs = False ) <EOL> for event in events : <EOL> self . assertEqual ( event [ '<STR_LIT:type>' ] , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_timestamp_range ( self ) : <EOL> all_events = self . client . events . list ( _sort = '<STR_LIT>' ) <EOL> first_event = all_events [ <NUM_LIT:0> ] <EOL> median_event = all_events [ len ( all_events ) / <NUM_LIT:2> - <NUM_LIT:1> ] <EOL> min_time = first_event [ '<STR_LIT>' ] <EOL> max_time = median_event [ '<STR_LIT>' ] <EOL> ranged_events = self . client . events . list ( from_datetime = min_time , <EOL> to_datetime = max_time ) <EOL> self . assertEquals ( len ( ranged_events ) , len ( all_events ) / <NUM_LIT:2> ) <EOL> def test_sorted_events ( self ) : <EOL> events = self . client . events . list ( _sort = '<STR_LIT>' ) <EOL> self . assertGreater ( len ( events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> sorted_events = sorted ( events , key = lambda x : x . get ( '<STR_LIT>' ) , reverse = True ) <EOL> self . assertListEqual ( events . items , sorted_events ) <EOL> def test_filtered_events ( self ) : <EOL> deployment_ids = [ ] <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> deployment_ids . append ( self . _create_deployment ( ) ) <EOL> expected_deployment_ids = deployment_ids [ : <NUM_LIT:2> ] <EOL> filters = { '<STR_LIT>' : expected_deployment_ids } <EOL> events = self . client . events . list ( ** filters ) <EOL> self . assertGreater ( len ( events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> deployments_with_events = { event [ '<STR_LIT>' ] [ '<STR_LIT>' ] for event in events } <EOL> self . assertEquals ( sorted ( deployments_with_events ) , <EOL> sorted ( expected_deployment_ids ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( expected_deployment_ids , <EOL> deployments_with_events ) ) <EOL> def test_paginated_events ( self ) : <EOL> size = <NUM_LIT:5> <EOL> offset = <NUM_LIT:3> <EOL> events = self . client . events . list ( _offset = offset , _size = size ) <EOL> pagination_info = events . metadata . pagination <EOL> self . assertEquals ( len ( events ) , size ) <EOL> self . assertEquals ( pagination_info . offset , offset ) <EOL> self . assertEquals ( pagination_info . size , size ) <EOL> def test_query_with_reserved_characters ( self ) : <EOL> message = '<STR_LIT>' <EOL> self . client . events . list ( message = message ) <EOL> def test_search_event_message_with_reserved_characters ( self ) : <EOL> from manager_rest . manager_elasticsearch import RESERVED_CHARS_REGEX <EOL> message = '<STR_LIT>' <EOL> searched_events = self . client . events . list ( message = message , <EOL> include_logs = True ) <EOL> self . assertGreater ( len ( searched_events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> message_without_reserved_chars = re . sub ( RESERVED_CHARS_REGEX , '<STR_LIT>' , message ) <EOL> keywords = message_without_reserved_chars . lower ( ) . split ( ) <EOL> for event in searched_events : <EOL> text = event [ '<STR_LIT:message>' ] [ '<STR_LIT:text>' ] . lower ( ) <EOL> self . assertTrue ( all ( [ keyword in text for keyword in keywords ] ) ) <EOL> def test_search_event_message ( self ) : <EOL> all_events = self . client . events . list ( ) <EOL> message = '<STR_LIT>' <EOL> searched_events = self . client . events . list ( message = message ) <EOL> self . assertGreater ( len ( searched_events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertLess ( len ( searched_events ) , len ( all_events ) ) <EOL> for event in searched_events : <EOL> self . assertIn ( message . lower ( ) , event [ '<STR_LIT:message>' ] [ '<STR_LIT:text>' ] . lower ( ) ) <EOL> def test_list_with_include_option ( self ) : <EOL> _include = [ '<STR_LIT>' , '<STR_LIT:type>' ] <EOL> events = self . client . events . list ( _include = _include ) <EOL> self . assertGreater ( len ( events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> for event in events : <EOL> self . assertListEqual ( _include , event . keys ( ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( _include , event . keys ( ) ) ) <EOL> def test_events_with_logs ( self ) : <EOL> events = self . client . events . list ( include_logs = True ) <EOL> self . assertGreater ( len ( events ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> for event in events : <EOL> if event [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> break <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def _create_deployment ( self ) : <EOL> dsl_path = resource ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> test_deployment , _ = deploy ( dsl_path ) <EOL> return test_deployment . id </s>
