source: .tests/btlib.py @ a18cf1b

Last change on this file since a18cf1b was a18cf1b, checked in by / <>, at 2021-03-29T12:18:57Z

prosody restart fix
  • Property mode set to 100644
File size: 8.5 KB
Line 
1import socket
2import sys
3import time
4import select
5
6SEPARATOR = "="*60
7SMOLPARATOR = "-"*60
8
9class IrcClient:
10    def __init__(self, nick, pwd):
11        self.nick = nick
12        self.pwd = pwd
13        self.log = ''
14        self.tmplog = ''
15        self.sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
16
17    def send_raw(self, msg, loud = False, log = True):
18        self.receive()
19        if loud:
20            print('FROM '+ self.nick + '|| ' + msg)
21        if log:
22            self.log += msg+'\r\n'
23            self.tmplog += msg+'\r\n'
24        self.sck.send((msg+'\r\n').encode())
25
26    def send_priv_msg(self, recip, msg, loud = False):
27        self.send_raw('PRIVMSG '+recip+' :'+msg, loud)
28
29    def connect(self):
30        connected = False
31        for x in range(5):
32            try:
33                self.sck.connect(('127.0.0.1', 6667))
34                connected = True
35                break
36            except:
37                time.sleep(1)
38
39        if not connected:
40            print("IRC connection failed for " + self.nick)
41            sys.exit(1)
42       
43        print("IRC connection established for " + self.nick)
44
45        self.send_raw('USER ' + (self.nick + " ")*3)
46        self.send_raw('NICK ' + self.nick)
47        self.send_raw('JOIN &bitlbee')
48
49    def jabber_login(self):
50        self.send_priv_msg("&bitlbee", "account add jabber "+self.nick+"@localhost "+self.pwd)
51        time.sleep(0.3)
52        self.send_priv_msg("&bitlbee", "account on")
53        time.sleep(1)
54        self.receive()
55        if self.log.find('Logged in') == -1:
56            print("Jabber login failed for " + self.nick)
57            sys.exit(1)
58        else:
59            print("Jabber login successful for " + self.nick)
60
61    def receive(self):
62        text = ''
63        while True:
64            readable, _, _ = select.select([self.sck], [], [], 5)
65            if self.sck in readable:
66                text += self.sck.recv(2040).decode()
67                for line in text.split('\n'):
68                    if line.find('PING') != -1:
69                        self.send_raw('PONG ' + line.split()[1])
70            else:
71                break
72        self.log += text
73        self.tmplog += text
74        return text
75
76    def add_jabber_buddy(self, nick):
77        self.send_priv_msg("&bitlbee", "add 0 " + nick+"@localhost")
78   
79    def block_jabber_buddy(self, nick):
80        self.send_priv_msg("&bitlbee", "block " + nick)
81
82    def unblock_jabber_buddy(self, nick):
83        self.send_priv_msg("&bitlbee", "allow " + nick)
84
85    def rename_jabber_buddy(self, oldnick, newnick):
86        self.send_priv_msg("&bitlbee", "rename " + oldnick + " " + newnick)
87       
88def msg_comes_thru(sender, receiver, message):
89    sender.send_priv_msg(receiver.nick, message)
90    received = receiver.receive().find(message) != -1
91    return received
92
93
94def perform_test(test_function):
95    clis = []
96    clis += [IrcClient('test1', 'asd')]
97    clis += [IrcClient('test2', 'asd')]
98    for cli in clis:
99        cli.connect()
100
101    fail = not test_function(clis)
102
103    for cli in clis:
104        cli.receive()
105        print(SEPARATOR)
106        print("Test Log "+ cli.nick+":")
107        print(cli.tmplog)
108    print(SEPARATOR)
109   
110
111    if fail:
112        sys.exit(1)
113
114
115def yes_test(clis):
116    ret = False
117    clis[0].send_priv_msg("&bitlbee", "yes")
118    clis[0].receive()
119    for x, fun in enumerate(FUN):
120        if (clis[0].log.find(fun) != -1):
121            ret = True
122            if x:
123                print("The RNG gods smile upon us")
124            break
125    return ret
126
127def add_buddy_test(clis):
128    clis[0].add_jabber_buddy(clis[1].nick)
129    clis[1].send_priv_msg("&bitlbee", "yes")
130
131    clis[1].add_jabber_buddy(clis[0].nick)
132    clis[0].send_priv_msg("&bitlbee", "yes")
133
134    clis[0].send_priv_msg("&bitlbee", "blist")
135    junk = clis[0].receive()
136    ret = junk.find(clis[1].nick) != -1
137    ret = ret & (junk.find("1 available") != -1)
138    '''
139    clis[0].send_priv_msg("&bitlbee", "remove " +clis[1].nick)
140    clis[0].send_priv_msg("&bitlbee", "blist")
141    ret = ret & (clis[0].receive().find(clis[1].nick) == -1)
142
143    clis[0].add_jabber_buddy(clis[1].nick)
144    clis[1].send_priv_msg("&bitlbee", "yes")
145    time.sleep(1)
146    clis[0].send_priv_msg("&bitlbee", "blist")
147    junk = clis[0].receive()
148    ret = ret & (junk.find("1 available") != -1)
149    ret = ret & (junk.find(clis[1].nick) != -1)
150    '''
151    return ret
152
153def message_test(clis):
154    ret = msg_comes_thru(clis[0], clis[1], 'ohai <3')
155    ret = ret & msg_comes_thru(clis[1], clis[0], 'uwu *pounces*')
156    return ret
157
158def block_test(clis):
159    clis[0].block_jabber_buddy(clis[1].nick)
160    ret = not msg_comes_thru(clis[1], clis[0], 'm-meow?')
161    clis[0].unblock_jabber_buddy(clis[1].nick)
162    ret = ret & msg_comes_thru(clis[1], clis[0], '*purrs*')
163    return ret
164
165def rename_test(clis):
166    newname = "xXx_pup_LINKENPARK4EVA"
167    message = "rawr meanmz i luv<3 u in dinosaur"
168
169    clis[0].rename_jabber_buddy(clis[1].nick, newname)
170    clis[0].send_priv_msg(newname, message)
171    ret = clis[1].receive().find(message) != -1
172
173    clis[0].rename_jabber_buddy("-del", newname)
174    ret = ret & msg_comes_thru(clis[0], clis[1], "rawr")
175    return ret
176
177def status_test(clis):
178    status = "get out of my room mom"
179    clis[1].send_priv_msg("&bitlbee", "set status '"+status+"'")
180    clis[0].send_priv_msg("&bitlbee", "info "+clis[1].nick)
181    ret = (clis[0].receive().find("jabber - Status message: "+status) != -1)
182
183    clis[1].send_priv_msg("&bitlbee", "set")
184    ret = ret & (clis[1].receive().find(status) != -1)
185
186    clis[1].send_priv_msg("&bitlbee", "set -del status")
187    clis[0].send_priv_msg("&bitlbee", "info "+clis[1].nick)
188    ret = ret & (clis[0].receive().find("jabber - Status message: (none)") != -1)
189    return ret
190
191def offline_test(clis):
192    clis[0].send_priv_msg("&bitlbee", "account off")
193
194    junk = clis[0].receive()
195    ret = (junk.find(clis[1].nick) != -1)
196    ret = ret & (junk.find("QUIT") != -1)
197
198    junk = clis[1].receive()
199    ret = ret & (junk.find(clis[0].nick) != -1)
200    ret = ret & (junk.find("QUIT") != -1)
201
202    clis[0].send_priv_msg(clis[1].nick, "i'm not ur mom")
203    ret = ret & (clis[0].receive().find("No such nick/channel") != -1)
204
205    clis[0].send_priv_msg("&bitlbee", "account on")
206
207    junk = clis[0].receive()
208    ret = ret & (junk.find(clis[1].nick) != -1)
209    ret = ret & (junk.find("JOIN") != -1)
210
211    junk = clis[1].receive()
212    ret = ret & (junk.find(clis[0].nick) != -1)
213    ret = ret & (junk.find("JOIN") != -1)
214
215    return ret
216
217def default_target_test(clis):
218    clis[0].send_priv_msg("&bitlbee", "set default_target last")
219    clis[0].send_priv_msg("&bitlbee", "test2: ur mah default now")
220   
221    ret = (clis[1].receive().find("ur mah default now") != -1)
222
223    clis[0].send_priv_msg("&bitlbee", "mrow")
224    ret = ret & (clis[1].receive().find("mrow") != -1)
225
226    clis[0].send_priv_msg("root", "set default_target root")
227    junk = clis[0].receive()
228    ret = ret & (junk.find("default_target") != -1)
229    ret = ret & (junk.find("root") != -1)
230
231    clis[0].send_priv_msg("&bitlbee", "yes")
232    ret = ret & (clis[1].receive().find("yes") == -1)
233    return ret
234
235def help_test(clis):
236    clis[0].send_priv_msg("&bitlbee", "help")
237    ret = (clis[0].receive().find("identify_methods") != -1)
238    clis[0].send_priv_msg("&bitlbee", "help commands")
239    ret = ret & (clis[0].receive().find("qlist") != -1)
240    return ret
241   
242def run_tests(failed):
243    clis = []
244    clis += [IrcClient('test1', 'asd')]
245    clis += [IrcClient('test2', 'asd')]
246    for cli in clis:
247        cli.connect()
248
249    perform_test(failed, clis, yes_test, "Yes", debug_file)
250
251    print("")
252    for cli in clis:
253        cli.jabber_login()
254
255    perform_test(failed, clis, add_buddy_test, "Add/remove buddy")
256    perform_test(failed, clis, message_test, "Send message")
257    #perform_test(failed, clis, block_test, "Block user", debug_file)
258    perform_test(failed, clis, rename_test, "Rename user")
259    perform_test(failed, clis, status_test, "Change status")
260    perform_test(failed, clis, offline_test, "Go offline")
261    perform_test(failed, clis, default_target_test, "Change default target")
262    perform_test(failed, clis, help_test, "Ask for help")
263
264    if failed or SHOWLOG:
265        print("")
266        for cli in clis:
267            print(SMOLPARATOR)
268            print("Log "+ cli.nick+":")
269            print(cli.log)
270        print(SMOLPARATOR)
271
272    if failed:
273        print("\n" + SEPARATOR + "\nSome test have failed:")
274        for fail in failed:
275            print(fail)
276    else:
277        print("\n" + SEPARATOR + "\nAll tests have passed")
278   
279if __name__ == "__main__":
280    failed = []
281    run_tests(failed)
282    if failed:
283        sys.exit(1)
Note: See TracBrowser for help on using the repository browser.