Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _resolve_url method in robotframework-pageobjects

Best Python code snippet using robotframework-pageobjects_python

__init__.py

Source:__init__.py

...143                           signature_type='query')144        else:145            oauth = None146        if send_json:147            response = self._session.request(method, self._resolve_url(url), headers=headers,148                                             json=data, auth=oauth, timeout=self.timeout)149        else:150            response = self._session.request(method, self._resolve_url(url), headers=headers,151                                             data=data, auth=oauth, timeout=self.timeout)152        self.log_request(response)153        response.raise_for_status()154        return response155156    def fetch_request_token(self):157        """Helper method to fetch and set request token.158159        Returns token string.160        """161        oauth = OAuth1(client_key=self.consumer_key,162                       client_secret=self.consumer_secret,163                       callback_uri=self.callback_url,164                       signature_type='auth_header')165        response = self._session.post(url=self.request_token_url, auth=oauth, timeout=self.timeout)166        self.log_request(response)167        if response.status_code != 200:168            raise OAuthException("OAuth token request failed (%s) %s" % (response.status_code, response.text))169        credentials = parse_qs(response.text)170        self._token = {'key': credentials['oauth_token'][0],171                       'secret': credentials['oauth_token_secret'][0]}172        return self._token173174    def authorize_token(self, email=None, password=None):175        """Helper method to authorize."""176        # Give precedence to credentials passed in methods calls over those set177        # in the instance. This allows you to override user creds that may have178        # been loaded from a file.179        if not email:180            email = self._email181182        if not password:183            password = self._password184185        if not email or not password:186            raise Exception('Missing email or password')187188        data = {189            'email': email,190            'password': password,191            'oauth_token': self._token['key']}192193        response = self._session.post(url=self.authorization_url, data=data, timeout=self.timeout)194        self.log_request(response)195        if response.status_code != 200:196            raise OAuthException("OAuth login failed (%s) %s" % (response.status_code, response.text))197198        # set token verifier199        self._token['verifier'] = parse_qs(response.text)['oauth_verifier'][0]200201    def fetch_access_token(self):202        """Helper method to fetch and set access token.203204        Returns token string.205        """206        oauth = OAuth1(client_key=self.consumer_key,207                       client_secret=self.consumer_secret,208                       resource_owner_key=self._token['key'],209                       resource_owner_secret=self._token['secret'],210                       verifier=self._token['verifier'],211                       callback_uri=self.callback_url,212                       signature_type='auth_header')213        response = self._session.post(url=self.access_token_url, auth=oauth, timeout=self.timeout)214        self.log_request(response)215        if response.status_code != 200:216            raise OAuthException("OAuth token verification failed (%s) %s" % (response.status_code, response.text))217        self._token = parse_qs(response.text)['oauth_token'][0]218        return self._token219220    def validate_session(self):221        """Validate an API session."""222223        # We need to store our access token as the openx3_access_token cookie.224        # This cookie will be passed to all future API requests.225        cookie = cookielib.Cookie(226            version=0,227            name='openx3_access_token',228            value=self._token,229            port=None,230            port_specified=False,231            domain=self.domain,232            domain_specified=True,233            domain_initial_dot=False,234            path='/',235            path_specified=True,236            secure=False,237            expires=None,238            discard=False,239            comment=None,240            comment_url=None,241            rest={})242        self._session.cookies.set_cookie(cookie)243244        # v2 doesn't need this extra step, just the cookie:245        if self.api_path == API_PATH_V1:246            response = self._session.put(url=self._resolve_url('/a/session/validate'), timeout=self.timeout)247            self.log_request(response)248            return response.text249250    def logon(self, email=None, password=None):251        """Returns self after authentication.252253        Single call to complete OAuth login process.254255        Keyword arguments:256        email -- user email address.257        password -- user password.258259        """260        self.fetch_request_token()261        self.authorize_token(email=email, password=password)262        self.fetch_access_token()263        self.validate_session()264        return self265266    def logoff(self):267        """Returns self after deleting authenticated session."""268        if self.api_path == API_PATH_V1:269            response = self._session.delete(self._resolve_url('/a/session'), timeout=self.timeout)270        elif self.api_path == API_PATH_V2:271            response = self._session.delete(self._resolve_url('/session'), timeout=self.timeout)272        elif self.api_path == API_PATH_SSO:273            oauth = OAuth1(client_key=self.consumer_key,274                           resource_owner_key=self._token,275                           callback_uri=self.callback_url,276                           signature_type='query')277278            response = self._session.delete(url=self.access_token_url, auth=oauth, timeout=self.timeout)279            if response.status_code != 204:280                raise OAuthException("OAuth token deletion failed (%s) %s" % (response.status_code, response.text))281        else:282            raise UnknownAPIFormatError(283                'Unrecognized API path: %s' % self.api_path)284        self.log_request(response)285        return self286287    def _resolve_url(self, url):288        """Converts an API path shorthand into a full URL unless289        given a full url already.290291        """292        parse_res = urlparse(url)293294        # 2.4 returns a tuple instead of ParseResult. Since ParseResult is a295        # subclass or tuple we can access URL components similarly across296        # 2.4 - 2.7. Yay!297298        # If there is no scheme specified we create a fully qualified URL.299        if not parse_res[0]:300            url = '%s://%s%s%s' % (self.scheme, self.domain, self.api_path,301                                   parse_res[2])302            if parse_res[4]:303                url = url + '?' + parse_res[4]304305        return url306307    def _response_value(self, response):308        """ Utility method. Returns decoded json. If the response content cannot be decoded, then309        the text is returned.310311        """312        try:313            return response.json()314        except ValueError:315            return response.text316317    def get(self, url):318        """Issue a GET request to the given URL or API shorthand319320        """321        response = self._session.get(self._resolve_url(url), timeout=self.timeout)322        self.log_request(response)323        response.raise_for_status()324        return self._response_value(response)325326    def options(self, url):327        """Send a request with HTTP method OPTIONS to the given328        URL or API shorthand.329330        OX3 v2 uses this method for showing help information.331332        """333        response = self._session.options(self._resolve_url(url), timeout=self.timeout)334        self.log_request(response)335        response.raise_for_status()336        return self._response_value(response)337338    def put(self, url, data=None):339        """Issue a PUT request to url (either a full URL or API340        shorthand) with the data.341342        """343        if self.api_path in JSON_PATHS:344            response = self._session.put(self._resolve_url(url), data=json.dumps(data), timeout=self.timeout)345        else:346            response = self._session.put(self._resolve_url(url), data=data, timeout=self.timeout)347        self.log_request(response)348        response.raise_for_status()349        return self._response_value(response)350351    def post(self, url, data=None):352        """Issue a POST request to url (either a full URL or API353        shorthand) with the data.354355        """356        if self.api_path in JSON_PATHS:357            response = self._session.post(self._resolve_url(url), data=json.dumps(data), timeout=self.timeout)358        else:359            response = self._session.post(self._resolve_url(url), data=data, timeout=self.timeout)360        self.log_request(response)361        response.raise_for_status()362        return self._response_value(response)363364    def delete(self, url):365        """Issue a DELETE request to the URL or API shorthand."""366        response = self._session.delete(self._resolve_url(url))367        self.log_request(response)368        response.raise_for_status()369        # Catch no content responses from some delete actions.370        if response.status_code == 204:371            return []372        return self._response_value(response)373374    def upload_creative(self, account_id, file_path):375        """Upload a media creative to the account with ID376        account_id from the local file_path.377378        """379        # Thanks to nosklo for his answer on SO:380        # http://stackoverflow.com/a/681182381        boundary = '-----------------------------' + str(int(random.random() * 1e10))382        parts = []383384        # Set account ID part.385        parts.append('--' + boundary)386        parts.append('Content-Disposition: form-data; name="account_id"')387        parts.append('')388        parts.append(str(account_id))389390        # Set creative contents part.391        parts.append('--' + boundary)392        parts.append('Content-Disposition: form-data; name="userfile"; filename="%s"' % file_path)393        parts.append('Content-Type: %s' % mimetypes.guess_type(file_path)[0] or 'application/octet-stream')394        parts.append('')395        # TODO: catch errors with opening file.396        with open(file_path, 'r') as f:397            parts.append(f.read())398399        parts.append('--' + boundary + '--')400        parts.append('')401402        body = '\r\n'.join(parts)403404        # TODO: Catch errors in attempt to upload.405        headers = {'content-type': 'multipart/form-data; boundary=' + boundary}406        if self.api_path == API_PATH_V1:407            url = self._resolve_url('/a/creative/uploadcreative')408        elif self.api_path == API_PATH_V2:409            url = self._resolve_url('/creative/uploadcreative')410        else:411            raise UnknownAPIFormatError(412                'Unrecognized API path: %s' % self.api_path)413        response = self._session.get(url, headers=headers, data=body, timeout=self.timeout)414        self.log_request(response)415        response.raise_for_status()416        return self._response_value(response)417418419def client_from_file(file_path='.ox3rc', env=None):420    """Return an instance of ox3apiclient.Client with data from file_path.421422    Keyword arguments:423    file_path -- the file to load. Default is '.ox3rc' form current dir.
...

