Python hypothesis.strategies 模块，fixed_dictionaries() 实例源码

def spark_application(app_id):
    """Mock of the Spark jobs REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'jobId': st.integers(0),
        'name': st.text(),
        'submissionTime': st.text(),
        'completionTime': st.text(),
        'stageIds': st.lists(st.integers(0), average_size=3),
        'status': st.sampled_from(['SUCCEEDED', 'RUNNING', 'FAILED']),
        'numTasks': st.integers(0),
        'numActiveTasks': st.integers(0),
        'numCompletedTasks': st.integers(0),
        'numSkippedTasks': st.integers(0),
        'numFailedTasks': st.integers(0),
        'numActiveStages': st.integers(0),
        'numCompletedStages': st.integers(0),
        'numSkippedStages': st.integers(0),
        'numFailedStages': st.integers(0),
    })
    result = json.dumps(st.lists(d, average_size=3).example())
    redis.set(request.base_url, result)
    return jsonify(result)

def default(self, obj):
        """
        """
        if isinstance(obj, dict) and _is_swagger_parameter(obj):
            parameter_type = obj.get('format', obj.get('type'))
            parameter_schema = obj.get('schema')
            parameter_ref = obj.get('$ref')

            if parameter_type in SWAGGER_FORMAT_MAPPING:
                return SWAGGER_FORMAT_MAPPING[parameter_type]
            elif parameter_ref:
                return self.transform(self.get_ref(parameter_ref, self.spec))
            elif parameter_type == 'array':
                if obj['items'].get('enum'):
                    return st.lists(elements=st.sampled_from(obj['items']['enum']))
                elif obj['items'].get('type'):
                    return st.lists(elements=SWAGGER_FORMAT_MAPPING[obj['items']['type']])
                elif obj['items'].get('$ref'):
                    schema = self.get_ref(obj['items']['$ref'], self.spec)
                    return st.lists(elements=self.transform(schema))
                raise Exception('array', obj)
            elif parameter_type == 'object':
                properties = {}
                for property_name, property_ in obj['properties'].items():
                    properties[property_name] = self.transform(property_)
                return st.fixed_dictionaries(properties)
            elif parameter_schema:
                if parameter_schema.get('type') == 'array':
                    schema = self.get_ref(parameter_schema['items']['$ref'], self.spec)
                    return st.lists(elements=self.transform(schema))
                else:
                    schema = self.get_ref(parameter_schema['$ref'], self.spec)
                    transformed = self.transform(schema)
                    return transformed
            else:
                raise Exception("Invalid", obj, parameter_type)

        return obj

def metrics():
    """Mock of the YARN cluster metrics REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'activeNodes': st.integers(0),
        'allocatedMB': st.integers(0),
        'allocatedVirtualCores': st.integers(0),
        'appsCompleted': st.integers(0),
        'appsFailed': st.integers(0),
        'appsKilled': st.integers(0),
        'appsPending': st.integers(0),
        'appsRunning': st.integers(0),
        'appsSubmitted': st.integers(0),
        'availableMB': st.integers(0),
        'availableVirtualCores': st.integers(0),
        'containersAllocated': st.integers(0),
        'containersPending': st.integers(0),
        'containersReserved': st.integers(0),
        'decommissionedNodes': st.integers(0),
        'lostNodes': st.integers(0),
        'rebootedNodes': st.integers(0),
        'reservedMB': st.integers(0),
        'reservedVirtualCores': st.integers(0),
        'totalMB': st.integers(0),
        'totalNodes': st.integers(0),
        'totalVirtualCores': st.integers(0),
        'unhealthyNodes': st.integers(0)
    })
    result = json.dumps({
        'clusterMetrics': d.example()
    })
    redis.set(request.base_url, result)
    return jsonify(result)

