<s> import datetime <EOL> import mock <EOL> from elastalert . ruletypes import AnyRule <EOL> from elastalert . ruletypes import BlacklistRule <EOL> from elastalert . ruletypes import CardinalityRule <EOL> from elastalert . ruletypes import ChangeRule <EOL> from elastalert . ruletypes import EventWindow <EOL> from elastalert . ruletypes import FlatlineRule <EOL> from elastalert . ruletypes import FrequencyRule <EOL> from elastalert . ruletypes import NewTermsRule <EOL> from elastalert . ruletypes import SpikeRule <EOL> from elastalert . ruletypes import WhitelistRule <EOL> from elastalert . util import ts_now <EOL> from elastalert . util import ts_to_dt <EOL> def hits ( size , ** kwargs ) : <EOL> ret = [ ] <EOL> for n in range ( size ) : <EOL> ts = ts_to_dt ( '<STR_LIT>' % ( n / <NUM_LIT> , n % <NUM_LIT> ) ) <EOL> n += <NUM_LIT:1> <EOL> event = create_event ( ts , ** kwargs ) <EOL> ret . append ( event ) <EOL> return ret <EOL> def create_event ( timestamp , timestamp_field = '<STR_LIT>' , ** kwargs ) : <EOL> event = { timestamp_field : timestamp } <EOL> event . update ( ** kwargs ) <EOL> return event <EOL> def assert_matches_have ( matches , terms ) : <EOL> assert len ( matches ) == len ( terms ) <EOL> for match , term in zip ( matches , terms ) : <EOL> assert term [ <NUM_LIT:0> ] in match <EOL> assert match [ term [ <NUM_LIT:0> ] ] == term [ <NUM_LIT:1> ] <EOL> def test_any ( ) : <EOL> event = hits ( <NUM_LIT:1> ) <EOL> rule = AnyRule ( { } ) <EOL> rule . add_data ( [ event ] ) <EOL> assert rule . matches == [ event ] <EOL> def test_freq ( ) : <EOL> events = hits ( <NUM_LIT> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> rules = { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : datetime . timedelta ( hours = <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> events = hits ( <NUM_LIT> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> rules [ '<STR_LIT>' ] = '<STR_LIT:username>' <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> events = hits ( <NUM_LIT> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT> <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> assert '<STR_LIT>' in rule . occurrences <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert rule . occurrences == { } <EOL> def test_freq_count ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:100> , <EOL> '<STR_LIT>' : datetime . timedelta ( hours = <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : True } <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:6> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:6> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_freq_out_of_order ( ) : <EOL> events = hits ( <NUM_LIT> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> rules = { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : datetime . timedelta ( hours = <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = FrequencyRule ( rules ) <EOL> rule . add_data ( events [ : <NUM_LIT:10> ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT:username>' : '<STR_LIT>' } ] * <NUM_LIT:50> ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_data ( events [ <NUM_LIT:15> : <NUM_LIT:20> ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_data ( events [ <NUM_LIT:10> : <NUM_LIT:15> ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_data ( events [ <NUM_LIT:20> : <NUM_LIT> ] ) <EOL> rule . add_data ( events [ <NUM_LIT> : ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_data ( events [ <NUM_LIT> : <NUM_LIT> ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_freq_terms ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : datetime . timedelta ( hours = <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : '<STR_LIT:username>' } <EOL> rule = FrequencyRule ( rules ) <EOL> terms1 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } <EOL> terms2 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:8> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } <EOL> terms3 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:3> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ] } <EOL> rule . add_terms_data ( terms1 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_terms_data ( terms2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] . get ( '<STR_LIT:username>' ) == '<STR_LIT>' <EOL> rule . add_terms_data ( terms3 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:2> <EOL> assert rule . matches [ <NUM_LIT:1> ] . get ( '<STR_LIT:username>' ) == '<STR_LIT>' <EOL> def test_eventwindow ( ) : <EOL> timeframe = datetime . timedelta ( minutes = <NUM_LIT:10> ) <EOL> window = EventWindow ( timeframe , getTimestamp = lambda e : e [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> timestamps = [ ts_to_dt ( x ) for x in [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ] <EOL> for ts in timestamps : <EOL> window . append ( [ { '<STR_LIT>' : ts } , <NUM_LIT:1> ] ) <EOL> timestamps . sort ( ) <EOL> for exp , actual in zip ( timestamps [ <NUM_LIT:1> : ] , window . data ) : <EOL> assert actual [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == exp <EOL> window . append ( [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) } , <NUM_LIT:1> ] ) <EOL> timestamps . append ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> for exp , actual in zip ( timestamps [ <NUM_LIT:3> : ] , window . data ) : <EOL> assert actual [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == exp <EOL> def test_spike_count ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:20> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:10> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:0> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_spike_deep_key ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT:foo>' : { '<STR_LIT:bar>' : { '<STR_LIT>' : '<STR_LIT>' } } } ] ) <EOL> assert '<STR_LIT>' in rule . cur_windows <EOL> def test_spike ( ) : <EOL> events = hits ( <NUM_LIT:100> , timestamp_field = '<STR_LIT>' ) <EOL> rules = { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> events2 = events [ : <NUM_LIT:50> ] <EOL> for event in events [ <NUM_LIT:50> : ] : <EOL> events2 . append ( event ) <EOL> events2 . append ( { '<STR_LIT>' : event [ '<STR_LIT>' ] + datetime . timedelta ( milliseconds = <NUM_LIT:1> ) } ) <EOL> rules [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:3> <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> events = events [ : <NUM_LIT:50> ] + events [ <NUM_LIT> : ] <EOL> rules [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:11> <EOL> rules [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:10> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:30> <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rules . pop ( '<STR_LIT>' ) <EOL> rules [ '<STR_LIT>' ] = datetime . timedelta ( seconds = <NUM_LIT:30> ) <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT> <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> rules [ '<STR_LIT>' ] = True <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_spike_query_key ( ) : <EOL> events = hits ( <NUM_LIT:100> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> rules = { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:username>' } <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> events_bob = hits ( <NUM_LIT:100> , timestamp_field = '<STR_LIT>' , username = '<STR_LIT>' ) <EOL> events2 = events [ : <NUM_LIT:50> ] <EOL> for num in range ( <NUM_LIT:50> , <NUM_LIT> ) : <EOL> events2 . append ( events_bob [ num ] ) <EOL> events2 . append ( events [ num ] ) <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> events2 = events [ : <NUM_LIT:50> ] <EOL> for num in range ( <NUM_LIT:50> , <NUM_LIT> ) : <EOL> events2 . append ( events_bob [ num ] ) <EOL> events2 . append ( events [ num ] ) <EOL> events2 . append ( events [ num ] ) <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_spike_terms ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( minutes = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:username>' , <EOL> '<STR_LIT>' : True } <EOL> terms1 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } <EOL> terms2 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } <EOL> terms3 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> } ] } <EOL> terms4 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:12> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } ] } <EOL> terms5 = { ts_to_dt ( '<STR_LIT>' ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } ] } <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_terms_data ( terms1 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_terms_data ( terms2 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_terms_data ( terms3 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] . get ( '<STR_LIT:username>' ) == '<STR_LIT>' <EOL> rules . pop ( '<STR_LIT>' ) <EOL> rules [ '<STR_LIT>' ] = <NUM_LIT:50> <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_terms_data ( terms1 ) <EOL> rule . add_terms_data ( terms2 ) <EOL> rule . add_terms_data ( terms3 ) <EOL> rule . add_terms_data ( terms4 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rules [ '<STR_LIT>' ] = True <EOL> rule = SpikeRule ( rules ) <EOL> rule . add_terms_data ( terms1 ) <EOL> rule . add_terms_data ( terms2 ) <EOL> rule . add_terms_data ( terms3 ) <EOL> rule . add_terms_data ( terms4 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rule . matches = [ ] <EOL> rule . add_terms_data ( terms5 ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:username>' ] == '<STR_LIT>' <EOL> def test_blacklist ( ) : <EOL> events = [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = BlacklistRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> def test_whitelist ( ) : <EOL> events = [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = WhitelistRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> def test_whitelist_dont_ignore_nulls ( ) : <EOL> events = [ { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : ts_to_dt ( '<STR_LIT>' ) , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rules [ '<STR_LIT>' ] = False <EOL> rule = WhitelistRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> def test_change ( ) : <EOL> events = hits ( <NUM_LIT:10> , username = '<STR_LIT>' , term = '<STR_LIT>' ) <EOL> events [ <NUM_LIT:8> ] . pop ( '<STR_LIT>' ) <EOL> events [ <NUM_LIT:9> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> rules = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:username>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = ChangeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> events2 = hits ( <NUM_LIT:10> , username = [ '<STR_LIT>' ] , term = '<STR_LIT>' ) <EOL> events2 [ <NUM_LIT:9> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> rule = ChangeRule ( rules ) <EOL> rule . add_data ( events2 ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> rules [ '<STR_LIT>' ] = False <EOL> rule = ChangeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT:username>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> rules [ '<STR_LIT>' ] = datetime . timedelta ( seconds = <NUM_LIT:2> ) <EOL> rules [ '<STR_LIT>' ] = True <EOL> rule = ChangeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert_matches_have ( rule . matches , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> events = events [ : <NUM_LIT:8> ] + events [ <NUM_LIT:9> : ] <EOL> rules [ '<STR_LIT>' ] = datetime . timedelta ( seconds = <NUM_LIT:1> ) <EOL> rule = ChangeRule ( rules ) <EOL> rule . add_data ( events ) <EOL> assert rule . matches == [ ] <EOL> def test_new_term ( ) : <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT:a>' , '<STR_LIT:b>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT:index>' : '<STR_LIT>' } <EOL> mock_res = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } } } } <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> assert rule . es . search . call_count == <NUM_LIT:2> <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' } ] ) <EOL> assert rule . matches == [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' } ] ) <EOL> assert rule . matches == [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT:b>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:b>' ] == '<STR_LIT>' <EOL> rule . matches = [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' } ] ) <EOL> assert rule . matches == [ ] <EOL> rules [ '<STR_LIT>' ] = True <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT:b>' <EOL> def test_new_term_nested_field ( ) : <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT:a>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT:index>' : '<STR_LIT>' } <EOL> mock_res = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } } } } <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> assert rule . es . search . call_count == <NUM_LIT:2> <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:b>' : { '<STR_LIT:c>' : '<STR_LIT>' } } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:b>' ] [ '<STR_LIT:c>' ] == '<STR_LIT>' <EOL> rule . matches = [ ] <EOL> def test_new_term_with_terms ( ) : <EOL> rules = { '<STR_LIT>' : [ '<STR_LIT:a>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT:index>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:a>' } <EOL> mock_res = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:5> } ] } } } } <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> assert rule . es . search . call_count == <NUM_LIT:1> <EOL> terms = { ts_now ( ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } ] } <EOL> rule . add_terms_data ( terms ) <EOL> assert rule . matches == [ ] <EOL> terms = { ts_now ( ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } ] } <EOL> rule . add_terms_data ( terms ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT:a>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:a>' ] == '<STR_LIT>' <EOL> rule . matches = [ ] <EOL> terms = { ts_now ( ) : [ { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } ] } <EOL> rule . add_terms_data ( terms ) <EOL> assert rule . matches == [ ] <EOL> def test_new_term_with_composite_fields ( ) : <EOL> rules = { '<STR_LIT>' : [ [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] , [ '<STR_LIT:d>' , '<STR_LIT>' ] ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT:index>' : '<STR_LIT>' } <EOL> mock_res = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:3> , <EOL> } , <EOL> { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } , <EOL> ] <EOL> } <EOL> } <EOL> ] <EOL> } <EOL> } <EOL> ] <EOL> } <EOL> } <EOL> } <EOL> } <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> assert rule . es . search . call_count == <NUM_LIT:2> <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' , '<STR_LIT:c>' : '<STR_LIT>' } ] ) <EOL> assert rule . matches == [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' , '<STR_LIT:c>' : '<STR_LIT>' } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == ( '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ) <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:a>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:b>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:c>' ] == '<STR_LIT>' <EOL> rule . matches = [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' , '<STR_LIT:b>' : '<STR_LIT>' , '<STR_LIT:c>' : '<STR_LIT>' , '<STR_LIT:d>' : '<STR_LIT>' } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . matches = [ ] <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:d>' : '<STR_LIT>' , '<STR_LIT:e>' : { '<STR_LIT:f>' : '<STR_LIT>' } } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == ( '<STR_LIT:d>' , '<STR_LIT>' ) <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:d>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:e>' ] [ '<STR_LIT:f>' ] == '<STR_LIT>' <EOL> rule . matches = [ ] <EOL> rules [ '<STR_LIT>' ] = True <EOL> with mock . patch ( '<STR_LIT>' ) as mock_es : <EOL> mock_es . return_value = mock . Mock ( ) <EOL> mock_es . return_value . search . return_value = mock_res <EOL> rule = NewTermsRule ( rules ) <EOL> rule . add_data ( [ { '<STR_LIT>' : ts_now ( ) , '<STR_LIT:a>' : '<STR_LIT>' } ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:2> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == ( '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ) <EOL> assert rule . matches [ <NUM_LIT:1> ] [ '<STR_LIT>' ] == ( '<STR_LIT:d>' , '<STR_LIT>' ) <EOL> def test_flatline ( ) : <EOL> events = hits ( <NUM_LIT:10> ) <EOL> rules = { '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:30> ) , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = FlatlineRule ( rules ) <EOL> rule . add_data ( hits ( <NUM_LIT:1> ) ) <EOL> assert rule . matches == [ ] <EOL> rule . add_data ( events ) <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert rule . matches == [ ] <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_flatline_count ( ) : <EOL> rules = { '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:30> ) , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = FlatlineRule ( rules ) <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:1> } ) <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:0> } ) <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . add_count_data ( { ts_to_dt ( '<STR_LIT>' ) : <NUM_LIT:0> } ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_flatline_query_key ( ) : <EOL> rules = { '<STR_LIT>' : datetime . timedelta ( seconds = <NUM_LIT:30> ) , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = FlatlineRule ( rules ) <EOL> rule . add_data ( hits ( <NUM_LIT:1> , qk = '<STR_LIT>' ) ) <EOL> rule . add_data ( hits ( <NUM_LIT:1> , qk = '<STR_LIT>' ) ) <EOL> rule . add_data ( hits ( <NUM_LIT:1> , qk = '<STR_LIT>' ) ) <EOL> assert rule . matches == [ ] <EOL> rule . garbage_collect ( ts_to_dt ( '<STR_LIT>' ) ) <EOL> assert rule . matches == [ ] <EOL> rule . add_data ( [ create_event ( ts_to_dt ( '<STR_LIT>' ) , qk = '<STR_LIT>' ) ] ) <EOL> timestamp = '<STR_LIT>' <EOL> rule . garbage_collect ( ts_to_dt ( timestamp ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:2> <EOL> assert set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) == set ( [ m [ '<STR_LIT:key>' ] for m in rule . matches if m [ '<STR_LIT>' ] == timestamp ] ) <EOL> timestamp = '<STR_LIT>' <EOL> rule . garbage_collect ( ts_to_dt ( timestamp ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:3> <EOL> assert set ( [ '<STR_LIT>' ] ) == set ( [ m [ '<STR_LIT:key>' ] for m in rule . matches if m [ '<STR_LIT>' ] == timestamp ] ) <EOL> def test_cardinality_max ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : datetime . timedelta ( minutes = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT:user>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = CardinalityRule ( rules ) <EOL> users = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for user in users : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) , '<STR_LIT:user>' : user } <EOL> rule . add_data ( [ event ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . garbage_collect ( datetime . datetime . now ( ) ) <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) , '<STR_LIT:user>' : '<STR_LIT>' } <EOL> rule . add_data ( [ event ] ) <EOL> rule . garbage_collect ( datetime . datetime . now ( ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) , '<STR_LIT:user>' : '<STR_LIT>' } <EOL> rule . add_data ( [ event ] ) <EOL> rule . garbage_collect ( datetime . datetime . now ( ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> rule . matches = [ ] <EOL> users = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for user in users : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:15> ) , '<STR_LIT:user>' : user } <EOL> rule . add_data ( [ event ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> def test_cardinality_min ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : datetime . timedelta ( minutes = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT:user>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> rule = CardinalityRule ( rules ) <EOL> users = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] <EOL> for user in users : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) , '<STR_LIT:user>' : user } <EOL> rule . add_data ( [ event ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . garbage_collect ( datetime . datetime . now ( ) ) <EOL> users = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for user in users : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:5> ) , '<STR_LIT:user>' : user } <EOL> rule . add_data ( [ event ] ) <EOL> rule . garbage_collect ( datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:5> ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> user = '<STR_LIT>' <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:15> ) , '<STR_LIT:user>' : user } <EOL> rule . add_data ( [ event ] ) <EOL> rule . garbage_collect ( datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:15> ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:1> <EOL> def test_cardinality_qk ( ) : <EOL> rules = { '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : datetime . timedelta ( minutes = <NUM_LIT:10> ) , <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:user>' } <EOL> rule = CardinalityRule ( rules ) <EOL> users = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ] <EOL> for user in users : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) , '<STR_LIT:user>' : user , '<STR_LIT:foo>' : '<STR_LIT:foo>' + user } <EOL> rule . add_data ( [ event ] ) <EOL> assert len ( rule . matches ) == <NUM_LIT:0> <EOL> rule . garbage_collect ( datetime . datetime . now ( ) ) <EOL> values = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for value in values : <EOL> event = { '<STR_LIT>' : datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:5> ) , '<STR_LIT:user>' : '<STR_LIT>' , '<STR_LIT:foo>' : value } <EOL> rule . add_data ( [ event ] ) <EOL> rule . garbage_collect ( datetime . datetime . now ( ) + datetime . timedelta ( minutes = <NUM_LIT:5> ) ) <EOL> assert len ( rule . matches ) == <NUM_LIT:2> <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:user>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:1> ] [ '<STR_LIT:user>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:0> ] [ '<STR_LIT:foo>' ] == '<STR_LIT>' <EOL> assert rule . matches [ <NUM_LIT:1> ] [ '<STR_LIT:foo>' ] == '<STR_LIT>' </s>
