python类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]