rename package name from ate to httprunner

httprunner/response.py

@@ -0,0 +1,173 @@
+import logging
+import re
+from collections import OrderedDict
+
+from httprunner import exception, utils
+from requests.structures import CaseInsensitiveDict
+
+text_extractor_regexp_compile = re.compile(r".*\(.*\).*")
+
+
+class ResponseObject(object):
+
+    def __init__(self, resp_obj):
+        """ initialize with a requests.Response object
+        @param (requests.Response instance) resp_obj
+        """
+        self.resp_obj = resp_obj
+        self.resp_text = resp_obj.text
+        self.resp_body = self.parsed_body()
+
+    def parsed_body(self):
+        try:
+            return self.resp_obj.json()
+        except ValueError:
+            return self.resp_text
+
+    def parsed_dict(self):
+        return {
+            'status_code': self.resp_obj.status_code,
+            'headers': self.resp_obj.headers,
+            'body': self.resp_body
+        }
+
+    def _extract_field_with_regex(self, field):
+        """ extract field from response content with regex.
+            requests.Response body could be json or html text.
+        @param (str) field should only be regex string that matched r".*\(.*\).*"
+        e.g.
+            self.resp_text: "LB123abcRB789"
+            field: "LB[\d]*(.*)RB[\d]*"
+            return: abc
+        """
+        matched = re.search(field, self.resp_text)
+        if not matched:
+            err_msg = u"Extractor error: failed to extract data with regex!\n"
+            err_msg += u"response body: {}\n".format(self.resp_text)
+            err_msg += u"regex: {}\n".format(field)
+            logging.error(err_msg)
+            raise exception.ParamsError(err_msg)
+
+        return matched.group(1)
+
+    def _extract_field_with_delimiter(self, field):
+        """ response content could be json or html text.
+        @param (str) field should be string joined by delimiter.
+        e.g.
+            "status_code"
+            "content"
+            "headers.content-type"
+            "content.person.name.first_name"
+        """
+        try:
+            # string.split(sep=None, maxsplit=-1) -> list of strings
+            # e.g. "content.person.name" => ["content", "person.name"]
+            try:
+                top_query, sub_query = field.split('.', 1)
+            except ValueError:
+                top_query = field
+                sub_query = None
+
+            if top_query in ["body", "content", "text"]:
+                top_query_content = self.parsed_body()
+            else:
+                top_query_content = getattr(self.resp_obj, top_query)
+
+            if sub_query:
+                if not isinstance(top_query_content, (dict, CaseInsensitiveDict, list)):
+                    err_msg = u"Extractor error: failed to extract data with regex!\n"
+                    err_msg += u"response: {}\n".format(self.parsed_dict())
+                    err_msg += u"regex: {}\n".format(field)
+                    logging.error(err_msg)
+                    raise exception.ParamsError(err_msg)
+
+                # e.g. key: resp_headers_content_type, sub_query = "content-type"
+                return utils.query_json(top_query_content, sub_query)
+            else:
+                # e.g. key: resp_status_code, resp_content
+                return top_query_content
+
+        except AttributeError:
+            err_msg = u"Failed to extract value from response!\n"
+            err_msg += u"response: {}\n".format(self.parsed_dict())
+            err_msg += u"extract field: {}\n".format(field)
+            logging.error(err_msg)
+            raise exception.ParamsError(err_msg)
+
+    def extract_field(self, field):
+        """ extract value from requests.Response.
+        """
+        if text_extractor_regexp_compile.match(field):
+            return self._extract_field_with_regex(field)
+        else:
+            return self._extract_field_with_delimiter(field)
+
+    def extract_response(self, extractors):
+        """ extract value from requests.Response and store in OrderedDict.
+        @param (list) extractors
+            [
+                {"resp_status_code": "status_code"},
+                {"resp_headers_content_type": "headers.content-type"},
+                {"resp_content": "content"},
+                {"resp_content_person_first_name": "content.person.name.first_name"}
+            ]
+        @return (OrderDict) variable binds ordered dict
+        """
+        extracted_variables_mapping = OrderedDict()
+        extract_binds_order_dict = utils.convert_to_order_dict(extractors)
+
+        for key, field in extract_binds_order_dict.items():
+            if not isinstance(field, utils.string_type):
+                raise exception.ParamsError("invalid extractors in testcase!")
+
+            extracted_variables_mapping[key] = self.extract_field(field)
+
+        return extracted_variables_mapping
+
+    def validate(self, validators, variables_mapping):
+        """ Bind named validators to value within the context.
+        @param (list) validators
+            [
+                {"check": "status_code", "comparator": "eq", "expected": 201},
+                {"check": "resp_body_success", "comparator": "eq", "expected": True}
+            ]
+        @param (dict) variables_mapping
+            {
+                "resp_body_success": True
+            }
+        @return (list) content differences
+            [
+                {
+                    "check": "status_code",
+                    "comparator": "eq", "expected": 201, "value": 200
+                }
+            ]
+        """
+        for validator_dict in validators:
+
+            check_item = validator_dict.get("check")
+            if not check_item:
+                raise exception.ParamsError("invalid check item in testcase validators!")
+
+            if "expected" not in validator_dict:
+                raise exception.ParamsError("expected item missed in testcase validators!")
+
+            expected = validator_dict.get("expected")
+            comparator = validator_dict.get("comparator", "eq")
+
+            if check_item in variables_mapping:
+                validator_dict["actual_value"] = variables_mapping[check_item]
+            else:
+                try:
+                    validator_dict["actual_value"] = self.extract_field(check_item)
+                except exception.ParseResponseError:
+                    raise exception.ParseResponseError("failed to extract check item in response!")
+
+            utils.match_expected(
+                validator_dict["actual_value"],
+                expected,
+                comparator,
+                check_item
+            )
+
+        return True