Python mock 模块，create_autospec() 实例源码

def setUp(self):
        class MockResponse:
            def __init__(self, status_code, data):
                self.status_code = status_code
                self.text = json.dumps(data)

        self.mock_response = MockResponse

        self.orig_directory_init = OktaDirectoryConnector.__init__

        OktaDirectoryConnector.__init__ = mock.Mock(return_value=None)
        directory = OktaDirectoryConnector({})
        directory.options = {'all_users_filter': None, 'group_filter_format': '{group}'}
        directory.logger = mock.create_autospec(logging.Logger)
        directory.groups_client = okta.UserGroupsClient('example.com', 'xyz')

        self.directory = directory

def setUp(self):
        class MockResponse:
            def __init__(self, status_code, data):
                self.status_code = status_code
                self.text = json.dumps(data)

        self.mock_response = MockResponse

        self.orig_directory_init = OktaDirectoryConnector.__init__

        OktaDirectoryConnector.__init__ = mock.Mock(return_value=None)
        directory = OktaDirectoryConnector({})

        directory.logger = mock.create_autospec(logging.Logger)
        directory.groups_client = okta.UserGroupsClient('example.com', 'xyz')

        self.directory = directory

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 gitlab_mock():
    all_users = [JOHN, BART, TIMOTHEE, THEO]

    def get_project_members(project_id, query):
        assert project_id in PROJECT_IDS.values()
        return [user for user in all_users if query in user["name"]]

    def get_group_members(group_name, query):
        return [user for user in all_users if query in user["name"]]

    gl_mock = mock.create_autospec(tsrc.gitlab.GitLabHelper, instance=True)
    gl_mock.get_project_members = get_project_members
    gl_mock.get_group_members = get_group_members
    gl_mock.get_project_id = lambda x: PROJECT_IDS[x]
    # Define a few helper methods to make tests nicer to read:
    new_defs = {
        "assert_mr_created": gl_mock.create_merge_request.assert_called_with,
        "assert_mr_not_created": gl_mock.create_merge_request.assert_not_called,
        "assert_mr_updated": gl_mock.update_merge_request.assert_called_with,
        "assert_mr_accepted": gl_mock.accept_merge_request.assert_called_with,
    }
    for name, func in new_defs.items():
        setattr(gl_mock, name, func)
    return gl_mock

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.getParam.return_value = 0 # so mock doesn't do lazy binds

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        #Mock cableController
        self.shot.cable = mock.create_autospec(CableController)
        self.shot.cable.currentU = 0.5  # half-way through spline
        self.shot.cable.currentSeg = 0

        # create two waypoints
        loc = LocationGlobalRelative(37.873168,-122.302062, 0)
        self.shot.waypoints.append(Waypoint(loc,-90,0))
        self.shot.waypoints.append(Waypoint(loc,0,90))

        self.shot.camSpline = CatmullRom([Vector2(-180, -90), Vector2(-90, 0), Vector2(0, 90), Vector2(90, 180)])

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.getParam.return_value = 0 # so mock doesn't do lazy binds
        shotmgr.appMgr = Mock()

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        #Mock setButtonMappings()
        self.shot.setButtonMappings = Mock()

        self.shot.duplicateCheck = Mock(return_value=False)

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.getParam.return_value = 0 # so mock doesn't do lazy binds
        shotmgr.appMgr = Mock()

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        #cableCamPlaying
        self.cableCamPlaying = False

        # Mock setButtonMappings
        self.shot.setButtonMappings = Mock()

        #Mock duplicate check
        self.shot.duplicateCheck = Mock(return_value=False)

        # default packet
        self.point = (0, -15.3, 0, 37.330674, -122.028759, 15, 0, 90, 0, 0)

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.buttonManager = Mock()

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        # 1 waypoint
        self.shot.waypoints = [1]

        # Play mode
        self.shot.cableCamPlaying = True

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.appMgr = Mock()

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        #value 
        self.seek = (.3, 0)

        #cable
        self.shot.cable = [1]

        #mock setCruiseSpeed
        self.shot.setCruiseSpeed = Mock()

        #mock checkToNotifyApp
        self.shot.checkToNotifyApp = Mock()

        #attaching
        self.shot.attaching = False

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)

        #Run the shot constructor
        self.shot = multipoint.MultipointShot(vehicle, shotmgr)

        #Mock updatePlaybackStatus
        self.shot.updatePlaybackStatus = Mock()

        #Mock cable
        self.shot.cable = mock.create_autospec(CableController)
        self.shot.cable.reachedTarget = Mock(return_value = False)
        self.shot.cable.currentSeg = 1
        self.shot.cable.speed = 1

        #Mock setCruiseSpeed
        self.shot.setCruiseSpeed = Mock()
        self.shot.targetP = 1

