<s> import json <EOL> from asyncflux import AsyncfluxClient <EOL> from asyncflux . testing import AsyncfluxTestCase , gen_test <EOL> from asyncflux . user import User <EOL> from asyncflux . errors import AsyncfluxError <EOL> class UserTestCase ( AsyncfluxTestCase ) : <EOL> @ gen_test <EOL> def test_update ( self ) : <EOL> client = AsyncfluxClient ( ) <EOL> db_name = '<STR_LIT:foo>' <EOL> db = client [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> user = User ( db , username ) <EOL> password = '<STR_LIT>' <EOL> is_admin = True <EOL> read_from = '<STR_LIT>' <EOL> write_to = '<STR_LIT>' <EOL> payload = { '<STR_LIT:password>' : password } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . update ( new_password = password ) <EOL> self . assertIsNone ( response ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> payload = { '<STR_LIT>' : is_admin } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . update ( is_admin = is_admin ) <EOL> self . assertIsNone ( response ) <EOL> self . assertEqual ( user . is_admin , is_admin ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> payload = { '<STR_LIT>' : read_from , '<STR_LIT>' : write_to } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . update ( read_from = read_from , <EOL> write_to = write_to ) <EOL> self . assertIsNone ( response ) <EOL> self . assertEqual ( user . read_from , read_from ) <EOL> self . assertEqual ( user . write_to , write_to ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> exc_msg = '<STR_LIT>' <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . update ( password , is_admin = is_admin , read_from = read_from ) <EOL> exc_msg = '<STR_LIT>' <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . update ( ) <EOL> @ gen_test <EOL> def test_change_password ( self ) : <EOL> client = AsyncfluxClient ( ) <EOL> db_name = '<STR_LIT:foo>' <EOL> db = client [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> password = '<STR_LIT>' <EOL> user = User ( db , username ) <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . change_password ( password ) <EOL> self . assertIsNone ( response ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , <EOL> body = json . dumps ( { '<STR_LIT:password>' : password } ) ) <EOL> response_body = '<STR_LIT>' % username <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT> , body = response_body ) <EOL> with self . assertRaisesRegexp ( AsyncfluxError , response_body ) : <EOL> yield user . change_password ( password ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , <EOL> body = json . dumps ( { '<STR_LIT:password>' : password } ) ) <EOL> password = '<STR_LIT:bar>' <EOL> response_body = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT> , body = response_body ) <EOL> with self . assertRaisesRegexp ( AsyncfluxError , response_body ) : <EOL> yield user . change_password ( password ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , <EOL> body = json . dumps ( { '<STR_LIT:password>' : password } ) ) <EOL> @ gen_test <EOL> def test_change_privileges ( self ) : <EOL> client = AsyncfluxClient ( ) <EOL> db_name = '<STR_LIT:foo>' <EOL> db = client [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> user = User ( db , username ) <EOL> is_admin = True <EOL> read_from = '<STR_LIT>' <EOL> write_to = '<STR_LIT>' <EOL> payload = { '<STR_LIT>' : is_admin , '<STR_LIT>' : read_from , <EOL> '<STR_LIT>' : write_to } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . change_privileges ( is_admin , <EOL> read_from = read_from , <EOL> write_to = write_to ) <EOL> self . assertIsNone ( response ) <EOL> self . assertEqual ( user . is_admin , is_admin ) <EOL> self . assertEqual ( user . read_from , read_from ) <EOL> self . assertEqual ( user . write_to , write_to ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> payload = { '<STR_LIT>' : is_admin } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> yield user . change_privileges ( is_admin , None , None ) <EOL> self . assertIsNone ( response ) <EOL> self . assertEqual ( user . is_admin , is_admin ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> payload = { '<STR_LIT>' : is_admin , '<STR_LIT>' : read_from , <EOL> '<STR_LIT>' : write_to } <EOL> response_body = "<STR_LIT>" % username <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT> , body = response_body ) <EOL> with self . assertRaisesRegexp ( AsyncfluxError , response_body ) : <EOL> yield user . change_privileges ( is_admin , read_from = read_from , <EOL> write_to = write_to ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> exc_msg = '<STR_LIT>' <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . change_privileges ( is_admin , read_from , None ) <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . change_privileges ( is_admin , None , write_to ) <EOL> @ gen_test <EOL> def test_change_permissions ( self ) : <EOL> client = AsyncfluxClient ( ) <EOL> db_name = '<STR_LIT:foo>' <EOL> db = client [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> user = User ( db , username ) <EOL> read_from = '<STR_LIT>' <EOL> write_to = '<STR_LIT>' <EOL> payload = { '<STR_LIT>' : read_from , '<STR_LIT>' : write_to } <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . change_permissions ( read_from = read_from , <EOL> write_to = write_to ) <EOL> self . assertIsNone ( response ) <EOL> self . assertEqual ( user . read_from , read_from ) <EOL> self . assertEqual ( user . write_to , write_to ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> response_body = "<STR_LIT>" % username <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT> , body = response_body ) <EOL> with self . assertRaisesRegexp ( AsyncfluxError , response_body ) : <EOL> yield user . change_permissions ( read_from = read_from , <EOL> write_to = write_to ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT:POST>' , body = json . dumps ( payload ) ) <EOL> exc_msg = '<STR_LIT>' <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . change_permissions ( None , None ) <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . change_permissions ( read_from , None ) <EOL> with self . assertRaisesRegexp ( ValueError , exc_msg ) : <EOL> yield user . change_permissions ( None , write_to ) <EOL> @ gen_test <EOL> def test_delete ( self ) : <EOL> client = AsyncfluxClient ( ) <EOL> db_name = '<STR_LIT:foo>' <EOL> db = client [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> user = User ( db , username ) <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT:200> ) <EOL> response = yield user . delete ( ) <EOL> self . assertIsNone ( response ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT>' ) <EOL> response_body = "<STR_LIT>" % username <EOL> with self . patch_fetch_mock ( client ) as m : <EOL> self . setup_fetch_mock ( m , <NUM_LIT> , body = response_body ) <EOL> with self . assertRaisesRegexp ( AsyncfluxError , response_body ) : <EOL> yield user . delete ( ) <EOL> self . assert_mock_args ( m , '<STR_LIT>' % ( db_name , username ) , <EOL> method = '<STR_LIT>' ) <EOL> def test_repr ( self ) : <EOL> host = '<STR_LIT:localhost>' <EOL> port = <NUM_LIT> <EOL> db_name = '<STR_LIT:foo>' <EOL> db = AsyncfluxClient ( host , port ) [ db_name ] <EOL> username = '<STR_LIT:foo>' <EOL> format_repr = "<STR_LIT>" <EOL> self . assertEqual ( repr ( User ( db , username ) ) , <EOL> ( format_repr % ( host , port , db_name , username ) ) ) </s>
