def _do(self, **parents):
if self.dist is not None:
k = self.dist(1)[0]
if k == 0:
k = 1
else:
k = self.n
recs = []
@settings(max_examples=k)
def gen(**kwargs):
rels = {}
for rv in self.relative_values:
rels.update({name: xform(**kwargs, **parents, **self.fixtures, **rels, **self.extras) for name, xform in rv.items()})
recs.append(self.create(self.model, **kwargs, **parents, **self.fixtures, **rels))
if self.strategy:
given(**self.strategy)(gen)()
else:
gen()
self.db.session.commit()
return recs
python类given()的实例源码
def test_covering_cidr(ips):
"""
covering_cidr() gets the minimal CIDR that covers given IPs.
In particular, that means any subnets should *not* cover all given IPs.
"""
cidr = telepresence.vpn.covering_cidr(ips)
assert isinstance(cidr, str)
cidr = ipaddress.IPv4Network(cidr)
assert cidr.prefixlen <= 24
# All IPs in given CIDR:
ips = [ipaddress.IPv4Address(i) for i in ips]
assert all([ip in cidr for ip in ips])
# Subnets do not contain all IPs if we're not in minimum 24 bit CIDR:
if cidr.prefixlen < 24:
for subnet in cidr.subnets():
assert not all([ip in subnet for ip in ips])
def test_log_result_too_low_level(self):
"""
A LoggedMessageId matching the logged message is returned by logging APIs
even when the given level is low enough that a message wasn't sent to
the MCP.
"""
mdk, tracer = create_mdk_with_faketracer()
session = mdk.session()
session.info("cat", "message")
lmid = session.debug("cat", "another message")
lmid2 = session.info("cat", "message")
# Debug message wasn't set:
self.assertEqual([d["level"] for d in tracer.messages],
["INFO", "INFO"])
# But we still got LoggedMessageId for debug message:
self.assertEqual((lmid.causalLevel, lmid.traceId,
lmid2.causalLevel, lmid2.traceId),
([2], session._context.traceId,
[3], session._context.traceId))
def test_creation_of_particular_halfspaces(self):
"""Test that can create a given halfspace array representation"""
A = np.array([[3, 3], [2, 5], [0, 6]])
expected_halfspace = np.array([[ 3., 3., -1.],
[ 2., 5., -1.],
[ 0., 6., -1.],
[-1., -0., 0.],
[-0., -1., 0.]])
halfspace = build_halfspaces(A)
self.assertTrue(np.array_equal(halfspace, expected_halfspace))
B = np.array([[3, 2], [2, 6], [3, 1]])
expected_halfspace = np.array([[ 3., 2., 3., -1.],
[ 2., 6., 1., -1.],
[-1., -0., -0., 0.],
[-0., -1., -0., 0.],
[-0., -0., -1., 0.]])
halfspace = build_halfspaces(B.transpose())
self.assertTrue(np.array_equal(halfspace, expected_halfspace))
def assert_all_examples(strategy, predicate):
'''
Checks that there are no examples with given strategy
that do not match predicate.
:param strategy: Hypothesis strategy to check
:param predicate: (callable) Predicate that takes string example and returns bool
'''
@h.settings(max_examples=1000, max_iterations=5000)
@h.given(strategy)
def assert_examples(s):
assert predicate(s),'Found %r using strategy %s which does not match' % (
s, strategy,
)
assert_examples()
def assert_valid_credentials(access_key: str, secret_key: str) -> None:
"""
Given credentials, assert that they can authenticate with a Vuforia
database.
Raises:
AssertionError: The given credentials fail to authenticate with a
Vuforia database.
"""
credentials = VuforiaServerCredentials(
database_name=uuid.uuid4().hex,
access_key=access_key,
secret_key=secret_key,
)
response = get_vws_target(
vuforia_server_credentials=credentials,
target_id=uuid.uuid4().hex,
)
# This shows that the response does not give an authentication
# error which is what would happen if the keys were incorrect.
assert response.status_code == codes.NOT_FOUND
def test_non_annotated_function_call_bad_arguments():
""" User tries to call a non-annotated function on arguments of the wrong type.
"""
program = f'def add_num(num1, num2):\n' \
f' return num1 + num2\n' \
f'\n' \
f'add_num("bob", 1.0)\n'
try:
module, inferer = cs._parse_text(program)
except:
raise SkipTest()
call_node = next(module.nodes_of_class(astroid.Call))
expected_msg = f'In the Call node in line 4, there was an error in calling the function "add_num":\n' \
f'in parameter (1), the function was expecting an object of inferred type ' \
f'int but was given an object of type str.\n' \
f'in parameter (1), the function was expecting an object of inferred type ' \
f'int but was given an object of type float.\n'
# TODO: should we use the term inferred?
assert call_node.type_constraints.type.msg == expected_msg
def test_user_defined_annotated_call_wrong_arguments_type():
""" User tries to call an annotated user-defined function on the wrongly-typed arguments.
"""
program = f'def add_3(num1: int, num2: int, num3: int) -> int:\n' \
f' return num1 + num2 + num3\n' \
f'\n' \
f'add_3(1, "bob", 1.0)\n'
try:
module, inferer = cs._parse_text(program)
except:
raise SkipTest()
call_node = list(module.nodes_of_class(astroid.Call))[0]
expected_msg = f'In the Call node in line 4, there was an error in calling the annotated function "add_3":\n' \
f'in parameter (2), the annotated type is int but was given an object of type str.\n' \
f'in parameter (3), the annotated type is int but was given an object of type float.\n'
assert call_node.type_constraints.type.msg == expected_msg
def gen_data_from_z(z):
"""
Given the z axis values (2d list of failures) generate data that would have
given us this z value.
:param z: a list of lists of failures.
:return: the data that would have given this z value, the z value, the x value
(stages) and the y value (projects).
"""
data = []
projects_len = len(z)
stages_len = 0 if projects_len == 0 else len(z[0])
projects = [uuid4() for _ in xrange(projects_len)]
stages = [uuid4() for _ in xrange(stages_len)]
for pidx, project in enumerate(projects):
for sidx, stage in enumerate(stages):
failures = z[pidx][sidx]
repo = strings.example()
for _ in xrange(failures):
data.append(MockPipelineRun(stage_failed=stage, project=project, repository=repo))
return data, z, stages, projects
def with_fixed_values_for(self, **fixtures):
"""
Use fixed values for the given set of attributes.
:param fixtures (attr_name -> attr_value): A mapping of attribute names to static values. These will be used
directly to assign to attribute values on the database record.
:return: Generate
"""
ret = copy.copy(self)
ret.fixtures = copy.copy(self.fixtures)
ret.fixtures.update(fixtures)
return ret
def with_relative_values_for(self, **kwargs):
"""
Use relative values for the given set of attributes.
:param kwargs (attr_name -> function): A mapping of attibute names to functions of a single dictionary argument
containing all values already set for any given record so far. Fixed, random, and "extra" values. The
function should return the value you wnat to use for the attribute for that record.
:return: Generate
"""
ret = copy.copy(self)
ret.relative_values = copy.copy(ret.relative_values)
ret.relative_values.append(kwargs)
return ret
def assertPolicyState(self, policies, successes, failures):
"""
Assert that the given FailurePolicy instances has the given number of
success() and failure() calls.
"""
for policy in policies:
self.assertEqual((policy.successes, policy.failures),
(successes, failures))
def assertSessionHas(self, session, trace_id, clock_level, **properties):
"""
Assert the given SessionImpl has the given trace, clock and properties.
"""
self.assertEqual(session._context.traceId, trace_id)
self.assertEqual(session._context.clock.clocks, clock_level)
self.assertEqual(session._context.properties, properties)
def assertEnvironmentEquals(test, environment, name, fallback):
"""
Assert the given environment has the given name and fallback name.
"""
test.assertEqual(environment.name, name)
test.assertEqual(environment.fallbackName, fallback)
def assert_log_level_enforced(self, minimum_level):
"""Only log messages at or above the given level are sent."""
mdk, tracer = create_mdk_with_faketracer()
session = mdk.session()
messages = ["a", "b", "c", "d", "e"]
# Set logging level of the session:
session.trace(minimum_level)
# Log messages at all levels:
for (l, m) in zip(self.LEVELS, messages):
getattr(session, l.lower())("category", m)
# Only messages at or above current level should actually be logged:
result = [d["level"] for d in tracer.messages]
expected_levels = self.LEVELS[self.LEVELS.index(minimum_level):]
self.assertEqual(result, expected_levels)
def do(settings):
PARSED_HOST = urlparse(settings.spec_url)
swagger_spec = requests.get(settings.spec_url)
swagger_spec.raise_for_status()
SPEC = swagger_spec.json()
validator = get_validator(SPEC, settings.spec_url)
validator.validate_spec(swagger_spec.json(), settings.spec_url)
SPEC_HOST = urlunparse(list(PARSED_HOST)[:2] + [SPEC['basePath']] + ['', '', ''])
s = requests.Session()
@given(data())
@hsettings(max_examples=settings.iterations)
def swagger_fuzzer(data):
request = get_request(data, SPEC, SPEC_HOST)
note("Curl command: {}".format(to_curl_command(request)))
result = s.send(request)
for validator in VALIDATORS:
validator(SPEC, request, result, settings)
# Call the function
swagger_fuzzer()
def assert_can_generate(pattern):
'''
Checks that regex strategy for given pattern generates examples
that match that regex pattern
'''
compiled_pattern = re.compile(pattern)
strategy = regex(pattern)
assert_all_examples(strategy, compiled_pattern.match)
def test_decorator_real_http(self) -> None:
"""
When the `real_http` parameter given to the decorator is set to `True`,
requests made to unmocked addresses are not stopped.
"""
with pytest.raises(requests.exceptions.ConnectionError):
request_unmocked_address()
def test_real_http(self) -> None:
"""
When the `real_http` parameter given to the context manager is set to
`True`, requests made to unmocked addresses are not stopped.
"""
with MockVWS(real_http=True):
with pytest.raises(requests.exceptions.ConnectionError):
request_unmocked_address()
def setUp(self) -> None:
"""
Overload the getter for SQLAlchemy sessions for a given database
model with a mock. This method should work similarly to
``Session.object_session``, which returns the SQLAlchemy ORM session
to which the model class is bound
"""
TestService.setUp(self)
self.session_getter = mock.MagicMock(spec=Session.object_session)
self.service = Service(
self.database_service, session_getter_for_model=self.session_getter
)
def test_name_sqlite(self) -> None:
"""
Tests that if the name ``sqlite`` is given to the database, then a
CHAR descriptor comes back out with 32 characters
"""
uuid_instance = DB_UUID()
dialect_impl = uuid_instance.load_dialect_impl(self.sqlite_dialect)
self.assertIsInstance(
dialect_impl, self.sqlite_dialect.type_descriptor(
CHAR(32)
).__class__
)
def _parse_to_function(function_name, args_list, return_statement):
"""Helper to parse given data into function definition."""
return f'def {function_name}({", ".join(args_list)}):' \
f' return {return_statement}'
def _parse_to_function_no_return(function_name, args_list, function_body):
"""Helper to parse given data into function definition."""
return f'def {function_name}({", ".join(args_list)}):\n' \
f' {function_body}'
def test_user_defined_annotated_call_wrong_arguments_number():
""" User tries to call an annotated function on the wrong number of arguments.
"""
program = f'def add_3(num1: int, num2: int, num3: int) -> int:\n' \
f' return num1 + num2 + num3\n' \
f'\n' \
f'add_3()\n'
try:
module, inferer = cs._parse_text(program)
except:
raise SkipTest()
call_node = list(module.nodes_of_class(astroid.Call))[0]
expected_msg = f'In the Call node in line 4, there was an error in calling the function "add_3":\n' \
f'the function was expecting 3 arguments, but was given 0.'
assert call_node.type_constraints.type.msg == expected_msg
def _parse_to_function(function_name, args_list, return_statement):
"""Helper to parse given data into function definition."""
return f'def {function_name}({", ".join(args_list)}):' \
f' return {return_statement}'
def _parse_to_function_no_return(function_name, args_list, function_body):
"""Helper to parse given data into function definition."""
return f'def {function_name}({", ".join(args_list)}):\n' \
f' {function_body}'
def test_inference_args_simple_return(function_name, arguments):
"""Test whether visitor was able to infer type of argument given function called on it in function body."""
assume(function_name not in arguments)
for argument in arguments:
program = _parse_to_function_no_return(function_name, arguments, ('return ' + argument + " + " + repr('bob')))
module, inferer = cs._parse_text(program)
# get the functionDef node - there is only one in this test case.
function_def_node = next(module.nodes_of_class(astroid.FunctionDef))
expected_arg_type_vars = [function_def_node.type_environment.lookup_in_env(argument) for argument in arguments]
expected_arg_types = [inferer.type_constraints.lookup_concrete(type_var) for type_var in expected_arg_type_vars]
function_type_var = module.type_environment.lookup_in_env(function_name)
function_type = inferer.type_constraints.lookup_concrete(function_type_var)
actual_arg_types, actual_return_type = inferer.type_constraints.types_in_callable(function_type)
assert expected_arg_types == actual_arg_types
def test_function_def_args_simple_return(function_name, arguments):
"""Test whether visitor was able to infer type of function given function called on it's arguments."""
assume(function_name not in arguments)
for argument in arguments:
program = _parse_to_function_no_return(function_name, arguments, ('return ' + argument + " + " + repr('bob')))
module, inferer = cs._parse_text(program)
function_def_node = next(module.nodes_of_class(astroid.FunctionDef))
expected_arg_type_vars = [function_def_node.type_environment.lookup_in_env(argument) for argument in arguments]
expected_arg_types = [inferer.type_constraints.lookup_concrete(type_var) for type_var in expected_arg_type_vars]
function_type_var = module.type_environment.lookup_in_env(function_name)
function_type = inferer.type_constraints.lookup_concrete(function_type_var)
actual_arg_types, actual_return_type = inferer.type_constraints.types_in_callable(function_type)
return_type_var = function_def_node.type_environment.lookup_in_env(argument)
expected_return_type = inferer.type_constraints.lookup_concrete(return_type_var)
assert Callable[actual_arg_types, actual_return_type] == Callable[expected_arg_types, expected_return_type]
def test_subscript_homogeneous_list_slice(node):
"""Test visitor of Subscript node representing slicing of homogeneous list."""
module, _ = cs._parse_text(node)
for subscript_node in module.nodes_of_class(astroid.Subscript):
list_node = subscript_node.value
assert subscript_node.type_constraints.type == List[list_node.elts[0].type_constraints.type]
# TODO: this test currently fails
# @given(cs.simple_homogeneous_dict_node(min_size=1))
# def test_inference_dict_subscript(node):
# """Note that this test only takes in a dictionary because the subscript index
# must be the same type as the dictionary's keys in order to type check.
# """
# for key, _ in node.items:
# new_node = astroid.Subscript()
# new_node.postinit(node, key)
# module, _ = cs._parse_text(new_node)
# for subscript_node in module.nodes_of_class(astroid.Subscript):
# dict_node = subscript_node.value
# assert subscript_node.type_constraints.type == list(dict_node.items)[0][1].type_constraints.type
# TODO: this test needs to be converted, but will also fail
# @given(cs.random_list(min_size=2), cs.random_slice_indices())
# def test_subscript_heterogeneous_list_slice(input_list, slice):
# """Test visitor of Subscript node representing slicing of heterogeneous list."""
# assume(not isinstance(input_list[0], type(input_list[1])))
# input_slice = ':'.join([str(index) if index else '' for index in slice])
# program = f'{input_list}[{input_slice}]'
# module, _ = cs._parse_text(program)
# subscript_node = list(module.nodes_of_class(astroid.Subscript))[0]
# assert subscript_node.type_constraints.type == List[Any]
def test_completeness_and_uniqueness(self, n, m):
solutions = set(weak_composition_iter(n, m))
if m == 0 and n > 0:
expected_count = 0
else:
# the total number of distinct partitions is given by (n+m-1)!/((m-1)!*n!)
expected_count = 1
for i in range(1, m):
expected_count *= n + m - i
for i in range(1, m):
expected_count /= i
assert len(solutions) == expected_count
assert len(set(solutions)) == expected_count