def setUp(self):
        self.v = mock.create_autospec(Vehicle)
        self.v.location.global_relative_frame = LocationGlobalRelative(-48.5468695, 5.68464, 10.5)
        self.v.mount_status = [-80]
        self.mockMgr = Mock()
        self.mockMgr.buttonManager = Mock()
        self.mockMgr.getParam = Mock(return_value=500.0)
        self.controller = CableCamShot(self.v, self.mockMgr)
        loc2 = location_helpers.newLocationFromAzimuthAndDistance(self.v.location.global_relative_frame, 23.4, 25.0)
        self.startYaw = 12.4
        self.startPitch = -16.7
        waypt1 = cable_cam.Waypoint( loc2, self.startYaw, self.startPitch )
        self.controller.waypoints.append( waypt1 )
        self.endYaw = 175.4
        self.endPitch = -83.4
        waypt2 = cable_cam.Waypoint( self.v.location.global_relative_frame, self.endYaw, self.endPitch )
        self.controller.waypoints.append( waypt2 )
        self.controller.deadReckoningTicks = 0
        self.controller.accel = 0.0
        self.controller.totalDistance = location_helpers.getDistanceFromPoints3d(
                self.v.location.global_relative_frame, loc2)
        self.v.message_factory = Mock()
        # turn off dead reckoning
        self.v.groundspeed = 0.0
        self.controller.desiredSpeed = 0.0

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = orbit.OrbitShot(vehicle, shotmgr)

        #Set ROI
        self.shot.roi =  LocationGlobalRelative(37.873168,-122.302062, 0)

        #Mock setButtonMappings()
        self.shot.setButtonMappings = Mock()

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = orbit.OrbitShot(vehicle, shotmgr)

        #Mock pathController
        self.shot.pathController = mock.create_autospec(OrbitController)
        self.shot.pathController.radius = self.DISTANCE

        #set vehicle location
        self.shot.vehicle.location.global_relative_frame = LocationGlobalRelative(-4.897465, 111.4894, 10.0)

        #set mount status
        self.shot.vehicle.mount_status = [-20.0, 40.0, 30.0]

        #set ROI
        self.shot.roi = location_helpers.newLocationFromAzimuthAndDistance(self.shot.vehicle.location.global_relative_frame, self.ARBITRARY_HEADING, self.DISTANCE)

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        self.shot.pathHandler = mock.create_autospec(pathHandler.PathHandler)
        self.shot.pathHandler.cruiseSpeed = 0

        #Mock the pathController object
        self.shot.followState = FOLLOW_ORBIT
        self.shot.pathController = mock.create_autospec(OrbitController)

        self.shot.ROIAltitudeOffset = 0.0


        self.shot.ROIAltitudeOffset = 0.0

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        #Mock Methods
        self.shot.updateMountStatus = Mock()
        self.shot.initLookAtMeController = Mock()
        self.shot.initOrbitController = Mock()
        self.shot.initLeashController = Mock()
        self.shot.initFreeLookController = Mock()
        self.shot.updateMountStatus = Mock()
        self.shot.updateAppOptions = Mock()
        self.shot.setButtonMappings = Mock()


        #Setup Attributes
        self.shot.followState = FOLLOW_WAIT
        self.shot.rawROI = location_helpers.newLocationFromAzimuthAndDistance(self.shot.vehicle.location.global_relative_frame, self.ARBITRARY_HEADING, self.DISTANCE)

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])
        shotmgr.buttonManager = Mock()
        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        # rawROI
        self.ROI = LocationGlobalRelative(37.873168,-122.302062, 0) #sample ROI, used throughout
        self.shot.rawROI = self.ROI

        # mock methods
        self.shot.initState = Mock()
        location_helpers.calcYawPitchFromLocations = Mock()
        location_helpers.calcYawPitchFromLocations.return_value = (0.0,0.0)

        #roiDeltaTime
        self.shot.roiDeltaTime = 0.04 # 25Hz (guess)

        #init vars
        self.shot.roiVelocity = Vector3()

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        # mock methods
        location_helpers.getDistanceFromPoints = Mock()
        location_helpers.calcAzimuthFromPoints = Mock()

        location_helpers.getDistanceFromPoints.return_value = 10.0
        location_helpers.calcAzimuthFromPoints.return_value = 0.0

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        #Neutral sticks
        throttle = 0.0
        roll = 0.0
        pitch = 0.0
        yaw = 0.0 
        self.channels = [throttle, roll, pitch, yaw, 0.0, 0.0, 0.0, 0.0]

