test.py (7777B)
1 import json 2 import sys 3 import requests 4 import asyncio 5 import websockets 6 7 async def main(): 8 ############# 9 # INIT CODE # 10 ############# 11 12 # same hash for everyone for succinctness 13 HASH = "6c6e2b0cfda80007e693d52b5956083ea68770e1310d0ed02d195cb14113b284" 14 15 # login as quantum for init step 16 r = requests.post( 17 f'http://localhost:3000/api/user?name=quantum&email=test@example.com', 18 data=HASH, 19 headers={"Content-Type": "text/plain"} 20 ) 21 quantum = r.json()["id"] 22 r = requests.post( 23 f'http://localhost:3000/api/login?id={quantum}', 24 data=HASH, 25 headers={"Content-Type": "text/plain"} 26 ) 27 tok = r.text 28 29 # init other users 30 r = requests.post( 31 f'http://localhost:3000/api/user?name=jemoka&email=test@example.com', 32 data=HASH, 33 headers={"Content-Type": "text/plain"} 34 ) 35 jemoka = r.json()["id"] 36 r = requests.post( 37 f'http://localhost:3000/api/user?name=exr0n&email=test@example.com', 38 data=HASH, 39 headers={"Content-Type": "text/plain"} 40 ) 41 exr0n = r.json()["id"] 42 r = requests.post( 43 f'http://localhost:3000/api/user?name=enquirer&email=test@example.com', 44 data=HASH, 45 headers={"Content-Type": "text/plain"} 46 ) 47 enquirer = r.json()["id"] 48 r = requests.post( 49 f'http://localhost:3000/api/user?name=zbuster&email=test@example.com', 50 data=HASH, 51 headers={"Content-Type": "text/plain"} 52 ) 53 zbuster = r.json()["id"] 54 55 # create group with multiple members 56 r = requests.post( 57 f'http://localhost:3000/api/group?name=testing', 58 headers={"Authorization": tok} 59 ) 60 testing = r.json()["id"] 61 r = requests.put( 62 f'http://localhost:3000/api/group/members?gid={testing}&uid={zbuster}', 63 headers={"Authorization": tok} 64 ) 65 r = requests.put( 66 f'http://localhost:3000/api/group/members?gid={testing}&uid={exr0n}', 67 headers={"Authorization": tok} 68 ) 69 r = requests.put( 70 f'http://localhost:3000/api/group/members?gid={testing}&uid={jemoka}', 71 headers={"Authorization": tok} 72 ) 73 # create another two groups 74 r = requests.post( 75 f'http://localhost:3000/api/group?name=whoo', 76 headers={"Authorization": tok} 77 ) 78 whoo = r.json()["id"] 79 r = requests.put( 80 f'http://localhost:3000/api/group/members?gid={whoo}&uid={zbuster}', 81 headers={"Authorization": tok} 82 ) 83 r = requests.put( 84 f'http://localhost:3000/api/group/members?gid={whoo}&uid={enquirer}', 85 headers={"Authorization": tok} 86 ) 87 88 r = requests.post( 89 f'http://localhost:3000/api/group?name=whee', 90 headers={"Authorization": tok} 91 ) 92 whee = r.json()["id"] 93 r = requests.put( 94 f'http://localhost:3000/api/group/members?gid={whee}&uid={enquirer}', 95 headers={"Authorization": tok} 96 ) 97 r = requests.put( 98 f'http://localhost:3000/api/group/members?gid={whee}&uid={jemoka}', 99 headers={"Authorization": tok} 100 ) 101 102 ############################## 103 # INTEGRATION TEST 1: GROUPS # 104 ############################## 105 106 # log two users in 107 r = requests.post( 108 f'http://localhost:3000/api/login?id={zbuster}', 109 data=HASH, 110 headers={"Content-Type": "text/plain"} 111 ) 112 z_tok = r.text 113 r = requests.post( 114 f'http://localhost:3000/api/login?id={jemoka}', 115 data=HASH, 116 headers={"Content-Type": "text/plain"} 117 ) 118 j_tok = r.text 119 120 # init websockets for both 121 z_sock = await websockets.connect("ws://localhost:3001/") 122 await z_sock.send(f'{{"hash": "{HASH}", "id": {zbuster}}}') 123 j_sock = await websockets.connect("ws://localhost:3001/") 124 await j_sock.send(f'{{"hash": "{HASH}", "id": {jemoka}}}') 125 126 # check zbuster's groups 127 r = requests.get( 128 f'http://localhost:3000/api/user/groups', 129 headers={"Authorization": z_tok} 130 ) 131 assert(len(r.json()) == 2) 132 group_ids = list(map(lambda x: x["id"], r.json())) 133 assert(testing in group_ids) 134 assert(whoo in group_ids) 135 136 # check jemoka's groups 137 r = requests.get( 138 f'http://localhost:3000/api/user/groups', 139 headers={"Authorization": j_tok} 140 ) 141 assert(len(r.json()) == 2) 142 group_ids = list(map(lambda x: x["id"], r.json())) 143 assert(testing in group_ids) 144 assert(whee in group_ids) 145 146 # get + check main channel of testing 147 r = requests.get( 148 f'http://localhost:3000/api/group?id={testing}', 149 headers={"Authorization": z_tok} 150 ) 151 testing_main = r.json()["channels"][0] 152 assert(r.json()["name"] == "testing") 153 assert(r.json()["is_dm"] == False) 154 assert(jemoka in r.json()["members"]) 155 assert(quantum in r.json()["members"]) 156 assert(exr0n in r.json()["members"]) 157 assert(zbuster in r.json()["members"]) 158 159 # test basic messaging 160 await j_sock.send(f'{{"content": "chickens", "channel": {testing_main}}}') 161 z_msg = json.loads(await z_sock.recv()) 162 assert(z_msg["author"] == jemoka) 163 assert(z_msg["content"] == "chickens") 164 assert(z_msg["channel"] == testing_main) 165 166 await z_sock.send(f'{{"content": "what?", "channel": {testing_main}}}') 167 j_msg = json.loads(await z_sock.recv()) 168 assert(j_msg["author"] == zbuster) 169 assert(j_msg["content"] == "what?") 170 assert(j_msg["channel"] == testing_main) 171 172 # log another user in 173 r = requests.post( 174 f'http://localhost:3000/api/login?id={enquirer}', 175 data=HASH, 176 headers={"Content-Type": "text/plain"} 177 ) 178 h_tok = r.text 179 180 # attempt invalid action 181 r = requests.delete( 182 f'http://localhost:3000/api/group?id={testing}', 183 headers={"Authorization": h_tok} 184 ) 185 assert(r.status_code == 401) 186 print("Test 1 done!") 187 188 ########################## 189 # INTEGRATION TEST 2: DM # 190 ########################## 191 192 # log yet another user in 193 r = requests.post( 194 f'http://localhost:3000/api/login?id={exr0n}', 195 data=HASH, 196 headers={"Content-Type": "text/plain"} 197 ) 198 e_tok = r.text 199 200 # init websockets for both exr0n and enquirer 201 e_sock = await websockets.connect("ws://localhost:3001/") 202 await e_sock.send(f'{{"hash": "{HASH}", "id": {exr0n}}}') 203 h_sock = await websockets.connect("ws://localhost:3001/") 204 await h_sock.send(f'{{"hash": "{HASH}", "id": {enquirer}}}') 205 206 # create dm 207 r = requests.post( 208 f'http://localhost:3000/api/dm?uid={exr0n}', 209 headers={"Authorization": h_tok} 210 ) 211 dm = r.json()["id"] 212 assert(r.json()["name"] == "") 213 assert(r.json()["is_dm"] == True) 214 assert(exr0n in r.json()["members"]) 215 assert(enquirer in r.json()["members"]) 216 assert(len(r.json()["admin"]) == 0) 217 assert(r.json()["owner"] == enquirer) 218 assert(len(r.json()["channels"]) == 1) 219 dm_main = r.json()["channels"][0] 220 221 # check that exr0n can see it 222 r = requests.get( 223 f'http://localhost:3000/api/user/dms', 224 headers={"Authorization": e_tok} 225 ) 226 assert(len(r.json()) == 1) 227 assert(dm == r.json()[0]["id"]) 228 229 # and enquirer for that matter 230 r = requests.get( 231 f'http://localhost:3000/api/user/dms', 232 headers={"Authorization": h_tok} 233 ) 234 assert(len(r.json()) == 1) 235 assert(dm == r.json()[0]["id"]) 236 # test basic messaging 237 await h_sock.send(f'{{"content": "videogames?", "channel": {dm_main}}}') 238 e_msg = json.loads(await e_sock.recv()) 239 assert(e_msg["author"] == enquirer) 240 assert(e_msg["content"] == "videogames?") 241 assert(e_msg["channel"] == dm_main) 242 print("Test 2 done!") 243 244 245 loop = asyncio.get_event_loop() 246 loop.run_until_complete(asyncio.wait_for(main(), 5)) 247