KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _get_opener method in tempest

Best Python code snippet using tempest_python

core.py

Source:core.py

...67		"""Deterministic checks for consistiency."""68		if not self.url:69			raise URLRequired70		71	def _get_opener(self):72		""" Creates appropriate opener object for urllib2.73		"""74		75		if self.auth:76			# create a password manager77			authr = urllib2.HTTPPasswordMgrWithDefaultRealm()78			authr.add_password(None, self.url, self.auth.username, self.auth.password)79			handler = urllib2.HTTPBasicAuthHandler(authr)80			opener = urllib2.build_opener(handler)81			# use the opener to fetch a URL82			return opener.open83		else:84			return urllib2.urlopen85	86	def send(self, anyway=False):87		"""Sends the request. Returns True of successfull, false if not.88		    If there was an HTTPError during transmission,89		    self.response.status_code will contain the HTTPError code.90		    Once a request is successfully sent, `sent` will equal True.91		92		    :param anyway: If True, request will be sent, even if it has93		    already been sent.94		"""95		self._checks()96		success = False97		98		if self.method in ('GET', 'HEAD', 'DELETE'):99			if (not self.sent) or anyway:100				# url encode GET params if it's a dict101				if isinstance(self.params, dict):102					params = urllib.urlencode(self.params)103				else:104					params = self.params105				req = _Request(("%s?%s" % (self.url, params)), method=self.method)106				if self.headers:107					req.headers = self.headers108				opener = self._get_opener()109				try:110					resp = opener(req)111					self.response.status_code = resp.code112					self.response.headers = resp.info().dict113					if self.method.lower() == 'get':114						self.response.content = resp.read()115					success = True116				except urllib2.HTTPError, why:117					self.response.status_code = why.code118		elif self.method == 'PUT':119			if (not self.sent) or anyway:120				req = _Request(self.url, method='PUT')121				if self.headers:122					req.headers = self.headers123				req.data = self.data124				try:125					opener = self._get_opener()126					resp =  opener(req)127					self.response.status_code = resp.code128					self.response.headers = resp.info().dict129					self.response.content = resp.read()130					success = True131				except urllib2.HTTPError, why:132					self.response.status_code = why.code133		elif self.method == 'POST':134			if (not self.sent) or anyway:135				req = _Request(self.url, method='POST')136				if self.headers:137					req.headers = self.headers138				# url encode form data if it's a dict139				if isinstance(self.data, dict):140					req.data = urllib.urlencode(self.data)141				else:142					req.data = self.data143				try:144					opener = self._get_opener()145					resp =  opener(req)146					self.response.status_code = resp.code147					self.response.headers = resp.info().dict148					self.response.content = resp.read()149					success = True150				except urllib2.HTTPError, why:151					self.response.status_code = why.code152		153		self.sent = True if success else False154		155		return success156		157class Response(object):158	"""The :class:`Request` object. All :class:`Request` objects contain a...

test_dashboard_basic_ops.py

Source:test_dashboard_basic_ops.py

...63    def setup_credentials(cls):64        cls.set_network_resources()65        super(TestDashboardBasicOps, cls).setup_credentials()66    def check_login_page(self):67        response = self._get_opener().open(CONF.dashboard.dashboard_url).read()68        self.assertIn("id_username", response.decode("utf-8"))69    def user_login(self, username, password):70        response = self._get_opener().open(CONF.dashboard.dashboard_url).read()71        # Grab the CSRF token and default region72        parser = HorizonHTMLParser()73        parser.feed(response.decode("utf-8"))74        # construct login url for dashboard, discovery accommodates non-/ web75        # root for dashboard76        login_url = parse.urljoin(CONF.dashboard.dashboard_url, parser.login)77        # Prepare login form request78        req = request.Request(login_url)79        req.add_header('Content-type', 'application/x-www-form-urlencoded')80        req.add_header('Referer', CONF.dashboard.dashboard_url)81        # Pass the default domain name regardless of the auth version in order82        # to test the scenario of when horizon is running with keystone v383        params = {'username': username,84                  'password': password,85                  'region': parser.region,86                  'domain': CONF.auth.default_credentials_domain_name,87                  'csrfmiddlewaretoken': parser.csrf_token}88        self._get_opener().open(req, parse.urlencode(params).encode())89    def check_home_page(self):90        response = self._get_opener().open(CONF.dashboard.dashboard_url).read()91        self.assertIn('Overview', response.decode("utf-8"))92    def _get_opener(self):93        if not self.opener:94            if (CONF.dashboard.disable_ssl_certificate_validation and95                    self._ssl_default_context_supported()):96                ctx = ssl.create_default_context()97                ctx.check_hostname = False98                ctx.verify_mode = ssl.CERT_NONE99                self.opener = request.build_opener(100                    request.HTTPSHandler(context=ctx),101                    request.HTTPCookieProcessor())102            else:103                self.opener = request.build_opener(104                    request.HTTPCookieProcessor())105        return self.opener106    def _ssl_default_context_supported(self):...

jsonlines.py

Source:jsonlines.py

...7	return any(8		str(path).endswith(ext) 9		for ext in _GZIP_EXTENSIONS10	)11def _get_opener(12	path: str, 13	use_gzip: bool|None = None,14) -> Callable:15	if use_gzip is None:16		use_gzip = _file_is_gzip(path)17	return open if not use_gzip else gzip.open18def jsonl_load(19		path: str, 20		/,*,21		use_gzip: bool|None = None,22	) -> list[JSONitem]:23	opener: Callable = _get_opener(path, use_gzip)24	data: list[JSONitem] = list()	25	with opener(path, 'rt', encoding="UTF-8") as f:26		for line in f:27			data.append(json.loads(line))28	29	return data30def jsonl_write(31		path: str, 32		items: Sequence[JSONitem],33		use_gzip: bool|None = None,34		gzip_compresslevel: int = 2,35	) -> None:36	opener: Callable = _get_opener(path, use_gzip)37	opener_kwargs: dict = dict()38	if use_gzip:39		opener_kwargs = dict(compresslevel = gzip_compresslevel)40	with opener(path, 'wt', encoding="UTF-8", **opener_kwargs) as f:41		for item in items:42			f.write(json.dumps(item) + '\n')...

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.