我们从Python开源项目中，提取了以下45个代码示例，用于说明如何使用unittest.mock.create_autospec()。
def test_autospec_side_effect(self): # Test for issue17826 results = [1, 2, 3] def effect(): return results.pop() def f(): pass mock = create_autospec(f) mock.side_effect = [1, 2, 3] self.assertEqual([mock(), mock(), mock()], [1, 2, 3], "side effect not used correctly in create_autospec") # Test where side effect is a callable results = [1, 2, 3] mock = create_autospec(f) mock.side_effect = effect self.assertEqual([mock(), mock(), mock()], [3, 2, 1], "callable side effect not used correctly")
def test_mock_calls_create_autospec(self): def f(a, b): pass obj = Iter() obj.f = f funcs = [ create_autospec(f), create_autospec(obj).f ] for func in funcs: func(1, 2) func(3, 4) self.assertEqual( func.mock_calls, [call(1, 2), call(3, 4)] ) #Issue21222
def setUp(self): self.fetcher = mock.create_autospec(Fetcher) self.fetcher.fetch.return_value = '/some/path' self.auth = mock.create_autospec(Auth) self.mailer = mock.create_autospec(Mailer) self.project = mock.create_autospec(Project) self.notifier = mock.create_autospec(Notifier) self.project.name = '123' self.service = mock.create_autospec(Service) self.service.options = {'ports': ['80:80']} self.project.services = [self.service] self.get_project = mock.Mock() self.get_project.return_value = self.project self.logger = mock.create_autospec(logging.Logger) self.tester = mock.create_autospec(Tester) self.ci = CI( fetcher=self.fetcher, auth=self.auth, mailer=self.mailer, get_project=self.get_project, logger=self.logger, tester=self.tester, notifier=self.notifier, )
def test_it_logins_against_the_registry(self): project = mock.create_autospec(Project) project.client = mock.create_autospec(docker.DockerClient) logger = mock.create_autospec(logging.Logger) auth = Auth( user='test', password='test', email='test', registry='example.org', logger=logger ) auth.login(project) project.client.login.assert_called_with('test', 'test', 'test', 'example.org')
def setUp(self): self.getuser_patch = mock.patch( 'slivka.scheduler.executors.getpass.getuser', return_value='mockuser' ) self.getuser_patch.start() self.subprocess_patch = mock.patch( 'slivka.scheduler.executors.subprocess', autospec=True ) self.mock_subprocess = self.subprocess_patch.start() mock_popen = self.mock_subprocess.Popen.return_value mock_popen.communicate.return_value = (self.qstat_output, '') mock_exec = mock.create_autospec(Executor) mock_exec.result_paths = [] self.job = GridEngineJob('', '', mock_exec)
def test_discover(self): """Tests discovery.""" answers = [] for i in range(1,3): r = mock.create_autospec(dns.rdtypes.IN.SRV.SRV) r.port = 2379 try: method = dns.name.from_unicode except AttributeError: method = dns.name.from_text r.target = method(u'etcd{}.example.com'.format(i)) answers.append(r) dns.resolver.query = mock.create_autospec(dns.resolver.query, return_value=answers) self.machines = etcd.Client.machines etcd.Client.machines = mock.create_autospec(etcd.Client.machines, return_value=[u'https://etcd2.example.com:2379']) c = etcd.Client(srv_domain="example.com", allow_reconnect=True, protocol="https") etcd.Client.machines = self.machines self.assertEquals(c.host, u'etcd1.example.com') self.assertEquals(c.port, 2379) self.assertEquals(c._machines_cache, [u'https://etcd2.example.com:2379'])
def test_acquired_no_timeout(self): self.locker._sequence = 4 returns = [('/_locks/test_lock/4', None), ('/_locks/test_lock/1', '/_locks/test_lock/4')] def side_effect(): return returns.pop() d = { u'action': u'get', u'node': { u'modifiedIndex': 190, u'key': u'/_locks/test_lock/4', u'value': self.locker.uuid } } self._mock_api(200, d) self.locker._get_locker = mock.create_autospec( self.locker._get_locker, side_effect=side_effect) self.assertTrue(self.locker._acquired())
def patch_autospec( self, obj, attribute, spec_set=False, instance=False) -> MagicMock: """Patch `obj.attribute` with an auto-spec of itself. See `mock.create_autospec` and `patch`. :return: The patched-in object. """ spec = getattr(obj, attribute) if isinstance(spec, mock.Base): raise TypeError( "Cannot use a mock object as a specification: %s.%s = %r" % (_get_name(obj), attribute, spec)) value = mock.create_autospec(spec, spec_set, instance) super(MAASTestCase, self).patch(obj, attribute, value) return value
def _test_combine(self, combine_type='transparent'): jsonresponse = self.test_put_share(session_type=combine_type) session_id = jsonresponse['session_id'] payload = { 'client_alias': 'molly', 'share': 'cafe02' } subscribe2 = self.client.put('/combine/%s' % session_id, data=json.dumps(payload)) print(subscribe2.json) self.assert200(subscribe2) self.assertEqual(subscribe2.json['session']['users'][2]['alias'], 'molly') self.user_with_uuid_have_share(subscribe2.json, 'cafe02') print('CombineSession: the second client join a combine session and present its share') payload = { 'client_alias': 'armitage', 'share': 'cafe03' } from ssshare import control control.fxc_web_api_service = create_autospec(FXCWebApiService) control.fxc_web_api_service.combine.return_value = 'secret' subscribe3 = self.client.put('/combine/%s' % session_id, data=json.dumps(payload)) self.assert200(subscribe3) self.user_with_uuid_have_share(subscribe3.json, 'cafe03') self.assertEqual(subscribe3.json['session']['users'][3]['alias'], 'armitage') expected_secret = { 'protocol': 'fxc1', 'quorum': 3, 'sha256': '2bb80d537b1da3e38bd30361aa855686bde0eacd7162fef6a25fe97bf527a25b', 'shares': 5, } if combine_type == 'transparent': expected_secret['secret'] = 'secret' self.assertEqual(subscribe3.json['session']['secret'], expected_secret) print('CombineSession: the third client join a combine session and present its share') return subscribe3.json
def make_fake_signal(*event_types): return webhook_handler(*event_types)( mock.create_autospec(signal_template) )
def test_run_handles_action_response(data_mock, store_mock, signal_mock, context_mock): run_result = Action(create_autospec(MultiTaskData, instance=True)) class Task(BaseTask): def run(self, *args, **kwargs): return run_result result = Task('task-name')._run(data_mock, store_mock, signal_mock, context_mock) assert result == run_result
def data_mock(): yield create_autospec(MultiTaskData, instance=True)
def store_mock(): yield create_autospec(DataStoreDocument, instance=True)
def signal_mock(): m = create_autospec(TaskSignal, instance=True) m.configure_mock(is_stopped=False) yield m
def context_mock(): yield create_autospec(TaskContext, instance=True)
def driver(): ''' Returns a mock selenium driver object ''' return mock.create_autospec(webdriver.Firefox)
def element(): ''' Returns a mock selenium element object ''' return mock.create_autospec(WebElement)
def setUp(self): super().setUp() os.environ['AWS_DEFAULT_REGION'] = self.REGION self.parent = CloudWatchClient('', '', '') self.cwl = self.parent.client = create_autospec(self.parent.client) self.client = LogGroupClient(self.GROUP, self.parent) self.cwl.reset_mock()
def setUp(self): super().setUp() os.environ['AWS_DEFAULT_REGION'] = self.REGION self.client = self.make_client(self.CURSOR, self.GROUP, self.STREAM) self.cwl = self.client.client = create_autospec(self.client.client)
def setUp(self): self.simulation_client = SimulationClient(None, None) self.mock_client = MagicMock() self.mock_client.run_command = create_autospec(acknowledged_command) self.simulation_client._client = self.mock_client
def test_get_block(self): command = acknowledged_command(GetBlockCommand()) self.mock_client.run_command = create_autospec(acknowledged_command, return_value=command) self.assertDictEqual({1: {"test": 1.8}}, self.simulation_client.get_block(2))
def test_get_block_outputs(self): command = acknowledged_command(GetOutputsCommand()) self.mock_client.run_command = create_autospec(acknowledged_command, return_value=command) self.assertIsNone( self.simulation_client.get_block_outputs(2) ) # TODO: return value dict(ObjectId, dict(str, float)
def test_create_autospec_with_name(self): m = mock.create_autospec(object(), name='sweet_func') self.assertIn('sweet_func', repr(m))
def setUp(self): self.project = mock.create_autospec(Project) self.project.name = '123' self.logger = mock.create_autospec(logging.Logger)
def test_it_runs_the_convention_based_tests_service(self): self.project.get_service.return_value = mock.create_autospec(Service) self.project.get_service('tests').create_container().logs.return_value = b'oh you' self.tester = Tester( hook=None, logger=self.logger, ) result = self.tester.test(self.project, '/some/path') self.assertEqual(result.output, 'oh you')
def test_options(self, mock_shlex): mock_shlex.split.return_value = [mock.sentinel.token] option = mock.create_autospec(CommandOption) option.name = mock.sentinel.option_name option.get_cmd_option.return_value = mock.sentinel.cmd_option exe = Executor(options=[option]) options_cmd = exe.get_options({ mock.sentinel.option_name: mock.sentinel.option_val }) option.get_cmd_option.assert_called_with(mock.sentinel.option_val) mock_shlex.split.assert_called_with(mock.sentinel.cmd_option) self.assertListEqual(options_cmd, [mock.sentinel.token])
def setUp(self): self.mock_exe = mock.create_autospec(Executor)
def test_file_results(self): mock_file_result1 = mock.create_autospec(PathWrapper) mock_file_result1.get_paths.return_value = ['/foo', '/bar'] mock_file_result2 = mock.create_autospec(PathWrapper) mock_file_result2.get_paths.return_value = ['/qux'] self.mock_exe.result_paths = [mock_file_result1, mock_file_result2] job = Job(None, mock.sentinel.cwd, self.mock_exe) self.assertListEqual(job.result_paths, ['/foo', '/bar', '/qux']) mock_file_result1.get_paths.assert_called_once_with(mock.sentinel.cwd) mock_file_result2.get_paths.assert_called_once_with(mock.sentinel.cwd)
def setUp(self): self._config = Config() self._client = create_autospec(docker.APIClient) self._helper = DockerHelper(self._config, self._client) self._cid = "cont_id1" self._cid2 = "cont_id2" self._params = {"Id": self._cid, "param1": "1"} self._params2 = {"Id": self._cid2, "param1": "2"}
def _mock_error(self, error_code, msg, cause, method='PUT', fields=None, cluster_id=None): resp = self._prepare_response( 500, {'errorCode': error_code, 'message': msg, 'cause': cause} ) resp.getheader.return_value = cluster_id or "abcdef1234" self.client.http.request_encode_body = mock.create_autospec( self.client.http.request_encode_body, return_value=resp ) self.client.http.request = mock.create_autospec( self.client.http.request, return_value=resp )
def test_watch_timeout(self): """ Exception will be raised if prevValue != value in test_set """ self.client.http.request = mock.create_autospec( self.client.http.request, side_effect=urllib3.exceptions.ReadTimeoutError(self.client.http, "foo", "Read timed out") ) self.assertRaises( etcd.EtcdWatchTimedOut, self.client.watch, '/testKey', )
def test_read_connection_error(self): self.client.http.request = mock.create_autospec( self.client.http.request, side_effect=socket.error() ) self.assertRaises(etcd.EtcdConnectionFailed, self.client.read, '/something') # Direct GET request self.assertRaises(etcd.EtcdConnectionFailed, self.client.api_execute, '/a', 'GET')
def setUp(self): plugins.git_stats.github3 = create_autospec(github3) self.mock_org = create_autospec(github3.orgs.Organization) self.mock_gh = create_autospec(github3.GitHub) self.mock_repo = create_autospec(IGitt.GitHub.GitHub.GitHubRepository) plugins.git_stats.github3.login.return_value = self.mock_gh self.mock_gh.organization.return_value = self.mock_org plugins.git_stats.github3.organization.return_value = self.mock_org self.plugin = plugins.git_stats.GitStats self.plugin.__bases__ = (BotPlugin, )
def setUp(self): plugins.labhub.github3 = create_autospec(github3) self.mock_org = create_autospec(github3.orgs.Organization) self.mock_gh = create_autospec(github3.GitHub) self.mock_team = create_autospec(github3.orgs.Team) self.mock_team.name = PropertyMock() self.mock_team.name = 'mocked team' self.mock_repo = create_autospec(IGitt.GitHub.GitHub.GitHubRepository) plugins.labhub.github3.login.return_value = self.mock_gh self.mock_gh.organization.return_value = self.mock_org self.mock_org.iter_teams.return_value = [self.mock_team] plugins.labhub.github3.organization.return_value = self.mock_org
def test_create_issue_cmd(self): plugins.labhub.GitHub = create_autospec(IGitt.GitHub.GitHub.GitHub) plugins.labhub.GitLab = create_autospec(IGitt.GitLab.GitLab.GitLab) plugins.labhub.GitHubToken = create_autospec(IGitt.GitHub.GitHubToken) plugins.labhub.GitLabPrivateToken = create_autospec(IGitt.GitLab.GitLabPrivateToken) labhub, testbot_private = plugin_testbot( plugins.labhub.LabHub, logging.ERROR, {'BACKEND': 'text', 'ACCESS_CONTROLS':{'create_issue_cmd' : {'allowprivate':False}}} ) labhub.activate() plugins.labhub.GitHubToken.assert_called_with(None) plugins.labhub.GitLabPrivateToken.assert_called_with(None) # Creating issue in private chat testbot_private.assertCommand('!new issue repository this is the title\nbo\ndy', 'You\'re not allowed') # Creating issue in public chat labhub, testbot_public = plugin_testbot( plugins.labhub.LabHub, logging.ERROR, {'BACKEND': 'text'} ) labhub.activate() labhub.REPOS = {'repository': self.mock_repo, 'repository.github.io': self.mock_repo} testbot_public.assertCommand('!new issue repository this is the title\nbo\ndy', 'Here you go') labhub.REPOS['repository'].create_issue.assert_called_once_with( 'this is the title', 'bo\ndy\nOpened by @None at [text]()' ) testbot_public.assertCommand('!new issue repository.github.io another title\nand body', 'Here you go') labhub.REPOS['repository.github.io'].create_issue.assert_called_with( 'another title', 'and body\nOpened by @None at [text]()' ) testbot_public.assertCommand('!new issue coala title', 'repository that does not exist')
def test_unassign_cmd(self): plugins.labhub.GitHub = create_autospec(IGitt.GitHub.GitHub.GitHub) plugins.labhub.GitLab = create_autospec(IGitt.GitLab.GitLab.GitLab) labhub, testbot = plugin_testbot(plugins.labhub.LabHub, logging.ERROR) labhub.activate() labhub.REPOS = {'name': self.mock_repo} mock_iss = create_autospec(IGitt.GitHub.GitHubIssue) self.mock_repo.get_issue.return_value = mock_iss mock_iss.assignees = PropertyMock() mock_iss.assignees = (None, ) mock_iss.unassign = MagicMock() testbot.assertCommand('!unassign https://github.com/coala/name/issues/23', 'you are unassigned now', timeout=10000) self.mock_repo.get_issue.assert_called_with(23) mock_iss.unassign.assert_called_once_with(None) mock_iss.assignees = ('meetmangukiya', ) testbot.assertCommand('!unassign https://github.com/coala/name/issues/23', 'not an assignee on the issue') testbot.assertCommand('!unassign https://github.com/coala/s/issues/52', 'Repository doesn\'t exist.') testbot.assertCommand('!unassign https://gitlab.com/ala/am/issues/532', 'Repository not owned by our org.')
def setUp(self): """setup mock stdin and stdout""" self.stdout = create_autospec(sys.stdout) self.stdin = create_autospec(sys.stdin)
def test_output(): ''' Test that sending a message behaves as expected ''' rtmbot = init_rtmbot() # Mock the slack_client object with Server, Channel objects and needed methods slackclient_mock = create_autospec(SlackClient) server_mock = create_autospec(_server.Server) # Mock Server with channels method and correct return value slackclient_mock.server = server_mock searchlist_mock = create_autospec(_util.SearchList) server_mock.channels = searchlist_mock channel_mock = create_autospec(_channel.Channel) slackclient_mock.server.channels.find.return_value = channel_mock rtmbot.slack_client = slackclient_mock # mock the plugin object to return a sample response plugin_mock = create_autospec(Plugin) plugin_mock.do_output.return_value = [['C12345678', 'test message']] rtmbot.bot_plugins.append(plugin_mock) rtmbot.output() # test that the output matches the expected value channel_mock.send_message.assert_called_with('test message') # test that emoji messages work as expected channel_mock.reset_mock() plugin_mock.reset_mock() plugin_mock.do_output.return_value = [['C12345678', '?? testing']] rtmbot.output() channel_mock.send_message.assert_called_with('?? testing') # test that unicode messages work as expected channel_mock.reset_mock() plugin_mock.reset_mock() plugin_mock.do_output.return_value = [['C12345678', 'ù hœø3ö']] rtmbot.output() channel_mock.send_message.assert_called_with('ù hœø3ö')
def test_master_put_secret_on_joined_session(self): from ssshare import control control.fxc_web_api_service = create_autospec(FXCWebApiService) shares = [ Share(value='cafe01'), Share(value='cafe02'), Share(value='cafe03'), Share(value='cafe04'), Share(value='cafe05') ] control.fxc_web_api_service.split.return_value = shares joined_session = self.test_join_session() print('SplitSession: a master is able to put a secret and its rules into a session') session_id = joined_session['session_id'] payload = { 'session': { 'secret': { 'value': 'my awesome secret', } }, 'client_alias': self.master_alias, 'auth': self._masterkey } response = self.client.put('/split/%s' % session_id, data=json.dumps(payload)) self.assert200(response) self.assertTrue(is_uuid(response.json['session_id'])) self.assertTrue(is_uuid(response.json['session']['users'][1]['auth'])) expected_response = { 'session': { 'ttl': response.json['session']['ttl'], 'alias': 'the session alias', 'secret': { 'quorum': 3, 'shares': 5, 'sha256': '6e1f1d4f6b6c900f3fb72466bbec4a3c7c049fc845a8751a5374227091c1f252', 'secret': 'my awesome secret', 'protocol': 'fxc1' }, 'users': [ { 'auth': self._masterkey, 'alias': self.master_alias, 'role': 'master', 'shareholder': False }, { 'auth': response.json['session']['users'][1]['auth'], 'alias': 'a shareholder', 'role': 'user', 'share': 'cafe01' } ] }, 'session_id': session_id } self.assertEqual(expected_response, response.json) self.assertTrue(control.fxc_web_api_service.split.called) self.assertEqual(shares[0].user, response.json['session']['users'][1]['auth']) return response.json
def test_mark_cmd(self): labhub, testbot = plugin_testbot(plugins.labhub.LabHub, logging.ERROR) labhub.activate() labhub.REPOS = {'a': self.mock_repo} mock_github_mr = create_autospec(GitHubMergeRequest) mock_gitlab_mr = create_autospec(GitLabMergeRequest) mock_github_mr.labels = PropertyMock() mock_gitlab_mr.labels = PropertyMock() mock_github_mr.author = 'johndoe' mock_gitlab_mr.author = 'johndoe' cmd_github = '!mark {} https://github.com/{}/{}/pull/{}' cmd_gitlab = '!mark {} https://gitlab.com/{}/{}/merge_requests/{}' self.mock_repo.get_mr.return_value = mock_github_mr # Non-eistent repo testbot.assertCommand(cmd_github.format('wip', 'a', 'b', '23'), 'Repository doesn\'t exist.') testbot.assertCommand('!mark wip https://gitlab.com/a/b/merge_requests/2', 'Repository doesn\'t exist.') mock_github_mr.web_url = 'https://github.com/coala/a/pull/23' mock_gitlab_mr.web_url = 'https://gitlab.com/coala/a/merge_requests/23' # mark wip mock_github_mr.labels = ['process/pending review'] mock_gitlab_mr.labels = ['process/pending review'] testbot.assertCommand(cmd_github.format('wip', 'coala', 'a', '23'), 'marked work in progress') testbot.assertCommand(cmd_github.format('wip', 'coala', 'a', '23'), '@johndoe, please check your pull request') testbot.assertCommand(cmd_github.format('wip', 'coala', 'a', '23'), 'https://github.com/coala/a/pull/23') self.mock_repo.get_mr.return_value = mock_gitlab_mr testbot.assertCommand(cmd_gitlab.format('wip', 'coala', 'a', '23'), '@johndoe, please check your pull request') testbot.assertCommand(cmd_gitlab.format('wip', 'coala', 'a', '23'), 'https://gitlab.com/coala/a/merge_requests/23') self.mock_repo.get_mr.return_value = mock_github_mr # mark pending mock_github_mr.labels = ['process/wip'] mock_gitlab_mr.labels = ['process/wip'] testbot.assertCommand(cmd_github.format('pending', 'coala', 'a', '23'), 'marked pending review') testbot.assertCommand(cmd_github.format('pending-review', 'coala', 'a', '23'), 'marked pending review') testbot.assertCommand(cmd_github.format('pending review', 'coala', 'a', '23'), 'marked pending review')
