我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用mock.mock.patch()。
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
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
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
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")
