Remove default OAuth implementation from Tesla Fleet (#132431)

homeassistant/components/tesla_fleet/__init__.py

@@ -34,7 +34,6 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.device_registry import DeviceInfo
 
-from .config_flow import OAuth2FlowHandler
 from .const import DOMAIN, LOGGER, MODELS
 from .coordinator import (
     TeslaFleetEnergySiteInfoCoordinator,
@@ -42,7 +41,6 @@ from .coordinator import (
     TeslaFleetVehicleDataCoordinator,
 )
 from .models import TeslaFleetData, TeslaFleetEnergyData, TeslaFleetVehicleData
-from .oauth import TeslaSystemImplementation
 
 PLATFORMS: Final = [
     Platform.BINARY_SENSOR,
@@ -73,11 +71,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslaFleetConfigEntry) -
     scopes: list[Scope] = [Scope(s) for s in token["scp"]]
     region: str = token["ou_code"].lower()
 
-    OAuth2FlowHandler.async_register_implementation(
-        hass,
-        TeslaSystemImplementation(hass),
-    )
-
     implementation = await async_get_config_entry_implementation(hass, entry)
     oauth_session = OAuth2Session(hass, entry, implementation)
     refresh_lock = asyncio.Lock()

homeassistant/components/tesla_fleet/config_flow.py

@@ -12,7 +12,6 @@ from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
 from homeassistant.helpers import config_entry_oauth2_flow
 
 from .const import DOMAIN, LOGGER
-from .oauth import TeslaSystemImplementation
 
 
 class OAuth2FlowHandler(
@@ -31,11 +30,6 @@ class OAuth2FlowHandler(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Handle a flow start."""
-        self.async_register_implementation(
-            self.hass,
-            TeslaSystemImplementation(self.hass),
-        )
-
         return await super().async_step_user()
 
     async def async_oauth_create_entry(

homeassistant/components/tesla_fleet/oauth.py

@@ -1,8 +1,5 @@
 """Provide oauth implementations for the Tesla Fleet integration."""
 
-import base64
-import hashlib
-import secrets
 from typing import Any
 
 from homeassistant.components.application_credentials import (
@@ -11,59 +8,8 @@ from homeassistant.components.application_credentials import (
     ClientCredential,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers import config_entry_oauth2_flow
 
-from .const import AUTHORIZE_URL, CLIENT_ID, DOMAIN, SCOPES, TOKEN_URL
-
-
-class TeslaSystemImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
-    """Tesla Fleet API open source Oauth2 implementation."""
-
-    code_verifier: str
-    code_challenge: str
-
-    def __init__(self, hass: HomeAssistant) -> None:
-        """Initialize open source Oauth2 implementation."""
-
-        # Setup PKCE
-        self.code_verifier = secrets.token_urlsafe(32)
-        hashed_verifier = hashlib.sha256(self.code_verifier.encode()).digest()
-        self.code_challenge = (
-            base64.urlsafe_b64encode(hashed_verifier).decode().replace("=", "")
-        )
-        super().__init__(
-            hass,
-            DOMAIN,
-            CLIENT_ID,
-            "",
-            AUTHORIZE_URL,
-            TOKEN_URL,
-        )
-
-    @property
-    def name(self) -> str:
-        """Name of the implementation."""
-        return "Built-in open source client ID"
-
-    @property
-    def extra_authorize_data(self) -> dict[str, Any]:
-        """Extra data that needs to be appended to the authorize url."""
-        return {
-            "prompt": "login",
-            "scope": " ".join(SCOPES),
-            "code_challenge": self.code_challenge,  # PKCE
-        }
-
-    async def async_resolve_external_data(self, external_data: Any) -> dict:
-        """Resolve the authorization code to tokens."""
-        return await self._token_request(
-            {
-                "grant_type": "authorization_code",
-                "code": external_data["code"],
-                "redirect_uri": external_data["state"]["redirect_uri"],
-                "code_verifier": self.code_verifier,  # PKCE
-            }
-        )
+from .const import AUTHORIZE_URL, SCOPES, TOKEN_URL
 
 
 class TeslaUserImplementation(AuthImplementation):

tests/components/tesla_fleet/test_config_flow.py

@@ -11,7 +11,6 @@ from homeassistant.components.application_credentials import (
 )
 from homeassistant.components.tesla_fleet.const import (
     AUTHORIZE_URL,
-    CLIENT_ID,
     DOMAIN,
     SCOPES,
     TOKEN_URL,
@@ -52,69 +51,18 @@ async def access_token(hass: HomeAssistant) -> str:
     )
 
 
-@pytest.mark.usefixtures("current_request_with_host")
-async def test_full_flow(
-    hass: HomeAssistant,
-    hass_client_no_auth: ClientSessionGenerator,
-    aioclient_mock: AiohttpClientMocker,
-    access_token: str,
-) -> None:
-    """Check full flow."""
-    result = await hass.config_entries.flow.async_init(
-        DOMAIN, context={"source": SOURCE_USER}
-    )
-
-    state = config_entry_oauth2_flow._encode_jwt(
+@pytest.fixture(autouse=True)
+async def create_credential(hass: HomeAssistant) -> None:
+    """Create a user credential."""
+    # Create user application credential
+    assert await async_setup_component(hass, "application_credentials", {})
+    await async_import_client_credential(
         hass,
-        {
-            "flow_id": result["flow_id"],
-            "redirect_uri": REDIRECT,
-        },
+        DOMAIN,
+        ClientCredential("user_client_id", "user_client_secret"),
+        "user_cred",
     )
 
-    assert result["type"] is FlowResultType.EXTERNAL_STEP
-
-    assert result["url"].startswith(AUTHORIZE_URL)
-    parsed_url = urlparse(result["url"])
-    parsed_query = parse_qs(parsed_url.query)
-    assert parsed_query["response_type"][0] == "code"
-    assert parsed_query["client_id"][0] == CLIENT_ID
-    assert parsed_query["redirect_uri"][0] == REDIRECT
-    assert parsed_query["state"][0] == state
-    assert parsed_query["scope"][0] == " ".join(SCOPES)
-    assert parsed_query["code_challenge"][0] is not None
-
-    client = await hass_client_no_auth()
-    resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
-    assert resp.status == 200
-    assert resp.headers["content-type"] == "text/html; charset=utf-8"
-
-    aioclient_mock.clear_requests()
-    aioclient_mock.post(
-        TOKEN_URL,
-        json={
-            "refresh_token": "mock-refresh-token",
-            "access_token": access_token,
-            "type": "Bearer",
-            "expires_in": 60,
-        },
-    )
-    with patch(
-        "homeassistant.components.tesla_fleet.async_setup_entry", return_value=True
-    ) as mock_setup:
-        result = await hass.config_entries.flow.async_configure(result["flow_id"])
-
-    assert len(hass.config_entries.async_entries(DOMAIN)) == 1
-    assert len(mock_setup.mock_calls) == 1
-
-    assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert result["title"] == UNIQUE_ID
-    assert "result" in result
-    assert result["result"].unique_id == UNIQUE_ID
-    assert "token" in result["result"].data
-    assert result["result"].data["token"]["access_token"] == access_token
-    assert result["result"].data["token"]["refresh_token"] == "mock-refresh-token"
-
 
 @pytest.mark.usefixtures("current_request_with_host")
 async def test_full_flow_user_cred(
@@ -125,24 +73,10 @@ async def test_full_flow_user_cred(
 ) -> None:
     """Check full flow."""
 
-    # Create user application credential
-    assert await async_setup_component(hass, "application_credentials", {})
-    await async_import_client_credential(
-        hass,
-        DOMAIN,
-        ClientCredential("user_client_id", "user_client_secret"),
-        "user_cred",
-    )
-
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": SOURCE_USER}
     )
 
-    assert result["type"] is FlowResultType.FORM
-
-    result = await hass.config_entries.flow.async_configure(
-        result["flow_id"], {"implementation": "user_cred"}
-    )
     assert result["type"] is FlowResultType.EXTERNAL_STEP
 
     state = config_entry_oauth2_flow._encode_jwt(