KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_local_port method in uiautomator
Best Python code snippet using uiautomator
test_coap_core.py
Source:test_coap_core.py 
1import unittest2import time3from piccata import core4from piccata import message5from piccata import resource6from piccata.constants import *7from transport import tester8from ipaddress import ip_address9import sys10TEST_PAYLOAD = b"testPayload"11TEST_TOKEN = b"abcd"12TEST_MID = 100013TEST_ADDRESS = ip_address(u"12.34.56.78")14TEST_PORT = 1234515TEST_LOCAL_ADDRESS = ip_address(u"10.10.10.10")16TEST_LOCAL_PORT = 2000017class TestResource(resource.CoapResource):18 def __init__(self):19 resource.CoapResource.__init__(self)20 self.resource_handler = None21 self.call_counter = 022 def render_GET(self, request):23 self.call_counter += 124 rsp = None25 if self.resource_handler != None:26 rsp = self.resource_handler(request)27 return rsp28class TestCoap(unittest.TestCase):29 def setUp(self):30 root = resource.CoapResource()31 self.test_resource = TestResource()32 root.put_child(b'test', self.test_resource)33 endpoint = resource.CoapEndpoint(root)34 self.transport = tester.TesterTransport()35 self.protocol = core.Coap(self.transport)36 self.request_handler = resource.ResourceManager(endpoint)37 self.transport.register_receiver(self.protocol)38 self.protocol.register_request_handler(self.request_handler)39 self.transport.open()40 self.resource_handler = None41 self.responseResult = None42 self.callbackCounter = 043 def tearDown(self):44 self.transport.close()45 def assertMessageInTransport(self, message, remote, count=None):46 data = message.encode()47 self.assertTupleEqual(self.transport.tester_remote, remote)48 self.assertEqual(self.transport.tester_data, data)49 if count != None:50 self.assertEqual(self.transport.output_count, count)51 def callback(self, result, request, response):52 self.responseResult = result53 self.callbackCounter += 154 def assertInRetransmissionList(self, message):55 self.assertIn(message.mid, self.protocol._message_layer._active_exchanges)56 self.assertEqual(self.protocol._message_layer._active_exchanges[message.mid][0], message)57 def assertNotInRetransmissionList(self, mid):58 self.assertNotIn(mid, self.protocol._message_layer._active_exchanges)59 def assertInOutgoingRequestList(self, request):60 key = (request.token, request.remote)61 self.assertIn(key, self.protocol._transaction_layer._outgoing_requests)62 self.assertEqual(self.protocol._transaction_layer._outgoing_requests[key][1], (self.callback, None, None))63 def assertNotInOutgoingRequestList(self, token, remote):64 self.assertNotIn((token, remote), self.protocol._transaction_layer._outgoing_requests)65 def assertInDeduplicationList(self, mid, remote, response=None):66 key = (mid, remote)67 self.assertIn(key, self.protocol._message_layer._recent_remote_ids)68 if response != None:69 self.assertEqual(len(self.protocol._message_layer._recent_remote_ids[key]), 3)70 self.assertEqual(self.protocol._message_layer._recent_remote_ids[key][2], response)71 def assertNotInDeduplicationList(self, mid, remote):72 self.assertIn((mid, remote), self.protocol._message_layer._recent_remote_ids)73class TestCoapSendRequestPath(TestCoap):74 def test_coap_core_shall_return_error_when_non_request_message_is_sent_as_request(self):75 req = message.Message(CON, TEST_MID, CHANGED, TEST_PAYLOAD, TEST_TOKEN)76 req.remote = (TEST_ADDRESS, TEST_PORT)77 self.assertRaises(ValueError, self.protocol.request, (req))78 def test_coap_core_shall_queue_CON_request_on_retransmission_list(self):79 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)80 req.remote = (TEST_ADDRESS, TEST_PORT)81 self.protocol.request(req)82 self.assertMessageInTransport(req, req.remote, 1)83 self.assertInRetransmissionList(req)84 def test_coap_core_shall_not_queue_NON_request_on_retransmission_list(self):85 req = message.Message(NON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)86 req.remote = (TEST_ADDRESS, TEST_PORT)87 self.protocol.request(req)88 self.assertMessageInTransport(req, req.remote, 1)89 self.assertNotInRetransmissionList(TEST_MID)90 def test_coap_core_shall_queue_request_on_pending_response_list_if_callback_is_registered(self):91 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)92 req.remote = (TEST_ADDRESS, TEST_PORT)93 self.protocol.request(req, self.callback)94 self.assertInOutgoingRequestList(req)95 def test_coap_core_shall_not_queue_request_on_pending_response_list_if_callback_is_not_registered(self):96 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)97 req.remote = (TEST_ADDRESS, TEST_PORT)98 self.protocol.request(req)99 self.assertNotInOutgoingRequestList(TEST_TOKEN, req.remote)100class TestCoapSendResponsePath(TestCoap):101 def setUp(self):102 super(TestCoapSendResponsePath, self).setUp()103 self.test_resource.resource_handler = self.responder104 def responder(self, request):105 return self.rsp106 def test_coap_core_shall_return_error_when_non_response_message_is_sent_as_response(self):107 self.rsp = message.Message(ACK, TEST_MID, GET, b"", TEST_TOKEN)108 # Prepare fake request to trigger response sending.109 req = message.Message(CON, TEST_MID, GET, b"", TEST_TOKEN)110 req.opt.uri_path = (b"test", )111 raw = req.encode()112 # Check that error is raised on incorrect response type113 self.assertRaises(ValueError, self.transport._receive, raw, (TEST_ADDRESS, TEST_PORT), (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))114 def test_coap_core_shall_queue_CON_response_on_retransmission_list(self):115 self.rsp = message.Message(CON, TEST_MID + 1, CONTENT, b"", TEST_TOKEN)116 # Prepare fake request to trigger response sending.117 req = message.Message(NON, TEST_MID, GET, b"", TEST_TOKEN)118 req.opt.uri_path = (b"test", )119 raw = req.encode()120 # Simulate fake request reception121 self.transport._receive(raw, (TEST_ADDRESS, TEST_PORT), (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))122 # Validate that response was handled properly123 self.assertInRetransmissionList(self.rsp)124 def test_coap_core_shall_not_queue_NON_response_on_retransmission_list(self):125 self.rsp = message.Message(NON, TEST_MID + 1, CONTENT, b"", TEST_TOKEN)126 # Prepare fake request to trigger response sending.127 req = message.Message(NON, TEST_MID, GET, b"", TEST_TOKEN)128 req.opt.uri_path = (b"test", )129 raw = req.encode()130 # Simulate fake request reception131 self.transport._receive(raw, (TEST_ADDRESS, TEST_PORT), (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))132 # Validate that response was handled properly133 self.assertNotInRetransmissionList(TEST_MID + 1)134 def test_coap_core_shall_queue_ACK_and_RST_response_on_responded_list(self):135 self.rsp = message.Message(ACK, TEST_MID, CONTENT, b"", TEST_TOKEN)136 # Prepare fake request to trigger response sending.137 req = message.Message(NON, TEST_MID, GET, b"", TEST_TOKEN)138 req.opt.uri_path = (b"test", )139 raw = req.encode()140 remote = (TEST_ADDRESS, TEST_PORT)141 # Simulate fake request reception142 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))143 # Validate that request was added to the deduplication list, and response was registered.144 self.assertInDeduplicationList(TEST_MID, remote, self.rsp)145class TestCoapReceiveRequestPath(TestCoap):146 def responder(self, request):147 return self.rsp148 def test_coap_core_shall_store_received_CON_request_on_deduplication_list(self):149 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)150 raw = req.encode()151 remote = (TEST_ADDRESS, TEST_PORT)152 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))153 self.assertInDeduplicationList(TEST_MID, remote)154 def test_coap_core_shall_store_received_NON_request_on_deduplication_list(self):155 req = message.Message(NON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)156 raw = req.encode()157 remote = (TEST_ADDRESS, TEST_PORT)158 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))159 self.assertInDeduplicationList(TEST_MID, remote)160 def check_that_duplicated_request_is_automatically_responded(self):161 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)162 req.opt.uri_path = (b"test", )163 raw = req.encode()164 remote = (TEST_ADDRESS, TEST_PORT)165 # Receive first request.166 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))167 # Verify that resource handler was called and response sent.168 self.assertEqual(self.test_resource.call_counter, 1)169 self.assertEqual(self.transport.output_count, 1)170 self.assertTupleEqual(self.transport.tester_remote, remote)171 self.assertEqual(self.transport.tester_data, self.rsp.encode())172 # Receive duplicated request.173 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))174 # Verify that resource handler was not called but reponse was resent.175 self.assertEqual(self.test_resource.call_counter, 1)176 self.assertEqual(self.transport.output_count, 2)177 self.assertTupleEqual(self.transport.tester_remote, remote)178 self.assertEqual(self.transport.tester_data, self.rsp.encode())179 def test_coap_core_shall_resend_ACK_on_duplicated_CON_request(self):180 self.test_resource.resource_handler = self.responder181 self.rsp = message.Message(ACK, TEST_MID, CONTENT, TEST_PAYLOAD, TEST_TOKEN)182 self.check_that_duplicated_request_is_automatically_responded()183 def test_coap_core_shall_resend_RST_on_duplicated_CON_request(self):184 self.test_resource.resource_handler = self.responder185 self.rsp = message.Message(RST, TEST_MID, CONTENT, TEST_PAYLOAD, TEST_TOKEN)186 self.check_that_duplicated_request_is_automatically_responded()187 def test_coap_core_shall_ignore_duplicated_CON_if_no_response_was_sent_to_the_original_message(self):188 # No resopnse is sent.189 req = message.Message(CON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)190 req.opt.uri_path = (b"test", )191 raw = req.encode()192 remote = (TEST_ADDRESS, TEST_PORT)193 # Receive first request.194 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))195 # Verify that resource handler was called..196 self.assertEqual(self.test_resource.call_counter, 1)197 # Receive duplicated request.198 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))199 # Verify that resource handler was not called and no response sent.200 self.assertEqual(self.test_resource.call_counter, 1)201 self.assertEqual(self.transport.output_count, 0)202 def test_coap_core_shall_ignore_duplicated_NON_request(self):203 self.test_resource.resource_handler = self.responder204 self.rsp = message.Message(NON, TEST_MID, CONTENT, TEST_PAYLOAD, TEST_TOKEN)205 req = message.Message(NON, TEST_MID, GET, TEST_PAYLOAD, TEST_TOKEN)206 req.opt.uri_path = (b"test", )207 raw = req.encode()208 remote = (TEST_ADDRESS, TEST_PORT)209 # Receive first request.210 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))211 # Verify that resource handler was called and response sent.212 self.assertEqual(self.test_resource.call_counter, 1)213 self.assertEqual(self.transport.output_count, 1)214 self.assertTupleEqual(self.transport.tester_remote, remote)215 self.assertEqual(self.transport.tester_data, self.rsp.encode())216 # Receive duplicated request.217 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))218 # Verify that resource handler was not called and nothing was transmitted.219 self.assertEqual(self.test_resource.call_counter, 1)220 self.assertEqual(self.transport.output_count, 1)221class TestCoapReceiveResponsePath(TestCoap):222 def send_initial_request(self, remote, timeout = None):223 self.req = message.Message(CON, TEST_MID, GET, b"", TEST_TOKEN)224 self.req.remote = remote225 if timeout != None:226 self.req.timeout = timeout227 self.protocol.request(self.req, self.callback)228 def receive_ack_response(self, remote):229 rsp = message.Message(ACK, TEST_MID, CONTENT, TEST_PAYLOAD, TEST_TOKEN)230 raw = rsp.encode()231 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))232 def receive_empty_ack_response(self, remote):233 rsp = message.Message(ACK, TEST_MID, EMPTY)234 raw = rsp.encode()235 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))236 def receive_rst_response(self, remote):237 rsp = message.Message(RST, TEST_MID, EMPTY)238 raw = rsp.encode()239 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))240 def receive_con_response(self, remote):241 rsp = message.Message(CON, TEST_MID + 1, CONTENT, TEST_PAYLOAD, TEST_TOKEN)242 raw = rsp.encode()243 self.transport._receive(raw, remote, (TEST_LOCAL_ADDRESS, TEST_LOCAL_PORT))244 def test_coap_core_shall_remove_CON_message_from_retransmission_list_if_ACK_is_received(self):245 remote = (TEST_ADDRESS, TEST_PORT)246 # Send request so it could be queued on retransmisison list.247 self.send_initial_request(remote)248 self.assertInRetransmissionList(self.req)249 # Simulate receiveing response and verify that retransmission was removed.250 self.receive_ack_response(remote)251 self.assertNotInRetransmissionList(TEST_MID);252 def test_coap_core_shall_remove_CON_message_from_retransmission_list_if_RST_is_received(self):253 remote = (TEST_ADDRESS, TEST_PORT)254 # Send request so it could be queued on retransmisison list.255 self.send_initial_request(remote)256 self.assertInRetransmissionList(self.req)257 # Simulate receiveing response and verify that retransmission was removed.258 self.receive_rst_response(remote)259 self.assertNotInRetransmissionList(TEST_MID);260 def test_coap_core_shall_remove_request_from_pending_response_list_if_response_is_received(self):261 remote = (TEST_ADDRESS, TEST_PORT)262 # Send request so it could be queued on retransmisison list.263 self.send_initial_request(remote)264 self.assertInOutgoingRequestList(self.req)265 # Simulate receiveing response and verify that retransmission was removed.266 self.receive_ack_response(remote)267 self.assertNotInOutgoingRequestList(TEST_TOKEN, remote)268 def test_coap_core_shall_remove_request_form_pending_response_list_if_RST_is_received(self):269 remote = (TEST_ADDRESS, TEST_PORT)270 # Send request so it could be queued on retransmisison list.271 self.send_initial_request(remote)272 self.assertInOutgoingRequestList(self.req)273 # Simulate receiveing response and verify that retransmission was removed.274 self.receive_rst_response(remote)275 self.assertNotInOutgoingRequestList(TEST_TOKEN, remote)276 def test_coap_core_shall_call_application_callback_with_success_on_response_received(self):277 remote = (TEST_ADDRESS, TEST_PORT)278 self.send_initial_request(remote)279 self.receive_ack_response(remote)280 self.assertEqual(self.responseResult, RESULT_SUCCESS)281 def test_coap_core_shall_call_application_callback_with_error_on_RST_received(self):282 remote = (TEST_ADDRESS, TEST_PORT)283 self.send_initial_request(remote)284 self.receive_rst_response(remote)285 self.assertEqual(self.responseResult, RESULT_RESET)286 def test_coap_core_shall_call_application_callback_with_error_on_request_cancelled(self):287 remote = (TEST_ADDRESS, TEST_PORT)288 self.send_initial_request(remote)289 self.protocol.cancel_request(self.req)290 self.assertEqual(self.responseResult, RESULT_CANCELLED)291 def test_coap_core_shall_call_application_callback_with_timeout_on_no_response_received(self):292 remote = (TEST_ADDRESS, TEST_PORT)293 self.send_initial_request(remote, timeout = 0.5)294 time.sleep(0.6)295 self.assertEqual(self.responseResult, RESULT_TIMEOUT)296 def test_coap_core_shall_resend_ACK_on_duplicated_CON_response(self):297 # Send initial request.298 remote = (TEST_ADDRESS, TEST_PORT)299 raw_empty_ack = message.Message(ACK, TEST_MID + 1, EMPTY).encode()300 self.send_initial_request(remote)301 self.assertEqual(self.transport.output_count, 1)302 # Simulate receiving empty ACK and verify that no callback was called.303 self.receive_empty_ack_response(remote)304 self.assertIsNone(self.responseResult)305 # Receive separete CON response amd verify that callback was called and empty ACK was automatically sent.306 self.receive_con_response(remote)307 self.assertEqual(self.transport.output_count, 2)308 self.assertEqual(self.transport.tester_data, raw_empty_ack)309 self.assertEqual(self.callbackCounter, 1)310 self.assertEqual(self.responseResult, RESULT_SUCCESS)311 # Receive separate response duplicate, verify that empty ACK was forwarded and no callback was called.312 self.receive_con_response(remote)313 self.assertEqual(self.transport.output_count, 3)314 self.assertEqual(self.transport.tester_data, raw_empty_ack)315 self.assertEqual(self.callbackCounter, 1)316if __name__ == "__main__":...
myfunc.py
Source:myfunc.py
...129 return ip130 return '127.0.0.1'131 # was before return str(gethostbyname(gethostname()))132 return get(ips_ls)133def test_local_port(port):134 # https://docs.python.org/2/library/socket.html#example135 try:136 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)137 s.bind(('', port)) # Try to open port138 s.close()139 return False140 except OSError:141 return True142def get_free_port(scope: Tuple[int, str] = (5000, 6000), exclude: List[int] = []):143 if not scope:144 i = 0145 while True:146 i += 1147 with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:148 s.bind(('', 0))149 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)150 if s.getsockname()[1] not in exclude:151 return s.getsockname()[1]152 if i > 1000:153 raise Exception('get_free_port search exceeded 1000')154 else:155 sc = []156 for el in scope:157 sc.append(int(el))158 scope = tuple(sc)159 for port in range(scope[0], scope[1]):160 if port not in exclude:161 if not test_local_port(port):162 return port163 raise Exception(f'Free port cannot be found in the range {scope}')164def verify_port(port: str, exclude=[]):165 if ':' in port:166 ports = tuple(port.split(':'))167 return get_free_port(scope=ports,exclude=exclude)168 else:169 port = int(port)170 if test_local_port(port) or port in exclude:171 return get_free_port(scope=(port, port+1000), exclude=exclude)172 return port173def ndarray_tostring(arr):174 '''175 Convert ndarray to tab separated string176 >>> import numpy as np177 >>> tests_devices = np.array([1, 2, 3])178 >>> b = ndarray_tostring(tests_devices)179 >>> print(b)180 1.000181 2.000182 3.000183 >>> c = np.array([[1, 2, 3],[1, 2, 3]])184 >>> d = ndarray_tostring(c)...
netmiko_cdp_neighbors.py
Source:netmiko_cdp_neighbors.py
...21 @pytest.mark.nuts("remote_host")22 def test_remote_host(self, single_result: NutsResult, remote_host: str) -> None:23 assert remote_host in single_result.result24 @pytest.mark.nuts("remote_host,local_port")25 def test_local_port(26 self, single_result: NutsResult, remote_host: str, local_port: str27 ) -> None:28 assert single_result.result[remote_host]["local_port"] == local_port29 @pytest.mark.nuts("remote_host,remote_port")30 def test_remote_port(31 self, single_result: NutsResult, remote_host: str, remote_port: str32 ) -> None:33 assert single_result.result[remote_host]["remote_port"] == remote_port34 @pytest.mark.nuts("remote_host,management_ip")35 def test_management_ip(36 self, single_result: NutsResult, remote_host: str, management_ip: str37 ) -> None:38 assert single_result.result[remote_host]["management_ip"] == management_ip39class TestNetmikoCdpNeighborsCount:...
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 uiautomator 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?
