<s> import os , json <EOL> from twisted . trial import unittest <EOL> from nacl . public import PrivateKey <EOL> from . common import TwoNodeMixin <EOL> from . . errors import CommandError <EOL> from . . mailbox . retrieval import HTTPRetriever <EOL> def fetchone ( db , tablename ) : <EOL> return db . execute ( "<STR_LIT>" % tablename ) . fetchone ( ) <EOL> def fetchall ( db , tablename ) : <EOL> return db . execute ( "<STR_LIT>" % tablename ) . fetchall ( ) <EOL> class Creation ( TwoNodeMixin , unittest . TestCase ) : <EOL> def test_unknown_retrieval_type ( self ) : <EOL> basedir1 = os . path . join ( self . make_basedir ( ) , "<STR_LIT>" ) <EOL> self . createNode ( basedir1 ) <EOL> n1 = self . startNode ( basedir1 ) <EOL> bogus_retriever = { "<STR_LIT:type>" : "<STR_LIT>" } <EOL> self . failUnlessRaises ( CommandError , <EOL> n1 . agent . build_retriever , <NUM_LIT:1> , bogus_retriever ) <EOL> def test_http_retriever ( self ) : <EOL> basedir1 = os . path . join ( self . make_basedir ( ) , "<STR_LIT>" ) <EOL> self . createNode ( basedir1 ) <EOL> n1 = self . startNode ( basedir1 ) <EOL> rrec = { "<STR_LIT:type>" : "<STR_LIT:http>" , <EOL> "<STR_LIT>" : "<STR_LIT:/>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> rc = n1 . agent . build_retriever ( <NUM_LIT:1> , rrec ) <EOL> self . failUnless ( isinstance ( rc , HTTPRetriever ) ) <EOL> class Invite ( TwoNodeMixin , unittest . TestCase ) : <EOL> def test_offer_mailbox ( self ) : <EOL> nA , nB = self . make_nodes ( transport = "<STR_LIT:none>" ) <EOL> self . failUnlessEqual ( len ( nA . agent . get_transports ( ) ) , <NUM_LIT:0> ) <EOL> d = self . add_new_channel_with_invitation ( nA , nB , offer_mailbox = True , <EOL> accept_mailbox_offer = True ) <EOL> def _then ( ( entA , entB ) ) : <EOL> self . failUnlessEqual ( entA [ "<STR_LIT>" ] , entB [ "<STR_LIT>" ] ) <EOL> self . failUnlessEqual ( entB [ "<STR_LIT>" ] , entA [ "<STR_LIT>" ] ) <EOL> st = fetchone ( nB . db , "<STR_LIT>" ) <EOL> sj = fetchone ( nB . db , "<STR_LIT>" ) <EOL> s = json . loads ( sj [ "<STR_LIT>" ] ) <EOL> r_privkey = PrivateKey ( s [ "<STR_LIT>" ] . decode ( "<STR_LIT>" ) ) <EOL> r_pubkey = r_privkey . public_key . encode ( ) . encode ( "<STR_LIT>" ) <EOL> t_privkey = PrivateKey ( s [ "<STR_LIT>" ] . decode ( "<STR_LIT>" ) ) <EOL> t_pubkey = t_privkey . public_key . encode ( ) . encode ( "<STR_LIT>" ) <EOL> transports = nA . agent . get_transports ( ) <EOL> self . failUnlessEqual ( len ( transports ) , <NUM_LIT:1> ) <EOL> self . failUnless ( <NUM_LIT:1> in transports ) <EOL> t1 = transports [ <NUM_LIT:1> ] [ "<STR_LIT>" ] <EOL> self . failUnlessEqual ( t1 [ "<STR_LIT>" ] [ "<STR_LIT:type>" ] , "<STR_LIT:http>" ) <EOL> self . failUnlessEqual ( t1 [ "<STR_LIT>" ] [ "<STR_LIT:url>" ] , nB . baseurl + "<STR_LIT>" ) <EOL> self . failUnlessEqual ( t1 [ "<STR_LIT>" ] [ "<STR_LIT>" ] , t_pubkey ) <EOL> self . failUnlessIn ( "<STR_LIT>" , t1 [ "<STR_LIT>" ] ) <EOL> r = transports [ <NUM_LIT:1> ] [ "<STR_LIT>" ] <EOL> self . failUnlessEqual ( r [ "<STR_LIT:type>" ] , "<STR_LIT:http>" ) <EOL> self . failUnlessEqual ( r [ "<STR_LIT>" ] , nB . baseurl + "<STR_LIT>" ) <EOL> self . failUnlessEqual ( r [ "<STR_LIT>" ] , st [ "<STR_LIT>" ] ) <EOL> self . failUnlessEqual ( r [ "<STR_LIT>" ] , r_pubkey ) <EOL> self . failUnlessEqual ( r [ "<STR_LIT>" ] , st [ "<STR_LIT>" ] ) <EOL> retrievers = nA . agent . mailbox_retrievers <EOL> self . failUnlessEqual ( len ( retrievers ) , <NUM_LIT:1> ) <EOL> self . failUnless ( isinstance ( list ( retrievers ) [ <NUM_LIT:0> ] , HTTPRetriever ) ) <EOL> return nA . disownServiceParent ( ) <EOL> d . addCallback ( _then ) <EOL> d . addCallback ( lambda _ : self . startNode ( nA . basedir ) ) <EOL> def _then2 ( nA2 ) : <EOL> retrievers = nA2 . agent . mailbox_retrievers <EOL> self . failUnlessEqual ( len ( retrievers ) , <NUM_LIT:1> ) <EOL> self . failUnless ( isinstance ( list ( retrievers ) [ <NUM_LIT:0> ] , HTTPRetriever ) ) <EOL> d . addCallback ( _then2 ) <EOL> return d <EOL> def test_offer_mailbox_rejected ( self ) : <EOL> nA , nB = self . make_nodes ( transport = "<STR_LIT:none>" ) <EOL> self . failUnlessEqual ( len ( nA . agent . get_transports ( ) ) , <NUM_LIT:0> ) <EOL> d = self . add_new_channel_with_invitation ( nA , nB , offer_mailbox = True ) <EOL> def _then ( ( entA , entB ) ) : <EOL> self . failUnlessEqual ( entA [ "<STR_LIT>" ] , entB [ "<STR_LIT>" ] ) <EOL> self . failUnlessEqual ( entB [ "<STR_LIT>" ] , entA [ "<STR_LIT>" ] ) <EOL> transports = nA . agent . get_transports ( ) <EOL> self . failIf ( transports ) <EOL> d . addCallback ( _then ) <EOL> return d </s>
