Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_with_gather():
np_orig = MockPopulation._simulator.state.num_processes
rank_orig = MockPopulation._simulator.state.mpi_rank
gd_orig = recording.gather_dict
MockPopulation._simulator.state.num_processes = 2
MockPopulation._simulator.state.mpi_rank = 0
def mock_gather_dict(D): # really hacky
assert isinstance(D[0], (list, numpy.ndarray))
D[1] = [i-1 for i in D[0]] + [D[0][-1] + 1]
return D
recording.gather_dict = mock_gather_dict
p = MockPopulation()
ps = Mock()
ps.evaluate = Mock(return_value=numpy.arange(11.0, 23.0, 2.0))
p._get_parameters = Mock(return_value={'tau_m': ps})
assert_arrays_equal(p.get("tau_m", gather=True),
numpy.arange(10.0, 23.0))
MockPopulation._simulator.state.num_processes = np_orig
MockPopulation._simulator.state.mpi_rank = rank_orig
recording.gather_dict = gd_orig
def test_normal(self):
(action, mock_vm) = self.mock_one_vm('vm-12345')
mockFileMgr = mock.Mock()
mockFileMgr.CreateTemporaryFileInGuest = mock.Mock()
mockFileMgr.CreateTemporaryFileInGuest.return_value = '/tmp/foo.txt'
action.si_content.guestOperationsManager = mock.Mock()
action.si_content.guestOperationsManager.fileManager = mockFileMgr
result = action.run(vm_id='vm-12345', username='u', password='p',
guest_directory='d', prefix='p', suffix='s')
mockFileMgr.CreateTemporaryFileInGuest.assert_called_once()
self.assertEqual(result, '/tmp/foo.txt')
hm = self.get_obj()
mock_load_source.side_effect = ImportError
attrs = dir(hm)
hm.HandleEvent()
mock_load_source.assert_called_with(
safe_module_name('TemplateHelper', hm._module_name),
hm.name)
self.assertEqual(attrs, dir(hm))
self.assertEqual(hm._attrs, [])
mock_load_source.reset()
mock_load_source.side_effect = None
# a regular Mock (not a MagicMock) won't automatically create
# __export__, so this triggers a failure condition in HandleEvent
mock_load_source.return_value = Mock()
attrs = dir(hm)
hm.HandleEvent()
mock_load_source.assert_called_with(
safe_module_name('TemplateHelper', hm._module_name),
hm.name)
self.assertEqual(attrs, dir(hm))
self.assertEqual(hm._attrs, [])
# test reserved attributes
module = Mock()
module.__export__ = ["_attrs", "HandleEvent", "__init__"]
mock_load_source.reset()
mock_load_source.return_value = module
attrs = dir(hm)
hm.HandleEvent()
mock_load_source.assert_called_with(
def test_get_snapshot_missing_repository_arg(self):
client = Mock()
self.assertRaises(
curator.MissingArgument,
curator.get_snapshot, client, snapshot=testvars.snap_name
)
def test_get_active_shas_for_service():
mock_pod_list = [
mock.Mock(
metadata=mock.Mock(
labels={
"yelp.com/paasta_config_sha": "a123",
"yelp.com/paasta_git_sha": "b456",
"paasta.yelp.com/config_sha": "a123",
"paasta.yelp.com/git_sha": "b456",
}
)
),
mock.Mock(
metadata=mock.Mock(
labels={
"yelp.com/paasta_config_sha": "a123!!!",
"yelp.com/paasta_git_sha": "b456!!!",
"paasta.yelp.com/config_sha": "a123!!!",
"paasta.yelp.com/git_sha": "b456!!!",
def test_client_with_response_parser_factory(self):
factory = mock.Mock()
creator = self.create_client_creator(response_parser_factory=factory)
creator.create_client('myservice', 'us-west-2')
call_kwargs = self.endpoint_creator.create_endpoint.call_args[1]
self.assertEqual(call_kwargs['response_parser_factory'], factory)
def setUp(self):
with mock.patch(
"paasta_tools.deployd.leader.Election.__init__", autospec=False
):
self.mock_client = mock.Mock()
self.mock_control = mock.Mock()
self.election = PaastaLeaderElection(
self.mock_client, control=self.mock_control
)
def test_bind_on_x_add_route_unbind_and_bind_on_otherhost(self, LoadBalancer):
lb = LoadBalancer.find.return_value
lb.hosts = [mock.Mock(), mock.Mock()]
lb.hosts[0].dns_name = "h1"
lb.hosts[1].dns_name = "h2"
manager = Manager(self.config)
manager.consul_manager = mock.Mock()
manager.bind("x", "apphost.com")
manager.add_route("x", "/", "my.custom.host", None, False)
manager.unbind("x")
manager.bind("x", "otherhost.com")
binding_data = self.storage.find_binding("x")
self.assertDictEqual(binding_data, {
"_id": "x",
"app_host": "otherhost.com",
"paths": [
{"path": "/", "destination": "my.custom.host", "content": None, "https_only": False}
]
})
def _unary_helper(self):
continuation = mock.Mock()
mock_response = mock.Mock()
continuation.return_value = mock_response
interceptor = client_interceptor.OpenCensusClientInterceptor()
interceptor._intercept_call = mock.Mock(
return_value=(None, iter([mock.Mock()]), None))
return interceptor, continuation, mock_response
def test_delay_overlimit_withnow(self, mock_time):
limit = mock.Mock(cost=10.0, unit_value=100)
bucket = limits.Bucket('db', limit, 'key', last=1000000.0, level=100.0)
result = bucket.delay({}, now=1000005.0)
self.assertEqual(result, 5.0)
self.assertEqual(bucket.last, 1000005.0)
self.assertEqual(bucket.next, 1000010.0)
self.assertEqual(bucket.level, 95.0)