test_marathon_pod.py 文件源码

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)