KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use background_app method in robotframework-appiumlibrary
Best Python code snippet using robotframework-appiumlibrary_python
test_websockets.py
Source:test_websockets.py 
1import json2import logging3import psycopg24import tornado.log as tornado_logging5import tornado.testing6import tornado.web7import tornado.ioloop8import tornado.websocket9from tornado.httpclient import HTTPRequest10from app import GandalfConfiguration11from app import PostgresAdapter12from app import WEBSOCKET13from app import make_app14tornado_logging.access_log.setLevel(logging.DEBUG)15tornado_logging.app_log.setLevel(logging.DEBUG)16tornado_logging.gen_log.setLevel(logging.DEBUG)17class WebsocketTest(tornado.testing.AsyncHTTPTestCase):18 def get_app(self):19 conn = psycopg2.connect(host="localhost", user="postgres")20 cursor = conn.cursor()21 cursor.execute("DROP SCHEMA IF EXISTS gandalf CASCADE")22 conn.commit()23 app = make_app(GandalfConfiguration('localhost:8889', PostgresAdapter(), 'localhost', mode=WEBSOCKET))24 app.listen(8888)25 return app26 @tornado.testing.gen_test27 async def test_websockets_times_out_without_auth(self):28 testself = self29 user_id = None30 class EchoWebSocket(tornado.websocket.WebSocketHandler):31 def on_message(self, message):32 testself.assertEqual(message, 'USER_ID: %s' % user_id)33 self.write_message("Boom")34 background_app = tornado.web.Application([35 (r".*", EchoWebSocket),36 ])37 background_app.listen(8889)38 response = await self.http_client.fetch("http://localhost:8888/auth/users", method="POST",39 body="username=test&password=test")40 self.assertEqual(response.code, 201)41 user_id = response.headers['USER_ID']42 response = await self.http_client.fetch("http://localhost:8888/auth/login", method="POST",43 body="username=test&password=test")44 self.assertEqual(response.code, 200)45 token = json.loads(response.body.decode())["access_token"]46 request = HTTPRequest(url="ws://localhost:8888/")47 conn = await tornado.websocket.websocket_connect(request)48 self.assertEqual(await conn.read_message(), None)49 self.assertEqual(conn.close_code, 401)50 request = HTTPRequest(url="ws://localhost:8888/")51 conn = await tornado.websocket.websocket_connect(request)52 conn.write_message("Authorization: Bearer {}".format(token))53 self.assertEqual(await conn.read_message(), 'Boom')54 conn.close()55 self.assertIsNone(await conn.read_message())56 @tornado.testing.gen_test57 async def test_websockets_requires_auth(self):58 testself = self59 user_id = None60 class EchoWebSocket(tornado.websocket.WebSocketHandler):61 def on_message(self, message):62 testself.assertEqual(message, 'USER_ID: %s' % user_id)63 self.write_message("Boom")64 background_app = tornado.web.Application([65 (r".*", EchoWebSocket),66 ])67 background_app.listen(8889)68 response = await self.http_client.fetch("http://localhost:8888/auth/users", method="POST",69 body="username=test&password=test")70 self.assertEqual(response.code, 201)71 user_id = response.headers['USER_ID']72 response = await self.http_client.fetch("http://localhost:8888/auth/login", method="POST",73 body="username=test&password=test")74 self.assertEqual(response.code, 200)75 token = json.loads(response.body.decode())["access_token"]76 request = HTTPRequest(url="ws://localhost:8888/")77 conn = await tornado.websocket.websocket_connect(request)78 conn.write_message("Authorization: Bearer {}".format(token))79 self.assertEqual(await conn.read_message(), 'Boom')80 conn.close()81 self.assertIsNone(await conn.read_message())82 @tornado.testing.gen_test83 async def test_websockets_closes_on_bogus_auth(self):84 testself = self85 user_id = None86 class EchoWebSocket(tornado.websocket.WebSocketHandler):87 def on_message(self, message):88 testself.assertEqual(message, 'USER_ID: %s' % user_id)89 self.write_message("Boom")90 background_app = tornado.web.Application([91 (r".*", EchoWebSocket),92 ])93 background_app.listen(8889)94 request = HTTPRequest(url="ws://localhost:8888/")95 conn = await tornado.websocket.websocket_connect(request)96 conn.write_message("Authorization: Bearer {}".format("bogus token"))97 self.assertEqual(await conn.read_message(), None)98 self.assertEqual(conn.close_code, 401)99 @tornado.testing.gen_test100 async def test_websockets_closes_when_message_sent_before_auth(self):101 testself = self102 user_id = None103 class EchoWebSocket(tornado.websocket.WebSocketHandler):104 def on_message(self, message):105 testself.assertEqual(message, 'USER_ID: %s' % user_id)106 self.write_message("Boom")107 background_app = tornado.web.Application([108 (r".*", EchoWebSocket),109 ])110 background_app.listen(8889)111 request = HTTPRequest(url="ws://localhost:8888/")112 conn = await tornado.websocket.websocket_connect(request)113 conn.write_message("Sup?")114 self.assertEqual(await conn.read_message(), None)115 self.assertEqual(conn.close_code, 401)116 @tornado.testing.gen_test117 async def test_websockets_allow_two_direct_traffic(self):118 testself = self119 user_id = None120 class EchoWebSocket(tornado.websocket.WebSocketHandler):121 def __init__(self, application, request, **kwargs):122 super().__init__(application, request, **kwargs)123 self.user_id = None124 def on_message(self, message):125 if self.user_id is None:126 testself.assertEqual(message, 'USER_ID: %s' % user_id)127 self.user_id = user_id128 elif message == "What does an explosion say?":129 self.write_message("Boom")130 else:131 self.write_message("Moo")132 background_app = tornado.web.Application([133 (r".*", EchoWebSocket),134 ])135 background_app.listen(8889)136 response = await self.http_client.fetch("http://localhost:8888/auth/users", method="POST",137 body="username=test&password=test")138 self.assertEqual(response.code, 201)139 user_id = response.headers['USER_ID']140 response = await self.http_client.fetch("http://localhost:8888/auth/login", method="POST",141 body="username=test&password=test")142 self.assertEqual(response.code, 200)143 token = json.loads(response.body.decode())["access_token"]144 request = HTTPRequest(url="ws://localhost:8888/")145 conn = await tornado.websocket.websocket_connect(request)146 conn.write_message("Authorization: Bearer {}".format(token))147 conn.write_message("What does an explosion say?")148 self.assertEqual(await conn.read_message(), 'Boom')149 conn.write_message("What does a cow say?")150 self.assertEqual(await conn.read_message(), 'Moo')151 conn.close()152 self.assertIsNone(await conn.read_message())153 @tornado.testing.gen_test154 async def test_websockets_block_outgoing_when_deactivated(self):155 class EchoWebSocket(tornado.websocket.WebSocketHandler):156 def on_message(self, message):157 self.write_message("Boom")158 background_app = tornado.web.Application([159 (r".*", EchoWebSocket),160 ])161 background_app.listen(8889)162 response = await self.http_client.fetch("http://localhost:8888/auth/users", method="POST",163 body="username=test&password=test")164 self.assertEqual(response.code, 201)165 user_id = response.headers['USER_ID']166 response = await self.http_client.fetch("http://localhost:8888/auth/login", method="POST",167 body="username=test&password=test")168 self.assertEqual(response.code, 200)169 token = json.loads(response.body.decode())["access_token"]170 request = HTTPRequest(url="ws://localhost:8888/")171 conn = await tornado.websocket.websocket_connect(request)172 conn.write_message("Authorization: Bearer {}".format(token))173 self.assertEqual(await conn.read_message(), 'Boom')174 response = await self.http_client.fetch("http://localhost:8888/auth/users/%s/deactivate" % user_id, method="POST",175 body="")176 self.assertEqual(response.code, 200)177 conn.write_message("What does an explosion say?")178 self.assertEqual(await conn.read_message(), None)179 self.assertEqual(conn.close_code, 401)180 @tornado.testing.gen_test181 async def test_websockets_block_incoming_when_deactivated(self):182 class EchoWebSocket(tornado.websocket.WebSocketHandler):183 def on_message(self, message):184 self.write_message("Boom")185 def delayed_message():186 self.write_message("Boom2")187 tornado.ioloop.IOLoop.current().call_later(1, delayed_message)188 background_app = tornado.web.Application([189 (r".*", EchoWebSocket),190 ])191 background_app.listen(8889)192 response = await self.http_client.fetch("http://localhost:8888/auth/users", method="POST",193 body="username=test&password=test")194 self.assertEqual(response.code, 201)195 user_id = response.headers['USER_ID']196 response = await self.http_client.fetch("http://localhost:8888/auth/login", method="POST",197 body="username=test&password=test")198 self.assertEqual(response.code, 200)199 token = json.loads(response.body.decode())["access_token"]200 request = HTTPRequest(url="ws://localhost:8888/")201 conn = await tornado.websocket.websocket_connect(request)202 conn.write_message("Authorization: Bearer {}".format(token))203 self.assertEqual(await conn.read_message(), 'Boom')204 response = await self.http_client.fetch("http://localhost:8888/auth/users/%s/deactivate" % user_id, method="POST",205 body="")206 self.assertEqual(response.code, 200)207 self.assertEqual(await conn.read_message(), None)...
test_user_management.py
Source:test_user_management.py
1import json2import logging3import os4import psycopg25import tornado.httpclient6import tornado.ioloop7import tornado.log as tornado_logging8import tornado.testing9import tornado.web10from app import GandalfConfiguration11from app.db.postgres_adapter import PostgresAdapter12from run import make_app13tornado_logging.access_log.setLevel(logging.DEBUG)14tornado_logging.app_log.setLevel(logging.DEBUG)15tornado_logging.gen_log.setLevel(logging.DEBUG)16class TestHandler(tornado.web.RequestHandler):17 def get(self):18 self.write("this works")19class UserManagementTest(tornado.testing.AsyncHTTPTestCase):20 def get_app(self):21 conn = psycopg2.connect(host="localhost", user="postgres")22 cursor = conn.cursor()23 cursor.execute("DROP SCHEMA IF EXISTS gandalf CASCADE")24 conn.commit()25 app = make_app(GandalfConfiguration('localhost:8889', PostgresAdapter(), 'localhost'))26 app.listen(8888)27 return app28 def test_create_user(self):29 os.putenv("NEXT_HOST", "http://localhost:8889")30 background_app = tornado.web.Application([31 (r".*", TestHandler),32 ])33 background_app.listen(8889)34 response = self.fetch("/")35 self.assertEqual(response.code, 401)36 response = self.fetch("/auth/users", method="POST", body="username=test&password=test")37 self.assertEqual(response.code, 201)38 response = self.fetch("/auth/login", method="POST", body="username=test&password=test")39 self.assertEqual(response.code, 200)40 self.assertIsNotNone(json.loads(response.body.decode())["access_token"])41 def test_change_user_password(self):42 os.putenv("NEXT_HOST", "http://localhost:8889")43 background_app = tornado.web.Application([44 (r".*", TestHandler),45 ])46 background_app.listen(8889)47 response = self.fetch("/")48 self.assertEqual(response.code, 401)49 response = self.fetch("/auth/users", method="POST", body="username=test&password=test")50 self.assertEqual(response.code, 201)51 user_id = response.headers["USER_ID"]52 self.assertIsNotNone(user_id)53 response = self.fetch("/auth/login", method="POST", body="username=test&password=test")54 self.assertEqual(response.code, 200)55 self.assertIsNotNone(json.loads(response.body.decode())["access_token"])56 response = self.fetch("/auth/users/{}".format(user_id), method="POST", body="password=test2")57 self.assertEqual(response.code, 200)58 def test_deactivate_reactivate_user(self):59 os.putenv("NEXT_HOST", "http://localhost:8889")60 background_app = tornado.web.Application([61 (r".*", TestHandler),62 ])63 background_app.listen(8889)64 response = self.fetch("/")65 self.assertEqual(response.code, 401)66 response = self.fetch("/auth/users", method="POST", body="username=test&password=test")67 self.assertEqual(response.code, 201)68 user_id = response.headers["USER_ID"]69 self.assertIsNotNone(user_id)70 response = self.fetch("/auth/login", method="POST", body="username=test&password=test")71 self.assertEqual(response.code, 200)72 token = json.loads(response.body.decode())["access_token"]73 self.assertIsNotNone(token)74 response = self.fetch("/", headers={"Authorization": "Bearer {}".format(token)})75 self.assertEqual(response.code, 200)76 response = self.fetch("/auth/users/{}/deactivate".format(user_id), method="POST", body="")77 self.assertEqual(response.code, 200)78 response = self.fetch("/", headers={"Authorization": "Bearer {}".format(token)})79 self.assertEqual(response.code, 401)80 response = self.fetch("/auth/login", method="POST", body="username=test&password=test")81 self.assertEqual(response.code, 401)82 response = self.fetch("/auth/users/{}/reactivate".format(user_id), method="POST", body="")83 self.assertEqual(response.code, 200)84 response = self.fetch("/auth/login", method="POST", body="username=test&password=test")85 self.assertEqual(response.code, 200)86 token = json.loads(response.body.decode())["access_token"]87 self.assertIsNotNone(token)88 response = self.fetch("/", headers={"Authorization": "Bearer {}".format(token)})...
background.py
Source:background.py
1"""2This script contains all classes for3Background processes4"""5from timetrigger import Repeat6from adafruit_hid.keycode import Keycode7class Background_app:8 """9 Base class for background procedures10 return 0 when not taking action11 """12 def __init__(self, freq=None, period=None):13 assert (freq is not None) != (period is not None) # != use as xor14 if freq is not None:15 self.freq = freq16 if period is not None:17 self.freq = 1 / period18 self.repeat_timer = Repeat(self.freq)19 def procedure(self):20 return 021 def __call__(self):22 if self.repeat_timer.check():23 return self.procedure()24 else:25 return 026class FpsControl(Background_app):27 """28 Control the frame rate29 """30 def __init__(self, fps=None):31 super().__init__(freq=fps)32 self.fps_now = 1033 def procedure(self):34 self.fps_now = self.repeat_timer.freq_measure35 return 136class FpsMonitor(Background_app):37 """38 print the current FPS to serial39 This is used for debug propose40 """41 def __init__(self, period, fps_app):42 self.fps_app = fps_app43 super().__init__(period=period)44 def procedure(self):45 print('FPS:', self.fps_app.fps_now)46 return 047class NumLocker(Background_app):48 """49 Keep the number lock on50 Should not be used on Mac51 Will detect Mac and stop52 """53 def __init__(self, keyboard):54 super().__init__(period=0.01)55 self.keyboard = keyboard56 self.key = 'key'57 self.counter = 058 self.is_win = True59 def procedure(self):60 ## keep number lock61 if self.is_win:62 if not int.from_bytes(self.keyboard.led_status, "big") & 1:63 # if NumLock LED on64 self.keyboard.send(Keycode.KEYPAD_NUMLOCK)65 if not int.from_bytes(self.keyboard.led_status, "big") & 1:66 # if key dose not have any effect67 # Either NumLock is being hold68 # or Mac69 self.counter += 170 if self.counter > 50:71 # If number raise up fastly72 # Confirmed Mac73 self.is_win = False74 print('Mac detected')75 else:76 self.counter = 077 return 078class MouseJitter(Background_app):79 """80 shake the cursor once a few seconds81 to keep the computer awake82 """83 def __init__(self, mouse, period):84 super().__init__(period=period)85 self.mouse = mouse86 def procedure(self):87 # Move mouse in a loop88 self.mouse.move(10, 0, 0)89 self.mouse.move(0, 10, 0)90 self.mouse.move(0, -10, 0)91 self.mouse.move(-10, 0, 0)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run robotframework-appiumlibrary automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
