test.py 文件源码

def _run(self, args, working_dir, output_filename, request_json_builder = None, live_request_json_builder = None):
        '''
        Runs the test given the command line arguments, and the output filename.
        :type args: list(str) representing the components of the command line argument.
        :c output_filename: str represents the filename of the file to write the command line output to.
        '''
        self.logger.info("%s" % (' '.join(args)))
        self._reset_output_file(output_filename)

        def match_um_requests_on(r1, r2):
            '''
            Custom uri matcher for use with vcrpy. Basically it provides custom matching for user and action UM
            requests, which ignores the org ID portion of the request. Otherwise, the request uri must match exactly.
            :param r1: 
            :param r2: 
            :return: 
            '''
            if re.match(self.test_suite_config['users_request_matcher'], r1.uri):
                if re.match(self.test_suite_config['users_request_matcher'], r2.uri):
                    return True

            if re.match(self.test_suite_config['actions_request_matcher'], r2.uri):
                if re.match(self.test_suite_config['actions_request_matcher'], r2.uri):
                    return True

            return r1.uri == r2.uri

        record_mode = 'all' if self.server_config['live_mode'] else 'none'
        mock_spec = 'proxy' if self.server_config['live_mode'] else 'playback'
        recorder = vcr.VCR(
            record_mode=record_mode,
            match_on=['um_request'],
            # match_on=match_um_requests_on,
            decode_compressed_response=True,
            filter_headers=['authorization']
        )
        recorder.register_matcher('um_request',match_um_requests_on)

        with recorder.use_cassette(self.server_config['cassette_filename']) as cassette:
            service = server.TestService(self.server_config, cassette, request_json_builder)
            service.run()

            with open(output_filename, 'w') as output_file:
                subprocess.call(args,
                                cwd=working_dir,
                                env=dict(os.environ, UMAPI_MOCK=mock_spec),
                                shell=IS_NT_PLATFORM,
                                stdin=None,
                                stdout=output_file,
                                stderr=output_file)
                # p = subprocess.Popen(args, cwd=working_dir, stdout=output_file, stderr=output_file)
                # output_bytes = subprocess.check_output(cmd, cwd=working_dir, shell=True)
                # output_file.write(output_bytes.decode())
                # p.communicate()

            service.stop()

            if live_request_json_builder:
                for stored_request, stored_response in cassette.data:
                    if stored_request.body:
                        live_request_json_builder.extend_with_json_string(stored_request.body)