Best Python code snippet using autotest_python
test_haproxy_control.py
Source:test_haproxy_control.py
...15 self.cleanup()16 def tearDown(self):17 self.cleanup()18 def cleanup(self):19 if self.control.get_server_status('A', 'A1') != STATUS_UP:20 self.control.set_server_status('A', 'A1', STATUS_READY)21 def test_set_to_drain(self):22 self.assertEqual(STATUS_UP, self.control.get_server_status('A', 'A1'))23 self.control.set_server_status('A', 'A1', STATUS_DRAIN)24 self.assertEqual(STATUS_DRAIN, self.control.get_server_status('A', 'A1'))25 def test_set_to_maint(self):26 self.assertEqual(STATUS_UP, self.control.get_server_status('A', 'A1'))27 self.control.set_server_status('A', 'A1', STATUS_MAINT)28 self.assertEqual(STATUS_MAINT, self.control.get_server_status('A', 'A1'))29 def test_set_to_ready(self):30 self.control.set_server_status('A', 'A1', STATUS_MAINT)31 self.assertEqual(STATUS_MAINT, self.control.get_server_status('A', 'A1'))32 self.control.set_server_status('A', 'A1', STATUS_READY)33 self.assertEqual(STATUS_UP, self.control.get_server_status('A', 'A1'))34 def test_connection_error_is_wrapped(self):35 # no haproxy running at port 990336 control = HaProxyControl('tcp://127.0.0.1:9903')37 with self.assertRaises(HaProxyConnectionRefused) as cm:38 control.set_server_status('A', 'A1', STATUS_MAINT)39 self.assertEquals('Connection refused',40 cm.exception.args[1])41if __name__ == '__main__':...
main.py
Source:main.py
...7class BoundThreadPoolExecutor(ThreadPoolExecutor):8 def __init__(self, *args, **kwargs) -> None:9 super(BoundThreadPoolExecutor, self).__init__(*args, **kwargs)10 self._work_queue = Queue(2)11def get_server_status(host: str = "farmer.xyz", port: int = 8088):12 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)13 try:14 server.connect((host, port))15 print('{0} port {1} is open'.format(host, port))16 return True17 except Exception as err:18 # print('{0} port {1} is not open'.format(host, port))19 return False20 finally:21 server.close()22with BoundThreadPoolExecutor(max_workers=300) as pool:23 for num in range(1, 255):24 host = "172.16.1." + str(num)25 print(host)26 for port in range(1, 8000):27 future1 = pool.submit(get_server_status, host, port)28 # print(pool._work_queue.qsize())29end_time = time.time()30runtime = end_time - star_time31print("runtime: ", runtime)32# pool = ThreadPoolExecutor(max_workers=400)33# for num in range(1, 255):34# host = "172.16.1." + str(num)35# for port in range(1, 65535):36# future1 = pool.submit(get_server_status, host, port)37# pool.shutdown()38# def get_que():39# t = que.get()40# t.start()41# def put_que(host, port):42# print("putting...")43# t = Thread(target=get_server_status,args=(host, port))44# que.put(t)45# get = Thread(target=get_que, name="Get Thread")46# get.start()47 # put_que(host=host, port=port)48 # put = Thread(target=put_que, args=[host, port], name="Put Thread")49 ...
scan.py
Source:scan.py
1import socket2import threading3def get_server_status(host: str = "farmer.xyz", port: int = 8088):4 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)5 try:6 server.connect((host, port))7 print('{0} port {1} is open'.format(host, port))8 return True9 except Exception as err:10 # print('{0} port {1} is not open'.format(host, port))11 return False12 finally:13 server.close()14for num in range(0, 255):15 host = "172.16.90." + str(num)16 t = threading.Thread(target=get_server_status,args=(host, 80))17 t.start()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!