<s> import unittest <EOL> import pybase <EOL> from collections import defaultdict <EOL> from time import sleep <EOL> from pybase . exceptions import * <EOL> table = "<STR_LIT:test>" <EOL> zkquorum = "<STR_LIT:localhost>" <EOL> cf1 = "<STR_LIT>" <EOL> cf2 = "<STR_LIT>" <EOL> class TestClient ( unittest . TestCase ) : <EOL> def test_new_client_good ( self ) : <EOL> c = pybase . NewClient ( zkquorum ) <EOL> self . assertEqual ( c . zkquorum , zkquorum ) <EOL> self . assertIsNotNone ( c . master_client ) <EOL> self . assertIsNotNone ( c . master_client . host ) <EOL> def test_new_client_bad ( self ) : <EOL> try : <EOL> c = pybase . NewClient ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except ZookeeperException : <EOL> pass <EOL> def test_client_close ( self ) : <EOL> c = pybase . NewClient ( zkquorum ) <EOL> c . get ( table , "<STR_LIT:U+0020>" ) <EOL> c . close ( ) <EOL> self . assertEqual ( len ( c . region_cache ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( c . reverse_client_cache ) , <NUM_LIT:0> ) <EOL> def test_client_close_then_reuse ( self ) : <EOL> c = pybase . NewClient ( zkquorum ) <EOL> values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> c . put ( table , self . __class__ . __name__ , values ) <EOL> c . close ( ) <EOL> res = c . get ( table , self . __class__ . __name__ ) <EOL> self . assertEqual ( result_to_dict ( res ) , { cf1 : { "<STR_LIT>" : "<STR_LIT>" } } ) <EOL> class TestGet ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . c . put ( table , cls . __name__ , cls . values ) <EOL> cls . row_prefix = cls . __name__ <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_get_entire_row ( self ) : <EOL> res = self . c . get ( table , self . row_prefix ) <EOL> self . assertEqual ( result_to_dict ( res ) , self . values ) <EOL> def test_get_specific_cell ( self ) : <EOL> families = { cf1 : "<STR_LIT>" } <EOL> res = self . c . get ( table , self . row_prefix , families = families ) <EOL> resd = result_to_dict ( res ) <EOL> self . assertEqual ( resd , { cf1 : { "<STR_LIT>" : "<STR_LIT>" } } ) <EOL> self . assertNotIn ( cf2 , resd . keys ( ) ) <EOL> def test_get_bad_table ( self ) : <EOL> try : <EOL> res = self . c . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchTableException : <EOL> pass <EOL> def test_get_bad_row ( self ) : <EOL> res = self . c . get ( table , "<STR_LIT>" ) <EOL> self . assertFalse ( res . exists ) <EOL> def test_get_bad_column_family ( self ) : <EOL> families = { "<STR_LIT>" : "<STR_LIT>" } <EOL> try : <EOL> self . c . get ( table , self . row_prefix , families = families ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> def test_get_bad_column_qualifier ( self ) : <EOL> families = { cf2 : "<STR_LIT>" } <EOL> res = self . c . get ( table , self . row_prefix , families = families ) <EOL> self . assertEqual ( result_to_dict ( res ) , { } ) <EOL> def test_get_with_filter ( self ) : <EOL> ft = pybase . filters . ColumnPrefixFilter ( "<STR_LIT>" ) <EOL> res = self . c . get ( table , self . row_prefix , filters = ft ) <EOL> resd = result_to_dict ( res ) <EOL> self . assertDictContainsSubset ( resd , self . values ) <EOL> self . assertNotIn ( cf2 , resd . keys ( ) ) <EOL> def test_get_with_bad_filter ( self ) : <EOL> ft = "<STR_LIT>" <EOL> try : <EOL> res = self . c . get ( table , self . row_prefix , filters = ft ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except ValueError : <EOL> pass <EOL> class TestPut ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . families = { <EOL> cf1 : [ "<STR_LIT>" ] , <EOL> cf2 : [ "<STR_LIT>" ] <EOL> } <EOL> cls . row_prefix = cls . __name__ <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_put_single_cell ( self ) : <EOL> values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> self . c . put ( table , self . row_prefix + "<STR_LIT:1>" , values ) <EOL> res = self . c . get ( table , self . row_prefix + "<STR_LIT:1>" , families = self . families ) <EOL> self . assertDictContainsSubset ( result_to_dict ( res ) , self . values ) <EOL> self . assertEqual ( len ( res . flatten_cells ( ) ) , <NUM_LIT:1> ) <EOL> def test_put_multiple_cells ( self ) : <EOL> self . c . put ( table , self . row_prefix + "<STR_LIT:2>" , self . values ) <EOL> res = self . c . get ( table , self . row_prefix + "<STR_LIT:2>" , families = self . families ) <EOL> self . assertEqual ( result_to_dict ( res ) , self . values ) <EOL> def test_put_bad_table ( self ) : <EOL> try : <EOL> self . c . put ( "<STR_LIT>" , "<STR_LIT>" , self . values ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchTableException : <EOL> pass <EOL> def test_put_malformed_values ( self ) : <EOL> values = { "<STR_LIT>" : "<STR_LIT>" } <EOL> try : <EOL> self . c . put ( table , self . row_prefix + "<STR_LIT:3>" , values ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except MalformedValues : <EOL> pass <EOL> def test_put_bad_column_family ( self ) : <EOL> values = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> try : <EOL> self . c . put ( table , self . row_prefix + "<STR_LIT:4>" , values ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> class TestScan ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . families = { <EOL> cf1 : [ "<STR_LIT>" ] , <EOL> cf2 : [ "<STR_LIT>" ] <EOL> } <EOL> cls . row_prefix = cls . __name__ <EOL> cls . pFilter = pybase . filters . PrefixFilter ( cls . row_prefix ) <EOL> cls . num_ops = <NUM_LIT:100> <EOL> for x in range ( cls . num_ops ) : <EOL> cls . c . put ( table , cls . row_prefix + str ( x ) , cls . values ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_scan_simple ( self ) : <EOL> rsp = self . c . scan ( table , filters = self . pFilter ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:200> ) <EOL> def test_scan_with_range ( self ) : <EOL> rsp = self . c . scan ( <EOL> table , start_key = self . row_prefix + "<STR_LIT:0>" , stop_key = self . row_prefix + "<STR_LIT>" , filters = self . pFilter ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT> ) <EOL> def test_scan_with_bad_range ( self ) : <EOL> try : <EOL> rsp = self . c . scan ( <EOL> table , start_key = "<STR_LIT>" , stop_key = <NUM_LIT> , filters = self . pFilter ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except TypeError : <EOL> pass <EOL> def test_scan_with_families ( self ) : <EOL> fam = { cf1 : [ "<STR_LIT>" ] } <EOL> rsp = self . c . scan ( table , filters = self . pFilter , families = fam ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:100> ) <EOL> def test_scan_with_bad_column_family ( self ) : <EOL> fam = { "<STR_LIT>" : [ "<STR_LIT>" ] } <EOL> try : <EOL> rsp = self . c . scan ( table , filters = self . pFilter , families = fam ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> def test_scan_with_bad_column_qualifier ( self ) : <EOL> fam = { cf1 : [ "<STR_LIT>" ] , cf2 : [ "<STR_LIT>" ] } <EOL> rsp = self . c . scan ( table , filters = self . pFilter , families = fam ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:100> ) <EOL> def test_scan_with_second_filter ( self ) : <EOL> new_filter = pybase . filters . ColumnPrefixFilter ( "<STR_LIT>" ) <EOL> f_list = pybase . filters . FilterList ( pybase . filters . MUST_PASS_ALL , new_filter , self . pFilter ) <EOL> rsp = self . c . scan ( table , filters = f_list ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:100> ) <EOL> def test_scan_with_bad_filter ( self ) : <EOL> new_filter = "<STR_LIT>" <EOL> try : <EOL> self . c . scan ( table , filters = new_filter ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except ValueError : <EOL> pass <EOL> class TestDelete ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . families = { <EOL> cf1 : [ "<STR_LIT>" ] , <EOL> cf2 : [ "<STR_LIT>" ] <EOL> } <EOL> cls . values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . row_prefix = cls . __name__ <EOL> for x in range ( <NUM_LIT:5> ) : <EOL> cls . c . put ( table , cls . row_prefix + str ( x ) , cls . values ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_delete_entire_row ( self ) : <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:0>" , self . values ) <EOL> self . assertTrue ( rsp . processed ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:0>" , self . families ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:0> ) <EOL> def test_delete_specific_cell ( self ) : <EOL> value = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:1>" , value ) <EOL> self . assertTrue ( rsp . processed ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:1>" , self . families ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:1> ) <EOL> def test_delete_bad_row ( self ) : <EOL> rsp = self . c . delete ( table , "<STR_LIT>" , self . values ) <EOL> self . assertFalse ( rsp . exists ) <EOL> def test_delete_bad_column_family ( self ) : <EOL> value = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> try : <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:2>" , value ) <EOL> self . assertEqual ( <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> def test_delete_bad_column_qualifier ( self ) : <EOL> value = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:3>" , value ) <EOL> self . assertFalse ( rsp . exists ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:3>" , self . families ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:1> ) <EOL> class TestAppend ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . families = { <EOL> cf1 : [ "<STR_LIT>" ] , <EOL> cf2 : [ "<STR_LIT>" ] <EOL> } <EOL> cls . values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . row_prefix = cls . __name__ <EOL> for x in range ( <NUM_LIT:5> ) : <EOL> cls . c . put ( table , cls . row_prefix + str ( x ) , cls . values ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_append_entire_row ( self ) : <EOL> rsp = self . c . append ( table , self . row_prefix + "<STR_LIT:0>" , self . values ) <EOL> rspd = result_to_dict ( rsp ) <EOL> self . assertEqual ( rspd [ cf1 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( rspd [ cf2 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:0>" , self . families ) <EOL> rspd = result_to_dict ( rsp ) <EOL> self . assertEqual ( rspd [ cf1 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( rspd [ cf2 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_append_specific_cell ( self ) : <EOL> values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> rsp = self . c . append ( table , self . row_prefix + "<STR_LIT:1>" , values ) <EOL> rspd = result_to_dict ( rsp ) <EOL> self . assertEqual ( rspd [ cf1 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:1>" , self . families ) <EOL> rspd = result_to_dict ( rsp ) <EOL> self . assertEqual ( rspd [ cf1 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( rspd [ cf2 ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_append_bad_row ( self ) : <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:2>" , self . values ) <EOL> rsp = self . c . append ( table , self . row_prefix + "<STR_LIT:2>" , self . values ) <EOL> self . assertFalse ( rsp . processed ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:2>" ) <EOL> self . assertEqual ( len ( rsp . cells ) , <NUM_LIT:0> ) <EOL> def test_append_bad_column_family ( self ) : <EOL> values = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } <EOL> try : <EOL> rsp = self . c . append ( table , self . row_prefix + "<STR_LIT:3>" , values ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> def test_append_bad_column_qualifier ( self ) : <EOL> values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:4>" , values ) <EOL> rsp = self . c . append ( table , self . row_prefix + "<STR_LIT:4>" , values ) <EOL> self . assertFalse ( rsp . processed ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:4>" ) <EOL> self . assertEqual ( len ( rsp . flatten_cells ( ) ) , <NUM_LIT:2> ) <EOL> class TestIncrement ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . c = pybase . NewClient ( zkquorum ) <EOL> cls . families = { <EOL> cf1 : [ "<STR_LIT>" ] , <EOL> cf2 : [ "<STR_LIT>" ] <EOL> } <EOL> cls . start_values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . inc_values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> cls . row_prefix = cls . __name__ <EOL> for x in range ( <NUM_LIT:6> ) : <EOL> cls . c . put ( table , cls . row_prefix + str ( x ) , cls . start_values ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . c . close ( ) <EOL> def test_increment_entire_row ( self ) : <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:0>" , self . inc_values ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:0>" , self . families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:0>" , self . inc_values ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:0>" , self . families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> cf2 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> def test_increment_specific_cell ( self ) : <EOL> new_values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> new_families = { <EOL> cf1 : [ "<STR_LIT>" ] <EOL> } <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:1>" , new_values ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:1>" , new_families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:1>" , new_values ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:1>" , new_families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> def test_increment_bad_column_family ( self ) : <EOL> new_values = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> try : <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:2>" , new_values ) <EOL> self . assertEqual ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> except NoSuchColumnFamilyException : <EOL> pass <EOL> def test_increment_new_column_qualifier ( self ) : <EOL> new_values = { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> new_families = { <EOL> cf1 : [ "<STR_LIT>" ] <EOL> } <EOL> rsp = self . c . delete ( table , self . row_prefix + "<STR_LIT:3>" , new_values ) <EOL> self . assertTrue ( rsp . processed ) <EOL> sleep ( <NUM_LIT> ) <EOL> rsp2 = self . c . increment ( table , self . row_prefix + "<STR_LIT:3>" , new_values ) <EOL> self . assertEqual ( result_to_dict ( rsp2 ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:3>" , new_families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . increment ( table , self . row_prefix + "<STR_LIT:3>" , new_values ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> rsp = self . c . get ( table , self . row_prefix + "<STR_LIT:3>" , new_families ) <EOL> self . assertEqual ( result_to_dict ( rsp ) , { <EOL> cf1 : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> } ) <EOL> def result_to_dict ( res ) : <EOL> hodor = defaultdict ( dict ) <EOL> for cell in res . flatten_cells ( ) : <EOL> hodor [ cell . family ] [ cell . qualifier ] = cell . value <EOL> return hodor <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
