← Back to branch summary

~ubuntu-branches/ubuntu/hardy/pymsn/hardy-proposed

« back to all changes in this revision

Viewing changes to test.py

  • Committer: Bazaar Package Importer
  • Author(s): Laurent Bigonville, Sjoerd Simons, Laurent Bigonville, Jonny Lamb
  • Date: 2008-01-17 18:23:14 UTC
  • mfrom: (1.1.2 upstream)
  • Revision ID: james.westby@ubuntu.com-20080117182314-lwymmpnk2ut3rvr1
Tags: 0.3.1-0ubuntu1
[ Sjoerd Simons ]
* debian/rules: remove dh_python, it's no longer needed

[ Laurent Bigonville ]
* New upstream release (0.3.1)
* debian/control:
  - Add myself as an Uploaders
  - Add python:Provides for binary package
  - Add python-ctypes and python-crypto to build-deps/deps
* debian/rules: remove binary-install rule
* Add watch file
* remove pycompat file, not needed anymore
* Modify Maintainer value to match the DebianMaintainerField
  specification.

[ Jonny Lamb ]
* Added python-adns to build-deps/deps.
* Added python-pyopenssl to build-deps/deps.
* Updated copyright.
* Upped Standards-Version to 3.7.3.
* Added "XS-Dm-Upload-Allowed: yes" under the request of Sjoerd Simons.
* Added myself to Uploaders.
* Added Homepage to control.
* Added Vcs-Bzr to control.

Show diffs side-by-side

added added

removed removed

Lines of Context:
test.py
 
1
#!/usr/bin/env python
 
2
 
1
3
import pymsn
2
 
import pymsn.msnp2p
3
 
import asyncore, getpass
 
4
import pymsn.event
 
5
 
 
6
 
4
7
import logging
5
 
import time
6
8
import gobject
7
 
import signal
8
 
import urllib
9
 
import re
10
 
import sys
11
9
 
12
10
logging.basicConfig(level=logging.DEBUG)
13
11
 
14
12
finished = False
15
 
def getproxies():
 
13
 
 
14
def get_proxies():
 
15
    import urllib
16
16
    proxies = urllib.getproxies()
17
17
    result = {}
 
18
    if 'https' not in proxies and \
 
19
            'http' in proxies:
 
20
        url = proxies['http'].replace("http://", "https://")
 
21
        result['https'] = pymsn.Proxy(url)
18
22
    for type, url in proxies.items():
19
 
        if type == 'no':
20
 
            continue
21
 
        result[type] = pymsn.network.ProxyInfos.build_from_string(url)
 
23
        if type == 'no': continue
 
24
        if type == 'https' and url.startswith('http://'):
 
25
            url = url.replace('http://', 'https://', 1)
 
26
        result[type] = pymsn.Proxy(url)
22
27
    return result
23
28
 
24
 
class NS(pymsn.Client):
25
 
    def __init__(self, server, account, mainloop, http_method=False):
26
 
        if http_method:
27
 
            transport = pymsn.transport.HTTPPollConnection
28
 
        else:
29
 
            transport = pymsn.transport.DirectConnection
30
 
        pymsn.Client.__init__(self, server, account, proxies = getproxies(),
31
 
                transport_class = transport)
32
 
        gobject.idle_add(self.connect_cb)
33
 
        self.mainloop = mainloop
34
 
 
35
 
    def connect_cb(self):
36
 
        self.login()
37
 
        return False
38
 
 
39
 
    def on_connect_failure(self, proto):
40
 
        print "Connect failed"
41
 
        self.mainloop.quit()
42
 
 
43
 
    def on_login_failure(self, proto):
44
 
        print "Login failed"
45
 
        self.mainloop.quit()
46
 
 
47
 
    def on_login_success(self, proto):
48
 
        self.dp_fetched = False
49
 
        #gobject.timeout_add(5000, self.__find_contact)
50
 
        gobject.timeout_add(5000, self.__publish_avatar)
51
 
 
52
 
    def on_switchboard_invitation(self, proto, server, key,
53
 
            session, passport, friendly_name):
54
 
        inv = [self.connection.get_contact_by_passport(passport),]
55
 
        self.switchboard = pymsn.Conversation(self, invitee=inv,
56
 
                                       server=server, key=key,
57
 
                                       session=session)
58
 
 
59
 
    def __publish_avatar(self):
60
 
        data = file("pymsn.png").read()
61
 
        self.profile.display_picture = data
62
 
        
63
 
    def __find_contact(self):
64
 
        for contact in self._protocol._contacts.values():
65
 
            passport = contact.get_property("passport")
66
 
            presence = contact.get_property("presence")
67
 
            if presence != pymsn.PresenceStatus.OFFLINE:
68
 
                print "Contact %s is online" % contact.get_property("passport")
69
 
                
70
 
                gobject.idle_add(self.__fetch_dp, contact)
71
 
                return False
72
 
 
 
29
 
 
30
class ClientEvents(pymsn.event.ClientEventInterface):
 
31
    def on_client_state_changed(self, state):
 
32
        if state == pymsn.event.ClientState.CLOSED:
 
33
            self._client.quit()
 
34
        elif state == pymsn.event.ClientState.OPEN:
 
35
            self._client.profile.display_name = "Kimbix"
 
36
            self._client.profile.presence = pymsn.Presence.ONLINE
 
37
            self._client.profile.current_media = ("I listen to", "Nothing")
 