def setUp(self):
        #Create a mock vehicle object
        vehicle = mock.create_autospec(Vehicle)

        #Create a mock shotManager object
        shotmgr = mock.create_autospec(ShotManager)
        shotmgr.rcMgr = Mock(specs=['remapper'])

        #Run the shot constructor
        self.shot = follow.FollowShot(vehicle, shotmgr)

        self.shot.camYaw = 0 #init to zero. Can be permuted below.
        #Neutral sticks, unless permuted in the methods below
        throttle = 0.0
        roll = 0.0
        pitch = 0.0
        yaw = 0.0
        self.channels = [throttle, roll, pitch, yaw, 0.0, 0.0, 0.0, 0.0]

def test_nodes(self, mock_connect):
        cloud_management_mock = mock.create_autospec(
            cloud_management.CloudManagement)
        mock_connect.return_value = cloud_management_mock
        cloud_management_mock.get_nodes.return_value.hosts = [
            node_collection.Host(
                ip='10.0.0.2', mac='09:7b:74:90:63:c1', fqdn='node1.local'),
            node_collection.Host(
                ip='10.0.0.3', mac='09:7b:74:90:63:c2', fqdn='node2.local')]

        with self.runner.isolated_filesystem():
            with open('my.yaml', 'w') as f:
                f.write('foo')
            myconf = os.path.abspath(f.name)
            result = self.runner.invoke(main.main, ['nodes', '-c', myconf])
        self.assertEqual(0, result.exit_code)
        self.assertEqual(
            '- fqdn: node1.local\n'
            '  ip: 10.0.0.2\n'
            '  mac: 09:7b:74:90:63:c1\n'
            '- fqdn: node2.local\n'
            '  ip: 10.0.0.3\n'
            '  mac: 09:7b:74:90:63:c2\n', result.output)

def test_header_challenge_auth(cred_auth, oidc_auth, req):
    resp = create_autospec(requests.Response)
    resp.status_code = 401
    resp.headers = {"www-authenticate": "oauthjwt"}
    req.return_value = resp

    auth.header_challenge_auth("url")
    oidc_auth.assert_called_once()

    resp2 = create_autospec(requests.Response)
    resp2.status_code = 401
    resp2.headers = {"www-authenticate": "acsjwt"}
    req.return_value = resp2

    auth.header_challenge_auth("url")
    cred_auth.assert_called_once()

def test_get_providers(config, req_mock):
    resp = create_autospec(requests.Response)
    resp.return_value = {}
    req_mock.return_value = resp
    config.return_value = "http://localhost"

    auth.get_providers()
    req_mock.assert_called_with(
        "http://localhost/acs/api/v1/auth/providers")

    # test url construction valid with trailing slash
    config.return_value = "http://localhost/"

    auth.get_providers()
    req_mock.assert_called_with(
        "http://localhost/acs/api/v1/auth/providers")

def _assert_pod_update_invoked_successfully(
        emitter, pod_id, force, resource, deployment_id, emitted):
    resource_reader = create_autospec(main.ResourceReader)
    resource_reader.get_resource.return_value = resource

    marathon_client = _marathon_client_fixture()
    marathon_client.update_pod.return_value = deployment_id

    subcmd = main.MarathonSubcommand(resource_reader, lambda: marathon_client)
    returncode = subcmd.pod_update(pod_id, force)

    assert returncode == 0
    marathon_client.show_pod.assert_called_with(pod_id)
    resource_reader.get_resource.assert_called_with(name=None)
    marathon_client.update_pod.assert_called_with(
        pod_id, pod_json=resource, force=force)
    emitter.publish.assert_called_with(emitted)

