How to use extract_internal method in pyresttest

Best Python code snippet using pyresttest_python

validators.py

Source:validators.py

...140    is_header_extractor = False  # Uses response headers141    args = None142    def __str__(self):143        return "Extractor type: {0}, query: {1}, is_templated: {2}, args: {3}".format(self.extractor_type, self.query, self.is_templated, self.args)144    def extract_internal(self, query=None, body=None, headers=None, args=None):145        """ Do extraction, query should be pre-templated """146        pass147    def extract(self, body=None, headers=None, context=None):148        """ Extract data """149        query = self.templated_query(context=context)150        args = self.args151        return self.extract_internal(query=query, body=body, headers=headers, args=self.args)152    def templated_query(self, context=None):153        query = self.query154        if context and self.is_templated:155            query = string.Template(query).safe_substitute(156                context.get_values())157        return query158    def get_readable_config(self, context=None):159        """ Print a human-readable version of the configuration """160        query = self.templated_query(context=context)161        output = 'Extractor Type: {0},  Query: "{1}", Templated?: {2}'.format(162            self.extractor_type, query, self.is_templated)163        args_string = None164        if self.args:165            args_string = ", Args: " + str(self.args)166            output = output + args_string167        return output168    @classmethod169    def configure_base(cls, config, extractor_base):170        """ Parse config object and do basic config on an Extractor171        """172        if isinstance(config, dict):173            try:174                config = config['template']175                extractor_base.is_templated = True176                extractor_base.query = config177            except KeyError:178                raise ValueError(179                    "Cannot define a dictionary config for abstract extractor without it having template key")180        elif isinstance(config, basestring):181            extractor_base.query = config182            extractor_base.is_templated = False183        else:184            raise TypeError(185                "Base extractor must have a string or {template: querystring} configuration node!")186        return extractor_base187class MiniJsonExtractor(AbstractExtractor):188    """ Extractor that uses jsonpath_mini syntax189        IE key.key or array_index.key extraction190    """191    extractor_type = 'jsonpath_mini'192    is_body_extractor = True193    def extract_internal(self, query=None, args=None, body=None, headers=None):194        if PYTHON_MAJOR_VERSION > 2 and isinstance(body, binary_type):195            body = text_type(body, 'utf-8')  # Default JSON encoding196        try:197            body = json.loads(body)198            return self.query_dictionary(query, body)199        except ValueError:200            raise ValueError("Not legal JSON!")201    @staticmethod202    def query_dictionary(query, dictionary, delimiter='.'):203        """ Do an xpath-like query with dictionary, using a template if relevant """204        # Based on205        # http://stackoverflow.com/questions/7320319/xpath-like-query-for-nested-python-dictionaries206        try:207            stripped_query = query.strip(delimiter)208            if stripped_query:209                for x in stripped_query.split(delimiter):210                    try:211                        x = int(x)212                        dictionary = dictionary[x]213                    except ValueError:214                        dictionary = dictionary[x]215        except:216            return None217        return dictionary218    @classmethod219    def parse(cls, config):220        base = MiniJsonExtractor()221        return cls.configure_base(config, base)222        return base223class HeaderExtractor(AbstractExtractor):224    """ Extractor that pulls out a named header value... or list of values if multiple values defined """225    extractor_type = 'header'226    is_header_extractor = True227    def extract_internal(self, query=None, args=None, body=None, headers=None):228        low = query.lower()229        # Value for all matching key names230        extracted = [y[1] for y in filter(lambda x: x[0] == low, headers)]231        if len(extracted) == 0:232            raise ValueError("Invalid header name {0}".format(query))233        elif len(extracted) == 1:234            return extracted[0]235        else:236            return extracted237    @classmethod238    def parse(cls, config, extractor_base=None):239        base = HeaderExtractor()240        return cls.configure_base(config, base)241class RawBodyExtractor(AbstractExtractor):242    """ Extractor that returns the full request body """243    extractor_type = 'raw_body'244    is_header_extractor = False245    is_body_extractor = True246    def extract_internal(self, query=None, args=None, body=None, headers=None):247        return body248    @classmethod249    def parse(cls, config, extractor_base=None):250        # Doesn't take any real configuration251        base = RawBodyExtractor()252        return base253def _get_extractor(config_dict):254    """ Utility function, get an extract function for a single valid extractor name in config255        and error if more than one or none """256    extractor = None257    extract_config = None258    for key, value in config_dict.items():259        if key in EXTRACTORS:260            return parse_extractor(key, value)...

get_impressionsv2.py

Source:get_impressionsv2.py

...68            shutil.move(d / f, local_dir / f"{f}_ext")69    f = extract_test if split == "test" else extract_dev70    with ThreadPoolExecutor(6) as executor:71        executor.map(f, list(range(1, n_video_zips + 1)))72    def extract_internal(i):73        local_zip = local_dir / f'{split_long}80_{i:02d}.zip'74        shutil.unpack_archive(local_zip, local_dir)75    def extract_internal_test(i):76        _unzip_encrypted(local_dir, f'test80_{i:02d}.zip_ext',77                         b'.chalearnLAPFirstImpressionsSECONDRoundICPRWorkshop2016.')78        extract_internal(i)79    f = extract_internal_test if split == "test" else extract_internal80    with ThreadPoolExecutor(6) as executor:81        executor.map(f, list(range(1, n_internal_video_zips + 1)))82    # clean up83    v_ext = ".mp4"84    p_ext = '.pkl'85    zip_ext = '.zip'86    zipe_ext = '.zip_ext'87    os.makedirs(IMPRESSIONSV2_DIR / split / 'video/', exist_ok=True)88    os.makedirs(IMPRESSIONSV2_DIR / split / 'meta/', exist_ok=True)89    for filename in os.listdir(IMPRESSIONSV2_DIR / split):90        if filename.endswith(v_ext):91            shutil.move(IMPRESSIONSV2_DIR / split / filename, IMPRESSIONSV2_DIR / split / 'video' / filename)92        elif filename.endswith(p_ext):...

test_jmespath_extractor.py

Source:test_jmespath_extractor.py

2from resttest3.ext.extractor_jmespath import JMESPathExtractor3class MTestJMESPathExtractor(unittest.TestCase):4    def setUp(self) -> None:5        self.ext = JMESPathExtractor()6    def test_extract_internal(self):7        unicoded_body = '{"test":"æäºå"}'8        b = bytes('{"test":23}', 'utf-8')9        self.ext.extract_internal('test', unicoded_body, None)10        data = self.ext.extract_internal('test', b)11        self.assertEqual(data, 23)12        self.assertRaises(ValueError, self.ext.extract_internal, 'test', None, 'abc')13if __name__ == '__main__':...

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.