""" :copyright: 2010 by Manuel Jacob :license: MIT """ from os import remove from tempfile import NamedTemporaryFile from mock import Mock, MagicMock, sentinel from nose.tools import assert_raises from tests.constants import (SERVER1, SERVER2, USER1_UNI, USER2_UNI, USER1_NICK, USER2_NICK, RESOURCE, RESOURCE1_UNI, PLACE_UNI, VARIABLES, CONTENT, PASSWORD, MESSAGE, ONLINE, PRESENCE_UNKNOWN, PRESENCE_ONLINE, OFFERED, ESTABLISHED, ERROR, EXCEPTION) from tests.helpers import mockified, PlaceHolder, inited_header, AsyncMethod from pypsyc.client.model import (Conversation, Member, Conference, Messaging, Conferencing, Account, FriendList, ClientCircuit, Circuit, Client) from pypsyc.core.psyc import PSYCPacket from pypsyc.protocol import (UnknownTargetError, DeliveryFailedError, AuthenticationError) class TestConversation(object): def test_send_message(self): message_ph = PlaceHolder() messaging = Mock() conversation = Conversation(messaging, USER2_UNI) conversation.send_message(MESSAGE) assert messaging.protocol.method_calls == [ ('send_private_message', (USER2_UNI, MESSAGE), {'relay': messaging.account.uni})] assert conversation.messages == [message_ph] assert message_ph.obj.source == messaging.account.uni class TestConference(object): def test_send_message(self): conferencing = Mock() conference = Conference(conferencing, PLACE_UNI) conference.send_message(MESSAGE) assert conferencing.protocol.method_calls == [ ('send_public_message', (PLACE_UNI, MESSAGE))] class TestMessaging(object): def test_recv_message(self): account = Mock() conversation = account.get_conversation.return_value conversation.messages = [] message_ph = PlaceHolder() messaging = Messaging(account) messaging.private_message(USER2_UNI, MESSAGE) assert account.get_conversation.call_args_list == [((USER2_UNI,),)] assert conversation.messages == [message_ph] assert message_ph.obj.source == USER2_UNI assert message_ph.obj.message == MESSAGE @mockified('pypsyc.client.model', ['MessagingProtocol']) def test_connected(self, MessagingProtocol): circuit = Mock() messaging = Messaging(Mock()) messaging.connected(circuit) assert circuit.psyc.method_calls == [ ('add_handler', (MessagingProtocol.return_value,))] def test_disconnected(self): Messaging(Mock).disconnected() class TestConferencing(object): def test_member_entered(self): account = Mock() conference = account.get_conference.return_value conference.members = [] conferencing = Conferencing(account) conferencing.member_entered(PLACE_UNI, USER2_UNI, USER2_NICK) assert account.get_conference.call_args_list == [((PLACE_UNI,),)] assert conference.members == [Member(USER2_UNI, USER2_NICK)] def test_member_left(self): account = Mock() conference = account.get_conference.return_value conference.members = [Member(USER1_UNI, USER1_NICK), Member(USER2_UNI, USER2_NICK)] conferencing = Conferencing(account) conferencing.member_left(PLACE_UNI, USER2_UNI) assert conference.members == [Member(USER1_UNI, USER1_NICK)] def test_recv_message(self): account = Mock() conference = account.get_conference.return_value conference.messages = [] message_ph = PlaceHolder() conferencing = Conferencing(account) conferencing.public_message(PLACE_UNI, USER2_UNI, MESSAGE) assert account.get_conference.call_args_list == [((PLACE_UNI,),)] assert conference.messages == [message_ph] assert message_ph.obj.source == USER2_UNI assert message_ph.obj.message == MESSAGE @mockified('pypsyc.client.model', ['ConferencingProtocol']) def test_connected(self, ConferencingProtocol): circuit = Mock() conferencing = Conferencing(Mock()) conferencing.connected(circuit) assert circuit.method_calls == [ ('psyc.add_handler', (ConferencingProtocol.return_value,)), ('add_pvar_handler', (ConferencingProtocol.return_value,))] def test_disconnected(self): Conferencing(Mock()).disconnected() class _Dict(dict): def __init__(self, *args, **kwds): dict.__init__(self, *args, **kwds) self.updated_item = Mock() class TestFriendList(object): def test_friendship(self): friend_ph = PlaceHolder() account = Mock() account.friends = _Dict() friend_list = FriendList(account) friend_list.friendship(USER1_UNI, OFFERED) assert account.friends == {USER1_UNI: friend_ph} friend = friend_ph.obj assert friend.account == account assert friend.uni == USER1_UNI assert friend.presence == PRESENCE_UNKNOWN assert friend.state == OFFERED friend_list.friendship(USER1_UNI, ESTABLISHED) assert friend.state == ESTABLISHED assert account.friends.updated_item.call_args_list == [((USER1_UNI,),)] friend_list.friendship_removed(USER1_UNI) assert account.friends == {} def test_presence(self): account = Mock() account.friends = _Dict({USER2_UNI: Mock()}) friend_list = FriendList(account) friend_list.presence(USER2_UNI, ONLINE) assert account.friends[USER2_UNI].presence == PRESENCE_ONLINE assert account.friends.updated_item.call_args_list == [((USER2_UNI,),)] @mockified('pypsyc.client.model', ['FriendshipProtocol']) def test_connected(self, FriendshipProtocol): circuit = Mock() friend_list = FriendList(Mock()) friend_list.connected(circuit) assert circuit.method_calls == [ ('psyc.add_handler', (FriendshipProtocol.return_value,)), ('add_pvar_handler', (FriendshipProtocol.return_value,))] def test_disconnected(self): account = Mock() account.friends = _Dict({USER2_UNI: Mock()}) friend_list = FriendList(account) friend_list.disconnected() assert account.friends == {} class TestAccount(object): @mockified('pypsyc.client.model', ['ObsDict']) def test_account(self, ObsDict): obs_dict = ObsDict.return_value = MagicMock() account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) assert account.uni == USER1_UNI assert ObsDict.call_args_list == [(), (), ()] assert account.conversations == obs_dict assert account.conferences == obs_dict assert account.friends == obs_dict @mockified('pypsyc.client.model', ['Messaging', 'Conversation']) def test_get_conversation(self, Messaging, Conversation): messaging = Messaging.return_value conversation = Conversation.return_value account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) assert Messaging.call_args_list == [((account,),)] assert account.get_conversation(USER2_UNI) == conversation assert account.get_conversation(USER2_UNI) == conversation assert Conversation.call_args_list == [((messaging, USER2_UNI),)] assert account.conversations == {USER2_UNI: conversation} @mockified('pypsyc.client.model', ['Conferencing', 'Conference']) def test_get_conference(self, Conferencing, Conference): conferencing = Conferencing.return_value conference = Conference.return_value account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) assert Conferencing.call_args_list == [((account,),)] assert account.get_conference(PLACE_UNI) == conference assert account.get_conference(PLACE_UNI) == conference assert Conference.call_args_list == [((conferencing, PLACE_UNI),)] assert account.conferences == {PLACE_UNI: conference} @mockified('pypsyc.client.model', ['Conferencing', 'Conference']) def test_get_conference_subscribe(self, Conferencing, Conference): conferencing = Conferencing.return_value conference = Conference.return_value account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) client_interface = account.client_interface = Mock() client_interface.subscribe.side_effect = AsyncMethod() assert account.get_conference(PLACE_UNI, subscribe=True) == conference assert account.get_conference(PLACE_UNI, subscribe=True) == conference assert Conference.call_args_list == [((conferencing, PLACE_UNI),)] assert account.conferences == {PLACE_UNI: conference} assert client_interface.method_calls == [('subscribe', (PLACE_UNI,))] @mockified('pypsyc.client.model', ['Conference']) def test_get_conference_subscribe_unknown_target(self, Conference): conference = Conference.return_value account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) client_interface = account.client_interface = Mock() client_interface.subscribe.side_effect = UnknownTargetError account.get_conference(PLACE_UNI, subscribe=True) assert conference.method_calls == [('unknown_target_evt',)] @mockified('pypsyc.client.model', ['Conference']) def test_get_conference_subscribe_delivery_failed(self, Conference): conference = Conference.return_value account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) client_interface = account.client_interface = Mock() client_interface.subscribe.side_effect = DeliveryFailedError(ERROR) account.get_conference(PLACE_UNI, subscribe=True) assert conference.method_calls == [('delivery_failed_evt', (ERROR,))] def test_remove_conference(self): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) account.conferences[PLACE_UNI] = Mock() client_interface = account.client_interface = Mock() client_interface.unsubscribe.side_effect = AsyncMethod() del account.conferences[PLACE_UNI] assert client_interface.method_calls == [('unsubscribe', (PLACE_UNI,))] def test_add_friend(self): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) client_interface = account.client_interface = Mock() account.add_friend(USER2_UNI) assert client_interface.method_calls == [('add_friend', (USER2_UNI,))] def test_remove_friend(self): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) client_interface = account.client_interface = Mock() account.remove_friend(USER2_UNI) assert client_interface.method_calls == [ ('remove_friend', (USER2_UNI,))] @mockified('pypsyc.client.model', ['FriendList']) def test_friendlist(self, FriendList): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) assert FriendList.call_args_list == [((account,),)] def test_unknown_target_error(self): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) account.conversations[USER1_UNI] = conversation = Mock() account.conferences[PLACE_UNI] = conference = Mock() account.unknown_target_error(USER1_UNI) account.unknown_target_error(USER2_UNI) account.unknown_target_error(PLACE_UNI) assert conversation.method_calls == [('unknown_target_evt',)] assert conference.method_calls == [('unknown_target_evt',)] def test_delivery_failed_error(self): account = Account(None, SERVER1, USER1_NICK, PASSWORD, True, False) account.conversations[USER1_UNI] = conversation = Mock() account.conferences[PLACE_UNI] = conference = Mock() account.delivery_failed_error(USER1_UNI, ERROR) account.delivery_failed_error(USER2_UNI, ERROR) account.delivery_failed_error(PLACE_UNI, ERROR) assert conversation.method_calls == [('delivery_failed_evt', (ERROR,))] assert conference.method_calls == [('delivery_failed_evt', (ERROR,))] @mockified('pypsyc.client.model', ['resolve_hostname', 'connect', 'LinkingProtocol', 'RoutingErrorProtocol', 'ClientInterfaceProtocol', 'Messaging', 'Conferencing', 'FriendList']) def test_active(self, resolve_hostname, connect, LinkingProtocol, RoutingErrorProtocol, ClientInterfaceProtocol, Messaging, Conferencing, FriendList): resolve_hostname.return_value = (sentinel.ip, sentinel.port) circuit = connect.return_value linking_protocol = LinkingProtocol.return_value linked = linking_protocol.link.side_effect = AsyncMethod() messaging = Messaging.return_value conferencing = Conferencing.return_value friend_list = FriendList.return_value client = Mock() account = Account(client, SERVER1, USER1_NICK, PASSWORD, True, False) account.active = True account.active = True assert resolve_hostname.call_args_list == [((account.server,),)] assert connect.call_args_list == [ ((sentinel.ip, sentinel.port, client),)] assert LinkingProtocol.call_args_list == [((circuit,),)] assert linking_protocol.method_calls == [ ('link', (USER1_UNI, RESOURCE, PASSWORD))] linked.callback() assert circuit.psyc.uni == RESOURCE1_UNI assert RoutingErrorProtocol.call_args_list == [((account,),)] assert circuit.psyc.method_calls == [ ('add_handler', (RoutingErrorProtocol.return_value,))] assert ClientInterfaceProtocol.call_args_list == [((account,),)] assert messaging.method_calls == [('connected', (circuit,))] assert conferencing.method_calls == [('connected', (circuit,))] assert friend_list.method_calls == [('connected', (circuit,))] messaging.reset_mock() conferencing.reset_mock() friend_list.reset_mock() account.active = False account.active = False assert circuit.transport.method_calls == [('loseConnection',)] assert not hasattr(account, 'circuit') assert messaging.method_calls == [('disconnected',)] assert conferencing.method_calls == [('disconnected',)] assert friend_list.method_calls == [('disconnected',)] @mockified('pypsyc.client.model', ['resolve_hostname']) def test_active_no_password(self, resolve_hostname): no_password_evt = Mock() account = Account(None, SERVER1, USER1_NICK, '', False, False) account.no_password_evt += no_password_evt account.active = True assert account.active == False assert no_password_evt.call_args_list == [()] assert resolve_hostname.call_args_list == [] @mockified('pypsyc.client.model', ['resolve_hostname', 'connect']) def test_active_dns_error(self, resolve_hostname, connect): resolve_hostname.side_effect = EXCEPTION client = Mock() connection_error_evt = Mock() account = Account(client, SERVER1, USER1_NICK, PASSWORD, False, False) account.connection_error_evt += connection_error_evt account.active = True assert account.active == False assert client.accounts.updated_item.call_args_list == [((account,),)] assert connection_error_evt.call_args_list == [((EXCEPTION,),)] assert connect.call_args_list == [] @mockified('pypsyc.client.model', ['resolve_hostname', 'connect', 'LinkingProtocol']) def test_active_connect_error(self, resolve_hostname, connect, LinkingProtocol): client = Mock() resolve_hostname.return_value = (sentinel.ip, sentinel.port) connect.side_effect = EXCEPTION connection_error_evt = Mock() account = Account(client, SERVER1, USER1_NICK, PASSWORD, False, False) account.connection_error_evt += connection_error_evt account.active = True assert account.active == False assert client.accounts.updated_item.call_args_list == [((account,),)] assert connection_error_evt.call_args_list == [((EXCEPTION,),)] assert LinkingProtocol.call_args_list == [] @mockified('pypsyc.client.model', ['resolve_hostname', 'connect', 'LinkingProtocol']) def test_active_no_such_user(self, resolve_hostname, connect, LinkingProtocol): ERROR = UnknownTargetError() client = Mock() resolve_hostname.return_value = (sentinel.ip, sentinel.port) circuit = connect.return_value LinkingProtocol.return_value.link.side_effect = ERROR no_such_user_evt = Mock() account = Account(client, SERVER1, USER1_NICK, PASSWORD, False, False) account.no_such_user_evt += no_such_user_evt account.active = True assert account.active == False assert client.accounts.updated_item.call_args_list == [((account,),)] assert no_such_user_evt.call_args_list == [((ERROR,),)] assert circuit.psyc.uni != account.uni.chain(account.resource) @mockified('pypsyc.client.model', ['resolve_hostname', 'connect', 'LinkingProtocol']) def test_active_incorrect_password(self, resolve_hostname, connect, LinkingProtocol): ERROR = AuthenticationError() client = Mock() resolve_hostname.return_value = (sentinel.ip, sentinel.port) circuit = connect.return_value LinkingProtocol.return_value.link.side_effect = ERROR auth_error_evt = Mock() account = Account(client, SERVER1, USER1_NICK, 'wrong', False, False) account.auth_error_evt += auth_error_evt account.active = True assert account.active == False assert client.accounts.updated_item.call_args_list == [((account,),)] assert auth_error_evt.call_args_list == [((ERROR,),)] assert circuit.psyc.uni != account.uni.chain(account.resource) def test_connection_error(self): client = Mock() connection_error_evt = Mock() account = Account(client, SERVER1, USER1_NICK, PASSWORD, False, False) account.connection_error_evt += connection_error_evt account.connection_error(EXCEPTION) assert account.active == False assert client.accounts.updated_item.call_args_list == [((account,),)] assert connection_error_evt.call_args_list == [((EXCEPTION,),)] class TestClientCircuit(object): @mockified(Circuit, ['connectionMade']) def test_dump_out(self, connectionMade): LINE1 = ':_target\t' + USER2_UNI LINE2 = '|' cc = ClientCircuit() dump_evt = Mock() cc.dump_evt += dump_evt cc.transport = Mock() orig_write = cc.transport.write cc.connectionMade() assert connectionMade.call_args_list == [((cc,),)] cc.send({'_target': USER2_UNI}, '') assert dump_evt.call_args_list == [(('o', LINE1),), (('o', LINE2),)] assert orig_write.call_args_list == [((LINE1+'\n'+LINE2+'\n',),)] @mockified(Circuit, ['lineReceived']) def test_dump_in(self, lineReceived): lineReceived.return_value = None cc = ClientCircuit() dump_evt = Mock() cc.dump_evt += dump_evt cc.dataReceived('a\nb\n') assert dump_evt.call_args_list == [(('i', 'a'),), (('i', 'b'),)] assert lineReceived.call_args_list == [((cc, 'a'),), ((cc, 'b'),)] @mockified('pypsyc.client.model', ['PSYCObject']) def test_packet_received(self, PSYCObject): psyc = PSYCObject.return_value psyc.handle_packet.return_value = PSYCPacket() cc = ClientCircuit() assert PSYCObject.call_args_list == [((cc.send,),)] cc.packet_received({}, CONTENT) assert psyc.method_calls == [('handle_packet', ({}, CONTENT))] @mockified('pypsyc.client.model', ['PSYCObject']) def test_relay(self, PSYCObject): psyc = PSYCObject.return_value psyc.uni = RESOURCE1_UNI psyc.handle_packet.return_value = PSYCPacket() cc = ClientCircuit() header = inited_header(_source=USER1_UNI, _source_relay=USER2_UNI) cc.packet_received(header, CONTENT) assert header.source == USER2_UNI header2 = inited_header(_source=USER2_UNI, _source_relay=USER2_UNI) assert_raises(AssertionError, cc.packet_received, header2, CONTENT) @mockified('pypsyc.client.model', ['PSYCObject']) def test_persistent(self, PSYCObject): HEADER = {'_context': USER2_UNI} packet = PSYCPacket({'=': VARIABLES, '+': VARIABLES, '-': VARIABLES}) packet.header = inited_header(HEADER) psyc = PSYCObject.return_value psyc.handle_packet.return_value = packet cc = ClientCircuit() handler = Mock() state_set_key = handler.state_set_key = Mock() state_add_key = handler.state_add_key = Mock() state_remove_key = handler.state_remove_key = Mock() cc.add_pvar_handler(handler) cc.packet_received(HEADER, CONTENT) assert psyc.method_calls == [('handle_packet', (HEADER, CONTENT))] assert state_set_key.call_args_list == [((USER2_UNI, VARIABLES),)] assert state_add_key.call_args_list == [((USER2_UNI, VARIABLES),)] assert state_remove_key.call_args_list == [((USER2_UNI, VARIABLES),)] class TestClient(object): @mockified('pypsyc.client.model', ['Account']) def test_accounts(self, Account): account = Account.return_value error_ph = PlaceHolder() account1 = Mock() account1.server = SERVER1 account1.person = USER1_NICK account1.password = PASSWORD account1.save_password = True account1.active = False account2 = Mock() account2.server = SERVER2 account2.person = USER2_NICK account2.password = PASSWORD account2.save_password = False account2.active = True try: with NamedTemporaryFile(delete=False) as f: f.write('[]') client = Client(accounts_file=f.name) assert client.accounts == [] client.accounts = [account1, account2] client.save_accounts() client = Client(accounts_file=f.name) client.load_accounts() assert client.accounts == [Account.return_value] * 2 assert Account.call_args_list == [ ((client, SERVER1, USER1_NICK, PASSWORD, True, False),), ((client, SERVER2, USER2_NICK, '', False, True),)] finally: remove(f.name) client.connection_lost(account.circuit, sentinel.error) assert account.method_calls == [('connection_error', (error_ph,))] * 2 assert error_ph.obj.args[0] == str(sentinel.error) @mockified('pypsyc.client.model', ['reactor']) def test_quit(self, reactor): account = Mock() client = Client() client.accounts = [account] client.quit() assert account.active == False assert reactor.method_calls == [('stop',)]