def test_kill_pod_instances_executes_successfully():
    pod_id = 'foo'
    instance_ids = ['instance1', 'instance2']
    path = 'v2/pods/foo::instances'
    response_json = {'some': ['instance', 'status']}

    mock_response = mock.create_autospec(requests.Response)
    mock_response.json.return_value = response_json

    marathon_client, rpc_client = _create_fixtures()
    rpc_client.http_req.return_value = mock_response

    path_format_method_name = 'dcos.marathon.Client._marathon_id_path_format'
    path_format = mock.MagicMock()
    path_format.return_value = path

    with mock.patch(path_format_method_name, new=path_format):
        actual_json = marathon_client.kill_pod_instances(pod_id, instance_ids)

    path_format.assert_called_with('v2/pods/{}::instances', pod_id)
    rpc_client.http_req.assert_called_with(
        http.delete, path, json=instance_ids)
    assert actual_json == response_json

def test_pod_feature_supported_gets_success_response(head_fn):
    def invoke_test_case(status_code):
        mock_response = mock.create_autospec(requests.Response)
        mock_response.status_code = status_code
        head_fn.return_value = mock_response

        rpc_client = rpcclient.RpcClient('http://base/url', timeout=42)
        marathon_client = marathon.Client(rpc_client)
        is_supported = marathon_client.pod_feature_supported()

        head_fn.assert_called_with('http://base/url/v2/pods', timeout=42)

        return is_supported

    assert invoke_test_case(status_code=200)
    assert invoke_test_case(status_code=204)

    assert not invoke_test_case(status_code=100)
    assert not invoke_test_case(status_code=302)

def test_rpc_client_http_req_propagates_method_fn_exception_1():
    request = requests.Request(method='ANY', url='http://arbitrary/url')
    # Need the mock so that the `.json()` method can be overridden
    response = mock.create_autospec(requests.Response)
    response.status_code = 403
    response.reason = 'Forbidden'
    response.request = request
    response.json.side_effect = Exception('not JSON')

    def method_fn(*args, **kwargs):
        raise DCOSHTTPException(response)

    rpc_client = rpcclient.RpcClient('http://base/url')
    with pytest.raises(DCOSException) as e:
        rpc_client.http_req(method_fn, 'some/path')

    expected_message = rpcclient.RpcClient.response_error_message(
        status_code=403,
        reason='Forbidden',
        request_method='ANY',
        request_url='http://arbitrary/url',
        json_body=None)
    assert str(e).endswith(expected_message)

