def test_redis_status(self, mock_redis):
setattr(settings, 'CACHEOPS_REDIS', {'host': 'foo', 'port': 1337})
mock_redis.StrictRedis.return_value.ping.return_value = 'PONG'
mock_redis.StrictRedis.return_value.info.return_value = {
'redis_version': '1.0.0'}
status = redis_status.check(request=None)
assert status['ping'] == 'PONG'
assert status['version'] == '1.0.0'
# @mock.patch('heartbeat.checkers.redis_status.redis')
# def test_redis_connection_error(self, mock_redis):
# setattr(settings, 'CACHEOPS_REDIS', {'host': 'foo', 'port': 1337})
# mock_ping = mock_redis.StrictRedis.return_value.ping
# mock_ping.side_effect = ConnectionError('foo')
# status = redis.check(request=None)
# assert status['error'] == 'foo', status
python类patch()的实例源码
def test_db_version(self):
import django
if django.VERSION >= (1, 7):
cursor = 'django.db.backends.utils.CursorWrapper'
else:
cursor = 'django.db.backends.util.CursorWrapper'
with mock.patch(cursor) as mock_cursor:
mock_cursor.return_value.fetchone.return_value = ['1.0.0']
dbs = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': 'foo'
}
}
setattr(settings, 'DATABASES', dbs)
dbs = databases.check(request=None)
assert len(dbs) == 1
assert dbs[0]['version'] == '1.0.0'
def test_create_logger(self):
with mock.patch.object(LogConfiguration, '__init__',
return_value=None):
with mock.patch('logging.getLogger') as mock_get:
with mock.patch.object(LogConfiguration, 'file_logger'):
with mock.patch.object(LogConfiguration, 'console_logger'):
log_config = LogConfiguration('foo', 'bar')
name = 'foobar'
result = log_config.create_logger(name)
mock_get.assert_called_once_with(name)
mock_get().setLevel.assert_called_once_with(
logging.DEBUG)
mock_get().addHandler.assert_has_calls([
mock.call(log_config.file_logger),
mock.call(log_config.console_logger)
])
self.assertEqual(result, mock_get())
def test_that_send_simple_message_creates_the_appropriate_request(self):
domain = 'domain'
api_key = 'APIKEY'
sender_name = 'foo'
mailer = Mailer(domain, api_key, sender_name)
with mock.patch('requests.post') as mock_post:
data = {'subject': 'foo'}
mailer.send_simple_message(data)
expected_data = data.copy()
expected_data['from'] = mailer.sender
mock_post.assert_called_once_with("%s/messages" % mailer.api_url,
auth=mailer.auth,
data=expected_data)
def test_mount_state_attach_to_other(self, mock_create_mp, mock_do_mount):
fd, fake_devpath = tempfile.mkstemp()
fake_link_path = fake_devpath
fake_mountpoint = 'fake-mount-point/'
with mock.patch.object(FakeCinderConnector, 'get_device_path',
return_value=fake_link_path):
with mock.patch.object(cinder.Cinder, '_get_mountpoint',
return_value=fake_mountpoint):
fake_c_vol = fake_object.FakeCinderVolume(multiattach=True)
with mock.patch.object(cinder.Cinder, '_get_docker_volume',
return_value=(fake_c_vol,
consts.ATTACH_TO_OTHER)):
self.assertEqual(fake_mountpoint,
self.cinderprovider.mount('fake-vol'))
fake_c_vol = fake_object.FakeCinderVolume(multiattach=False)
with mock.patch.object(cinder.Cinder, '_get_docker_volume',
return_value=(fake_c_vol,
consts.ATTACH_TO_OTHER)):
self.assertRaises(exceptions.FuxiException,
self.cinderprovider.mount, 'fake-vol')
def test_normalized_distance(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 1
self.bot.config.walk_min = 1
sw = StepWalker(self.bot, 0.1, 0.1, precision=0.0)
self.assertGreater(sw.dest_lat, 0)
self.assertGreater(sw.dest_lng, 0)
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step()
stayInPlace = run_step()
self.assertFalse(stayInPlace)
self.assertAlmostEqual(self.bot.position[0], NORMALIZED_LAT_LNG_DISTANCE[0], places=6)
self.assertAlmostEqual(self.bot.position[1], NORMALIZED_LAT_LNG_DISTANCE[1], places=6)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_normalized_distance_times_2(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 2
self.bot.config.walk_min = 2
sw = StepWalker(self.bot, 0.1, 0.1, precision=0.0)
self.assertTrue(sw.dest_lat > 0)
self.assertTrue(sw.dest_lng > 0)
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step()
stayInPlace = run_step()
self.assertFalse(stayInPlace)
self.assertAlmostEqual(self.bot.position[0], NORMALIZED_LAT_LNG_DISTANCE[0] * 2, places=6)
self.assertAlmostEqual(self.bot.position[1], NORMALIZED_LAT_LNG_DISTANCE[1] * 2, places=6)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_small_distance_same_spot(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 1
self.bot.config.walk_min = 1
sw = StepWalker(self.bot, 0, 0, precision=0.0)
self.assertEqual(sw.dest_lat, 0, 'dest_lat should be 0')
self.assertEqual(sw.dest_lng, 0, 'dest_lng should be 0')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step()
moveInprecision = run_step()
self.assertTrue(moveInprecision, 'step should return True')
distance = Geodesic.WGS84.Inverse(0.0, 0.0, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (sw.precision + sw.epsilon))
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_big_distances(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 1
self.bot.config.walk_min = 1
sw = StepWalker(self.bot, 10, 10, precision=0.0)
self.assertEqual(sw.dest_lat, 10)
self.assertEqual(sw.dest_lng, 10)
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step()
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
self.assertAlmostEqual(self.bot.position[0], NORMALIZED_LAT_LNG_DISTANCE[0], places=6)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_stay_put(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 4
self.bot.config.walk_min = 2
sw = StepWalker(self.bot, 10, 10, precision=0.0)
self.assertEqual(sw.dest_lat, 10)
self.assertEqual(sw.dest_lng, 10)
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step(speed=0)
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(0.0, 0.0, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (sw.precision + sw.epsilon))
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_teleport(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
self.bot.config.walk_max = 4
self.bot.config.walk_min = 2
sw = StepWalker(self.bot, 10, 10, precision=0.0)
self.assertEqual(sw.dest_lat, 10)
self.assertEqual(sw.dest_lng, 10)
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return sw.step(speed=float("inf"))
finishedWalking = run_step()
self.assertTrue(finishedWalking, 'step should return True')
total_distance = Geodesic.WGS84.Inverse(0.0, 0.0, 10, 10)["s12"]
distance = Geodesic.WGS84.Inverse(0.0, 0.0, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= abs(total_distance - distance) <= (sw.precision + sw.epsilon))
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def testInit(self):
"""
Testing the init method of the neurontransmitter.
"""
with mock.patch("kalliope.core.NeuronModule.run_synapse_by_name") as mock_run_synapse_by_name:
# Test direct link
parameters = {
"default": self.default,
"direct_link": self.direct_link
}
Neurotransmitter(**parameters)
mock_run_synapse_by_name.assert_called_once_with(self.direct_link, high_priority=True)
with mock.patch("kalliope.core.NeuronModule.get_audio_from_stt") as mock_get_audio_from_stt:
# Test get_audio_from_stt
parameters = {
"default": self.default,
"from_answer_link": self.from_answer_link,
}
Neurotransmitter(**parameters)
mock_get_audio_from_stt.assert_called_once()
def test_convert_to_wav(self):
"""
Test the api function to convert incoming sound file to wave.
"""
with mock.patch("os.system") as mock_os_system:
# Scenario 1 : input wav file
temp_file = "/tmp/kalliope/tempfile.wav" # tempfile.NamedTemporaryFile(suffix=".wav")
result_file = FlaskAPI._convert_to_wav(temp_file)
self.assertEqual(temp_file, result_file)
mock_os_system.assert_not_called()
# Scenario 2 : input not a wav file
temp_file = "/tmp/kalliope/tempfile.amr" # tempfile.NamedTemporaryFile(suffix=".wav")
expected_result = "/tmp/kalliope/tempfile.wav"
result_file = FlaskAPI._convert_to_wav(temp_file)
self.assertEqual(expected_result, result_file)
mock_os_system.assert_called_once_with("avconv -y -i " + temp_file + " " + expected_result)
def test_old_redis_version(request, version):
"""Test how fixture behaves in case of old redis version."""
config = get_config(request)
with mock.patch(
'os.popen',
lambda *args: StringIO(version)
):
with pytest.raises(RedisUnsupported):
RedisExecutor(
config['exec'],
databases=4,
redis_timeout=config['timeout'],
loglevel=config['loglevel'],
logsdir=config['logsdir'],
port=get_port(None),
host=config['host'],
timeout=30,
).start()
def test_upload(self):
self.client.post.return_value = mock_upload_response('/files/content',
httplib.CREATED,
None,
'files', 'POST_response.json')
file_name = 'upload.jpg'
file_path = path.join(path.dirname(__file__), '..', 'fixtures', 'files', file_name)
folder_id = 'folder-id'
@mock.patch('orangecloud_client.files.guess_type', return_value=('image/jpeg', 'binary'))
@mock.patch('__builtin__.open', spec=open, return_value=MockFile())
@mock.patch('orangecloud_client.files.MultipartEncoder', return_value=mock.Mock())
def fire_test(mock_multipart_encoder, mock_open, _):
data = mock_multipart_encoder()
data.content_type = 'upload content type'
response_file = self.files.upload(file_path, folder_id)
self.client.post.assert_called_with(self.client.post.return_value.url,
data=data,
headers={'Content-Type': data.content_type})
self.assertEqual('file-id', response_file.fileId)
self.assertEqual('file-name', response_file.fileName)
fire_test()
def test_download(self):
url_download = 'http://some-url-for-dowload/som-path/file'
self.client.get.return_value = mock_response('http://some-url-for-dowload/som-path/file',
httplib.OK,
None,
'files', 'download.txt')
def check_data(data, json, **kwargs):
self.assertIn('stream', kwargs)
self.assertTrue(kwargs['stream'])
mock_response.check_data = check_data
@mock.patch('__builtin__.open', spec=open, return_value=MockFile())
def fire_test(mock_open):
self.files.download(url_download, 'somewhere.txt')
self.client.get.assert_called_with(self.client.get.return_value.url,
verify=False,
stream=True)
self.assertEqual(''.join(mock_open.return_value.buffer), 'Some data downloaded')
fire_test()
def test_short_code_collision(self):
# Create a second node that has already been shared.
self.post('/api/nodes', expected_status=201, auth_token='token', json={
'project': self.project_id,
'node_type': 'asset',
'name': 'collider',
'properties': {},
'short_code': 'takenX',
})
# Mock create_short_code so that it returns predictable short codes.
codes = ['takenX', 'takenX', 'freeXX']
with mock.patch('pillar.api.nodes.create_short_code',
side_effect=codes) as create_short_link:
resp = self.post('/api/nodes/%s/share' % self.node_id, auth_token='token',
expected_status=201)
share_data = resp.json()
self._check_share_data(share_data)
self.assertEqual(3, create_short_link.call_count)
def test_register_template(self):
class Foo:
pass
render.register_template(Foo)
result = []
def f(*args, **kwargs):
result.append((args, kwargs))
return ''
with mock.patch('bricks.helpers.render._render_template', f):
foo = Foo()
x = render(foo)
args, kwargs = result.pop()
assert args == (['bricks/foo.html', 'bricks/foo.jinja2'],)
assert sorted(kwargs) == ['context', 'request']
request = kwargs['context']['request']
assert hasattr(request, 'POST')
assert kwargs['context'] == {'foo': foo, 'request': request}
def test_register_template_using_decorator(self):
class Foo:
pass
@render.register_template(Foo)
def get_context(x, request, **kwargs):
return {'x': x, 'request': None}
result = []
def f(*args, **kwargs):
result.append((args, kwargs))
return ''
with mock.patch('bricks.helpers.render._render_template', f):
foo = Foo()
x = render(foo)
args, kwargs = result.pop()
assert kwargs['context'] == {'x': foo, 'request': None}
def test_build_version_with_valid_package_name(self):
package = Mock(project_name='foo', version='1.0.0')
setattr(settings, 'HEARTBEAT', {'package_name': 'foo'})
with mock.patch.object(build.WorkingSet, 'find', return_value=package):
distro = build.check(request=None)
assert distro == {'name': 'foo', 'version': '1.0.0'}
def hwd(self):
"""
Patching hw_detection_method in the __init__ function
of HardwareDetections to allow sudoless test execution
Patched method are tested separately.
scope: class
return: instance of
<class srv.salt._modules.cephdisks.HardwareDetections>
"""
self.hw_detection_method = patch('srv.salt._modules.cephdisks.HardwareDetections._find_detection_tool')
self.hw_dtctr = self.hw_detection_method.start()
self.hw_dtctr.return_value = '/a/valid/path'
yield cephdisks.HardwareDetections()
self.hw_detection_method.stop()
def test_is_rotational(self, hwd):
read_data = "1"
with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
expect = read_data
out = hwd._is_rotational('disk/in/question')
assert expect == out
def test_is_rotational_not(self, hwd):
read_data = "0"
with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
expect = read_data
out = hwd._is_rotational('disk/in/question')
assert expect == out
def test_is_removable_not(self, hwd):
read_data = "0"
with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
expect = None
out = hwd._is_removable('disk/in/question')
assert expect == out
def test_is_removable(self, hwd):
read_data = "1"
with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
expect = True
out = hwd._is_removable('disk/in/question')
assert expect == out
def osd_o(self):
with patch.object(osd.OSDConfig, '__init__', lambda self: None):
print("Constructing the OSDConfig object")
cnf = osd.OSDConfig()
cnf.device = '/dev/sdx'
# monkeypatching the device in the object since __init__
# is mocked -> skipping the readlink()
yield cnf
# everything after the yield is a teardown code
print("Teardown OSDConfig object")
def zypp(self):
"""
Fixture to always get Zypper.
"""
self.linux_dist = patch('srv.salt._modules.packagemanager.linux_distribution')
self.lnx_dist_object = self.linux_dist.start()
self.lnx_dist_object.return_value = ('opensuse', '42.2', 'x86_64')
args = {'debug': False, 'kernel': False, 'reboot': False}
yield PackageManager(**args).pm
self.linux_dist.stop()
def test__handle_patches(self, patches_needed, po, _reboot, zypp):
"""
Given there are no updates patches.
Zypper returns 102 which should lead to a reboot.
But the reboot block should not be reached, therefore no reboot.
"""
patches_needed.return_value = False
po.return_value.returncode = 102
po.return_value.communicate.return_value = ("packages out", "error")
zypp._handle(strat='patch')
assert patches_needed.called is True
assert po.called is False
assert _reboot.called is False
def apt(self):
"""
Fixture to always get Apt.
"""
self.linux_dist = patch('srv.salt._modules.packagemanager.linux_distribution')
self.lnx_dist_object = self.linux_dist.start()
self.lnx_dist_object.return_value = ('ubuntu', '42.2', 'x86_64')
# Test all permutations of :debug :kernel and :reboot
args = {'debug': False, 'kernel': False, 'reboot': False}
yield PackageManager(**args).pm
self.linux_dist.stop()
def test__handle_updates_present_reboot_file_present(self, updates_needed, po, _reboot, apt):
"""
Given there are pending updates.
And Apt returns with 0
And Apt touches the /var/run/reboot-required file
A reboot will be triggered
"""
updates_needed.return_value = True
po.return_value.communicate.return_value = ("packages out", "error")
po.return_value.returncode = 0
with patch("srv.salt._modules.packagemanager.os.path.isfile") as mock_file:
mock_file.return_value = True
apt._handle()
assert po.called is True
assert _reboot.called is True