test_API_Client.py

Source:test_API_Client.py

...16    def tearDownClass(cls) -> None:17        cls.api_server.stop_server()18    def setUp(self):19        self.client = API_Client(url_server=self.url_server)20    def test__resolve_url(self):21        assert self.client._resolve_url('aaa' ) == f"{self.url_server}/aaa"22        assert self.client._resolve_url(''    ) == f"{self.url_server}"23        assert self.client._resolve_url(None  ) == f"{self.url_server}"24        assert self.client._resolve_url(      ) == f"{self.url_server}"25        assert self.client._resolve_url('/a'  ) == f"{self.url_server}/a"26        assert self.client._resolve_url('/a/b') == f"{self.url_server}/a/b"27        assert self.client._resolve_url('a/b' ) == f"{self.url_server}/a/b"28        self.client.server_ip += '/'29        assert self.client.server_ip == self.url_server + '/'30        assert self.client._resolve_url('aaa') == f"{self.url_server}/aaa"31        assert self.client._resolve_url(     ) == f"{self.url_server}/"32        assert self.client._resolve_url(''   ) == f"{self.url_server}/"33        assert self.client._resolve_url('///') == f"{self.url_server}/"34        self.client.server_ip += '/'35        assert self.client.server_ip == self.url_server + '//'                      # note: cases when two // in the server_ip cause some side effects36        assert self.client._resolve_url('aaa') == f"{self.url_server}/aaa"          # ok here37        assert self.client._resolve_url(     ) == f"{self.url_server}//"            # not ok here38        assert self.client._resolve_url(''   ) == f"{self.url_server}//"            # and here39        self.client.server_ip += '/123'40        assert self.client.server_ip == self.url_server + '///123'                  # when there is an extra segment41        assert self.client._resolve_url('aaa') == f"{self.url_server}/aaa"          # it is removed here42        assert self.client._resolve_url(''   ) == f"{self.url_server}///123"        # but it shows up here43    def test_health(self):44        result = self.client.health()45        assert result['status'] == 'ok'46    # def test_file_distributor_hd1(self):47    #     num_of_files = 148    #     result = self.client.file_distributor_hd1(num_of_files)49    #     pprint(result)50    def test_version(self):51        result = self.client.version()52        assert result['version'] == API_VERSION53    def test_configure_environment(self):54        data = { "hd1_path"  : "./test_data/scenario-1/hd1",55                 "hd2_path"  : "./test_data/scenario-1/hd1",56                 "hd3_path"  : "./test_data/scenario-1/hd1"}...

