<s> import datetime <EOL> import os <EOL> import time <EOL> import unittest <EOL> from nose . plugins . attrib import attr <EOL> import nose . tools as nt <EOL> from nose . tools import assert_equal as eq <EOL> from nose . tools import assert_true as ok <EOL> import simplejson as json <EOL> import dogapi <EOL> from dogapi . exceptions import * <EOL> from tests . util . snapshot_test_utils import ( <EOL> assert_snap_not_blank , assert_snap_has_no_events <EOL> ) <EOL> TEST_USER = os . environ . get ( '<STR_LIT>' ) <EOL> API_KEY = os . environ . get ( '<STR_LIT>' ) <EOL> APP_KEY = os . environ . get ( '<STR_LIT>' ) <EOL> dog = None <EOL> class TestDatadog ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> global dog <EOL> dog = dogapi . DogHttpApi ( ) <EOL> dog . api_key = API_KEY <EOL> dog . application_key = APP_KEY <EOL> dog . swallow = False <EOL> def test_tags ( self ) : <EOL> hostname = '<STR_LIT>' <EOL> dog . metric ( '<STR_LIT>' , <NUM_LIT:1> , host = hostname ) <EOL> dog . all_tags ( ) <EOL> dog . detach_tags ( hostname ) <EOL> assert len ( dog . host_tags ( hostname ) ) == <NUM_LIT:0> <EOL> dog . add_tags ( hostname , [ '<STR_LIT>' , '<STR_LIT>' ] , source = '<STR_LIT>' ) <EOL> new_tags = dog . host_tags ( hostname ) <EOL> assert len ( new_tags ) == <NUM_LIT:2> <EOL> assert '<STR_LIT>' in new_tags <EOL> assert '<STR_LIT>' in new_tags <EOL> dog . add_tags ( hostname , [ '<STR_LIT>' ] , source = '<STR_LIT>' ) <EOL> new_tags = dog . host_tags ( hostname ) <EOL> assert len ( new_tags ) == <NUM_LIT:3> <EOL> assert '<STR_LIT>' in new_tags <EOL> assert '<STR_LIT>' in new_tags <EOL> assert '<STR_LIT>' in new_tags <EOL> dog . change_tags ( hostname , [ '<STR_LIT>' ] , source = '<STR_LIT>' ) <EOL> new_tags = dog . host_tags ( hostname ) <EOL> assert len ( new_tags ) == <NUM_LIT:1> <EOL> assert '<STR_LIT>' in new_tags <EOL> dog . detach_tags ( hostname , source = '<STR_LIT>' ) <EOL> assert len ( dog . host_tags ( hostname ) ) == <NUM_LIT:0> <EOL> def test_events ( self ) : <EOL> wait_time = <NUM_LIT:10> <EOL> now = datetime . datetime . now ( ) <EOL> now_ts = int ( time . mktime ( now . timetuple ( ) ) ) <EOL> now_title = '<STR_LIT>' + str ( now_ts ) <EOL> now_message = '<STR_LIT>' + str ( now_ts ) <EOL> before_ts = int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:5> ) ) . timetuple ( ) ) ) <EOL> before_title = '<STR_LIT>' + str ( before_ts ) <EOL> before_message = '<STR_LIT>' + str ( before_ts ) <EOL> now_event_id = dog . event_with_response ( now_title , now_message , now_ts ) <EOL> before_event_id = dog . event_with_response ( before_title , before_message , before_ts ) <EOL> time . sleep ( wait_time ) <EOL> now_event = dog . get_event ( now_event_id ) <EOL> before_event = dog . get_event ( before_event_id ) <EOL> self . assertEquals ( now_event [ '<STR_LIT:text>' ] , now_message ) <EOL> self . assertEquals ( before_event [ '<STR_LIT:text>' ] , before_message ) <EOL> event_id = dog . event_with_response ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> host = '<STR_LIT>' , <EOL> device_name = '<STR_LIT>' ) <EOL> time . sleep ( wait_time ) <EOL> event = dog . get_event ( event_id ) <EOL> self . assertEquals ( event [ '<STR_LIT:host>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( event [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> event_id = dog . event_with_response ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> tags = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> time . sleep ( wait_time ) <EOL> event = dog . get_event ( event_id ) <EOL> assert '<STR_LIT>' in event [ '<STR_LIT>' ] <EOL> assert '<STR_LIT>' in event [ '<STR_LIT>' ] <EOL> def test_aggregate_events ( self ) : <EOL> now_ts = int ( time . time ( ) ) <EOL> agg_key = '<STR_LIT>' + str ( now_ts ) <EOL> msg_1 = '<STR_LIT>' <EOL> msg_2 = '<STR_LIT>' <EOL> event1_id = dog . event_with_response ( msg_1 , msg_1 , aggregation_key = agg_key ) <EOL> event2_id = dog . event_with_response ( msg_2 , msg_2 , aggregation_key = agg_key ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> event1 = dog . get_event ( event1_id ) <EOL> event2 = dog . get_event ( event2_id ) <EOL> self . assertEquals ( msg_1 , event1 [ '<STR_LIT:text>' ] ) <EOL> self . assertEquals ( msg_2 , event2 [ '<STR_LIT:text>' ] ) <EOL> def test_git_commits ( self ) : <EOL> """<STR_LIT>""" <EOL> event_id = dog . event_with_response ( "<STR_LIT>" , """<STR_LIT>""" , event_type = "<STR_LIT>" , source_type_name = "<STR_LIT>" , event_object = "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> event = dog . get_event ( event_id ) <EOL> self . assertEquals ( event . get ( "<STR_LIT:title>" , "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> def test_comments ( self ) : <EOL> wait_time = <NUM_LIT:3> <EOL> now = datetime . datetime . now ( ) <EOL> now_ts = int ( time . mktime ( now . timetuple ( ) ) ) <EOL> before_ts = int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:15> ) ) . timetuple ( ) ) ) <EOL> message = '<STR_LIT>' + str ( now_ts ) <EOL> comment_id = dog . comment ( TEST_USER , message ) <EOL> time . sleep ( wait_time ) <EOL> event = dog . get_event ( comment_id ) <EOL> eq ( event [ '<STR_LIT:text>' ] , message ) <EOL> dog . update_comment ( TEST_USER , message + '<STR_LIT>' , comment_id ) <EOL> time . sleep ( wait_time ) <EOL> event = dog . get_event ( comment_id ) <EOL> eq ( event [ '<STR_LIT:text>' ] , message + '<STR_LIT>' ) <EOL> reply_id = dog . comment ( TEST_USER , message + '<STR_LIT>' , related_event_id = comment_id ) <EOL> time . sleep ( wait_time ) <EOL> stream = dog . stream ( before_ts , now_ts + <NUM_LIT:100> ) <EOL> ok ( stream is not None , msg = "<STR_LIT>" ) <EOL> ok ( isinstance ( stream , list ) , msg = "<STR_LIT>" ) <EOL> ok ( len ( stream ) > <NUM_LIT:0> , msg = "<STR_LIT>" ) <EOL> comment_ids = [ x [ '<STR_LIT:id>' ] for x in stream [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] <EOL> ok ( reply_id in comment_ids , <EOL> msg = "<STR_LIT>" . format ( reply_id , comment_ids ) ) <EOL> dog . delete_comment ( reply_id ) <EOL> dog . delete_comment ( comment_id ) <EOL> time . sleep ( wait_time ) <EOL> try : <EOL> dog . get_event ( comment_id ) <EOL> except : <EOL> pass <EOL> else : <EOL> assert False <EOL> @ attr ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_dash_validation ( self ) : <EOL> graph = { <EOL> "<STR_LIT:title>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <EOL> { <EOL> "<STR_LIT>" : [ { "<STR_LIT:q>" : "<STR_LIT>" } ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> try : <EOL> dog . create_dashboard ( None , '<STR_LIT>' , [ graph ] ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . create_dashboard ( '<STR_LIT>' , None , [ graph ] ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . create_dashboard ( '<STR_LIT>' , '<STR_LIT>' , None ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . create_dashboard ( '<STR_LIT>' , '<STR_LIT>' , graph ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . create_dashboard ( '<STR_LIT>' , '<STR_LIT>' , [ ] ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . create_dashboard ( '<STR_LIT>' , '<STR_LIT>' , [ graph , None ] ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> try : <EOL> dog . dashboard ( <NUM_LIT> ) <EOL> assert False , "<STR_LIT>" <EOL> except ApiError , e : <EOL> exception_msg = e . message [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> eq ( exception_msg , "<STR_LIT>" ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_dash ( self ) : <EOL> graph = { <EOL> "<STR_LIT:title>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <EOL> { <EOL> "<STR_LIT>" : [ { "<STR_LIT:q>" : "<STR_LIT>" } ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> dash_id = dog . create_dashboard ( '<STR_LIT>' , '<STR_LIT>' , [ graph ] ) <EOL> remote_dash = dog . dashboard ( dash_id ) <EOL> eq ( '<STR_LIT>' , remote_dash [ '<STR_LIT:title>' ] ) <EOL> eq ( '<STR_LIT>' , remote_dash [ '<STR_LIT:description>' ] ) <EOL> eq ( graph [ '<STR_LIT>' ] [ '<STR_LIT>' ] , remote_dash [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> graph = { <EOL> "<STR_LIT:title>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ { "<STR_LIT:q>" : "<STR_LIT>" } ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> dash_id = dog . update_dashboard ( dash_id , '<STR_LIT>' , <EOL> '<STR_LIT>' , [ graph ] , <EOL> template_variables = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> remote_dash = dog . dashboard ( dash_id ) <EOL> eq ( '<STR_LIT>' , remote_dash [ '<STR_LIT:title>' ] ) <EOL> eq ( '<STR_LIT>' , remote_dash [ '<STR_LIT:description>' ] ) <EOL> self . assertEqual ( [ <EOL> { '<STR_LIT:default>' : None , '<STR_LIT:name>' : '<STR_LIT:foo>' , '<STR_LIT>' : None } , <EOL> { '<STR_LIT:default>' : None , '<STR_LIT:name>' : '<STR_LIT:bar>' , '<STR_LIT>' : None } , <EOL> ] , remote_dash [ '<STR_LIT>' ] ) <EOL> p = graph [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> eq ( p , remote_dash [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> dashes = dog . dashboards ( ) <EOL> ids = [ dash [ "<STR_LIT:id>" ] for dash in dashes ] <EOL> assert dash_id in ids or str ( dash_id ) in ids <EOL> dog . delete_dashboard ( dash_id ) <EOL> try : <EOL> d = dog . dashboard ( dash_id ) <EOL> except : <EOL> pass <EOL> else : <EOL> assert False <EOL> def test_search ( self ) : <EOL> results = dog . search ( '<STR_LIT:e>' ) <EOL> assert len ( results [ '<STR_LIT>' ] ) > <NUM_LIT:0> <EOL> assert len ( results [ '<STR_LIT>' ] ) > <NUM_LIT:0> <EOL> def test_metrics ( self ) : <EOL> now = datetime . datetime . now ( ) <EOL> now_ts = int ( time . mktime ( now . timetuple ( ) ) ) <EOL> dog . metric ( '<STR_LIT>' + str ( now_ts ) , <NUM_LIT:1> , host = "<STR_LIT>" + str ( now_ts ) ) <EOL> time . sleep ( <NUM_LIT:4> ) <EOL> results = dog . search ( '<STR_LIT>' + str ( now_ts ) ) <EOL> matt_series = [ <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:10> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:20> ) ) . timetuple ( ) ) ) , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:15> ) ) . timetuple ( ) ) ) , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:15> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:12> ) ) . timetuple ( ) ) ) , <NUM_LIT:15> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:12> ) ) . timetuple ( ) ) ) + <NUM_LIT:1> , <NUM_LIT:5> ) , <EOL> ( int ( time . mktime ( ( now - datetime . timedelta ( minutes = <NUM_LIT:11> ) ) . timetuple ( ) ) ) , <NUM_LIT:5> ) , <EOL> ] <EOL> dog . metric ( '<STR_LIT>' , matt_series , host = "<STR_LIT>" ) <EOL> dog . metrics ( { <EOL> '<STR_LIT>' : [ ( <NUM_LIT> , <NUM_LIT:1> ) , ( <NUM_LIT> , <NUM_LIT:2> ) ] , <EOL> '<STR_LIT>' : [ ( <NUM_LIT> , <NUM_LIT:2> ) , ( <NUM_LIT> , <NUM_LIT:4> ) ] , <EOL> } ) <EOL> def test_type_check ( self ) : <EOL> dog . metric ( "<STR_LIT>" , [ ( time . time ( ) - <NUM_LIT> , <NUM_LIT:1.0> ) ] ) <EOL> dog . metric ( "<STR_LIT>" , <NUM_LIT:1.0> ) <EOL> dog . metric ( "<STR_LIT>" , ( time . time ( ) , <NUM_LIT:1.0> ) ) <EOL> def test_alerts ( self ) : <EOL> query = "<STR_LIT>" <EOL> alert_id = dog . alert ( query ) <EOL> alert = dog . get_alert ( alert_id ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> assert alert [ '<STR_LIT>' ] == query , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == False , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == False , alert [ '<STR_LIT>' ] <EOL> dog . update_alert ( alert_id , query , silenced = True ) <EOL> alert = dog . get_alert ( alert_id ) <EOL> assert alert [ '<STR_LIT>' ] == query , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == True , alert [ '<STR_LIT>' ] <EOL> dog . update_alert ( alert_id , query , silenced = True , notify_no_data = True , timeout_h = <NUM_LIT:1> ) <EOL> alert = dog . get_alert ( alert_id ) <EOL> assert alert [ '<STR_LIT>' ] == query , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == True , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == True , alert [ '<STR_LIT>' ] <EOL> assert alert [ '<STR_LIT>' ] == <NUM_LIT:1> , alert [ '<STR_LIT>' ] <EOL> dog . delete_alert ( alert_id ) <EOL> try : <EOL> dog . get_alert ( alert_id ) <EOL> except : <EOL> pass <EOL> else : <EOL> assert False , '<STR_LIT>' <EOL> query1 = "<STR_LIT>" <EOL> query2 = "<STR_LIT>" <EOL> alert_id1 = dog . alert ( query1 ) <EOL> alert_id2 = dog . alert ( query2 ) <EOL> alerts = dog . get_all_alerts ( ) <EOL> alert1 = [ a for a in alerts if a [ '<STR_LIT:id>' ] == alert_id1 ] [ <NUM_LIT:0> ] <EOL> alert2 = [ a for a in alerts if a [ '<STR_LIT:id>' ] == alert_id2 ] [ <NUM_LIT:0> ] <EOL> assert alert1 [ '<STR_LIT>' ] == query1 , alert1 <EOL> assert alert2 [ '<STR_LIT>' ] == query2 , alert2 <EOL> def test_user_error ( self ) : <EOL> query = "<STR_LIT>" <EOL> dog . swallow = True <EOL> dog . json_responses = True <EOL> alert = dog . alert ( query ) <EOL> assert '<STR_LIT:id>' in alert , alert <EOL> result = dog . update_alert ( alert [ '<STR_LIT:id>' ] , '<STR_LIT>' , silenced = True ) <EOL> assert '<STR_LIT>' in result , result <EOL> dog . swallow = True <EOL> dog . json_responses = False <EOL> alert_id = dog . alert ( query ) <EOL> assert alert_id == int ( alert_id ) , alert_id <EOL> result = dog . update_alert ( alert_id , '<STR_LIT>' , silenced = True ) <EOL> assert '<STR_LIT>' in result , result <EOL> dog . swallow = False <EOL> dog . json_responses = False <EOL> alert_id = dog . alert ( query ) <EOL> assert alert_id == int ( alert_id ) , alert_id <EOL> try : <EOL> result = dog . update_alert ( alert_id , '<STR_LIT>' , silenced = True ) <EOL> except ApiError , e : <EOL> pass <EOL> else : <EOL> raise False , "<STR_LIT>" <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_graph_snapshot ( self ) : <EOL> metric_query = "<STR_LIT>" <EOL> event_query = "<STR_LIT:*>" <EOL> end = int ( time . time ( ) ) <EOL> start = end - <NUM_LIT> * <NUM_LIT> <EOL> snap = dog . graph_snapshot ( metric_query , start , end ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' not in snap , msg = snap ) <EOL> eq ( snap [ '<STR_LIT>' ] , metric_query ) <EOL> snapshot_url = snap [ '<STR_LIT>' ] <EOL> while dog . snapshot_status ( snapshot_url ) != <NUM_LIT:200> : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> if '<STR_LIT:localhost>' in dog . api_host : <EOL> snapshot_url = '<STR_LIT>' % ( dog . api_host , snapshot_url ) <EOL> assert_snap_not_blank ( snapshot_url ) <EOL> assert_snap_has_no_events ( snapshot_url ) <EOL> snap = dog . graph_snapshot ( metric_query , start , end , <EOL> event_query = event_query ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> eq ( snap [ '<STR_LIT>' ] , metric_query ) <EOL> eq ( snap [ '<STR_LIT>' ] , event_query ) <EOL> snapshot_url = snap [ '<STR_LIT>' ] <EOL> while dog . snapshot_status ( snapshot_url ) != <NUM_LIT:200> : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> if '<STR_LIT:localhost>' in dog . api_host : <EOL> snapshot_url = '<STR_LIT>' % ( dog . api_host , snapshot_url ) <EOL> assert_snap_not_blank ( snapshot_url ) <EOL> graph_def = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT:q>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:>>" , <EOL> "<STR_LIT:value>" : <NUM_LIT> <EOL> } , { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:>>" , <EOL> "<STR_LIT:value>" : <NUM_LIT> <EOL> } ] <EOL> } ] <EOL> } <EOL> graph_def = json . dumps ( graph_def ) <EOL> snap = dog . graph_snapshot_from_def ( graph_def , start , end ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' not in snap , msg = snap ) <EOL> ok ( '<STR_LIT>' not in snap , msg = snap ) <EOL> eq ( snap [ '<STR_LIT>' ] , graph_def ) <EOL> snapshot_url = snap [ '<STR_LIT>' ] <EOL> while dog . snapshot_status ( snapshot_url ) != <NUM_LIT:200> : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> if '<STR_LIT:localhost>' in dog . api_host : <EOL> snapshot_url = '<STR_LIT>' % ( dog . api_host , snapshot_url ) <EOL> assert_snap_not_blank ( snapshot_url ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_screenboard ( self ) : <EOL> def _compare_screenboard ( board1 , board2 ) : <EOL> compare_keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:width>' , '<STR_LIT>' ] <EOL> for key in compare_keys : <EOL> assert board1 [ key ] == board2 [ key ] , key <EOL> board = { <EOL> "<STR_LIT:width>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:title>" : False , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:width>" : <NUM_LIT> , <EOL> "<STR_LIT:y>" : <NUM_LIT> , <EOL> "<STR_LIT:x>" : <NUM_LIT> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:type>" : "<STR_LIT:image>" , <EOL> "<STR_LIT>" : <NUM_LIT:20> , <EOL> "<STR_LIT:width>" : <NUM_LIT:32> , <EOL> "<STR_LIT:y>" : <NUM_LIT:7> , <EOL> "<STR_LIT:x>" : <NUM_LIT:32> , <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> updated_board = { <EOL> "<STR_LIT:width>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:type>" : "<STR_LIT:image>" , <EOL> "<STR_LIT>" : <NUM_LIT:20> , <EOL> "<STR_LIT:width>" : <NUM_LIT:32> , <EOL> "<STR_LIT:y>" : <NUM_LIT:7> , <EOL> "<STR_LIT:x>" : <NUM_LIT:32> , <EOL> "<STR_LIT:url>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> create_res = dog . create_screenboard ( board ) <EOL> _compare_screenboard ( board , create_res ) <EOL> get_res = dog . get_screenboard ( create_res [ '<STR_LIT:id>' ] ) <EOL> _compare_screenboard ( get_res , create_res ) <EOL> assert get_res [ '<STR_LIT:id>' ] == create_res [ '<STR_LIT:id>' ] <EOL> get_all_res = dog . get_all_screenboards ( ) <EOL> created = [ s for s in get_all_res if s [ '<STR_LIT:id>' ] == create_res [ '<STR_LIT:id>' ] ] <EOL> self . assertEquals ( len ( created ) , <NUM_LIT:1> ) <EOL> update_res = dog . update_screenboard ( get_res [ '<STR_LIT:id>' ] , updated_board ) <EOL> _compare_screenboard ( update_res , updated_board ) <EOL> assert get_res [ '<STR_LIT:id>' ] == update_res [ '<STR_LIT:id>' ] <EOL> share_res = dog . share_screenboard ( get_res [ '<STR_LIT:id>' ] ) <EOL> assert share_res [ '<STR_LIT>' ] == get_res [ '<STR_LIT:id>' ] <EOL> delete_res = dog . delete_screenboard ( update_res [ '<STR_LIT:id>' ] ) <EOL> assert delete_res [ '<STR_LIT:id>' ] == update_res [ '<STR_LIT:id>' ] <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_monitor_crud ( self ) : <EOL> query = "<STR_LIT>" <EOL> options = { <EOL> '<STR_LIT>' : { '<STR_LIT:*>' : time . time ( ) + <NUM_LIT> * <NUM_LIT> } , <EOL> '<STR_LIT>' : False <EOL> } <EOL> monitor_id = dog . monitor ( '<STR_LIT>' , query , options = options ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> options [ '<STR_LIT>' ] ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , options [ '<STR_LIT>' ] ) <EOL> query2 = "<STR_LIT>" <EOL> updated_monitor_id = dog . update_monitor ( monitor_id , query2 , options = options ) <EOL> monitor = dog . get_monitor ( updated_monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query2 ) <EOL> name = '<STR_LIT>' <EOL> monitor_id = dog . update_monitor ( monitor_id , query2 , name = name , <EOL> options = { '<STR_LIT>' : True } ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT:name>' ] , name ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , True ) <EOL> dog . delete_monitor ( monitor_id ) <EOL> try : <EOL> dog . get_monitor ( monitor_id ) <EOL> except ApiError : <EOL> pass <EOL> else : <EOL> assert False , '<STR_LIT>' <EOL> query1 = "<STR_LIT>" <EOL> query2 = "<STR_LIT>" <EOL> query3 = "<STR_LIT>" <EOL> monitor_id1 = dog . monitor ( '<STR_LIT>' , query1 ) <EOL> monitor_id2 = dog . monitor ( '<STR_LIT>' , query2 ) <EOL> monitor_id3 = dog . monitor ( '<STR_LIT>' , query3 ) <EOL> monitors = dog . get_all_monitors ( group_states = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> monitor1 = [ m for m in monitors if m [ '<STR_LIT:id>' ] == monitor_id1 ] [ <NUM_LIT:0> ] <EOL> monitor2 = [ m for m in monitors if m [ '<STR_LIT:id>' ] == monitor_id2 ] [ <NUM_LIT:0> ] <EOL> assert monitor1 [ '<STR_LIT>' ] == query1 , monitor1 <EOL> assert monitor2 [ '<STR_LIT>' ] == query2 , monitor2 <EOL> monitors = dog . get_all_monitors ( tags = "<STR_LIT>" ) <EOL> monitor3 = [ m for m in monitors if m [ '<STR_LIT:id>' ] == monitor_id3 ] [ <NUM_LIT:0> ] <EOL> assert next ( ( m for m in monitors if m [ '<STR_LIT:id>' ] == monitor_id1 ) , None ) is None <EOL> assert next ( ( m for m in monitors if m [ '<STR_LIT:id>' ] == monitor_id2 ) , None ) is None <EOL> assert monitor3 [ '<STR_LIT>' ] == query3 , monitor3 <EOL> query = '<STR_LIT>' <EOL> options = { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:3> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:3> <EOL> } <EOL> } <EOL> monitor_id = dog . monitor ( '<STR_LIT>' , query , options = options ) <EOL> monitor = dog . get_monitor ( monitor_id , group_states = [ '<STR_LIT:all>' ] ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> options [ '<STR_LIT>' ] ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , options [ '<STR_LIT>' ] ) <EOL> query2 = '<STR_LIT>' <EOL> monitor_id = dog . update_monitor ( monitor_id , query2 ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query2 ) <EOL> dog . delete_monitor ( monitor_id ) <EOL> try : <EOL> dog . get_monitor ( monitor_id ) <EOL> except ApiError : <EOL> pass <EOL> else : <EOL> assert False , '<STR_LIT>' <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_monitor_muting ( self ) : <EOL> query = "<STR_LIT>" <EOL> monitor_id = dog . monitor ( '<STR_LIT>' , query ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query ) <EOL> dt = dog . mute_monitors ( ) <EOL> nt . assert_equal ( dt [ '<STR_LIT>' ] , True ) <EOL> nt . assert_equal ( dt [ '<STR_LIT>' ] , [ '<STR_LIT:*>' ] ) <EOL> dt = dog . unmute_monitors ( ) <EOL> nt . assert_equal ( dt , None ) <EOL> nt . assert_raises ( ApiError , dog . mute_monitor , monitor_id , scope = '<STR_LIT>' ) <EOL> query2 = "<STR_LIT>" <EOL> monitor_id = dog . monitor ( '<STR_LIT>' , query2 ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] , query2 ) <EOL> dog . mute_monitor ( monitor_id , scope = '<STR_LIT>' ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , { '<STR_LIT>' : None } ) <EOL> dog . unmute_monitor ( monitor_id , scope = '<STR_LIT>' ) <EOL> monitor = dog . get_monitor ( monitor_id ) <EOL> nt . assert_equal ( monitor [ '<STR_LIT>' ] [ '<STR_LIT>' ] , { } ) <EOL> dog . delete_monitor ( monitor_id ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_downtime ( self ) : <EOL> start = int ( time . time ( ) ) <EOL> end = start + <NUM_LIT:1000> <EOL> downtime_id = dog . schedule_downtime ( '<STR_LIT>' , start , end , "<STR_LIT>" ) <EOL> dt = dog . get_downtime ( downtime_id ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:start>' ] , start ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:end>' ] , end ) <EOL> nt . assert_equal ( dt [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:message>' ] , "<STR_LIT>" ) <EOL> dog . update_downtime ( '<STR_LIT>' , start + <NUM_LIT:1> , end + <NUM_LIT:1> , '<STR_LIT>' ) <EOL> dt = dog . get_downtime ( downtime_id ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:start>' ] , start + <NUM_LIT:1> ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:end>' ] , end + <NUM_LIT:1> ) <EOL> nt . assert_equal ( dt [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) <EOL> nt . assert_equal ( dt [ '<STR_LIT:message>' ] , "<STR_LIT>" ) <EOL> dog . cancel_downtime ( downtime_id ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_service_check ( self ) : <EOL> res = dog . service_check ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> , <EOL> message = '<STR_LIT>' , tags = [ '<STR_LIT>' ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
