<s> import datetime <EOL> import os . path <EOL> import unittest <EOL> import json <EOL> import logging <EOL> import esbench . analyze <EOL> import esbench . api <EOL> import esbench . test . test_api <EOL> class ApiCallsTest ( unittest . TestCase ) : <EOL> def test_get_benchmarks ( self ) : <EOL> conn = esbench . api . Conn ( conn_cls = esbench . test . test_api . MockHTTPConnection ) <EOL> resp = esbench . analyze . _get_benchmarks ( conn ) <EOL> self . assertEqual ( resp . curl , '<STR_LIT>' % esbench . STATS_INDEX_NAME ) <EOL> def test_get_observations ( self ) : <EOL> conn = esbench . api . Conn ( conn_cls = esbench . test . test_api . MockHTTPConnection ) <EOL> resp = esbench . analyze . _get_observations ( conn = conn , benchmark_id = '<STR_LIT:foo>' ) <EOL> self . assertEqual ( resp . curl , '<STR_LIT>' % ( esbench . STATS_INDEX_NAME , ) ) <EOL> class DataTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . tmp_get_benchmarks = esbench . analyze . _get_benchmarks <EOL> esbench . analyze . _get_benchmarks = lambda ** kwargs : esbench . api . ApiResponse ( <EOL> status = <NUM_LIT:200> , <EOL> reason = '<STR_LIT:OK>' , <EOL> data = r"""<STR_LIT>""" , <EOL> curl = '<STR_LIT>' , <EOL> ) <EOL> self . tmp_get_observations = esbench . analyze . _get_observations <EOL> esbench . analyze . _get_observations = lambda ** kwargs : esbench . api . ApiResponse ( <EOL> status = <NUM_LIT:200> , <EOL> reason = '<STR_LIT:OK>' , <EOL> data = r"""<STR_LIT>""" , <EOL> curl = '<STR_LIT>' <EOL> ) <EOL> def tearDown ( self ) : <EOL> esbench . analyze . _get_benchmarks = self . tmp_get_benchmarks <EOL> esbench . analyze . _get_observations = self . tmp_get_observations <EOL> def test_benchmarks ( self ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( list ( esbench . analyze . _benchmarks ( resp = esbench . analyze . _get_benchmarks ( ) ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( list ( esbench . analyze . _benchmarks ( resp = esbench . analyze . _get_benchmarks ( ) , benchmark_ids = [ '<STR_LIT>' , '<STR_LIT:foo>' ] ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( list ( esbench . analyze . _benchmarks ( resp = esbench . analyze . _get_benchmarks ( ) , benchmark_ids = [ '<STR_LIT:foo>' ] ) ) ) ) <EOL> resp = esbench . api . ApiResponse ( status = <NUM_LIT:200> , reason = '<STR_LIT:OK>' , data = '<STR_LIT>' , curl = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( list ( esbench . analyze . _benchmarks ( resp = resp ) ) ) ) <EOL> r = esbench . api . ApiResponse ( <EOL> status = <NUM_LIT:200> , <EOL> reason = '<STR_LIT:OK>' , <EOL> data = '<STR_LIT>' , <EOL> curl = '<STR_LIT>' , <EOL> ) <EOL> self . assertEqual ( [ { u'<STR_LIT>' : u'<STR_LIT:foo>' } , { u'<STR_LIT>' : u'<STR_LIT:bar>' } , { u'<STR_LIT>' : u'<STR_LIT>' } ] , list ( esbench . analyze . _benchmarks ( resp = r ) ) ) <EOL> self . assertEqual ( [ { u'<STR_LIT>' : u'<STR_LIT:foo>' } , { u'<STR_LIT>' : u'<STR_LIT:bar>' } , { u'<STR_LIT>' : u'<STR_LIT>' } ] , list ( esbench . analyze . _benchmarks ( resp = r , benchmark_ids = [ '<STR_LIT:all>' ] ) ) ) <EOL> self . assertEqual ( [ { u'<STR_LIT>' : u'<STR_LIT:foo>' } , { u'<STR_LIT>' : u'<STR_LIT>' } ] , list ( esbench . analyze . _benchmarks ( resp = r , benchmark_ids = [ '<STR_LIT>' , '<STR_LIT>' ] ) ) ) <EOL> self . assertEqual ( [ { u'<STR_LIT>' : u'<STR_LIT:bar>' } , { u'<STR_LIT>' : u'<STR_LIT>' } ] , list ( esbench . analyze . _benchmarks ( resp = r , benchmark_ids = [ '<STR_LIT:1>' , '<STR_LIT>' ] ) ) ) <EOL> self . assertEqual ( [ { u'<STR_LIT>' : u'<STR_LIT>' } , { u'<STR_LIT>' : u'<STR_LIT:foo>' } , { u'<STR_LIT>' : u'<STR_LIT:bar>' } , { u'<STR_LIT>' : u'<STR_LIT>' } ] , list ( esbench . analyze . _benchmarks ( resp = r , benchmark_ids = [ '<STR_LIT>' , '<STR_LIT:all>' ] ) ) ) <EOL> def test_observations ( self ) : <EOL> resp = esbench . api . ApiResponse ( status = <NUM_LIT:200> , reason = '<STR_LIT:OK>' , data = '<STR_LIT>' , curl = '<STR_LIT>' ) <EOL> obs = list ( esbench . analyze . _observations ( resp = esbench . analyze . _get_observations ( ) ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( obs ) ) <EOL> def test_get ( self ) : <EOL> for data in esbench . analyze . get_data ( None ) : <EOL> self . assertEqual ( data . keys ( ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_flatten_container ( self ) : <EOL> obs = [ esbench . analyze . flatten_container ( container = o ) for o in esbench . analyze . get_data ( conn = None ) ] <EOL> self . assertEqual ( len ( obs ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( obs [ <NUM_LIT:0> ] ) , len ( obs [ <NUM_LIT:1> ] ) ) <EOL> obs = [ [ t for t in esbench . analyze . flatten_container ( container = o ) if t [ <NUM_LIT:0> ] == u'<STR_LIT>' ] for o in esbench . analyze . get_data ( conn = None ) ] <EOL> self . assertEqual ( obs , [ [ ( u'<STR_LIT>' , u'<STR_LIT>' ) ] , [ ( u'<STR_LIT>' , u'<STR_LIT>' ) ] ] ) <EOL> def test_filter ( self ) : <EOL> for data in esbench . analyze . get_data ( None ) : <EOL> f = esbench . analyze . flatten_container ( container = data ) <EOL> d = esbench . analyze . filter_tuples ( f ) <EOL> self . assertEqual ( sorted ( f ) , d ) <EOL> self . assertEqual ( esbench . analyze . filter_tuples ( f , pattern = '<STR_LIT>' ) , esbench . analyze . filter_tuples ( f , pattern = '<STR_LIT>' ) ) <EOL> def test_group_observations ( self ) : <EOL> data = list ( esbench . analyze . get_data ( None ) ) <EOL> benchmarks = esbench . analyze . group_observations ( data = data , fields = esbench . analyze . FIELDS ) <EOL> self . assertEqual ( '<STR_LIT>' , benchmarks [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , benchmarks [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] [ <NUM_LIT:1> ] ) <EOL> class FlattenContainerTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . d = { '<STR_LIT:foo>' : [ { '<STR_LIT:bar>' : '<STR_LIT>' } , <NUM_LIT:2> , '<STR_LIT>' , [ '<STR_LIT:foo>' , { '<STR_LIT:a>' : <NUM_LIT:1> , '<STR_LIT:b>' : <NUM_LIT:2> , '<STR_LIT:c>' : <NUM_LIT:3> } ] , ] } <EOL> self . a = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' , { '<STR_LIT>' : True } , set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def test_flatten ( self ) : <EOL> self . assertRaises ( ValueError , esbench . analyze . flatten_container , container = iter ( range ( <NUM_LIT:10> ) ) ) <EOL> f = esbench . analyze . flatten_container ( container = self . d ) <EOL> self . assertEqual ( sorted ( f ) , [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT:foo>' ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , <NUM_LIT:3> ) ] ) <EOL> f = esbench . analyze . flatten_container ( container = self . a ) <EOL> self . assertEqual ( sorted ( f ) , [ ( '<STR_LIT:0>' , '<STR_LIT:foo>' ) , ( '<STR_LIT:1>' , '<STR_LIT:bar>' ) , ( '<STR_LIT>' , True ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> fmt = '<STR_LIT>' <EOL> logging . basicConfig ( level = logging . DEBUG , format = fmt ) <EOL> logging . getLogger ( '<STR_LIT>' ) . setLevel ( logging . ERROR ) <EOL> unittest . main ( ) </s>