base.py

Source:base.py

...7    Base client for accessing web-based APIs.8    """9    def __init__(self, base_url: str):10        self.base_url = base_url11    def _resolve_url(self, url: str) -> str:12        if url.startswith("/"):13            url = self.base_url + url14        return url15    def _head(self, url: str, **kwargs) -> requests.Response:16        url = self._resolve_url(url)17        response = requests.head(url, **kwargs)18        response.raise_for_status()19        return response20    def _get(self, url: str, **kwargs) -> requests.Response:21        url = self._resolve_url(url)22        response = requests.get(url, **kwargs)23        response.raise_for_status()24        return response25    def _post(26        self, url: str, form: dict = None, json: dict = None, buff=None, **kwargs27    ) -> requests.Response:28        url = self._resolve_url(url)29        response = None30        if form:31            response = requests.post(url, data=form, **kwargs)32        elif buff:33            response = requests.post(url, data=buff, **kwargs)34        elif json:35            response = requests.post(url, json=json, **kwargs)36        else:37            response = requests.post(url, **kwargs)38        response.raise_for_status()39        return response40    def _put(41        self, url: str, form: dict = None, json: dict = None, buff=None, **kwargs42    ) -> requests.Response:43        url = self._resolve_url(url)44        response = None45        if form:46            response = requests.put(url, data=form, **kwargs)47        elif buff:48            response = requests.put(url, data=buff, **kwargs)49        elif json:50            response = requests.put(url, json=json, **kwargs)51        else:52            response = requests.put(url, **kwargs)53        response.raise_for_status()54        return response55    def _delete(self, url: str, **kwargs) -> requests.Response:56        url = self._resolve_url(url)57        response = requests.delete(url, **kwargs)58        response.raise_for_status()...

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.