How to use onOpen method in Playwright Python

Best Python code snippet using playwright-python

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

websocket_client_threaded.py

Source: websocket_client_threaded.py Github

copy
1import threading, Queue
2from amitu import websocket_client
3
4class _Writer(threading.Thread):
5    def __init__(self, ws):
6        super(_Writer, self).__init__()
7        self.daemon = True
8        self.ws = ws
9        self.queue = Queue.Queue()
10
11    def send(self, data):
12        self.queue.put(data)
13
14    def run(self):
15        while True:
16            self.ws._send(self.queue.get(block=True))
17
18class WebSocket(websocket_client.WebSocket):
19    """
20    Threaded WebSocket class
21
22    Use this class to use a threaded websocket. It reads data from server
23    on the current thread, and sends data on a separate daemon thread.
24
25    >>> def onmessage(message): print "onmessage", message
26    ...
27    >>> def onopen(): print "onopen"
28    ...
29    >>> def onclose(): print "onclose"
30    ...
31    >>> ws = WebSocket("ws://server.com:8080/path")
32    >>> ws.onopen(onopen)
33    >>> ws.onclose(onclose)
34    >>> ws.onmessage(onmessage)
35
36    >>> ws.run() # blocks
37    """
38    def __init__(self, *args, **kw):
39        websocket_client.WebSocket.__init__(self, *args, **kw)
40
41        self.writer = _Writer(self)
42
43        self.onopen_handlers = []
44        self.onclose_handlers = []
45        self.onmessage_handlers = []
46
47    def run(self):
48        self.writer.start()
49        websocket_client.WebSocket.run(self)
50
51    def send(self, data):
52        self.writer.send(data)
53
54    def _fire_onopen(self):
55        for cb in self.onopen_handlers: cb()
56    def _fire_onmessage(self, data):
57        for cb in self.onmessage_handlers: cb(data)
58    def _fire_onclose(self):
59        for cb in self.onclose_handlers: cb()
60
61    def onopen(self, cb): self.onopen_handlers.append(cb)
62    def onmessage(self, cb): self.onmessage_handlers.append(cb)
63    def onclose(self, cb): self.onclose_handlers.append(cb)
64
65class WebSocketThreaded(WebSocket, threading.Thread):
66    """
67    WebSocketThreaded
68
69    This is a thread that runs in the background, reading and writing both
70    in two different threads.
71
72    >>> def onmessage(message): print "onmessage", message
73    ...
74    >>> def onopen(): print "onopen"
75    ...
76    >>> def onclose(): print "onclose"
77    ...
78    >>> ws = WebSocketThreaded("ws://server.com:8080/path")
79    >>> ws.onopen(onopen)
80    >>> ws.onclose(onclose)
81    >>> ws.onmessage(onmessage)
82
83    >>> ws.start()
84    >>> ws.wait()
85    """
86    def __init__(self, *args, **kw):
87        WebSocket.__init__(self, *args, **kw)
88        threading.Thread.__init__(self)
89
90
Full Screen

Examples_test.py

Source: Examples_test.py Github

