def test__handle_updates_present_reboot_file_not_present(self, updates_needed, po, _reboot, apt):
"""
Given there are pending updates.
And Apt returns with 0
And Apt does not touch the /var/run/reboot-required file
Then no reboot should 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 = False
apt._handle()
assert po.called is True
assert _reboot.called is False
python类patch()的实例源码
def test_load_root_default_config(self, mock_isfile, mock_yaml, mock_open):
'''
tests ConfigFileLoader.load_root_config by inputing a root configuration
file path, and asserts that the resulting processed content has the
file references properly updated
:type mock_isfile: Mock
:type mock_yaml: Mock
:type mock_open: Mock
'''
mock_isfile.return_value = True
mocked_open = mock_open('test')
mocked_open_name = '%s.open' % __name__
with patch(mocked_open_name, mocked_open, create=True):
mock_yaml.return_value = {
'adobe_users': {'connectors': {'umapi': 'test-123'}},
'logging': {'log_to_file': True},
}
yml = ConfigFileLoader.load_root_config('config-test-2/user-sync-config-test.yml')
# assert default values are preserved
self.assert_eq(yml['logging']['file_log_directory'], os.path.abspath('config-test-2/logs'),
'default log path is missing or incorrect')
# assert file paths are still updated properly
self.assert_eq(yml['adobe_users']['connectors']['umapi'], os.path.abspath('config-test-2/test-123'),
'default primary umapi configuration path is incorrect')
def test_load_sub_config(self, mock_isfile, mock_yaml, mock_open):
'''
same purpose as test_load_root_config, but tests against sub
configuration path updates (which is currently only the private key
path in the umapi configuration file)
:type mock_isfile: Mock
:type mock_yaml: Mock
:type mock_open: Mock
'''
mock_isfile.return_value = True
mocked_open = mock_open('test')
mocked_open_name = '%s.open' % __name__
with patch(mocked_open_name, mocked_open, create=True):
mock_yaml.return_value = {
'enterprise':{
'priv_key_path':'../keys/test-key.key',
'test':'value should not change'
},
'other': {
'test-2': 123
}
}
yml = ConfigFileLoader.load_sub_config('sub-config-test/user-sync-config-test.yml')
# test path updating
self.assert_eq(yml['enterprise']['priv_key_path'], os.path.abspath('keys/test-key.key'),
'private key path is incorrect')
# test control keys
self.assert_eq(yml['enterprise']['test'], 'value should not change',
'/enterprise/test value should not change')
self.assert_eq(yml['other']['test-2'], 123, '/other/test-2 value should not change')
def test_should_get_existing_token(self):
with patch('os.path.isfile', return_value=True),\
patch('six.moves.builtins.open', mock_open(read_data='123TOKEN'),
create=True):
self.assertEqual(self.token_request._get_existing_token(),
self.token_file_contents)
def test_should_not_get_existing_token(self):
with patch('os.path.isfile', return_value=False):
self.assertEqual(self.token_request._get_existing_token(), None)
def _test_init_with_log_value(self, debug, result_level):
joined_path = 'baz'
folder = 'foo'
filename = 'bar'
console_format = '[%(levelname)s] %(message)s'
file_format = '[%(name)s][%(levelname)s][%(asctime)s] %(message)s'
with mock.patch('logging.handlers.RotatingFileHandler') as mock_fh:
with mock.patch('logging.StreamHandler') as mock_sh:
with mock.patch('logging.Formatter') as mock_formatter:
with mock.patch('os.path.join',
return_value=joined_path) as mock_join:
log_config = LogConfiguration(folder, filename, debug)
mock_sh().setLevel.assert_called_once_with(
result_level)
mock_sh().setFormatter.assert_called_once_with(
mock_formatter())
mock_fh.assert_called_once_with(joined_path,
maxBytes=1024 * 1024,
backupCount=20)
mock_fh().setLevel.assert_called_once_with(
logging.DEBUG)
mock_fh().setFormatter.assert_called_once_with(
mock_formatter())
mock_formatter.assert_has_calls([
mock.call(console_format),
mock.call(file_format)
])
mock_join.assert_called_once_with(folder, 'logs',
'%s.log' % filename)
self.assertEqual(log_config._consoleLogger, mock_sh())
self.assertEqual(log_config.console_logger, mock_sh())
self.assertEqual(log_config._fileLogger, mock_fh())
self.assertEqual(log_config.file_logger, mock_fh())
return log_config
def test_create_with_multi_matched_volumes(self):
fake_vol_name = 'fake-vol'
fake_vols = [fake_object.FakeCinderVolume(name=fake_vol_name),
fake_object.FakeCinderVolume(name=fake_vol_name)]
with mock.patch.object(fake_client.FakeCinderClient.Volumes, 'list',
return_value=fake_vols):
self.assertRaises(exceptions.TooManyResources,
self.cinderprovider.create,
fake_vol_name,
{})
def test_list(self):
fake_vols = [fake_object.FakeCinderVolume(name='fake-vol1')]
with mock.patch.object(fake_client.FakeCinderClient.Volumes, 'list',
return_value=fake_vols):
self.assertEqual([{'Name': 'fake-vol1', 'Mountpoint': ''}],
self.cinderprovider.list())
def test_mount(self, mock_docker_volume, 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):
self.assertEqual(fake_mountpoint,
self.cinderprovider.mount('fake-vol'))
def test_mount_state_not_attach(self, mock_docker_volume,
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):
self.assertEqual(fake_mountpoint,
self.cinderprovider.mount('fake-vol'))
def setUp(self):
self.patcherSleep = patch('pokemongo_bot.walkers.step_walker.sleep')
self.patcherSleep.start()
self.bot = MagicMock()
self.bot.api = MagicMock()
# let us get back the position set by the PolylineWalker
def api_set_position(lat, lng, alt):
self.bot.position = [lat, lng, alt]
def hearbeat():
return True
self.bot.config.gmapkey = ''
self.bot.api.set_position = api_set_position
self.bot.heartbeat = hearbeat
directions_path = os.path.join(os.path.dirname(__file__), 'resources', ex_resp_directions)
with open(directions_path, 'rb') as directions:
ex_directions = pickle.load(directions)
elevations_path = os.path.join(os.path.dirname(__file__), 'resources', ex_resp_elevations)
with open(elevations_path, 'rb') as elevations:
ex_elevations = pickle.load(elevations)
with requests_mock.Mocker() as m:
m.get(
"https://maps.googleapis.com/maps/api/directions/json?mode=walking&origin={},{}&destination={},{}".format(
ex_orig[0], ex_orig[1], ex_dest[0], ex_dest[1]
), json=ex_directions, status_code=200)
m.get("https://maps.googleapis.com/maps/api/elevation/json?path=enc:{}&samples={}".format(
ex_enc_polyline, ex_nr_samples
), json=ex_elevations, status_code=200)
self.polyline = PolylineObjectHandler.cached_polyline(ex_orig, ex_dest)
self.bot.position = [ex_orig[0], ex_orig[1], self.polyline.get_alt(ex_orig)]
def test_one_small_speed(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
speed = 0.247503233266
precision = 0.0
dlat = 47.17064
dlng = 8.51674
self.bot.config.walk_max = speed
self.bot.config.walk_min = speed
pw = PolylineWalker(self.bot, ex_dest[0], ex_dest[1], precision=precision)
self.assertEqual(pw.dest_lat, ex_dest[0], 'dest_lat did not match')
self.assertEqual(pw.dest_lng, ex_dest[1], 'dest_lng did not match')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return pw.step()
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(dlat, dlng, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (pw.precision + pw.epsilon))
self.polyline._last_pos = (dlat, dlng)
self.assertTrue(abs(self.polyline.get_alt() - self.bot.position[2]) <= 1)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_one_small_speed_big_precision(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
speed = 0.247503233266
precision = 2.5
dlat = 47.170635631
dlng = 8.51673976413
self.bot.config.walk_max = speed
self.bot.config.walk_min = speed
pw = PolylineWalker(self.bot, ex_dest[0], ex_dest[1], precision=precision)
self.assertEqual(pw.dest_lat, ex_dest[0], 'dest_lat did not match')
self.assertEqual(pw.dest_lng, ex_dest[1], 'dest_lng did not match')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return pw.step()
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(dlat, dlng, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (pw.precision + pw.epsilon))
self.polyline._last_pos = (dlat, dlng)
self.assertTrue(abs(self.polyline.get_alt() - self.bot.position[2]) <= 1)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_intermediary_speed(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
speed = 166.8285172348795
precision = 0.0
dlat = 47.17022
dlng = 8.51789
self.bot.config.walk_max = speed
self.bot.config.walk_min = speed
pw = PolylineWalker(self.bot, ex_dest[0], ex_dest[1], precision=precision)
self.assertEqual(pw.dest_lat, ex_dest[0], 'dest_lat did not match')
self.assertEqual(pw.dest_lng, ex_dest[1], 'dest_lng did not match')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return pw.step()
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(dlat, dlng, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (pw.precision + pw.epsilon))
self.polyline._last_pos = (dlat, dlng)
self.assertTrue(abs(self.polyline.get_alt() - self.bot.position[2]) <= 1)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_intermediary_speed_big_precision(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
speed = 166.8285172348795
precision = 2.5
dlat = 47.17022
dlng = 8.51789
self.bot.config.walk_max = speed
self.bot.config.walk_min = speed
pw = PolylineWalker(self.bot, ex_dest[0], ex_dest[1], precision=precision)
self.assertEqual(pw.dest_lat, ex_dest[0], 'dest_lat did not match')
self.assertEqual(pw.dest_lng, ex_dest[1], 'dest_lng did not match')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return pw.step()
finishedWalking = run_step()
self.assertFalse(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(dlat, dlng, self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (pw.precision + pw.epsilon))
self.polyline._last_pos = (dlat, dlng)
self.assertTrue(abs(self.polyline.get_alt() - self.bot.position[2]) <= 1)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def test_bigger_then_total_speed_big_precision_offset(self):
walk_max = self.bot.config.walk_max
walk_min = self.bot.config.walk_min
speed = 300
precision = 2.5
self.bot.config.walk_max = speed
self.bot.config.walk_min = speed
pw = PolylineWalker(self.bot, ex_dest[0], ex_dest[1], precision=precision)
self.assertEqual(pw.dest_lat, ex_dest[0], 'dest_lat did not match')
self.assertEqual(pw.dest_lng, ex_dest[1], 'dest_lng did not match')
@mock.patch('random.uniform')
def run_step(mock_random):
mock_random.return_value = 0.0
return pw.step()
finishedWalking = run_step()
self.assertTrue(finishedWalking, 'step should return False')
distance = Geodesic.WGS84.Inverse(ex_dest[0], ex_dest[1], self.bot.position[0], self.bot.position[1])["s12"]
self.assertTrue(0.0 <= distance <= (pw.precision + pw.epsilon))
self.polyline._last_pos = self.polyline.destination
self.assertTrue(abs(self.polyline.get_alt() - self.bot.position[2]) <= 1)
self.bot.config.walk_max = walk_max
self.bot.config.walk_min = walk_min
def setUp(self):
self.patcherSleep = patch('pokemongo_bot.walkers.step_walker.sleep')
self.patcherSleep.start()
self.bot = MagicMock()
self.bot.position = [0, 0, 0]
self.bot.api = MagicMock()
# let us get back the position set by the StepWalker
def api_set_position(lat, lng, alt):
self.bot.position = [lat, lng, alt]
self.bot.api.set_position = api_set_position
def testCallback(self):
"""
Testing the callback provided when audio has been provided by the User as an answer.
"""
parameters = {
"default": self.default,
"from_answer_link": self.from_answer_link
}
with mock.patch("kalliope.core.NeuronModule.get_audio_from_stt") as mock_get_audio_from_stt:
with mock.patch("kalliope.core.NeuronModule.run_synapse_by_name") as mock_run_synapse_by_name:
# testing running the default when no order matching
nt = Neurotransmitter(**parameters)
mock_get_audio_from_stt.assert_called_once()
mock_get_audio_from_stt.reset_mock()
# testing running the default when audio None
audio_text = None
nt.callback(audio=audio_text)
mock_run_synapse_by_name.assert_called_once_with(self.default, high_priority=True, is_api_call=False)
mock_run_synapse_by_name.reset_mock()
# testing running the default when no order matching
audio_text = "try test audio "
nt.callback(audio=audio_text)
mock_run_synapse_by_name.assert_called_once_with(self.default, high_priority=True, is_api_call=False)
mock_run_synapse_by_name.reset_mock()
# Testing calling the right synapse
audio_text = "answer one"
nt.callback(audio=audio_text)
mock_run_synapse_by_name.assert_called_once_with(synapse_name="synapse2",
user_order=audio_text,
synapse_order="answer one",
high_priority=True,
is_api_call=False)
def test_check_supported_version(self):
# version ok
current_version = '0.4.0'
supported_version = ['0.4', '0.3', '0.2']
self.assertTrue(ResourcesManager._check_supported_version(current_version=current_version,
supported_versions=supported_version))
# version ok
current_version = '11.23.0'
supported_version = ['11.23', '12.3', '2.23']
self.assertTrue(ResourcesManager._check_supported_version(current_version=current_version,
supported_versions=supported_version))
# version non ok, user does not config
# Testing with integer values instead of string
current_version = '0.4.0'
supported_version = [0.3, 0.2]
with mock.patch('kalliope.Utils.query_yes_no', return_value=True):
self.assertTrue(ResourcesManager._check_supported_version(current_version=current_version,
supported_versions=supported_version))
with mock.patch('kalliope.Utils.query_yes_no', return_value=False):
self.assertFalse(ResourcesManager._check_supported_version(current_version=current_version,
supported_versions=supported_version))
def test_load_stt_plugin(self):
# Test getting default stt
ol = OrderListener()
stt1 = Stt(name="default-stt",
parameters=dict())
stt2 = Stt(name="second-stt",
parameters=dict())
stt3 = Stt(name="third-stt",
parameters=dict())
resources = Resources(stt_folder="/tmp")
ol.settings = mock.MagicMock(default_stt_name="default-stt",
stts=[stt1,stt2,stt3],
resources=resources)
callback = mock.MagicMock()
ol.callback = callback
with mock.patch("kalliope.core.Utils.get_dynamic_class_instantiation") as mock_get_dynamic_class_instantiation:
mock_get_dynamic_class_instantiation.return_value = 'class_instance'
self.assertEqual(ol.load_stt_plugin(),
"class_instance",
"Fail getting the proper value")
mock_get_dynamic_class_instantiation.assert_called_once_with(package_name= "stt",
module_name= "Default-stt",
parameters={'callback' : callback},
resources_dir= "/tmp")