<s> import time <EOL> import unittest <EOL> from Adafruit_IO import Client , Data , Feed , Group , RequestError <EOL> import base <EOL> BASE_URL = '<STR_LIT>' <EOL> PROXIES = None <EOL> class TestClient ( base . IOTestCase ) : <EOL> def get_client ( self ) : <EOL> return Client ( self . get_test_key ( ) , proxies = PROXIES , base_url = BASE_URL ) <EOL> def ensure_feed_deleted ( self , client , feed ) : <EOL> try : <EOL> client . delete_feed ( feed ) <EOL> except RequestError : <EOL> pass <EOL> def ensure_group_deleted ( self , client , group ) : <EOL> try : <EOL> client . delete_group ( group ) <EOL> except RequestError : <EOL> pass <EOL> def empty_feed ( self , client , feed ) : <EOL> data = client . data ( feed ) <EOL> for d in data : <EOL> client . delete ( feed , d . id ) <EOL> def test_set_key ( self ) : <EOL> key = "<STR_LIT>" <EOL> io = Client ( key ) <EOL> self . assertEqual ( key , io . key ) <EOL> def test_send_and_receive ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> response = io . send ( '<STR_LIT>' , '<STR_LIT:foo>' ) <EOL> self . assertEqual ( response . value , '<STR_LIT:foo>' ) <EOL> data = io . receive ( '<STR_LIT>' ) <EOL> self . assertEqual ( data . value , '<STR_LIT:foo>' ) <EOL> def test_receive_next ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> data = io . receive_next ( '<STR_LIT>' ) <EOL> self . assertEqual ( int ( data . value ) , <NUM_LIT:1> ) <EOL> @ unittest . expectedFailure <EOL> def test_receive_previous ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> io . receive_next ( '<STR_LIT>' ) <EOL> io . receive_next ( '<STR_LIT>' ) <EOL> data = io . receive_previous ( '<STR_LIT>' ) <EOL> self . assertEqual ( int ( data . value ) , <NUM_LIT:2> ) <EOL> data = io . receive_previous ( '<STR_LIT>' ) <EOL> self . assertEqual ( int ( data . value ) , <NUM_LIT:1> ) <EOL> def test_data_on_feed_returns_all_data ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> result = io . data ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( result ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( int ( result [ <NUM_LIT:0> ] . value ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( int ( result [ <NUM_LIT:1> ] . value ) , <NUM_LIT:2> ) <EOL> def test_data_on_feed_and_data_id_returns_data ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> data = io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> result = io . data ( '<STR_LIT>' , data . id ) <EOL> self . assertEqual ( data . id , result . id ) <EOL> self . assertEqual ( int ( data . value ) , int ( result . value ) ) <EOL> def test_create_data ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> data = Data ( value = <NUM_LIT> ) <EOL> result = io . create_data ( '<STR_LIT>' , data ) <EOL> self . assertEqual ( int ( result . value ) , <NUM_LIT> ) <EOL> def test_append_by_feed_name ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> feed = io . create_feed ( Feed ( name = '<STR_LIT>' ) ) <EOL> result = io . append ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . assertEqual ( int ( result . value ) , <NUM_LIT> ) <EOL> def test_append_by_feed_key ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> feed = io . create_feed ( Feed ( name = '<STR_LIT>' ) ) <EOL> result = io . append ( feed . key , <NUM_LIT> ) <EOL> self . assertEqual ( int ( result . value ) , <NUM_LIT> ) <EOL> def test_append_by_feed_id ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> feed = io . create_feed ( Feed ( name = '<STR_LIT>' ) ) <EOL> result = io . append ( feed . id , <NUM_LIT> ) <EOL> self . assertEqual ( int ( result . value ) , <NUM_LIT> ) <EOL> def test_create_feed ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> feed = Feed ( name = '<STR_LIT>' ) <EOL> result = io . create_feed ( feed ) <EOL> self . assertEqual ( result . name , '<STR_LIT>' ) <EOL> def test_feeds_returns_all_feeds ( self ) : <EOL> io = self . get_client ( ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> feeds = io . feeds ( ) <EOL> self . assertGreaterEqual ( len ( feeds ) , <NUM_LIT:1> ) <EOL> names = set ( map ( lambda x : x . name , feeds ) ) <EOL> self . assertTrue ( '<STR_LIT>' in names ) <EOL> def test_feeds_returns_requested_feed ( self ) : <EOL> io = self . get_client ( ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> result = io . feeds ( '<STR_LIT>' ) <EOL> self . assertEqual ( result . name , '<STR_LIT>' ) <EOL> self . assertEqual ( int ( result . last_value ) , <NUM_LIT:1> ) <EOL> def test_delete_feed ( self ) : <EOL> io = self . get_client ( ) <EOL> io . send ( '<STR_LIT>' , '<STR_LIT:foo>' ) <EOL> io . delete_feed ( '<STR_LIT>' ) <EOL> self . assertRaises ( RequestError , io . receive , '<STR_LIT>' ) <EOL> def test_delete_nonexistant_feed_fails ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> self . assertRaises ( RequestError , io . delete_feed , '<STR_LIT>' ) <EOL> def test_send_and_receive_group ( self ) : <EOL> io = self . get_client ( ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> response = io . send_group ( '<STR_LIT>' , { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> } ) <EOL> self . assertEqual ( len ( response . feeds ) , <NUM_LIT:2> ) <EOL> self . assertSetEqual ( set ( map ( lambda x : int ( x . stream . value ) , response . feeds ) ) , <EOL> set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> data = io . receive_group ( response . id ) <EOL> self . assertEqual ( len ( data . feeds ) , <NUM_LIT:2> ) <EOL> self . assertSetEqual ( set ( map ( lambda x : int ( x . stream . value ) , data . feeds ) ) , <EOL> set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> @ unittest . expectedFailure <EOL> def test_receive_next_group ( self ) : <EOL> io = self . get_client ( ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> response = io . send_group ( '<STR_LIT>' , { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } ) <EOL> data = io . receive_next_group ( response . id ) <EOL> self . assertEqual ( len ( data . feeds ) , <NUM_LIT:2> ) <EOL> self . assertSetEqual ( set ( map ( lambda x : int ( x . stream . value ) , data . feeds ) ) , <EOL> set ( [ <NUM_LIT> , <NUM_LIT> ] ) ) <EOL> @ unittest . expectedFailure <EOL> def test_receive_previous_group ( self ) : <EOL> io = self . get_client ( ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> self . empty_feed ( io , '<STR_LIT>' ) <EOL> response = io . send_group ( '<STR_LIT>' , { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:20> } ) <EOL> response = io . send_group ( '<STR_LIT>' , { '<STR_LIT>' : <NUM_LIT:30> , '<STR_LIT>' : <NUM_LIT> } ) <EOL> io . receive_next_group ( response . id ) <EOL> io . receive_next_group ( response . id ) <EOL> data = io . receive_previous_group ( response . id ) <EOL> self . assertEqual ( len ( data . feeds ) , <NUM_LIT:2> ) <EOL> self . assertSetEqual ( set ( map ( lambda x : int ( x . stream . value ) , data . feeds ) ) , <EOL> set ( [ <NUM_LIT:30> , <NUM_LIT> ] ) ) <EOL> data = io . receive_previous_group ( response . id ) <EOL> self . assertEqual ( len ( data . feeds ) , <NUM_LIT:2> ) <EOL> self . assertSetEqual ( set ( map ( lambda x : int ( x . stream . value ) , data . feeds ) ) , <EOL> set ( [ <NUM_LIT:10> , <NUM_LIT:20> ] ) ) <EOL> def test_groups_returns_all_groups ( self ) : <EOL> io = self . get_client ( ) <EOL> groups = io . groups ( ) <EOL> self . assertGreaterEqual ( len ( groups ) , <NUM_LIT:1> ) <EOL> names = set ( map ( lambda x : x . name , groups ) ) <EOL> self . assertTrue ( '<STR_LIT>' in names ) <EOL> def test_groups_retrieves_requested_group ( self ) : <EOL> io = self . get_client ( ) <EOL> response = io . groups ( '<STR_LIT>' ) <EOL> self . assertEqual ( response . name , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( response . feeds ) , <NUM_LIT:2> ) <EOL> @ unittest . expectedFailure <EOL> def test_create_group ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_group_deleted ( io , '<STR_LIT>' ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> self . ensure_feed_deleted ( io , '<STR_LIT>' ) <EOL> feed1 = io . create_feed ( Feed ( name = '<STR_LIT>' ) ) <EOL> feed2 = io . create_feed ( Feed ( name = '<STR_LIT>' ) ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:10> ) <EOL> io . send ( '<STR_LIT>' , <NUM_LIT:20> ) <EOL> group = Group ( name = '<STR_LIT>' , feeds = [ feed1 , feed2 ] ) <EOL> response = io . create_group ( group ) <EOL> self . assertEqual ( response . name , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( response . feeds ) , <NUM_LIT:2> ) <EOL> @ unittest . expectedFailure <EOL> def test_delete_group ( self ) : <EOL> io = self . get_client ( ) <EOL> self . ensure_group_deleted ( io , '<STR_LIT>' ) <EOL> group = io . create_group ( Group ( name = '<STR_LIT>' ) ) <EOL> io . delete_group ( '<STR_LIT>' ) <EOL> self . assertRaises ( RequestError , io . groups , '<STR_LIT>' ) </s>
