source: geniuswebcore/test/geniusweb/profileconnection/WebSocketProfileConnectorTest.py@ 88

Last change on this file since 88 was 88, checked in by Bart Vastenhouw, 3 years ago

Added python SimpleRunner GUI

File size: 2.7 KB
Line 
1import logging
2from pathlib import Path # type:ignore
3import threading
4from time import sleep
5from typing import Optional
6import unittest
7from unittest.mock import Mock
8
9from tudelft_utilities_logging.Reporter import Reporter
10from uri.uri import URI
11
12from geniusweb.profileconnection.FileProfileConnector import FileProfileConnector
13from geniusweb.profileconnection.Session import Session
14from geniusweb.profileconnection.WebSocketContainer import WebSocketContainer, \
15 DefaultWebSocketContainer
16from geniusweb.profileconnection.WebsocketProfileConnector import WebsocketProfileConnector
17from test.geniusweb.profileconnection.ProfileServerStub import ProfileServerStub
18
19
20class DummyReporter(Reporter):
21 def log(self, level:int, msg:str, thrown:BaseException=None):
22 print(msg)
23
24
25
26class WebSocketProfileConnectorTest(unittest.TestCase):
27
28 profiletext = Path("test/resources/japantrip1.json").read_text("utf-8")
29 reporter=DummyReporter()
30
31 def t1estConnect(self):
32 session=Mock(Session)
33 # this session does nothing, we manipulate it ourselves.
34
35 #mock websocket
36 wsContainer = Mock(WebSocketContainer)
37 wsContainer.connectToServer.return_value=session
38
39 profint= WebsocketProfileConnector("ws://blabla", self.reporter, wsContainer )
40
41 def pumpevents():
42 profint.onOpen(session)
43 sleep(1)
44 profint.onMessage(self.profiletext)
45 print ("Message was sent")
46
47 threading.Thread(target=pumpevents).start()
48 profint.close()
49
50
51 self.assertEqual("japantrip1", profint.getProfile().getName())
52 print("Received profile succesfully!")
53
54 def t1estConnectTimeout(self):
55 session=Mock(Session)
56 #mock websocket
57 wsContainer = Mock(WebSocketContainer)
58 wsContainer.connectToServer.return_value=session
59
60 profint= WebsocketProfileConnector("ws://blabla", self.reporter, wsContainer )
61
62 # since we don't pump events, this should timeout.
63 self.assertRaises(IOError, lambda:profint.getProfile())
64 profint.close()
65
66
67 def testWithRealProfilesServer(self):
68 '''
69 TO RUN THIS, ENSURE YOU HAVE A RUNNING PROFILESSERVER ON LOCALHOST
70 '''
71 server=ProfileServerStub()
72 server.start()
73 try:
74 profint= WebsocketProfileConnector(URI("ws://localhost:8080/profilesserver/websocket/get/party/party1"), self.reporter, DefaultWebSocketContainer() )
75 sleep(1)
76 profile=profint.getProfile()
77 self.assertTrue(profile)
78 profint.close()
79 finally:
80 server.stop()
81
Note: See TracBrowser for help on using the repository browser.