feat: add operations rest client to support long-running operations. (#311)

* feat: add operations rest client to support long-running operations.

* fix: address test coverage gaps in operations rest client.

* fix: removed stray print statement.

* fix: address lint, blacken, and mypy issues.

* fix: address pytype, more coverage issues

* fix: addressed additional pytype issues and one coverage line.

* fix: renamed OperationsRestClient to AbstractOperationsClient.

Co-authored-by: Kenneth Bandes <kbandes@google.com>

7 files changed, 2319 insertions, 1 deletions

diff --git a/google/api_core/operations_v1/__init__.py b/google/api_core/operations_v1/__init__.py
index d7f963e..6118645 100644
--- a/google/api_core/operations_v1/__init__.py
+++ b/google/api_core/operations_v1/__init__.py
@@ -14,7 +14,14 @@
 
 """Package for interacting with the google.longrunning.operations meta-API."""
 
+from google.api_core.operations_v1.abstract_operations_client import AbstractOperationsClient
 from google.api_core.operations_v1.operations_async_client import OperationsAsyncClient
 from google.api_core.operations_v1.operations_client import OperationsClient
+from google.api_core.operations_v1.transports.rest import OperationsRestTransport
 
-__all__ = ["OperationsAsyncClient", "OperationsClient"]
+__all__ = [
+    "AbstractOperationsClient",
+    "OperationsAsyncClient",
+    "OperationsClient",
+    "OperationsRestTransport"
+]
diff --git a/google/api_core/operations_v1/abstract_operations_client.py b/google/api_core/operations_v1/abstract_operations_client.py
new file mode 100644
index 0000000..631094e
--- /dev/null
+++ b/google/api_core/operations_v1/abstract_operations_client.py
@@ -0,0 +1,564 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from collections import OrderedDict
+from distutils import util
+import os
+import re
+from typing import Dict, Optional, Sequence, Tuple, Type, Union
+
+from google.api_core import client_options as client_options_lib  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core import retry as retries  # type: ignore
+from google.api_core.operations_v1 import pagers
+from google.api_core.operations_v1.transports.base import (
+    DEFAULT_CLIENT_INFO,
+    OperationsTransport,
+)
+from google.api_core.operations_v1.transports.rest import OperationsRestTransport
+from google.auth import credentials as ga_credentials  # type: ignore
+from google.auth.exceptions import MutualTLSChannelError  # type: ignore
+from google.auth.transport import mtls  # type: ignore
+from google.longrunning import operations_pb2
+from google.oauth2 import service_account  # type: ignore
+
+OptionalRetry = Union[retries.Retry, object]
+
+
+class AbstractOperationsClientMeta(type):
+    """Metaclass for the Operations client.
+
+    This provides class-level methods for building and retrieving
+    support objects (e.g. transport) without polluting the client instance
+    objects.
+    """
+
+    _transport_registry = OrderedDict()  # type: Dict[str, Type[OperationsTransport]]
+    _transport_registry["rest"] = OperationsRestTransport
+
+    def get_transport_class(
+        cls, label: Optional[str] = None,
+    ) -> Type[OperationsTransport]:
+        """Returns an appropriate transport class.
+
+        Args:
+            label: The name of the desired transport. If none is
+                provided, then the first transport in the registry is used.
+
+        Returns:
+            The transport class to use.
+        """
+        # If a specific transport is requested, return that one.
+        if label:
+            return cls._transport_registry[label]
+
+        # No transport is requested; return the default (that is, the first one
+        # in the dictionary).
+        return next(iter(cls._transport_registry.values()))
+
+
+class AbstractOperationsClient(metaclass=AbstractOperationsClientMeta):
+    """Manages long-running operations with an API service.
+
+    When an API method normally takes long time to complete, it can be
+    designed to return [Operation][google.api_core.operations_v1.Operation] to the
+    client, and the client can use this interface to receive the real
+    response asynchronously by polling the operation resource, or pass
+    the operation resource to another API (such as Google Cloud Pub/Sub
+    API) to receive the response. Any API service that returns
+    long-running operations should implement the ``Operations``
+    interface so developers can have a consistent client experience.
+    """
+
+    @staticmethod
+    def _get_default_mtls_endpoint(api_endpoint):
+        """Converts api endpoint to mTLS endpoint.
+
+        Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to
+        "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively.
+        Args:
+            api_endpoint (Optional[str]): the api endpoint to convert.
+        Returns:
+            str: converted mTLS api endpoint.
+        """
+        if not api_endpoint:
+            return api_endpoint
+
+        mtls_endpoint_re = re.compile(
+            r"(?P<name>[^.]+)(?P<mtls>\.mtls)?(?P<sandbox>\.sandbox)?(?P<googledomain>\.googleapis\.com)?"
+        )
+
+        m = mtls_endpoint_re.match(api_endpoint)
+        name, mtls, sandbox, googledomain = m.groups()
+        if mtls or not googledomain:
+            return api_endpoint
+
+        if sandbox:
+            return api_endpoint.replace(
+                "sandbox.googleapis.com", "mtls.sandbox.googleapis.com"
+            )
+
+        return api_endpoint.replace(".googleapis.com", ".mtls.googleapis.com")
+
+    DEFAULT_ENDPOINT = "longrunning.googleapis.com"
+    DEFAULT_MTLS_ENDPOINT = _get_default_mtls_endpoint.__func__(  # type: ignore
+        DEFAULT_ENDPOINT
+    )
+
+    @classmethod
+    def from_service_account_info(cls, info: dict, *args, **kwargs):
+        """Creates an instance of this client using the provided credentials
+            info.
+
+        Args:
+            info (dict): The service account private key info.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Returns:
+            AbstractOperationsClient: The constructed client.
+        """
+        credentials = service_account.Credentials.from_service_account_info(info)
+        kwargs["credentials"] = credentials
+        return cls(*args, **kwargs)
+
+    @classmethod
+    def from_service_account_file(cls, filename: str, *args, **kwargs):
+        """Creates an instance of this client using the provided credentials
+            file.
+
+        Args:
+            filename (str): The path to the service account private key json
+                file.
+            args: Additional arguments to pass to the constructor.
+            kwargs: Additional arguments to pass to the constructor.
+
+        Returns:
+            AbstractOperationsClient: The constructed client.
+        """
+        credentials = service_account.Credentials.from_service_account_file(filename)
+        kwargs["credentials"] = credentials
+        return cls(*args, **kwargs)
+
+    from_service_account_json = from_service_account_file
+
+    @property
+    def transport(self) -> OperationsTransport:
+        """Returns the transport used by the client instance.
+
+        Returns:
+            OperationsTransport: The transport used by the client
+                instance.
+        """
+        return self._transport
+
+    @staticmethod
+    def common_billing_account_path(billing_account: str,) -> str:
+        """Returns a fully-qualified billing_account string."""
+        return "billingAccounts/{billing_account}".format(
+            billing_account=billing_account,
+        )
+
+    @staticmethod
+    def parse_common_billing_account_path(path: str) -> Dict[str, str]:
+        """Parse a billing_account path into its component segments."""
+        m = re.match(r"^billingAccounts/(?P<billing_account>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_folder_path(folder: str,) -> str:
+        """Returns a fully-qualified folder string."""
+        return "folders/{folder}".format(folder=folder,)
+
+    @staticmethod
+    def parse_common_folder_path(path: str) -> Dict[str, str]:
+        """Parse a folder path into its component segments."""
+        m = re.match(r"^folders/(?P<folder>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_organization_path(organization: str,) -> str:
+        """Returns a fully-qualified organization string."""
+        return "organizations/{organization}".format(organization=organization,)
+
+    @staticmethod
+    def parse_common_organization_path(path: str) -> Dict[str, str]:
+        """Parse a organization path into its component segments."""
+        m = re.match(r"^organizations/(?P<organization>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_project_path(project: str,) -> str:
+        """Returns a fully-qualified project string."""
+        return "projects/{project}".format(project=project,)
+
+    @staticmethod
+    def parse_common_project_path(path: str) -> Dict[str, str]:
+        """Parse a project path into its component segments."""
+        m = re.match(r"^projects/(?P<project>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    @staticmethod
+    def common_location_path(project: str, location: str,) -> str:
+        """Returns a fully-qualified location string."""
+        return "projects/{project}/locations/{location}".format(
+            project=project, location=location,
+        )
+
+    @staticmethod
+    def parse_common_location_path(path: str) -> Dict[str, str]:
+        """Parse a location path into its component segments."""
+        m = re.match(r"^projects/(?P<project>.+?)/locations/(?P<location>.+?)$", path)
+        return m.groupdict() if m else {}
+
+    def __init__(
+        self,
+        *,
+        credentials: Optional[ga_credentials.Credentials] = None,
+        transport: Union[str, OperationsTransport, None] = None,
+        client_options: Optional[client_options_lib.ClientOptions] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+    ) -> None:
+        """Instantiates the operations client.
+
+        Args:
+            credentials (Optional[google.auth.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+            transport (Union[str, OperationsTransport]): The
+                transport to use. If set to None, a transport is chosen
+                automatically.
+            client_options (google.api_core.client_options.ClientOptions): Custom options for the
+                client. It won't take effect if a ``transport`` instance is provided.
+                (1) The ``api_endpoint`` property can be used to override the
+                default endpoint provided by the client. GOOGLE_API_USE_MTLS_ENDPOINT
+                environment variable can also be used to override the endpoint:
+                "always" (always use the default mTLS endpoint), "never" (always
+                use the default regular endpoint) and "auto" (auto switch to the
+                default mTLS endpoint if client certificate is present, this is
+                the default value). However, the ``api_endpoint`` property takes
+                precedence if provided.
+                (2) If GOOGLE_API_USE_CLIENT_CERTIFICATE environment variable
+                is "true", then the ``client_cert_source`` property can be used
+                to provide client certificate for mutual TLS transport. If
+                not provided, the default SSL client certificate will be used if
+                present. If GOOGLE_API_USE_CLIENT_CERTIFICATE is "false" or not
+                set, no client certificate will be used.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
+                creation failed for any reason.
+        """
+        if isinstance(client_options, dict):
+            client_options = client_options_lib.from_dict(client_options)
+        if client_options is None:
+            client_options = client_options_lib.ClientOptions()
+
+        # Create SSL credentials for mutual TLS if needed.
+        use_client_cert = bool(
+            util.strtobool(os.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"))
+        )
+
+        client_cert_source_func = None
+        is_mtls = False
+        if use_client_cert:
+            if client_options.client_cert_source:
+                is_mtls = True
+                client_cert_source_func = client_options.client_cert_source
+            else:
+                is_mtls = mtls.has_default_client_cert_source()
+                if is_mtls:
+                    client_cert_source_func = mtls.default_client_cert_source()
+                else:
+                    client_cert_source_func = None
+
+        # Figure out which api endpoint to use.
+        if client_options.api_endpoint is not None:
+            api_endpoint = client_options.api_endpoint
+        else:
+            use_mtls_env = os.getenv("GOOGLE_API_USE_MTLS_ENDPOINT", "auto")
+            if use_mtls_env == "never":
+                api_endpoint = self.DEFAULT_ENDPOINT
+            elif use_mtls_env == "always":
+                api_endpoint = self.DEFAULT_MTLS_ENDPOINT
+            elif use_mtls_env == "auto":
+                if is_mtls:
+                    api_endpoint = self.DEFAULT_MTLS_ENDPOINT
+                else:
+                    api_endpoint = self.DEFAULT_ENDPOINT
+            else:
+                raise MutualTLSChannelError(
+                    "Unsupported GOOGLE_API_USE_MTLS_ENDPOINT value. Accepted "
+                    "values: never, auto, always"
+                )
+
+        # Save or instantiate the transport.
+        # Ordinarily, we provide the transport, but allowing a custom transport
+        # instance provides an extensibility point for unusual situations.
+        if isinstance(transport, OperationsTransport):
+            # transport is a OperationsTransport instance.
+            if credentials or client_options.credentials_file:
+                raise ValueError(
+                    "When providing a transport instance, "
+                    "provide its credentials directly."
+                )
+            if client_options.scopes:
+                raise ValueError(
+                    "When providing a transport instance, provide its scopes "
+                    "directly."
+                )
+            self._transport = transport
+        else:
+            Transport = type(self).get_transport_class(transport)
+            self._transport = Transport(
+                credentials=credentials,
+                credentials_file=client_options.credentials_file,
+                host=api_endpoint,
+                scopes=client_options.scopes,
+                client_cert_source_for_mtls=client_cert_source_func,
+                quota_project_id=client_options.quota_project_id,
+                client_info=client_info,
+                always_use_jwt_access=True,
+            )
+
+    def list_operations(
+        self,
+        name: str,
+        filter_: Optional[str] = None,
+        *,
+        page_size: Optional[int] = None,
+        page_token: Optional[str] = None,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> pagers.ListOperationsPager:
+        r"""Lists operations that match the specified filter in the request.
+        If the server doesn't support this method, it returns
+        ``UNIMPLEMENTED``.
+
+        NOTE: the ``name`` binding allows API services to override the
+        binding to use different resource name schemes, such as
+        ``users/*/operations``. To override the binding, API services
+        can add a binding such as ``"/v1/{name=users/*}/operations"`` to
+        their service configuration. For backwards compatibility, the
+        default name includes the operations collection id, however
+        overriding users must ensure the name binding is the parent
+        resource, without the operations collection id.
+
+        Args:
+            name (str):
+                The name of the operation's parent
+                resource.
+            filter_ (str):
+                The standard list filter.
+                This corresponds to the ``filter`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.api_core.operations_v1.pagers.ListOperationsPager:
+                The response message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+
+                Iterating over this object will yield results and
+                resolve additional pages automatically.
+
+        """
+        # Create a protobuf request object.
+        request = operations_pb2.ListOperationsRequest(name=name, filter=filter_)
+        if page_size is not None:
+            request.page_size = page_size
+        if page_token is not None:
+            request.page_token = page_token
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.list_operations]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
+
+        # This method is paged; wrap the response in a pager, which provides
+        # an `__iter__` convenience method.
+        response = pagers.ListOperationsPager(
+            method=rpc, request=request, response=response, metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+
+    def get_operation(
+        self,
+        name: str,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.Operation:
+        r"""Gets the latest state of a long-running operation.
+        Clients can use this method to poll the operation result
+        at intervals as recommended by the API service.
+
+        Args:
+            name (str):
+                The name of the operation resource.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.longrunning.operations_pb2.Operation:
+                This resource represents a long-
+                unning operation that is the result of a
+                network API call.
+
+        """
+
+        request = operations_pb2.GetOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.get_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
+
+        # Done; return the response.
+        return response
+
+    def delete_operation(
+        self,
+        name: str,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> None:
+        r"""Deletes a long-running operation. This method indicates that the
+        client is no longer interested in the operation result. It does
+        not cancel the operation. If the server doesn't support this
+        method, it returns ``google.rpc.Code.UNIMPLEMENTED``.
+
+        Args:
+            name (str):
+                The name of the operation resource to
+                be deleted.
+
+                This corresponds to the ``name`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        # Create the request object.
+        request = operations_pb2.DeleteOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.delete_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        rpc(
+            request, retry=retry, timeout=timeout, metadata=metadata,
+        )
+
+    def cancel_operation(
+        self,
+        name: Optional[str] = None,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> None:
+        r"""Starts asynchronous cancellation on a long-running operation.
+        The server makes a best effort to cancel the operation, but
+        success is not guaranteed. If the server doesn't support this
+        method, it returns ``google.rpc.Code.UNIMPLEMENTED``. Clients
+        can use
+        [Operations.GetOperation][google.api_core.operations_v1.Operations.GetOperation]
+        or other methods to check whether the cancellation succeeded or
+        whether the operation completed despite cancellation. On
+        successful cancellation, the operation is not deleted; instead,
+        it becomes an operation with an
+        [Operation.error][google.api_core.operations_v1.Operation.error] value with
+        a [google.rpc.Status.code][google.rpc.Status.code] of 1,
+        corresponding to ``Code.CANCELLED``.
+
+        Args:
+            name (str):
+                The name of the operation resource to
+                be cancelled.
+
+                This corresponds to the ``name`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        # Create the request object.
+        request = operations_pb2.CancelOperationRequest(name=name)
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.cancel_operation]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata or ()) + (
+            gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
+        )
+
+        # Send the request.
+        rpc(
+            request, retry=retry, timeout=timeout, metadata=metadata,
+        )
diff --git a/google/api_core/operations_v1/pagers.py b/google/api_core/operations_v1/pagers.py
new file mode 100644
index 0000000..b8a4775
--- /dev/null
+++ b/google/api_core/operations_v1/pagers.py
@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from typing import (
+    Any,
+    Callable,
+    Iterator,
+    Sequence,
+    Tuple,
+)
+
+from google.longrunning import operations_pb2
+
+
+class ListOperationsPager:
+    """A pager for iterating through ``list_operations`` requests.
+
+    This class thinly wraps an initial
+    :class:`google.longrunning.operations_pb2.ListOperationsResponse` object, and
+    provides an ``__iter__`` method to iterate through its
+    ``operations`` field.
+
+    If there are more pages, the ``__iter__`` method will make additional
+    ``ListOperations`` requests and continue to iterate
+    through the ``operations`` field on the
+    corresponding responses.
+
+    All the usual :class:`google.longrunning.operations_pb2.ListOperationsResponse`
+    attributes are available on the pager. If multiple requests are made, only
+    the most recent response is retained, and thus used for attribute lookup.
+    """
+
+    def __init__(
+        self,
+        method: Callable[..., operations_pb2.ListOperationsResponse],
+        request: operations_pb2.ListOperationsRequest,
+        response: operations_pb2.ListOperationsResponse,
+        *,
+        metadata: Sequence[Tuple[str, str]] = ()
+    ):
+        """Instantiate the pager.
+
+        Args:
+            method (Callable): The method that was originally called, and
+                which instantiated this pager.
+            request (google.longrunning.operations_pb2.ListOperationsRequest):
+                The initial request object.
+            response (google.longrunning.operations_pb2.ListOperationsResponse):
+                The initial response object.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+        self._method = method
+        self._request = request
+        self._response = response
+        self._metadata = metadata
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._response, name)
+
+    @property
+    def pages(self) -> Iterator[operations_pb2.ListOperationsResponse]:
+        yield self._response
+        while self._response.next_page_token:
+            self._request.page_token = self._response.next_page_token
+            self._response = self._method(self._request, metadata=self._metadata)
+            yield self._response
+
+    def __iter__(self) -> Iterator[operations_pb2.Operation]:
+        for page in self.pages:
+            yield from page.operations
+
+    def __repr__(self) -> str:
+        return "{0}<{1!r}>".format(self.__class__.__name__, self._response)
diff --git a/google/api_core/operations_v1/transports/__init__.py b/google/api_core/operations_v1/transports/__init__.py
new file mode 100644
index 0000000..b443c07
--- /dev/null
+++ b/google/api_core/operations_v1/transports/__init__.py
@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from collections import OrderedDict
+from typing import Dict, Type
+
+from .base import OperationsTransport
+from .rest import OperationsRestTransport
+
+
+# Compile a registry of transports.
+_transport_registry = OrderedDict()  # type: Dict[str, Type[OperationsTransport]]
+_transport_registry["rest"] = OperationsRestTransport
+
+__all__ = (
+    "OperationsTransport",
+    "OperationsRestTransport",
+)
diff --git a/google/api_core/operations_v1/transports/base.py b/google/api_core/operations_v1/transports/base.py
new file mode 100644
index 0000000..460e646
--- /dev/null
+++ b/google/api_core/operations_v1/transports/base.py
@@ -0,0 +1,232 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import abc
+from typing import Awaitable, Callable, Optional, Sequence, Union
+
+import pkg_resources
+
+import google.api_core  # type: ignore
+from google.api_core import exceptions as core_exceptions  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core import retry as retries  # type: ignore
+import google.auth  # type: ignore
+from google.auth import credentials as ga_credentials  # type: ignore
+from google.longrunning import operations_pb2
+from google.oauth2 import service_account  # type: ignore
+from google.protobuf import empty_pb2  # type: ignore
+
+try:
+    DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
+        gapic_version=pkg_resources.get_distribution(
+            "google.api_core.operations_v1",
+        ).version,
+    )
+except pkg_resources.DistributionNotFound:
+    DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo()
+
+
+class OperationsTransport(abc.ABC):
+    """Abstract transport class for Operations."""
+
+    AUTH_SCOPES = ()
+
+    DEFAULT_HOST: str = "longrunning.googleapis.com"
+
+    def __init__(
+        self,
+        *,
+        host: str = DEFAULT_HOST,
+        credentials: ga_credentials.Credentials = None,
+        credentials_file: Optional[str] = None,
+        scopes: Optional[Sequence[str]] = None,
+        quota_project_id: Optional[str] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
+        **kwargs,
+    ) -> None:
+        """Instantiate the transport.
+
+        Args:
+            host (Optional[str]):
+                 The hostname to connect to.
+            credentials (Optional[google.auth.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+            credentials_file (Optional[str]): A file with credentials that can
+                be loaded with :func:`google.auth.load_credentials_from_file`.
+                This argument is mutually exclusive with credentials.
+            scopes (Optional[Sequence[str]]): A list of scopes.
+            quota_project_id (Optional[str]): An optional project to use for billing
+                and quota.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
+        """
+        # Save the hostname. Default to port 443 (HTTPS) if none is specified.
+        if ":" not in host:
+            host += ":443"
+        self._host = host
+
+        scopes_kwargs = {"scopes": scopes, "default_scopes": self.AUTH_SCOPES}
+
+        # Save the scopes.
+        self._scopes = scopes
+
+        # If no credentials are provided, then determine the appropriate
+        # defaults.
+        if credentials and credentials_file:
+            raise core_exceptions.DuplicateCredentialArgs(
+                "'credentials_file' and 'credentials' are mutually exclusive"
+            )
+
+        if credentials_file is not None:
+            credentials, _ = google.auth.load_credentials_from_file(
+                credentials_file, **scopes_kwargs, quota_project_id=quota_project_id
+            )
+
+        elif credentials is None:
+            credentials, _ = google.auth.default(
+                **scopes_kwargs, quota_project_id=quota_project_id
+            )
+
+        # If the credentials are service account credentials, then always try to use self signed JWT.
+        if (
+            always_use_jwt_access
+            and isinstance(credentials, service_account.Credentials)
+            and hasattr(service_account.Credentials, "with_always_use_jwt_access")
+        ):
+            credentials = credentials.with_always_use_jwt_access(True)
+
+        # Save the credentials.
+        self._credentials = credentials
+
+    def _prep_wrapped_messages(self, client_info):
+        # Precompute the wrapped methods.
+        self._wrapped_methods = {
+            self.list_operations: gapic_v1.method.wrap_method(
+                self.list_operations,
+                default_retry=retries.Retry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+            ),
+            self.get_operation: gapic_v1.method.wrap_method(
+                self.get_operation,
+                default_retry=retries.Retry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+            ),
+            self.delete_operation: gapic_v1.method.wrap_method(
+                self.delete_operation,
+                default_retry=retries.Retry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+            ),
+            self.cancel_operation: gapic_v1.method.wrap_method(
+                self.cancel_operation,
+                default_retry=retries.Retry(
+                    initial=0.5,
+                    maximum=10.0,
+                    multiplier=2.0,
+                    predicate=retries.if_exception_type(
+                        core_exceptions.ServiceUnavailable,
+                    ),
+                    deadline=10.0,
+                ),
+                default_timeout=10.0,
+                client_info=client_info,
+            ),
+        }
+
+    def close(self):
+        """Closes resources associated with the transport.
+
+       .. warning::
+            Only call this method if the transport is NOT shared
+            with other clients - this may cause errors in other clients!
+        """
+        raise NotImplementedError()
+
+    @property
+    def list_operations(
+        self,
+    ) -> Callable[
+        [operations_pb2.ListOperationsRequest],
+        Union[
+            operations_pb2.ListOperationsResponse,
+            Awaitable[operations_pb2.ListOperationsResponse],
+        ],
+    ]:
+        raise NotImplementedError()
+
+    @property
+    def get_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.GetOperationRequest],
+        Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]],
+    ]:
+        raise NotImplementedError()
+
+    @property
+    def delete_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.DeleteOperationRequest],
+        Union[empty_pb2.Empty, Awaitable[empty_pb2.Empty]],
+    ]:
+        raise NotImplementedError()
+
+    @property
+    def cancel_operation(
+        self,
+    ) -> Callable[
+        [operations_pb2.CancelOperationRequest],
+        Union[empty_pb2.Empty, Awaitable[empty_pb2.Empty]],
+    ]:
+        raise NotImplementedError()
+
+
+__all__ = ("OperationsTransport",)
diff --git a/google/api_core/operations_v1/transports/rest.py b/google/api_core/operations_v1/transports/rest.py
new file mode 100644
index 0000000..27ed766
--- /dev/null
+++ b/google/api_core/operations_v1/transports/rest.py
@@ -0,0 +1,455 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from typing import Callable, Dict, Optional, Sequence, Tuple, Union
+
+from requests import __version__ as requests_version
+
+from google.api_core import exceptions as core_exceptions  # type: ignore
+from google.api_core import gapic_v1  # type: ignore
+from google.api_core import path_template  # type: ignore
+from google.api_core import rest_helpers  # type: ignore
+from google.api_core import retry as retries  # type: ignore
+from google.auth import credentials as ga_credentials  # type: ignore
+from google.auth.transport.requests import AuthorizedSession  # type: ignore
+from google.longrunning import operations_pb2  # type: ignore
+from google.protobuf import empty_pb2  # type: ignore
+from google.protobuf import json_format  # type: ignore
+from .base import DEFAULT_CLIENT_INFO as BASE_DEFAULT_CLIENT_INFO, OperationsTransport
+
+OptionalRetry = Union[retries.Retry, object]
+
+DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
+    gapic_version=BASE_DEFAULT_CLIENT_INFO.gapic_version,
+    grpc_version=None,
+    rest_version=requests_version,
+)
+
+
+class OperationsRestTransport(OperationsTransport):
+    """REST backend transport for Operations.
+
+    Manages long-running operations with an API service.
+
+    When an API method normally takes long time to complete, it can be
+    designed to return [Operation][google.api_core.operations_v1.Operation] to the
+    client, and the client can use this interface to receive the real
+    response asynchronously by polling the operation resource, or pass
+    the operation resource to another API (such as Google Cloud Pub/Sub
+    API) to receive the response. Any API service that returns
+    long-running operations should implement the ``Operations``
+    interface so developers can have a consistent client experience.
+
+    This class defines the same methods as the primary client, so the
+    primary client can load the underlying transport implementation
+    and call it.
+
+    It sends JSON representations of protocol buffers over HTTP/1.1
+    """
+
+    def __init__(
+        self,
+        *,
+        host: str = "longrunning.googleapis.com",
+        credentials: ga_credentials.Credentials = None,
+        credentials_file: Optional[str] = None,
+        scopes: Optional[Sequence[str]] = None,
+        client_cert_source_for_mtls: Optional[Callable[[], Tuple[bytes, bytes]]] = None,
+        quota_project_id: Optional[str] = None,
+        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+        always_use_jwt_access: Optional[bool] = False,
+        url_scheme: str = "https",
+        http_options: Optional[Dict] = None,
+    ) -> None:
+        """Instantiate the transport.
+
+        Args:
+            host (Optional[str]):
+                 The hostname to connect to.
+            credentials (Optional[google.auth.credentials.Credentials]): The
+                authorization credentials to attach to requests. These
+                credentials identify the application to the service; if none
+                are specified, the client will attempt to ascertain the
+                credentials from the environment.
+
+            credentials_file (Optional[str]): A file with credentials that can
+                be loaded with :func:`google.auth.load_credentials_from_file`.
+                This argument is ignored if ``channel`` is provided.
+            scopes (Optional(Sequence[str])): A list of scopes. This argument is
+                ignored if ``channel`` is provided.
+            client_cert_source_for_mtls (Callable[[], Tuple[bytes, bytes]]): Client
+                certificate to configure mutual TLS HTTP channel. It is ignored
+                if ``channel`` is provided.
+            quota_project_id (Optional[str]): An optional project to use for billing
+                and quota.
+            client_info (google.api_core.gapic_v1.client_info.ClientInfo):
+                The client info used to send a user-agent string along with
+                API requests. If ``None``, then default info will be used.
+                Generally, you only need to set this if you're developing
+                your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
+            url_scheme: the protocol scheme for the API endpoint.  Normally
+                "https", but for testing or local servers,
+                "http" can be specified.
+            http_options: a dictionary of http_options for transcoding, to override
+                the defaults from operatons.proto.  Each method has an entry
+                with the corresponding http rules as value.
+
+        """
+        # Run the base constructor
+        # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
+        # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
+        # credentials object
+        super().__init__(
+            host=host,
+            credentials=credentials,
+            client_info=client_info,
+            always_use_jwt_access=always_use_jwt_access,
+        )
+        self._session = AuthorizedSession(
+            self._credentials, default_host=self.DEFAULT_HOST
+        )
+        if client_cert_source_for_mtls:
+            self._session.configure_mtls_channel(client_cert_source_for_mtls)
+        self._prep_wrapped_messages(client_info)
+        self._http_options = http_options or {}
+
+    def _list_operations(
+        self,
+        request: operations_pb2.ListOperationsRequest,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.ListOperationsResponse:
+        r"""Call the list operations method over HTTP.
+
+        Args:
+            request (~.operations_pb2.ListOperationsRequest):
+                The request object. The request message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            ~.operations_pb2.ListOperationsResponse:
+                The response message for
+                [Operations.ListOperations][google.api_core.operations_v1.Operations.ListOperations].
+
+        """
+
+        http_options = [
+            {"method": "get", "uri": "/v1/{name=operations}"},
+        ]
+        if "google.longrunning.Operations.ListOperations" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.ListOperations"
+            ]
+
+        request_kwargs = json_format.MessageToDict(
+            request,
+            preserving_proto_field_name=True,
+            including_default_value_fields=True,
+        )
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.ListOperationsRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            including_default_value_fields=False,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        response = getattr(self._session, method)(
+            "https://{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            raise core_exceptions.from_http_response(response)
+
+        # Return the response
+        api_response = operations_pb2.ListOperationsResponse()
+        json_format.Parse(response.content, api_response, ignore_unknown_fields=False)
+        return api_response
+
+    def _get_operation(
+        self,
+        request: operations_pb2.GetOperationRequest,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operations_pb2.Operation:
+        r"""Call the get operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.GetOperationRequest):
+                The request object. The request message for
+                [Operations.GetOperation][google.api_core.operations_v1.Operations.GetOperation].
+
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            ~.operations_pb2.Operation:
+                This resource represents a long-
+                unning operation that is the result of a
+                network API call.
+
+        """
+
+        http_options = [
+            {"method": "get", "uri": "/v1/{name=operations/**}"},
+        ]
+        if "google.longrunning.Operations.GetOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.GetOperation"
+            ]
+
+        request_kwargs = json_format.MessageToDict(
+            request,
+            preserving_proto_field_name=True,
+            including_default_value_fields=True,
+        )
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.GetOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            including_default_value_fields=False,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        response = getattr(self._session, method)(
+            "https://{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            raise core_exceptions.from_http_response(response)
+
+        # Return the response
+        api_response = operations_pb2.Operation()
+        json_format.Parse(response.content, api_response, ignore_unknown_fields=False)
+        return api_response
+
+    def _delete_operation(
+        self,
+        request: operations_pb2.DeleteOperationRequest,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> empty_pb2.Empty:
+        r"""Call the delete operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.DeleteOperationRequest):
+                The request object. The request message for
+                [Operations.DeleteOperation][google.api_core.operations_v1.Operations.DeleteOperation].
+
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+
+        http_options = [
+            {"method": "delete", "uri": "/v1/{name=operations/**}"},
+        ]
+        if "google.longrunning.Operations.DeleteOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.DeleteOperation"
+            ]
+
+        request_kwargs = json_format.MessageToDict(
+            request,
+            preserving_proto_field_name=True,
+            including_default_value_fields=True,
+        )
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.DeleteOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            including_default_value_fields=False,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        response = getattr(self._session, method)(
+            "https://{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            raise core_exceptions.from_http_response(response)
+
+        return empty_pb2.Empty()
+
+    def _cancel_operation(
+        self,
+        request: operations_pb2.CancelOperationRequest,
+        *,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: Optional[float] = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> empty_pb2.Empty:
+        r"""Call the cancel operation method over HTTP.
+
+        Args:
+            request (~.operations_pb2.CancelOperationRequest):
+                The request object. The request message for
+                [Operations.CancelOperation][google.api_core.operations_v1.Operations.CancelOperation].
+
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+        """
+
+        http_options = [
+            {"method": "post", "uri": "/v1/{name=operations/**}:cancel", "body": "*"},
+        ]
+        if "google.longrunning.Operations.CancelOperation" in self._http_options:
+            http_options = self._http_options[
+                "google.longrunning.Operations.CancelOperation"
+            ]
+
+        request_kwargs = json_format.MessageToDict(
+            request,
+            preserving_proto_field_name=True,
+            including_default_value_fields=True,
+        )
+        transcoded_request = path_template.transcode(http_options, **request_kwargs)
+
+        # Jsonify the request body
+        body_request = operations_pb2.CancelOperationRequest()
+        json_format.ParseDict(transcoded_request["body"], body_request)
+        body = json_format.MessageToDict(
+            body_request,
+            including_default_value_fields=False,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+        uri = transcoded_request["uri"]
+        method = transcoded_request["method"]
+
+        # Jsonify the query params
+        query_params_request = operations_pb2.CancelOperationRequest()
+        json_format.ParseDict(transcoded_request["query_params"], query_params_request)
+        query_params = json_format.MessageToDict(
+            query_params_request,
+            including_default_value_fields=False,
+            preserving_proto_field_name=False,
+            use_integers_for_enums=False,
+        )
+
+        # Send the request
+        headers = dict(metadata)
+        headers["Content-Type"] = "application/json"
+        response = getattr(self._session, method)(
+            "https://{host}{uri}".format(host=self._host, uri=uri),
+            timeout=timeout,
+            headers=headers,
+            params=rest_helpers.flatten_query_params(query_params),
+            data=body,
+        )
+
+        # In case of error, raise the appropriate core_exceptions.GoogleAPICallError exception
+        # subclass.
+        if response.status_code >= 400:
+            raise core_exceptions.from_http_response(response)
+
+        return empty_pb2.Empty()
+
+    @property
+    def list_operations(
+        self,
+    ) -> Callable[
+        [operations_pb2.ListOperationsRequest], operations_pb2.ListOperationsResponse
+    ]:
+        return self._list_operations
+
+    @property
+    def get_operation(
+        self,
+    ) -> Callable[[operations_pb2.GetOperationRequest], operations_pb2.Operation]:
+        return self._get_operation
+
+    @property
+    def delete_operation(
+        self,
+    ) -> Callable[[operations_pb2.DeleteOperationRequest], empty_pb2.Empty]:
+        return self._delete_operation
+
+    @property
+    def cancel_operation(
+        self,
+    ) -> Callable[[operations_pb2.CancelOperationRequest], empty_pb2.Empty]:
+        return self._cancel_operation
+
+
+__all__ = ("OperationsRestTransport",)
diff --git a/tests/unit/operations_v1/test_operations_rest_client.py b/tests/unit/operations_v1/test_operations_rest_client.py
new file mode 100644
index 0000000..dddf6b7
--- /dev/null
+++ b/tests/unit/operations_v1/test_operations_rest_client.py
@@ -0,0 +1,944 @@
+# -*- coding: utf-8 -*-
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import os
+
+import mock
+import pytest
+
+try:
+    import grpc  # noqa: F401
+except ImportError:
+    pytest.skip("No GRPC", allow_module_level=True)
+from requests import Response  # noqa I201
+from requests.sessions import Session
+
+from google.api_core import client_options
+from google.api_core import exceptions as core_exceptions
+from google.api_core import gapic_v1
+from google.api_core.operations_v1 import AbstractOperationsClient
+from google.api_core.operations_v1 import pagers
+from google.api_core.operations_v1 import transports
+import google.auth
+from google.auth import credentials as ga_credentials
+from google.auth.exceptions import MutualTLSChannelError
+from google.longrunning import operations_pb2
+from google.oauth2 import service_account
+from google.protobuf import json_format  # type: ignore
+from google.rpc import status_pb2  # type: ignore
+
+
+HTTP_OPTIONS = {
+    "google.longrunning.Operations.CancelOperation": [
+        {"method": "post", "uri": "/v3/{name=operations/*}:cancel", "body": "*"},
+    ],
+    "google.longrunning.Operations.DeleteOperation": [
+        {"method": "delete", "uri": "/v3/{name=operations/*}"},
+    ],
+    "google.longrunning.Operations.GetOperation": [
+        {"method": "get", "uri": "/v3/{name=operations/*}"},
+    ],
+    "google.longrunning.Operations.ListOperations": [
+        {"method": "get", "uri": "/v3/{name=operations}"},
+    ],
+}
+
+
+def client_cert_source_callback():
+    return b"cert bytes", b"key bytes"
+
+
+def _get_operations_client(http_options=HTTP_OPTIONS):
+    transport = transports.rest.OperationsRestTransport(
+        credentials=ga_credentials.AnonymousCredentials(), http_options=http_options
+    )
+
+    return AbstractOperationsClient(transport=transport)
+
+
+# If default endpoint is localhost, then default mtls endpoint will be the same.
+# This method modifies the default endpoint so the client can produce a different
+# mtls endpoint for endpoint testing purposes.
+def modify_default_endpoint(client):
+    return (
+        "foo.googleapis.com"
+        if ("localhost" in client.DEFAULT_ENDPOINT)
+        else client.DEFAULT_ENDPOINT
+    )
+
+
+def test__get_default_mtls_endpoint():
+    api_endpoint = "example.googleapis.com"
+    api_mtls_endpoint = "example.mtls.googleapis.com"
+    sandbox_endpoint = "example.sandbox.googleapis.com"
+    sandbox_mtls_endpoint = "example.mtls.sandbox.googleapis.com"
+    non_googleapi = "api.example.com"
+
+    assert AbstractOperationsClient._get_default_mtls_endpoint(None) is None
+    assert (
+        AbstractOperationsClient._get_default_mtls_endpoint(api_endpoint)
+        == api_mtls_endpoint
+    )
+    assert (
+        AbstractOperationsClient._get_default_mtls_endpoint(api_mtls_endpoint)
+        == api_mtls_endpoint
+    )
+    assert (
+        AbstractOperationsClient._get_default_mtls_endpoint(sandbox_endpoint)
+        == sandbox_mtls_endpoint
+    )
+    assert (
+        AbstractOperationsClient._get_default_mtls_endpoint(sandbox_mtls_endpoint)
+        == sandbox_mtls_endpoint
+    )
+    assert (
+        AbstractOperationsClient._get_default_mtls_endpoint(non_googleapi)
+        == non_googleapi
+    )
+
+
+@pytest.mark.parametrize("client_class", [AbstractOperationsClient])
+def test_operations_client_from_service_account_info(client_class):
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(
+        service_account.Credentials, "from_service_account_info"
+    ) as factory:
+        factory.return_value = creds
+        info = {"valid": True}
+        client = client_class.from_service_account_info(info)
+        assert client.transport._credentials == creds
+        assert isinstance(client, client_class)
+
+        assert client.transport._host == "longrunning.googleapis.com:443"
+
+
+@pytest.mark.parametrize(
+    "transport_class,transport_name", [(transports.OperationsRestTransport, "rest")]
+)
+def test_operations_client_service_account_always_use_jwt(
+    transport_class, transport_name
+):
+    with mock.patch.object(
+        service_account.Credentials, "with_always_use_jwt_access", create=True
+    ) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport_class(credentials=creds, always_use_jwt_access=True)
+        use_jwt.assert_called_once_with(True)
+
+    with mock.patch.object(
+        service_account.Credentials, "with_always_use_jwt_access", create=True
+    ) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport_class(credentials=creds, always_use_jwt_access=False)
+        use_jwt.assert_not_called()
+
+
+@pytest.mark.parametrize("client_class", [AbstractOperationsClient])
+def test_operations_client_from_service_account_file(client_class):
+    creds = ga_credentials.AnonymousCredentials()
+    with mock.patch.object(
+        service_account.Credentials, "from_service_account_file"
+    ) as factory:
+        factory.return_value = creds
+        client = client_class.from_service_account_file("dummy/file/path.json")
+        assert client.transport._credentials == creds
+        assert isinstance(client, client_class)
+
+        client = client_class.from_service_account_json("dummy/file/path.json")
+        assert client.transport._credentials == creds
+        assert isinstance(client, client_class)
+
+        assert client.transport._host == "longrunning.googleapis.com:443"
+
+
+def test_operations_client_get_transport_class():
+    transport = AbstractOperationsClient.get_transport_class()
+    available_transports = [
+        transports.OperationsRestTransport,
+    ]
+    assert transport in available_transports
+
+    transport = AbstractOperationsClient.get_transport_class("rest")
+    assert transport == transports.OperationsRestTransport
+
+
+@pytest.mark.parametrize(
+    "client_class,transport_class,transport_name",
+    [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
+)
+@mock.patch.object(
+    AbstractOperationsClient,
+    "DEFAULT_ENDPOINT",
+    modify_default_endpoint(AbstractOperationsClient),
+)
+def test_operations_client_client_options(
+    client_class, transport_class, transport_name
+):
+    # Check that if channel is provided we won't create a new one.
+    with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc:
+        transport = transport_class(credentials=ga_credentials.AnonymousCredentials())
+        client = client_class(transport=transport)
+        gtc.assert_not_called()
+
+    # Check that if channel is provided via str we will create a new one.
+    with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc:
+        client = client_class(transport=transport_name)
+        gtc.assert_called()
+
+    # Check the case api_endpoint is provided.
+    options = client_options.ClientOptions(api_endpoint="squid.clam.whelk")
+    with mock.patch.object(transport_class, "__init__") as patched:
+        patched.return_value = None
+        client = client_class(client_options=options)
+        patched.assert_called_once_with(
+            credentials=None,
+            credentials_file=None,
+            host="squid.clam.whelk",
+            scopes=None,
+            client_cert_source_for_mtls=None,
+            quota_project_id=None,
+            client_info=transports.base.DEFAULT_CLIENT_INFO,
+            always_use_jwt_access=True,
+        )
+
+    # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT is
+    # "never".
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}):
+        with mock.patch.object(transport_class, "__init__") as patched:
+            patched.return_value = None
+            client = client_class()
+            patched.assert_called_once_with(
+                credentials=None,
+                credentials_file=None,
+                host=client.DEFAULT_ENDPOINT,
+                scopes=None,
+                client_cert_source_for_mtls=None,
+                quota_project_id=None,
+                client_info=transports.base.DEFAULT_CLIENT_INFO,
+                always_use_jwt_access=True,
+            )
+
+    # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT is
+    # "always".
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
+        with mock.patch.object(transport_class, "__init__") as patched:
+            patched.return_value = None
+            client = client_class()
+            patched.assert_called_once_with(
+                credentials=None,
+                credentials_file=None,
+                host=client.DEFAULT_MTLS_ENDPOINT,
+                scopes=None,
+                client_cert_source_for_mtls=None,
+                quota_project_id=None,
+                client_info=transports.base.DEFAULT_CLIENT_INFO,
+                always_use_jwt_access=True,
+            )
+
+    # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT has
+    # unsupported value.
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "Unsupported"}):
+        with pytest.raises(MutualTLSChannelError):
+            client = client_class()
+
+    # Check the case GOOGLE_API_USE_CLIENT_CERTIFICATE has unsupported value.
+    with mock.patch.dict(
+        os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported"}
+    ):
+        with pytest.raises(ValueError):
+            client = client_class()
+
+    # Check the case quota_project_id is provided
+    options = client_options.ClientOptions(quota_project_id="octopus")
+    with mock.patch.object(transport_class, "__init__") as patched:
+        patched.return_value = None
+        client = client_class(client_options=options)
+        patched.assert_called_once_with(
+            credentials=None,
+            credentials_file=None,
+            host=client.DEFAULT_ENDPOINT,
+            scopes=None,
+            client_cert_source_for_mtls=None,
+            quota_project_id="octopus",
+            client_info=transports.base.DEFAULT_CLIENT_INFO,
+            always_use_jwt_access=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "client_class,transport_class,transport_name,use_client_cert_env",
+    [
+        (AbstractOperationsClient, transports.OperationsRestTransport, "rest", "true"),
+        (AbstractOperationsClient, transports.OperationsRestTransport, "rest", "false"),
+    ],
+)
+@mock.patch.object(
+    AbstractOperationsClient,
+    "DEFAULT_ENDPOINT",
+    modify_default_endpoint(AbstractOperationsClient),
+)
+@mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"})
+def test_operations_client_mtls_env_auto(
+    client_class, transport_class, transport_name, use_client_cert_env
+):
+    # This tests the endpoint autoswitch behavior. Endpoint is autoswitched to the default
+    # mtls endpoint, if GOOGLE_API_USE_CLIENT_CERTIFICATE is "true" and client cert exists.
+
+    # Check the case client_cert_source is provided. Whether client cert is used depends on
+    # GOOGLE_API_USE_CLIENT_CERTIFICATE value.
+    with mock.patch.dict(
+        os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env}
+    ):
+        options = client_options.ClientOptions(
+            client_cert_source=client_cert_source_callback
+        )
+
+        def fake_init(client_cert_source_for_mtls=None, **kwargs):
+            """Invoke client_cert source if provided."""
+
+            if client_cert_source_for_mtls:
+                client_cert_source_for_mtls()
+                return None
+
+        with mock.patch.object(transport_class, "__init__") as patched:
+            patched.side_effect = fake_init
+            client = client_class(client_options=options)
+
+            if use_client_cert_env == "false":
+                expected_client_cert_source = None
+                expected_host = client.DEFAULT_ENDPOINT
+            else:
+                expected_client_cert_source = client_cert_source_callback
+                expected_host = client.DEFAULT_MTLS_ENDPOINT
+
+            patched.assert_called_once_with(
+                credentials=None,
+                credentials_file=None,
+                host=expected_host,
+                scopes=None,
+                client_cert_source_for_mtls=expected_client_cert_source,
+                quota_project_id=None,
+                client_info=transports.base.DEFAULT_CLIENT_INFO,
+                always_use_jwt_access=True,
+            )
+
+    # Check the case ADC client cert is provided. Whether client cert is used depends on
+    # GOOGLE_API_USE_CLIENT_CERTIFICATE value.
+    with mock.patch.dict(
+        os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env}
+    ):
+        with mock.patch.object(transport_class, "__init__") as patched:
+            with mock.patch(
+                "google.auth.transport.mtls.has_default_client_cert_source",
+                return_value=True,
+            ):
+                with mock.patch(
+                    "google.auth.transport.mtls.default_client_cert_source",
+                    return_value=client_cert_source_callback,
+                ):
+                    if use_client_cert_env == "false":
+                        expected_host = client.DEFAULT_ENDPOINT
+                        expected_client_cert_source = None
+                    else:
+                        expected_host = client.DEFAULT_MTLS_ENDPOINT
+                        expected_client_cert_source = client_cert_source_callback
+
+                    patched.return_value = None
+                    client = client_class()
+                    patched.assert_called_once_with(
+                        credentials=None,
+                        credentials_file=None,
+                        host=expected_host,
+                        scopes=None,
+                        client_cert_source_for_mtls=expected_client_cert_source,
+                        quota_project_id=None,
+                        client_info=transports.base.DEFAULT_CLIENT_INFO,
+                        always_use_jwt_access=True,
+                    )
+
+    # Check the case client_cert_source and ADC client cert are not provided.
+    with mock.patch.dict(
+        os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env}
+    ):
+        with mock.patch.object(transport_class, "__init__") as patched:
+            with mock.patch(
+                "google.auth.transport.mtls.has_default_client_cert_source",
+                return_value=False,
+            ):
+                patched.return_value = None
+                client = client_class()
+                patched.assert_called_once_with(
+                    credentials=None,
+                    credentials_file=None,
+                    host=client.DEFAULT_ENDPOINT,
+                    scopes=None,
+                    client_cert_source_for_mtls=None,
+                    quota_project_id=None,
+                    client_info=transports.base.DEFAULT_CLIENT_INFO,
+                    always_use_jwt_access=True,
+                )
+
+
+@pytest.mark.parametrize(
+    "client_class,transport_class,transport_name",
+    [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
+)
+def test_operations_client_client_options_scopes(
+    client_class, transport_class, transport_name
+):
+    # Check the case scopes are provided.
+    options = client_options.ClientOptions(scopes=["1", "2"],)
+    with mock.patch.object(transport_class, "__init__") as patched:
+        patched.return_value = None
+        client = client_class(client_options=options)
+        patched.assert_called_once_with(
+            credentials=None,
+            credentials_file=None,
+            host=client.DEFAULT_ENDPOINT,
+            scopes=["1", "2"],
+            client_cert_source_for_mtls=None,
+            quota_project_id=None,
+            client_info=transports.base.DEFAULT_CLIENT_INFO,
+            always_use_jwt_access=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "client_class,transport_class,transport_name",
+    [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")],
+)
+def test_operations_client_client_options_credentials_file(
+    client_class, transport_class, transport_name
+):
+    # Check the case credentials file is provided.
+    options = client_options.ClientOptions(credentials_file="credentials.json")
+    with mock.patch.object(transport_class, "__init__") as patched:
+        patched.return_value = None
+        client = client_class(client_options=options)
+        patched.assert_called_once_with(
+            credentials=None,
+            credentials_file="credentials.json",
+            host=client.DEFAULT_ENDPOINT,
+            scopes=None,
+            client_cert_source_for_mtls=None,
+            quota_project_id=None,
+            client_info=transports.base.DEFAULT_CLIENT_INFO,
+            always_use_jwt_access=True,
+        )
+
+
+def test_list_operations_rest(
+    transport: str = "rest", request_type=operations_pb2.ListOperationsRequest
+):
+    client = _get_operations_client()
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(Session, "request") as req:
+        # Designate an appropriate value for the returned response.
+        return_value = operations_pb2.ListOperationsResponse(
+            next_page_token="next_page_token_value",
+        )
+
+        # Wrap the value into a proper Response obj
+        response_value = Response()
+        response_value.status_code = 200
+        json_return_value = json_format.MessageToJson(return_value)
+        response_value._content = json_return_value.encode("UTF-8")
+        req.return_value = response_value
+        response = client.list_operations(
+            name="operations", filter_="my_filter", page_size=10, page_token="abc"
+        )
+
+        actual_args = req.call_args
+        assert actual_args.args[0] == "GET"
+        assert (
+            actual_args.args[1]
+            == "https://longrunning.googleapis.com:443/v3/operations"
+        )
+        assert actual_args.kwargs["params"] == [
+            ("filter", "my_filter"),
+            ("pageSize", 10),
+            ("pageToken", "abc"),
+        ]
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, pagers.ListOperationsPager)
+    assert response.next_page_token == "next_page_token_value"
+
+
+def test_list_operations_rest_failure():
+    client = _get_operations_client(http_options=None)
+
+    with mock.patch.object(Session, "request") as req:
+        response_value = Response()
+        response_value.status_code = 400
+        mock_request = mock.MagicMock()
+        mock_request.method = "GET"
+        mock_request.url = "https://longrunning.googleapis.com:443/v1/operations"
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            client.list_operations(name="operations")
+
+
+def test_list_operations_rest_pager():
+    client = AbstractOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(Session, "request") as req:
+        # TODO(kbandes): remove this mock unless there's a good reason for it.
+        # with mock.patch.object(path_template, 'transcode') as transcode:
+        # Set the response as a series of pages
+        response = (
+            operations_pb2.ListOperationsResponse(
+                operations=[
+                    operations_pb2.Operation(),
+                    operations_pb2.Operation(),
+                    operations_pb2.Operation(),
+                ],
+                next_page_token="abc",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[], next_page_token="def",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[operations_pb2.Operation()], next_page_token="ghi",
+            ),
+            operations_pb2.ListOperationsResponse(
+                operations=[operations_pb2.Operation(), operations_pb2.Operation()],
+            ),
+        )
+        # Two responses for two calls
+        response = response + response
+
+        # Wrap the values into proper Response objs
+        response = tuple(json_format.MessageToJson(x) for x in response)
+        return_values = tuple(Response() for i in response)
+        for return_val, response_val in zip(return_values, response):
+            return_val._content = response_val.encode("UTF-8")
+            return_val.status_code = 200
+        req.side_effect = return_values
+
+        pager = client.list_operations(name="operations")
+
+        results = list(pager)
+        assert len(results) == 6
+        assert all(isinstance(i, operations_pb2.Operation) for i in results)
+
+        pages = list(client.list_operations(name="operations").pages)
+        for page_, token in zip(pages, ["abc", "def", "ghi", ""]):
+            assert page_.next_page_token == token
+
+
+def test_get_operation_rest(
+    transport: str = "rest", request_type=operations_pb2.GetOperationRequest
+):
+    client = _get_operations_client()
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(Session, "request") as req:
+        # Designate an appropriate value for the returned response.
+        return_value = operations_pb2.Operation(
+            name="operations/sample1", done=True, error=status_pb2.Status(code=411),
+        )
+
+        # Wrap the value into a proper Response obj
+        response_value = Response()
+        response_value.status_code = 200
+        json_return_value = json_format.MessageToJson(return_value)
+        response_value._content = json_return_value.encode("UTF-8")
+        req.return_value = response_value
+        response = client.get_operation("operations/sample1")
+
+    actual_args = req.call_args
+    assert actual_args.args[0] == "GET"
+    assert (
+        actual_args.args[1]
+        == "https://longrunning.googleapis.com:443/v3/operations/sample1"
+    )
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, operations_pb2.Operation)
+    assert response.name == "operations/sample1"
+    assert response.done is True
+
+
+def test_get_operation_rest_failure():
+    client = _get_operations_client(http_options=None)
+
+    with mock.patch.object(Session, "request") as req:
+        response_value = Response()
+        response_value.status_code = 400
+        mock_request = mock.MagicMock()
+        mock_request.method = "GET"
+        mock_request.url = (
+            "https://longrunning.googleapis.com:443/v1/operations/sample1"
+        )
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            client.get_operation("operations/sample1")
+
+
+def test_delete_operation_rest(
+    transport: str = "rest", request_type=operations_pb2.DeleteOperationRequest
+):
+    client = _get_operations_client()
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(Session, "request") as req:
+        # Wrap the value into a proper Response obj
+        response_value = Response()
+        response_value.status_code = 200
+        json_return_value = ""
+        response_value._content = json_return_value.encode("UTF-8")
+        req.return_value = response_value
+        client.delete_operation(name="operations/sample1")
+        assert req.call_count == 1
+        actual_args = req.call_args
+        assert actual_args.args[0] == "DELETE"
+        assert (
+            actual_args.args[1]
+            == "https://longrunning.googleapis.com:443/v3/operations/sample1"
+        )
+
+
+def test_delete_operation_rest_failure():
+    client = _get_operations_client(http_options=None)
+
+    with mock.patch.object(Session, "request") as req:
+        response_value = Response()
+        response_value.status_code = 400
+        mock_request = mock.MagicMock()
+        mock_request.method = "DELETE"
+        mock_request.url = (
+            "https://longrunning.googleapis.com:443/v1/operations/sample1"
+        )
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            client.delete_operation(name="operations/sample1")
+
+
+def test_cancel_operation_rest(transport: str = "rest"):
+    client = _get_operations_client()
+
+    # Mock the http request call within the method and fake a response.
+    with mock.patch.object(Session, "request") as req:
+        # Wrap the value into a proper Response obj
+        response_value = Response()
+        response_value.status_code = 200
+        json_return_value = ""
+        response_value._content = json_return_value.encode("UTF-8")
+        req.return_value = response_value
+        client.cancel_operation(name="operations/sample1")
+        assert req.call_count == 1
+        actual_args = req.call_args
+        assert actual_args.args[0] == "POST"
+        assert (
+            actual_args.args[1]
+            == "https://longrunning.googleapis.com:443/v3/operations/sample1:cancel"
+        )
+
+
+def test_cancel_operation_rest_failure():
+    client = _get_operations_client(http_options=None)
+
+    with mock.patch.object(Session, "request") as req:
+        response_value = Response()
+        response_value.status_code = 400
+        mock_request = mock.MagicMock()
+        mock_request.method = "POST"
+        mock_request.url = (
+            "https://longrunning.googleapis.com:443/v1/operations/sample1:cancel"
+        )
+        response_value.request = mock_request
+        req.return_value = response_value
+        with pytest.raises(core_exceptions.GoogleAPIError):
+            client.cancel_operation(name="operations/sample1")
+
+
+def test_credentials_transport_error():
+    # It is an error to provide credentials and a transport instance.
+    transport = transports.OperationsRestTransport(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+    with pytest.raises(ValueError):
+        AbstractOperationsClient(
+            credentials=ga_credentials.AnonymousCredentials(), transport=transport,
+        )
+
+    # It is an error to provide a credentials file and a transport instance.
+    transport = transports.OperationsRestTransport(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+    with pytest.raises(ValueError):
+        AbstractOperationsClient(
+            client_options={"credentials_file": "credentials.json"},
+            transport=transport,
+        )
+
+    # It is an error to provide scopes and a transport instance.
+    transport = transports.OperationsRestTransport(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+    with pytest.raises(ValueError):
+        AbstractOperationsClient(
+            client_options={"scopes": ["1", "2"]}, transport=transport,
+        )
+
+
+def test_transport_instance():
+    # A client may be instantiated with a custom transport instance.
+    transport = transports.OperationsRestTransport(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+    client = AbstractOperationsClient(transport=transport)
+    assert client.transport is transport
+
+
+@pytest.mark.parametrize("transport_class", [transports.OperationsRestTransport])
+def test_transport_adc(transport_class):
+    # Test default credentials are used if not provided.
+    with mock.patch.object(google.auth, "default") as adc:
+        adc.return_value = (ga_credentials.AnonymousCredentials(), None)
+        transport_class()
+        adc.assert_called_once()
+
+
+def test_operations_base_transport_error():
+    # Passing both a credentials object and credentials_file should raise an error
+    with pytest.raises(core_exceptions.DuplicateCredentialArgs):
+        transports.OperationsTransport(
+            credentials=ga_credentials.AnonymousCredentials(),
+            credentials_file="credentials.json",
+        )
+
+
+def test_operations_base_transport():
+    # Instantiate the base transport.
+    with mock.patch(
+        "google.api_core.operations_v1.transports.OperationsTransport.__init__"
+    ) as Transport:
+        Transport.return_value = None
+        transport = transports.OperationsTransport(
+            credentials=ga_credentials.AnonymousCredentials(),
+        )
+
+    # Every method on the transport should just blindly
+    # raise NotImplementedError.
+    methods = (
+        "list_operations",
+        "get_operation",
+        "delete_operation",
+        "cancel_operation",
+    )
+    for method in methods:
+        with pytest.raises(NotImplementedError):
+            getattr(transport, method)(request=object())
+
+    with pytest.raises(NotImplementedError):
+        transport.close()
+
+
+def test_operations_base_transport_with_credentials_file():
+    # Instantiate the base transport with a credentials file
+    with mock.patch.object(
+        google.auth, "load_credentials_from_file", autospec=True
+    ) as load_creds, mock.patch(
+        "google.api_core.operations_v1.transports.OperationsTransport._prep_wrapped_messages"
+    ) as Transport:
+        Transport.return_value = None
+        load_creds.return_value = (ga_credentials.AnonymousCredentials(), None)
+        transports.OperationsTransport(
+            credentials_file="credentials.json", quota_project_id="octopus",
+        )
+        load_creds.assert_called_once_with(
+            "credentials.json",
+            scopes=None,
+            default_scopes=(),
+            quota_project_id="octopus",
+        )
+
+
+def test_operations_base_transport_with_adc():
+    # Test the default credentials are used if credentials and credentials_file are None.
+    with mock.patch.object(google.auth, "default", autospec=True) as adc, mock.patch(
+        "google.api_core.operations_v1.transports.OperationsTransport._prep_wrapped_messages"
+    ) as Transport:
+        Transport.return_value = None
+        adc.return_value = (ga_credentials.AnonymousCredentials(), None)
+        transports.OperationsTransport()
+        adc.assert_called_once()
+
+
+def test_operations_auth_adc():
+    # If no credentials are provided, we should use ADC credentials.
+    with mock.patch.object(google.auth, "default", autospec=True) as adc:
+        adc.return_value = (ga_credentials.AnonymousCredentials(), None)
+        AbstractOperationsClient()
+        adc.assert_called_once_with(
+            scopes=None, default_scopes=(), quota_project_id=None,
+        )
+
+
+def test_operations_http_transport_client_cert_source_for_mtls():
+    cred = ga_credentials.AnonymousCredentials()
+    with mock.patch(
+        "google.auth.transport.requests.AuthorizedSession.configure_mtls_channel"
+    ) as mock_configure_mtls_channel:
+        transports.OperationsRestTransport(
+            credentials=cred, client_cert_source_for_mtls=client_cert_source_callback
+        )
+        mock_configure_mtls_channel.assert_called_once_with(client_cert_source_callback)
+
+
+def test_operations_host_no_port():
+    client = AbstractOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        client_options=client_options.ClientOptions(
+            api_endpoint="longrunning.googleapis.com"
+        ),
+    )
+    assert client.transport._host == "longrunning.googleapis.com:443"
+
+
+def test_operations_host_with_port():
+    client = AbstractOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        client_options=client_options.ClientOptions(
+            api_endpoint="longrunning.googleapis.com:8000"
+        ),
+    )
+    assert client.transport._host == "longrunning.googleapis.com:8000"
+
+
+def test_common_billing_account_path():
+    billing_account = "squid"
+    expected = "billingAccounts/{billing_account}".format(
+        billing_account=billing_account,
+    )
+    actual = AbstractOperationsClient.common_billing_account_path(billing_account)
+    assert expected == actual
+
+
+def test_parse_common_billing_account_path():
+    expected = {
+        "billing_account": "clam",
+    }
+    path = AbstractOperationsClient.common_billing_account_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = AbstractOperationsClient.parse_common_billing_account_path(path)
+    assert expected == actual
+
+
+def test_common_folder_path():
+    folder = "whelk"
+    expected = "folders/{folder}".format(folder=folder,)
+    actual = AbstractOperationsClient.common_folder_path(folder)
+    assert expected == actual
+
+
+def test_parse_common_folder_path():
+    expected = {
+        "folder": "octopus",
+    }
+    path = AbstractOperationsClient.common_folder_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = AbstractOperationsClient.parse_common_folder_path(path)
+    assert expected == actual
+
+
+def test_common_organization_path():
+    organization = "oyster"
+    expected = "organizations/{organization}".format(organization=organization,)
+    actual = AbstractOperationsClient.common_organization_path(organization)
+    assert expected == actual
+
+
+def test_parse_common_organization_path():
+    expected = {
+        "organization": "nudibranch",
+    }
+    path = AbstractOperationsClient.common_organization_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = AbstractOperationsClient.parse_common_organization_path(path)
+    assert expected == actual
+
+
+def test_common_project_path():
+    project = "cuttlefish"
+    expected = "projects/{project}".format(project=project,)
+    actual = AbstractOperationsClient.common_project_path(project)
+    assert expected == actual
+
+
+def test_parse_common_project_path():
+    expected = {
+        "project": "mussel",
+    }
+    path = AbstractOperationsClient.common_project_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = AbstractOperationsClient.parse_common_project_path(path)
+    assert expected == actual
+
+
+def test_common_location_path():
+    project = "winkle"
+    location = "nautilus"
+    expected = "projects/{project}/locations/{location}".format(
+        project=project, location=location,
+    )
+    actual = AbstractOperationsClient.common_location_path(project, location)
+    assert expected == actual
+
+
+def test_parse_common_location_path():
+    expected = {
+        "project": "scallop",
+        "location": "abalone",
+    }
+    path = AbstractOperationsClient.common_location_path(**expected)
+
+    # Check that the path construction is reversible.
+    actual = AbstractOperationsClient.parse_common_location_path(path)
+    assert expected == actual
+
+
+def test_client_withDEFAULT_CLIENT_INFO():
+    client_info = gapic_v1.client_info.ClientInfo()
+
+    with mock.patch.object(
+        transports.OperationsTransport, "_prep_wrapped_messages"
+    ) as prep:
+        AbstractOperationsClient(
+            credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
+        )
+        prep.assert_called_once_with(client_info)
+
+    with mock.patch.object(
+        transports.OperationsTransport, "_prep_wrapped_messages"
+    ) as prep:
+        transport_class = AbstractOperationsClient.get_transport_class()
+        transport_class(
+            credentials=ga_credentials.AnonymousCredentials(), client_info=client_info,
+        )
+        prep.assert_called_once_with(client_info)