def get_request(data, spec, spec_host):
    endpoint_path = data.draw(st.sampled_from(spec['paths'].keys()))
    endpoint = spec['paths'][endpoint_path]

    method_name = data.draw(st.sampled_from(endpoint.keys()))
    endpoint = endpoint[method_name]

    path_params = _get_filtered_parameter(endpoint, 'path', spec)
    path_args = data.draw(st.fixed_dictionaries(path_params))

    query_params = _get_filtered_parameter(endpoint, 'query', spec)
    query_args = data.draw(st.fixed_dictionaries(query_params))

    body_params = _get_filtered_parameter(endpoint, 'body', spec)
    if body_params:
        body_args = data.draw(body_params['body'])
    else:
        body_args = None

    valid_request_body_format = get_item_path_acceptable_format(endpoint, spec)

    request_data = None
    request_headers = {}

    if body_args:
        # no_body_format_declaration(body_args, valid_request_body_format, endpoint)
        if body_args and valid_request_body_format is None:
            # Force a request format, swagger ui seems to force json format
            valid_request_body_format = ["application/json"]

        request_body_format = data.draw(st.sampled_from(valid_request_body_format), 'request_body_format')

        request_headers['Content-Type'] = request_body_format
        if request_body_format == 'application/x-www-form-urlencoded':
            request_data = body_args
        elif request_body_format == 'application/json':
            request_data = json.dumps(body_args, cls=CustomJsonEncoder)
        elif request_body_format == 'application/xml':
            pass
            # TODO Implement XML
        else:
            raise Exception(request_body_format)

    endpoint_url = endpoint_path.format(**path_args)
    assume('\x00' not in endpoint_url)

    # Generate request
    URL = furl(spec_host)
    URL = URL.join(endpoint_url.lstrip('/'))

    if query_args:
        URL = URL.add(args=query_args)

    request = Request(method_name, URL.url, data=request_data,
                      headers=request_headers).prepare()
    request.build_context = locals()
    return request

def applications():
    """Mock of the YARN cluster apps REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'allocatedMB': st.integers(-1),
        'allocatedVCores': st.integers(-1),
        'amContainerLogs': st.text(),
        'amHostHttpAddress': st.text(),
        'applicationTags': st.text(),
        'applicationType': st.sampled_from(['MAPREDUCE', 'SPARK']),
        'clusterId': st.integers(0),
        'diagnostics': st.text(),
        'elapsedTime': st.integers(0),
        'finalStatus': st.sampled_from(['UNDEFINED', 'SUCCEEDED', 'FAILED', 'KILLED']),
        'finishedTime': st.integers(0),
        'id': st.text(string.ascii_letters, min_size=5, max_size=25),
        'memorySeconds': st.integers(0),
        'name': st.text(min_size=5),
        'numAMContainerPreempted': st.integers(0),
        'numNonAMContainerPreempted': st.integers(0),
        'preemptedResourceMB': st.integers(0),
        'preemptedResourceVCores': st.integers(0),
        'progress': st.floats(0, 100),
        'queue': st.text(),
        'runningContainers': st.integers(-1),
        'startedTime': st.integers(0),
        'state': st.sampled_from(['NEW', 'NEW_SAVING', 'SUBMITTED', 'ACCEPTED', 'RUNNING', 'FINISHED', 'FAILED', 'KILLED']),
        'trackingUI': st.text(),
        'trackingUrl': st.just(os.environ['YARN_ENDPOINT']),
        'user': st.text(),
        'vcoreSeconds': st.integers(0)
    })

    result = json.dumps({
        'apps': {
            'app': st.lists(d, min_size=4, average_size=10).example()
        }
    })
    redis.set(request.base_url, result)
    return jsonify(result)

def mapreduce_application():
    """Mock of the mapreduce jobs REST resource."""
    if 'last' in request.args:
        return jsonify(redis.get(request.base_url))

    d = st.fixed_dictionaries({
        'startTime': st.integers(0),
        'finishTime': st.integers(0),
        'elapsedTime': st.integers(0),
        'id': st.integers(0),
        'name': st.text(),
        'user': st.text(),
        'state': st.sampled_from(['NEW', 'SUCCEEDED', 'RUNNING', 'FAILED', 'KILLED']),
        'mapsTotal': st.integers(0),
        'mapsCompleted': st.integers(0),
        'reducesTotal': st.integers(0),
        'reducesCompleted': st.integers(0),
        'mapProgress': st.floats(0, 100),
        'reduceProgress': st.floats(0, 100),
        'mapsPending': st.integers(0),
        'mapsRunning': st.integers(0),
        'reducesPending': st.integers(0),
        'reducesRunning': st.integers(0),
        'uberized': st.booleans(),
        'diagnostics': st.text(),
        'newReduceAttempts': st.integers(0),
        'runningReduceAttempts': st.integers(0),
        'failedReduceAttempts': st.integers(0),
        'killedReduceAttempts': st.integers(0),
        'successfulReduceAttempts': st.integers(0),
        'newMapAttempts': st.integers(0),
        'runningMapAttempts': st.integers(0),
        'failedMapAttempts': st.integers(0),
        'killedMapAttempts': st.integers(0),
        'successfulMapAttempts': st.integers(0)
    })
    result = json.dumps({
        'jobs': {
            'job': st.lists(d, average_size=3).example()
        }
    })
    redis.set(request.base_url, result)
    return jsonify(result)