Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use wait_for_login method in autotest
Best Python code snippet using autotest_python
test_nvp_api_request_eventlet.py
Source:test_nvp_api_request_eventlet.py 
1# Copyright (C) 2009-2012 Nicira Networks, Inc. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"); you may4# not use this file except in compliance with the License. You may obtain5# a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12# License for the specific language governing permissions and limitations13# under the License.14import httplib15import logging16import new17import random18import unittest19import eventlet20from eventlet.green import urllib221from mock import Mock22from mock import patch23from quantum.plugins.nicira.nicira_nvp_plugin.api_client import (24 client_eventlet as nace,25 request_eventlet as nare,26)27logging.basicConfig(level=logging.DEBUG)28LOG = logging.getLogger("test_nvp_api_request_eventlet")29REQUEST_TIMEOUT = 130def fetch(url):31 return urllib2.urlopen(url).read()32class NvpApiRequestEventletTest(unittest.TestCase):33 def setUp(self):34 self.client = nace.NvpApiClientEventlet(35 [("127.0.0.1", 4401, True)], "admin", "admin")36 self.url = "/ws.v1/_debug"37 self.req = nare.NvpApiRequestEventlet(self.client, self.url)38 def tearDown(self):39 self.client = None40 self.req = None41 def test_construct_eventlet_api_request(self):42 e = nare.NvpApiRequestEventlet(self.client, self.url)43 self.assertTrue(e is not None)44 def test_apirequest_spawn(self):45 def x(id):46 eventlet.greenthread.sleep(random.random())47 LOG.info('spawned: %d' % id)48 for i in range(10):49 nare.NvpApiRequestEventlet._spawn(x, i)50 def test_apirequest_start(self):51 for i in range(10):52 a = nare.NvpApiRequestEventlet(53 self.client, self.url, request_timeout=0.1)54 a._handle_request = Mock()55 a.start()56 eventlet.greenthread.sleep(0.1)57 logging.info('_handle_request called: %s' %58 a._handle_request.called)59 nare.NvpApiRequestEventlet.joinall()60 def test_join_with_handle_request(self):61 self.req._handle_request = Mock()62 self.req.start()63 self.req.join()64 self.assertTrue(self.req._handle_request.called)65 def test_join_without_handle_request(self):66 self.req._handle_request = Mock()67 self.req.join()68 self.assertFalse(self.req._handle_request.called)69 def test_copy(self):70 req = self.req.copy()71 for att in [72 '_api_client', '_url', '_method', '_body', '_headers',73 '_http_timeout', '_request_timeout', '_retries',74 '_redirects', '_auto_login']:75 self.assertTrue(getattr(req, att) is getattr(self.req, att))76 def test_request_error(self):77 self.assertTrue(self.req.request_error is None)78 def test_run_and_handle_request(self):79 self.req._request_timeout = None80 self.req._handle_request = Mock()81 self.req.start()82 self.req.join()83 self.assertTrue(self.req._handle_request.called)84 def test_run_and_timeout(self):85 def my_handle_request(self):86 LOG.info('my_handle_request() self: %s' % self)87 LOG.info('my_handle_request() dir(self): %s' % dir(self))88 eventlet.greenthread.sleep(REQUEST_TIMEOUT * 2)89 self.req._request_timeout = REQUEST_TIMEOUT90 self.req._handle_request = new.instancemethod(91 my_handle_request, self.req, nare.NvpApiRequestEventlet)92 self.req.start()93 self.assertTrue(self.req.join() is None)94 def prep_issue_request(self):95 mysock = Mock()96 mysock.gettimeout.return_value = 424297 myresponse = Mock()98 myresponse.read.return_value = 'body'99 myresponse.getheaders.return_value = 'headers'100 myresponse.status = httplib.MOVED_PERMANENTLY101 myconn = Mock()102 myconn.request.return_value = None103 myconn.sock = mysock104 myconn.getresponse.return_value = myresponse105 myconn.__str__ = Mock()106 myconn.__str__.return_value = 'myconn string'107 req = self.req108 req._request_timeout = REQUEST_TIMEOUT = 1109 req._redirect_params = Mock()110 req._redirect_params.return_value = (myconn, 'url')111 req._request_str = Mock()112 req._request_str.return_value = 'http://cool/cool'113 client = self.client114 client.need_login = False115 client._auto_login = False116 client._auth_cookie = False117 client.acquire_connection = Mock()118 client.acquire_connection.return_value = myconn119 client.release_connection = Mock()120 return (mysock, myresponse, myconn)121 def test_issue_request_trigger_exception(self):122 (mysock, myresponse, myconn) = self.prep_issue_request()123 self.client.acquire_connection.return_value = None124 self.req._issue_request()125 LOG.info('request_error: %s' % self.req._request_error)126 self.assertTrue(isinstance(self.req._request_error, Exception))127 self.assertTrue(self.client.acquire_connection.called)128 def test_issue_request_handle_none_sock(self):129 (mysock, myresponse, myconn) = self.prep_issue_request()130 myconn.sock = None131 self.req.start()132 self.assertTrue(self.req.join() is None)133 self.assertTrue(self.client.acquire_connection.called)134 def test_issue_request_exceed_maximum_retries(self):135 (mysock, myresponse, myconn) = self.prep_issue_request()136 self.req.start()137 self.assertTrue(self.req.join() is None)138 self.assertTrue(self.client.acquire_connection.called)139 def test_issue_request_trigger_non_redirect(self):140 (mysock, myresponse, myconn) = self.prep_issue_request()141 myresponse.status = httplib.OK142 self.req.start()143 self.assertTrue(self.req.join() is None)144 self.assertTrue(self.client.acquire_connection.called)145 def test_issue_request_trigger_internal_server_error(self):146 (mysock, myresponse, myconn) = self.prep_issue_request()147 self.req._redirect_params.return_value = (myconn, None)148 self.req.start()149 self.assertTrue(self.req.join() is None)150 self.assertTrue(self.client.acquire_connection.called)151 def test_redirect_params_break_on_location(self):152 myconn = Mock()153 (conn, retval) = self.req._redirect_params(154 myconn, [('location', None)])155 self.assertTrue(retval is None)156 def test_redirect_params_parse_a_url(self):157 myconn = Mock()158 (conn, retval) = self.req._redirect_params(159 myconn, [('location', '/path/a/b/c')])160 self.assertTrue(retval is not None)161 def test_redirect_params_invalid_redirect_location(self):162 myconn = Mock()163 (conn, retval) = self.req._redirect_params(164 myconn, [('location', '+path/a/b/c')])165 self.assertTrue(retval is None)166 def test_redirect_params_invalid_scheme(self):167 myconn = Mock()168 (conn, retval) = self.req._redirect_params(169 myconn, [('location', 'invalidscheme://hostname:1/path')])170 self.assertTrue(retval is None)171 def test_redirect_params_setup_https_with_cooki(self):172 with patch('quantum.plugins.nicira.nicira_nvp_plugin.api_client.'173 'client_eventlet.NvpApiClientEventlet') as mock:174 api_client = mock.return_value175 api_client.wait_for_login.return_value = None176 api_client.auth_cookie = 'mycookie'177 api_client.acquire_connection.return_value = True178 myconn = Mock()179 (conn, retval) = self.req._redirect_params(180 myconn, [('location', 'https://host:1/path')])181 self.assertTrue(retval is not None)182 self.assertTrue(api_client.wait_for_login.called)183 self.assertTrue(api_client.acquire_connection.called)184 def test_redirect_params_setup_htttps_and_query(self):185 with patch('quantum.plugins.nicira.nicira_nvp_plugin.api_client.'186 'client_eventlet.NvpApiClientEventlet') as mock:187 api_client = mock.return_value188 api_client.wait_for_login.return_value = None189 api_client.auth_cookie = 'mycookie'190 api_client.acquire_connection.return_value = True191 myconn = Mock()192 (conn, retval) = self.req._redirect_params(myconn, [193 ('location', 'https://host:1/path?q=1')])194 self.assertTrue(retval is not None)195 self.assertTrue(api_client.wait_for_login.called)196 self.assertTrue(api_client.acquire_connection.called)197 def test_redirect_params_setup_https_connection_no_cookie(self):198 with patch('quantum.plugins.nicira.nicira_nvp_plugin.api_client.'199 'client_eventlet.NvpApiClientEventlet') as mock:200 api_client = mock.return_value201 api_client.wait_for_login.return_value = None202 api_client.auth_cookie = None203 api_client.acquire_connection.return_value = True204 myconn = Mock()205 (conn, retval) = self.req._redirect_params(myconn, [206 ('location', 'https://host:1/path')])207 self.assertTrue(retval is not None)208 self.assertTrue(api_client.wait_for_login.called)209 self.assertTrue(api_client.acquire_connection.called)210 def test_redirect_params_setup_https_and_query_no_cookie(self):211 with patch('quantum.plugins.nicira.nicira_nvp_plugin.api_client.'212 'client_eventlet.NvpApiClientEventlet') as mock:213 api_client = mock.return_value214 api_client.wait_for_login.return_value = None215 api_client.auth_cookie = None216 api_client.acquire_connection.return_value = True217 myconn = Mock()218 (conn, retval) = self.req._redirect_params(219 myconn, [('location', 'https://host:1/path?q=1')])220 self.assertTrue(retval is not None)221 self.assertTrue(api_client.wait_for_login.called)222 self.assertTrue(api_client.acquire_connection.called)223 def test_redirect_params_path_only_with_query(self):224 with patch('quantum.plugins.nicira.nicira_nvp_plugin.api_client.'225 'client_eventlet.NvpApiClientEventlet') as mock:226 api_client = mock.return_value227 api_client.wait_for_login.return_value = None228 api_client.auth_cookie = None229 api_client.acquire_connection.return_value = True230 myconn = Mock()231 (conn, retval) = self.req._redirect_params(myconn, [232 ('location', '/path?q=1')])233 self.assertTrue(retval is not None)234 def test_handle_request_auto_login(self):235 self.req._auto_login = True236 self.req._api_client = Mock()237 self.req._api_client.need_login = True238 self.req._request_str = Mock()239 self.req._request_str.return_value = 'http://cool/cool'240 self.req.spawn = Mock()241 self.req._handle_request()242 def test_handle_request_auto_login_unauth(self):243 self.req._auto_login = True244 self.req._api_client = Mock()245 self.req._api_client.need_login = True246 self.req._request_str = Mock()247 self.req._request_str.return_value = 'http://cool/cool'248 import socket249 resp = httplib.HTTPResponse(socket.socket())250 resp.status = httplib.UNAUTHORIZED251 mywaiter = Mock()252 mywaiter.wait = Mock()253 mywaiter.wait.return_value = resp254 self.req.spawn = Mock(return_value=mywaiter)255 self.req._handle_request()256 # NvpLoginRequestEventlet tests.257 def test_construct_eventlet_login_request(self):258 r = nare.NvpLoginRequestEventlet(self.client, 'user', 'password')259 self.assertTrue(r is not None)260 def test_session_cookie_session_cookie_retrieval(self):261 r = nare.NvpLoginRequestEventlet(self.client, 'user', 'password')262 r.successful = Mock()263 r.successful.return_value = True264 r.value = Mock()265 r.value.get_header = Mock()266 r.value.get_header.return_value = 'cool'267 self.assertTrue(r.session_cookie() is not None)268 def test_session_cookie_not_retrieved(self):269 r = nare.NvpLoginRequestEventlet(self.client, 'user', 'password')270 r.successful = Mock()271 r.successful.return_value = False272 r.value = Mock()273 r.value.get_header = Mock()274 r.value.get_header.return_value = 'cool'275 self.assertTrue(r.session_cookie() is None)276 # NvpGetApiProvidersRequestEventlet tests.277 def test_construct_eventlet_get_api_providers_request(self):278 r = nare.NvpGetApiProvidersRequestEventlet(self.client)279 self.assertTrue(r is not None)280 def test_api_providers_none_api_providers(self):281 r = nare.NvpGetApiProvidersRequestEventlet(self.client)282 r.successful = Mock(return_value=False)283 self.assertTrue(r.api_providers() is None)284 def test_api_providers_non_none_api_providers(self):285 r = nare.NvpGetApiProvidersRequestEventlet(self.client)286 r.value = Mock()287 r.value.body = """{288 "results": [289 { "roles": [290 { "role": "api_provider",291 "listen_addr": "pssl:1.1.1.1:1" }]}]}"""292 r.successful = Mock(return_value=True)293 LOG.info('%s' % r.api_providers())...
testcase_01_lifecycle.py
Source:testcase_01_lifecycle.py
...32 def test_password_start_ecs(self):33 self.log.info("Start ECS")34 self.vm_test01.start()35 self.vm_test01.wait_for_running()36 self.assertTrue(self.vm_test01.wait_for_login(authentication="password"),37 "Fail to ssh login")38 def test_reset_password_ecs(self):39 self.log.info("Reset password for ECS")40 self.vm_test01.reset_password(new_password="Redhat123$")41 self.vm_test01.password="Redhat123$"42 self.vm_params["Password"]="Redhat123$"43 self.vm_test01.restart()44 self.vm_test01.wait_for_running()45 self.assertTrue(self.vm_test01.wait_for_login(authentication="password"),46 "Fail to ssh login")47 def test_create_ecs(self):48 self.log.info("Create ECS with keypair")49 self.vm_test01.create(self.vm_params)50 self.vm_test01.wait_for_created()51 self.vm_test01.allocate_public_address()52 self.assertIsNotNone(self.vm_test01.get_public_address(),53 "Fail to allocate public ip address")54 def test_start_ecs(self):55 self.log.info("Start ECS")56 self.vm_test01.start()57 self.vm_test01.wait_for_running()58 self.assertTrue(self.vm_test01.wait_for_login(),59 "Fail to ssh login")60 def test_stop_ecs(self):61 """62 Stop ECS63 1. Stop ECS64 2. Force stop ECS65 """66 self.log.info("Stop ECS")67 self.log.info("1. Stop ECS")68 self.vm_test01.stop()69 self.vm_test01.wait_for_stopped()70 self.assertFalse(self.vm_test01.wait_for_login(timeout=10),71 "Should not be able to login when ECS is stopped")72 # Recovery: Start VM73 self.vm_test01.start()74 self.vm_test01.wait_for_running()75 self.vm_test01.wait_for_login()76 self.log.info("2. Force stop ECS")77 self.vm_test01.stop(force=True)78 self.vm_test01.wait_for_stopped()79 self.assertFalse(self.vm_test01.wait_for_login(timeout=10),80 "Should not be able to login when ECS is force stopped")81 def test_restart_ecs(self):82 """83 Restart ECS84 1. Restart ECS85 2. Force restart ECS86 """87 self.log.info("Restart ECS")88 self.log.info("1. Restart ECS")89 self.vm_test01.wait_for_login()90 before = self.vm_test01.get_output("who -b")91 time.sleep(60)92 self.vm_test01.restart()93 self.vm_test01.wait_for_running()94 self.vm_test01.wait_for_login()95 after = self.vm_test01.get_output("who -b")96 self.assertNotEqual(before, after,97 "Restart error: ECS is not restarted")98 self.log.info("2. Force restart ECS")99 before = after100 time.sleep(60)101 self.vm_test01.restart(force=True)102 self.vm_test01.wait_for_running()103 self.vm_test01.wait_for_login()104 after = self.vm_test01.get_output("who -b")105 self.assertNotEqual(before, after,106 "Force restart error: ECS is not restarted")107 def test_reboot_inside_vm(self):108 self.log.info("Reboot inside VM")109 self.vm_test01.wait_for_login()110 before = self.vm_test01.get_output("who -b")111 time.sleep(60)112 self.vm_test01.send_line("reboot")113 self.vm_test01.wait_for_running()114 self.vm_test01.wait_for_login()115 after = self.vm_test01.get_output("who -b")116 self.assertNotEqual(before, after,117 "Restart error: ECS is not restarted")118 def test_modify_instance_type(self):119 self.log.info("Modify ECS instance type")120 self.vm_test01.modify_instance_type(new_type="ecs.n1.medium")121 self.vm_test01.start()122 self.vm_test01.wait_for_running()123 self.vm_test01.wait_for_login()124 cpu = self.vm_test01.get_output("cat /proc/cpuinfo|grep processor|wc -l")125 memory = self.vm_test01.get_output("cat /proc/meminfo|grep MemTotal")126 cpu_std = self.params.get("cpu")127 def test_delete_ecs(self):128 self.log.info("Delete ECS")129 self.vm_test01.delete()130 self.vm_test01.wait_for_deleted()131 def tearDown(self):...
helpers.py
Source:helpers.py
...62 stdout=output_pipe,63 stderr=output_pipe,64 text=True,65 )66def wait_for_login(client_process):67 """Wait until the user has logged in"""68 # Read from stdout until the process claims to have logged in69 output = deque(maxlen=1000)70 while client_process.poll() is None and "Connection established" not in (71 line := client_process.stdout.readline()72 ):73 if line:74 output.append(line)75 if client_process.poll() is not None:76 client_output = "\n\t".join(output)77 remaining_stdout, stderr = client_process.communicate()78 stdout = "\n\t\t".join(client_output) + f"\n\t\t{remaining_stdout}"79 raise RuntimeError(80 f"Client process exited with code {client_process.returncode}."81 f"\n\tStdout: {stdout}"82 f"\n\tStderr: {stderr}"83 )84def setup_for_fishing(tmp_path, server, water=True):85 """Log in to receive items and get teleported"""86 client_process = start_client(tmp_path, server)87 wait_for_login(client_process)88 height = 253 if water else 25089 with MCRcon(server.host, server.rcon_password, server.rcon_port) as mcr:90 mcr.command(f"/clear {FISHING_USERNAME}")91 mcr.command(f"/tp {FISHING_USERNAME} 0 {height} 0 -90 0")92 mcr.command(f"/give {FISHING_USERNAME} minecraft:fishing_rod")93 mcr.command(f"/enchant {FISHING_USERNAME} minecraft:lure 3")94 stop_client(client_process)95def stop_client(client_process):96 """Stop the client process and return remaining output"""97 client_process.terminate()98 return client_process.communicate()99@dataclass100class ManagedClient:101 """102 Context manager handling setup and shutdown of fishing clients103 """104 client_dir: Path105 server: Server106 host_override: Optional[dict] = None107 options_override: Optional[dict] = None108 capture_output: bool = True109 process: Optional[subprocess.Popen] = None110 def __enter__(self):111 self.process = start_client(112 client_dir=self.client_dir,113 server=self.server,114 host_override=self.host_override or {},115 options_override=self.options_override or {},116 capture_output=self.capture_output,117 )118 wait_for_login(self.process)119 return self.process120 def __exit__(self, *args, **kwargs):...
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 autotest 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?