def _assert_method_raises_dcos_exception_for_json_parse_errors(invoke_method):
    def assert_test_case(non_json):
        mock_response = mock.create_autospec(requests.Response)
        mock_response.json.side_effect = Exception()
        mock_response.text = non_json

        marathon_client, rpc_client = _create_fixtures()
        rpc_client.http_req.return_value = mock_response

        with pytest.raises(DCOSException) as exception_info:
            invoke_method(marathon_client)

        pattern = ('Error: Response from Marathon was not in expected JSON '
                   'format:\n(.*)')
        actual_error = str(exception_info.value)
        _assert_matches_with_groups(pattern, actual_error, (non_json,))

    assert_test_case('not-json')
    assert_test_case('{"oops"}')

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'instances'
        self.maxDiff = None
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(instance_dao.InstanceDao)
        self.pipeline = (
            load_instances_pipeline.LoadInstancesPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_instances \
            .FAKE_PROJECT_INSTANCES_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'backend_services'
        self.maxDiff = None
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(bsdao.BackendServiceDao)
        self.pipeline = (
            load_backend_services_pipeline.LoadBackendServicesPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_backend_services \
            .FAKE_PROJECT_BACKEND_SERVICES_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'instance_templates'
        self.maxDiff = None
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(instance_template_dao.InstanceTemplateDao)
        self.pipeline = (
            load_instance_templates_pipeline.LoadInstanceTemplatesPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_instance_templates \
            .FAKE_PROJECT_INSTANCE_TEMPLATES_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.mock_appengine = mock.create_autospec(appengine.AppEngineClient)
        self.mock_dao = mock.create_autospec(appengine_dao.AppEngineDao)
        self.pipeline = (
            load_appengine_pipeline.LoadAppenginePipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_appengine,
                self.mock_dao))
        self.project_ids = fake_appengine_applications \
            .FAKE_PROJECT_APPLICATIONS_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'instance_group_managers'
        self.maxDiff = None
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(instance_group_manager_dao.InstanceGroupManagerDao)
        self.pipeline = (
            load_instance_group_managers_pipeline.LoadInstanceGroupManagersPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_instance_group_managers \
            .FAKE_PROJECT_INSTANCE_GROUP_MANAGERS_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'forwarding_rules'
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(frdao.ForwardingRulesDao)
        self.pipeline = (
            load_forwarding_rules_pipeline.LoadForwardingRulesPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_forwarding_rules \
            .FAKE_PROJECT_FWD_RULES_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.mock_iam = mock.create_autospec(iam.IAMClient)
        self.mock_dao = mock.create_autospec(
            service_account_dao.ServiceAccountDao)
        self.pipeline = (
            load_service_accounts_pipeline.LoadServiceAccountsPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_iam,
                self.mock_dao))
        self.project_ids = fake_service_accounts \
            .FAKE_PROJECT_SERVICE_ACCOUNTS_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
                         for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
                         if p['project_id'] in self.project_ids]

def setUp(self):
        """Set up."""
        self.cycle_timestamp = '20001225T120000Z'
        self.configs = fake_configs.FAKE_CONFIGS
        self.resource_name = 'instance_groups'
        self.maxDiff = None
        self.mock_compute_client = mock.create_autospec(compute.ComputeClient)
        self.mock_dao = mock.create_autospec(instance_group_dao.InstanceGroupDao)
        self.pipeline = (
            load_instance_groups_pipeline.LoadInstanceGroupsPipeline(
                self.cycle_timestamp,
                self.configs,
                self.mock_compute_client,
                self.mock_dao))

        self.project_ids = fake_instance_groups \
            .FAKE_PROJECT_INSTANCE_GROUPS_MAP.keys()
        self.projects = [project_dao.ProjectDao.map_row_to_object(p)
             for p in fake_projects.EXPECTED_LOADABLE_PROJECTS
             if p['project_id'] in self.project_ids]

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_before_request_refreshes(self, jwt_grant):
        credentials = self.make_credentials()
        token = 'token'
        jwt_grant.return_value = (
            token, _helpers.utcnow() + datetime.timedelta(seconds=500), None)
        request = mock.create_autospec(transport.Request, instance=True)

        # Credentials should start as invalid
        assert not credentials.valid

        # before_request should cause a refresh
        credentials.before_request(
            request, 'GET', 'http://example.com?a=1#3', {})

        # The refresh endpoint should've been called.
        assert jwt_grant.called

        # Credentials should now be valid.
        assert credentials.valid

def test_call_no_refresh(self):
        credentials = CredentialsStub()
        request = mock.create_autospec(transport.Request)

        plugin = google.auth.transport.grpc.AuthMetadataPlugin(
            credentials, request)

        context = mock.create_autospec(grpc.AuthMetadataContext, instance=True)
        context.method_name = mock.sentinel.method_name
        context.service_url = mock.sentinel.service_url
        callback = mock.create_autospec(grpc.AuthMetadataPluginCallback)

        plugin(context, callback)

        callback.assert_called_once_with(
            [(u'authorization', u'Bearer {}'.format(credentials.token))], None)

def test_call_refresh(self):
        credentials = CredentialsStub()
        credentials.expiry = datetime.datetime.min + _helpers.CLOCK_SKEW
        request = mock.create_autospec(transport.Request)

        plugin = google.auth.transport.grpc.AuthMetadataPlugin(
            credentials, request)

        context = mock.create_autospec(grpc.AuthMetadataContext, instance=True)
        context.method_name = mock.sentinel.method_name
        context.service_url = mock.sentinel.service_url
        callback = mock.create_autospec(grpc.AuthMetadataPluginCallback)

        plugin(context, callback)

        assert credentials.token == 'token1'
        callback.assert_called_once_with(
            [(u'authorization', u'Bearer {}'.format(credentials.token))], None)

def test_remind(self):
        """
        remind should send remind and mark job as done.
        """
        job = self.create_job(state='queued')
        with mock_instance('sndlatr.gmail.Mailman') as mailman:
            mail = mock.create_autospec(spec=gae_mail.EmailMessage)
            mailman.build_reply.return_value = mail
            job.remind('token')
            mailman.send_mail.assert_called_with(mail)
            mailman.build_reply.assert_called_with(job.thread_id_int,
                                                   mock.ANY)
            mailman.quit.assert_called_with()
        job = job.key.get()
        self.assertEquals(job.state, 'done')

def test_find_reply(self):
        """ Should find first unknown reply """
        job = self.create_job(known_message_ids=['abc1'])
        mailman = mock.create_autospec(gmail.Mailman)
        mail1 = create_thread_mail(message_id='abc1')
        mail2 = create_thread_mail(message_id='abc2')
        mail3 = create_thread_mail(message_id='abc3')
        mailman.get_thread.return_value = [mail1, mail2, mail3]
        reply = job.find_reply(mailman)
        # should match first unknown reply
        self.assertEqual(reply, mail2)

def test_find_reply_none(self):
        """ Return value should be none if there is no unkown reply """
        job = self.create_job(known_message_ids=['abc1', 'abc2'])
        mailman = mock.create_autospec(gmail.Mailman)
        mailman.get_thread.return_value = [
            create_thread_mail(message_id='abc1'),
            create_thread_mail(message_id='abc2')]
        self.assertIsNone(job.find_reply(mailman))

def test_BrowserReturnPositioner():
    fake_browser = mock.create_autospec(EpiphanyBrowser)
    html_template = jinja2.Template('genre: {{ genre }}')
    tempdir = tempfile.mkdtemp()
    try:
        positioner = BrowserReturnPositioner(fake_browser, html_template, tempdir)
        positioner.show('infraA[?3]')

        assert fake_browser.open.called

        positioner.hide()

        assert fake_browser.close.called
    finally:
        shutil.rmtree(tempdir)

def test_KintoneLogger():
    kintone = mock.create_autospec(Kintone)
    logger = KintoneLogger('system1', kintone)
    logger.log_nfc_connected('0123', 'user-hoge')

    assert kintone.add_log.called

    system_id, json_msg = kintone.add_log.call_args[0]
    json_obj = json.loads(json_msg)

    assert system_id == 'system1'
    assert json_obj['employee_id'] == '0123'
    assert json_obj['user_code'] == 'user-hoge'

def setUp(self):
        self.orig_directory_init = OktaDirectoryConnector.__init__

        OktaDirectoryConnector.__init__ = mock.Mock(return_value=None)
        directory = OktaDirectoryConnector({})
        directory.options = {'all_users_filter': None, 'group_filter_format': '{group}'}
        directory.logger = mock.create_autospec(logging.Logger)
        directory.groups_client = okta.UserGroupsClient('example.com', 'xyz')

        self.directory = directory

def setUp(self):
        class MockResponse:
            def __init__(self, status_code, data):
                self.status_code = status_code
                self.text = json.dumps(data)
                self.links = {}

        self.mock_response = MockResponse
        self.orig_directory_init = OktaDirectoryConnector.__init__
        OktaDirectoryConnector.__init__ = mock.Mock(return_value=None)
        directory = OktaDirectoryConnector({})
        directory.logger = mock.create_autospec(logging.Logger)
        directory.groups_client = okta.UserGroupsClient('example.com', 'xyz')
        directory.user_identity_type = 'enterpriseID'
        self.directory = directory

def recruiter(self, active_config):
        from dallinger.mturk import MTurkService
        from dallinger.recruiters import MTurkRecruiter
        with mock.patch.multiple('dallinger.recruiters',
                                 os=mock.DEFAULT,
                                 get_base_url=mock.DEFAULT) as mocks:
            mocks['get_base_url'].return_value = 'http://fake-domain'
            mocks['os'].getenv.return_value = 'fake-host-domain'
            mockservice = mock.create_autospec(MTurkService)
            active_config.extend({'mode': u'sandbox'})
            r = MTurkRecruiter()
            r.mturkservice = mockservice('fake key', 'fake secret')
            r.mturkservice.check_credentials.return_value = True
            r.mturkservice.create_hit.return_value = {'type_id': 'fake type id'}
            return r