38
            for contact in self._client.address_book.contacts:
 
39
                print contact
 
40
            #self._client.profile.personal_message = "Testing pymsn, and freeing the pandas!"
 
41
            gobject.timeout_add(5000, self._client.start_conversation)
 
42
 
 
43
    def on_client_error(self, error_type, error):
 
44
        print "ERROR :", error_type, " ->", error
 
45
 
 
46
class AnnoyingConversation(pymsn.event.ConversationEventInterface):
 
47
    def on_conversation_user_joined(self, contact):
 
48
        gobject.timeout_add(5000, self.annoy_user)
 
49
 
 
50
    def annoy_user(self):
 
51
        msg = "Let's free the pandas ! (testing pymsn)"
 
52
        formatting = pymsn.TextFormat("Comic Sans MS", 
 
53
                         pymsn.TextFormat.UNDERLINE | pymsn.TextFormat.BOLD,
 
54
                         'FF0000')
 
55
        self._client.send_text_message(pymsn.ConversationMessage(msg, formatting))
 
56
#         self._client.send_nudge()
 
57
#         self._client.send_typing_notification()
73
58
        return True
74
59
 
75
 
    def __fetch_dp(self, contact):
76
 
        if self.dp_fetched:
 
60
    def on_conversation_user_typing(self, contact):
 
61
        pass
 
62
 
 
63
    def on_conversation_message_received(self, sender, message):
 
64
        pass
 
65
 
 
66
    def on_conversation_error(self, error_type, error):
 
67
        print "ERROR :", error_type, " ->", error
 
68
 
 
69
class Client(pymsn.Client):
 
70
    def __init__(self, account, quit, http_mode=False):
 
71
        server = ('messenger.hotmail.com', 1863)
 
72
        self.quit = quit
 
73
        self.account = account
 
74
        if http_mode:
 
75
            from pymsn.transport import HTTPPollConnection
 
76
            pymsn.Client.__init__(self, server, get_proxies(), HTTPPollConnection)
 
77
        else:
 
78
            pymsn.Client.__init__(self, server, proxies = get_proxies())
 
79
        self._event_handler = ClientEvents(self)
 
80
        gobject.idle_add(self._connect)
 
81
 
 
82
    def _connect(self):
 
83
        self.login(*self.account)
 
84
        return False
 
85
 
 
86
    def start_conversation(self):
 
87
        contacts = self.address_book.contacts.\
 
88
                search_by_presence(pymsn.Presence.ONLINE)
 
89
        for c in self.address_book.contacts:
 
90
            if c.account == "@hotmail.com":
 
91
                print "Fetching space of : %s with cid %s\n" % (c.display_name, c.cid)
 
92
                self.spaces_service.get_contact_card(c)
 
93
 
 
94
        if len(contacts) == 0:
 
95
            print "No online contacts"
 
96
            return True
 
97
        else:
 
98
            for contact in contacts:
 
99
                if contact.account == "johann.prieur@gmail.com":
 
100
                    print "Inviting %s for a conversation" % contact.display_name
 
101
                    self.conv = pymsn.Conversation(self, [contact])
 
102
                    self._convo_events = AnnoyingConversation(self.conv)
77
103
            return False
78
 
        
79
 
        self.dp_fetched = True
80
 
        
81
 
        dpc = pymsn.msnp2p.DisplayPictureCall(self)
82
 
        dpc.request(contact, self.__on_dp_request_done)
83
 
        
84
 
        return False
85
 
    
86
 
    def __on_dp_request_done(self, result):
87
 
        if result == None:
88
 
            print "_on_dp_request_done: Failed to fetch DP"
89
 
            return
90
 
        
91
 
        print "_on_dp_request_done: Got DP! %d bytes of data" % len(result)
92
 
 
93
104
 
94
105
def main():
 
106
    import sys
 
107
    import getpass
 
108
    import signal
 
109
 
 
110
    if "--http" in sys.argv:
 
111
        http_mode = True
 
112
        sys.argv.remove('--http')
 
113
    else:
 
114
        http_mode = False
 
115
 
95
116
    if len(sys.argv) < 2:
96
117
        account = raw_input('Account: ')
97
118
    else:
98
119
        account = sys.argv[1]
99
 
    
 
120
 
100
121
    if len(sys.argv) < 3:
101
122
        passwd = getpass.getpass('Password: ')
102
123
    else:
103
124
        passwd = sys.argv[2]
104
 
    
 
125
 
105
126
    mainloop = gobject.MainLoop(is_running=True)
106
 
    http_method = '--http' in sys.argv
107
 
    n = NS(('messenger.hotmail.com', 1863), (account, passwd), mainloop, http_method)
108
 
 
109
 
 
110
 
    def quit_cb():
 
127
 
 
128
    def quit():
111
129
        mainloop.quit()
112
130
 
113
131
    def sigterm_cb():
114
 
        gobject.idle_add(quit_cb)
 
132
        gobject.idle_add(quit)
115
133
 
116
134
    signal.signal(signal.SIGTERM, sigterm_cb)
117
135
 
 
136
    n = Client((account, passwd), quit, http_mode)
 
137
 
118
138
    while mainloop.is_running():
119
139
        try:
120
140
            mainloop.run()
121
141
        except KeyboardInterrupt:
122
 
            quit_cb()
 
142
            quit()
123
143
 
124
144
if __name__ == '__main__':
125
145
    main()