copy
1import pytest
2from unittest.mock import Mock
3
4import sys
5# insert at 1, 0 is the script path (or '' in REPL)
6if not '../event-notifier' in sys.path:
7    sys.path.insert(1, '../event-notifier')
8from EventNotifier.SubscriberManager import SubscriberManager
9
10
11
12class TestExamples:
13    def test_subscribe_to_all(self):
14        from EventNotifier import Notifier
15        class CallableFileWatchdog:
16            def __init__(self, pathToWatch):
17                self.pathToWatch = pathToWatch
18
19
20            def __call__(self, *args, **kwargs):
21                if len(args) > 0:
22                    print \
23                        (f"Event {args[0]} at path {self.pathToWatch} is called with following simple args: {[*args]} and with following keyword args: { {**kwargs} }")
24
25        callable_watchog = CallableFileWatchdog("some\\path\\here")
26        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])
27
28        notifier.subscribe_to_all(callable_watchog)
29
30        notifier.raise_event("onCreate", "onCreate", fileName="test_file.txt")
31        notifier.raise_event("onOpen", "onOpen", openMode="w+", fileName="test_file.txt")
32
33
34    def test_get_registered_events(self):
35        from EventNotifier import Notifier
36        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])
37        print(notifier.get_registered_events())
38
39
40    def test_remove_subscribers_by_event_name(self):
41        from EventNotifier import Notifier
42        class FileWatchDog():
43            def onOpen(self, fileName, openMode):
44                print(f"File {fileName} opened with {openMode} mode")
45
46            def onClose(self, fileName):
47                print(f"File {fileName} closed")
48
49
50        def onOpenStandaloneMethod(fileName, openMode):
51            print(f"StandaloneMethod: File {fileName} opened with {openMode} mode")
52
53        watchDog = FileWatchDog()
54
55        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])
56
57        notifier.subscribe("onOpen", watchDog.onOpen)
58        notifier.subscribe("onOpen", onOpenStandaloneMethod)
59        notifier.subscribe("onClose", watchDog.onClose)
60
61        print("\nAfter subscription:")
62        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")
63        notifier.raise_event("onClose", fileName="test_file.txt")
64
65        notifier.remove_subscribers_by_event_name("onOpen")
66
67        print("\nAfter removal of onOpen subscribers:")
68        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")
69        notifier.raise_event("onClose", fileName="test_file.txt")
70
71        notifier.remove_subscribers_by_event_name("onClose")
72
73        print("\nAfter removal of onClose subscribers:")
74        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")
75        notifier.raise_event("onClose", fileName="test_file.txt")
76
77
78
79    def test_remove_all_subscribers(self):
80        from EventNotifier import Notifier
81        class FileWatchDog():
82            def onOpen(self, fileName, openMode):
83                print(f"File {fileName} opened with {openMode} mode")
84
85            def onClose(self, fileName):
86                print(f"File {fileName} closed")
87
88
89        def onOpenStandaloneMethod(fileName, openMode):
90            print(f"StandaloneMethod: File {fileName} opened with {openMode} mode")
91
92        watchDog = FileWatchDog()
93
94        notifier = Notifier(["onCreate", "onOpen", "onModify", "onClose", "onDelete"])
95
96        notifier.subscribe("onOpen", watchDog.onOpen)
97        notifier.subscribe("onOpen", onOpenStandaloneMethod)
98        notifier.subscribe("onClose", watchDog.onClose)
99
100        print("\nAfter subscription:")
101        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")
102        notifier.raise_event("onClose", fileName="test_file.txt")
103
104        notifier.remove_all_subscribers()
105
106        print("\nAfter removal of all subscribers:")
107        notifier.raise_event("onOpen", openMode="w+", fileName="test_file.txt")
108        notifier.raise_event("onClose", fileName="test_file.txt")
Full Screen

pwave.py

Source: pwave.py Github

copy
1from websocket import *
2import websocket, httplib
3
4'''
5connect to the socketio server
61. perform the HTTP handshake
72. open a websocket connection
8'''
9def connect(self) :
10    conn  = httplib.HTTPConnection('localhost:8124')
11    conn.request('POST','/socket.io/1/')
12    resp  = conn.getresponse() 
13    hskey = resp.read().split(':')[0]
14
15    self._ws = websocket.WebSocket(
16                    'ws://localhost:8124/socket.io/1/websocket/'+hskey,
17                    onopen   = self._onopen,
18                    onmessage = self._onmessage)
19
20def my_msg_handler(msg):
21  print 'Got "%s"!' % msg
22
23def encode_for_socketio(message):
24    """
25    Encode 'message' string or dictionary to be able
26    to be transported via a Python WebSocket client to 
27    a Socket.IO server (which is capable of receiving 
28    WebSocket communications). This method taken from 
29    gevent-socketio.
30    """
31    MSG_FRAME = "~m~"
32    HEARTBEAT_FRAME = "~h~"
33    JSON_FRAME = "~j~"
34
35    if isinstance(message, basestring):
36            encoded_msg = message
37    elif isinstance(message, (object, dict)):
38            return encode_for_socketio(JSON_FRAME + json.dumps(message))
39    else:
40            raise ValueError("Can't encode message.")
41
42    return MSG_FRAME + str(len(encoded_msg)) + MSG_FRAME + encoded_msg
43
44# socket = websocket.WebSocket('ws://localhost:5000/', onmessage=my_msg_handler)
45
46# self._ws = websocket.WebSocket(
47#                     'ws://localhost:8124/socket.io/1/websocket/'+hskey,
48#                     onopen   = self._onopen,
49#                     onmessage = self._onmessage)
50
51conn  = httplib.HTTPConnection('localhost:5000')
52conn.request('POST','/socket.io/1/')
53resp  = conn.getresponse() 
54respdata = resp.read()
55hskey = respdata.split(':')[0]
56print respdata
57print hskey
58socket = websocket.WebSocket('ws://localhost:5000/socket.io/1/websocket/'+hskey, onmessage=my_msg_handler)
59# # socket.onopen = lambda: socket.send(encode_for_socketio('Hello world!'))
60senddata = '5:::{"message":"hi"}'
61socket.onopen = lambda: socket.send(senddata)
62
63# ws = websocket.create_connection('ws://localhost:5000/data')
64# msg = "Hello, world!"
65# msg = encode_for_socketio(msg)
66# ws.send(msg)
67
68
69try:
70  asyncore.loop()
71except KeyboardInterrupt:
72  socket.close()
73
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Playwright Python on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)