DatabricksHook: fix status property to work with ClientResponse used …

…in async mode (#43333)

* fix status property to work with ClientResponse used in async mode in Databricks Hook

* add unittest for Databricks async api call

providers/src/airflow/providers/databricks/hooks/databricks_base.py

@@ -645,7 +645,7 @@ async def _a_do_api_call(self, endpoint_info: tuple[str, str], json: dict[str, A
                         headers={**headers, **self.user_agent_header},
                         timeout=self.timeout_seconds,
                     ) as response:
-                        self.log.debug("Response Status Code: %s", response.status_code)
+                        self.log.debug("Response Status Code: %s", response.status)
                         self.log.debug("Response text: %s", response.text)
                         response.raise_for_status()
                         return await response.json()

providers/tests/databricks/hooks/test_databricks.py

@@ -1309,6 +1309,20 @@ async def test_async_do_api_call_respects_schema(self, mock_get):
         mock_get.assert_called_once()
         assert mock_get.call_args.args == (f"http://{HOST}:7908/api/2.1/foo/bar",)
 
+    @pytest.mark.asyncio
+    @mock.patch("airflow.providers.databricks.hooks.databricks_base.aiohttp.ClientSession.get")
+    async def test_async_do_api_call_only_existing_response_properties_are_read(self, mock_get):
+        self.hook.log.setLevel("DEBUG")
+        response = mock_get.return_value.__aenter__.return_value
+        response.mock_add_spec(aiohttp.ClientResponse, spec_set=True)
+        response.json = AsyncMock(return_value={"bar": "baz"})
+        async with self.hook:
+            run_page_url = await self.hook._a_do_api_call(("GET", "api/2.1/foo/bar"))
+
+        assert run_page_url == {"bar": "baz"}
+        mock_get.assert_called_once()
+        assert mock_get.call_args.args == (f"http://{HOST}:7908/api/2.1/foo/bar",)
+
 
 class TestRunState:
     def test_is_terminal_true(self):