fix: revert "fix: do not error on LROs with no response or error" (#294)

Reverts #258


From @TrucHLe 
> A while ago you helped me submit this pull request for the Python LRO client library. This was about making the LRO library not throw an error when receiving an empty LRO response. It turns out that the Python LRO library has always been behaving correctly by accepting empty LRO responses. It would just throw an error if the response field is NULL (see screenshot).
>
> Since then I've consulted other LRO client library owners (@vam-google ) and it turns out that the fault was in the CCAI Insights server code. CCAI Insights was returning NULL responses instead of empty responses (more context here b/202432905). Since then the issue has been fixed in our server and has rolled out to production, so reverting the Python LRO false fix wouldn't break our code sample.

google/api_core/operation.py

@@ -140,11 +140,11 @@ def _set_result_from_operation(self):
                 )
                 self.set_exception(exception)
             else:
-                # Some APIs set `done: true`, with an empty response.
-                # Set the result to an empty message of the expected
-                # result type.
-                # https://google.aip.dev/151
-                self.set_result(self._result_type())
+                exception = exceptions.GoogleAPICallError(
+                    "Unexpected state: Long-running operation had neither "
+                    "response nor error set."
+                )
+                self.set_exception(exception)
 
     def _refresh_and_update(self, retry=polling.DEFAULT_RETRY):
         """Refresh the operation and update the result if needed.

google/api_core/operation_async.py

@@ -136,11 +136,11 @@ def _set_result_from_operation(self):
                 )
                 self.set_exception(exception)
             else:
-                # Some APIs set `done: true`, with an empty response.
-                # Set the result to an empty message of the expected
-                # result type.
-                # https://google.aip.dev/151
-                self.set_result(self._result_type())
+                exception = exceptions.GoogleAPICallError(
+                    "Unexpected state: Long-running operation had neither "
+                    "response nor error set."
+                )
+                self.set_exception(exception)
 
     async def _refresh_and_update(self, retry=async_future.DEFAULT_RETRY):
         """Refresh the operation and update the result if needed.

tests/asyncio/test_operation_async.py

@@ -158,17 +158,17 @@ async def test_exception():
 
 @mock.patch("asyncio.sleep", autospec=True)
 @pytest.mark.asyncio
-async def test_done_with_no_error_or_response(unused_sleep):
+async def test_unexpected_result(unused_sleep):
     responses = [
         make_operation_proto(),
         # Second operation response is done, but has not error or response.
         make_operation_proto(done=True),
     ]
     future, _, _ = make_operation_future(responses)
 
-    result = await future.result()
+    exception = await future.exception()
 
-    assert isinstance(result, struct_pb2.Struct)
+    assert "Unexpected state" in "{!r}".format(exception)
 
 
 def test_from_gapic():

tests/unit/test_operation.py

@@ -169,15 +169,17 @@ def test_exception_with_error_code():
     assert isinstance(exception, exceptions.NotFound)
 
 
-def test_done_with_no_error_or_response():
+def test_unexpected_result():
     responses = [
         make_operation_proto(),
         # Second operation response is done, but has not error or response.
         make_operation_proto(done=True),
     ]
     future, _, _ = make_operation_future(responses)
 
-    assert isinstance(future.result(), struct_pb2.Struct)
+    exception = future.exception()
+
+    assert "Unexpected state" in "{!r}".format(exception)
 
 
 def test__refresh_http():