Use http.HTTPStatus in components/[wxyz]* (#58330)

homeassistant/components/wallbox/__init__.py

@@ -1,5 +1,6 @@
 """The Wallbox integration."""
 from datetime import timedelta
+from http import HTTPStatus
 import logging
 
 import requests
@@ -45,7 +46,7 @@ class WallboxHub:
             self._wallbox.authenticate()
             return True
         except requests.exceptions.HTTPError as wallbox_connection_error:
-            if wallbox_connection_error.response.status_code == 403:
+            if wallbox_connection_error.response.status_code == HTTPStatus.FORBIDDEN:
                 raise InvalidAuth from wallbox_connection_error
             raise ConnectionError from wallbox_connection_error
 

homeassistant/components/withings/common.py

@@ -7,6 +7,7 @@ from dataclasses import dataclass
 import datetime
 from datetime import timedelta
 from enum import Enum, IntEnum
+from http import HTTPStatus
 import logging
 import re
 from typing import Any, Dict
@@ -32,7 +33,6 @@ from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
 from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
 from homeassistant.const import (
     CONF_WEBHOOK_ID,
-    HTTP_UNAUTHORIZED,
     MASS_KILOGRAMS,
     PERCENTAGE,
     SPEED_METERS_PER_SECOND,
@@ -58,7 +58,7 @@ from .const import Measurement
 
 _LOGGER = logging.getLogger(const.LOG_NAMESPACE)
 NOT_AUTHENTICATED_ERROR = re.compile(
-    f"^{HTTP_UNAUTHORIZED},.*",
+    f"^{HTTPStatus.UNAUTHORIZED},.*",
     re.IGNORECASE,
 )
 DATA_UPDATED_SIGNAL = "withings_entity_state_updated"

homeassistant/components/wsdot/sensor.py

@@ -1,5 +1,6 @@
 """Support for Washington State Department of Transportation (WSDOT) data."""
 from datetime import datetime, timedelta, timezone
+from http import HTTPStatus
 import logging
 import re
 
@@ -13,7 +14,6 @@ from homeassistant.const import (
     CONF_API_KEY,
     CONF_ID,
     CONF_NAME,
-    HTTP_OK,
     TIME_MINUTES,
 )
 import homeassistant.helpers.config_validation as cv
@@ -111,7 +111,7 @@ class WashingtonStateTravelTimeSensor(WashingtonStateTransportSensor):
         }
 
         response = requests.get(RESOURCE, params, timeout=10)
-        if response.status_code != HTTP_OK:
+        if response.status_code != HTTPStatus.OK:
             _LOGGER.warning("Invalid response from WSDOT API")
         else:
             self._data = response.json()

homeassistant/components/xiaomi/device_tracker.py

@@ -1,4 +1,5 @@
 """Support for Xiaomi Mi routers."""
+from http import HTTPStatus
 import logging
 
 import requests
@@ -9,7 +10,7 @@ from homeassistant.components.device_tracker import (
     PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
     DeviceScanner,
 )
-from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, HTTP_OK
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
 import homeassistant.helpers.config_validation as cv
 
 _LOGGER = logging.getLogger(__name__)
@@ -112,7 +113,7 @@ def _retrieve_list(host, token, **kwargs):
     except requests.exceptions.Timeout:
         _LOGGER.exception("Connection to the router timed out at URL %s", url)
         return
-    if res.status_code != HTTP_OK:
+    if res.status_code != HTTPStatus.OK:
         _LOGGER.exception("Connection failed with http code %s", res.status_code)
         return
     try:
@@ -150,7 +151,7 @@ def _get_token(host, username, password):
     except requests.exceptions.Timeout:
         _LOGGER.exception("Connection to the router timed out")
         return
-    if res.status_code == HTTP_OK:
+    if res.status_code == HTTPStatus.OK:
         try:
             result = res.json()
         except ValueError:

homeassistant/components/xmpp/notify.py

@@ -1,5 +1,6 @@
 """Jabber (XMPP) notification service."""
 from concurrent.futures import TimeoutError as FutTimeoutError
+from http import HTTPStatus
 import logging
 import mimetypes
 import pathlib
@@ -29,7 +30,6 @@ from homeassistant.const import (
     CONF_RESOURCE,
     CONF_ROOM,
     CONF_SENDER,
-    HTTP_BAD_REQUEST,
 )
 import homeassistant.helpers.config_validation as cv
 import homeassistant.helpers.template as template_helper
@@ -267,7 +267,7 @@ async def async_send_message(  # noqa: C901
 
             result = await hass.async_add_executor_job(get_url, url)
 
-            if result.status_code >= HTTP_BAD_REQUEST:
+            if result.status_code >= HTTPStatus.BAD_REQUEST:
                 _LOGGER.error("Could not load file from %s", url)
                 return None
 

homeassistant/components/yandextts/tts.py

@@ -1,5 +1,6 @@
 """Support for the yandex speechkit tts  service."""
 import asyncio
+from http import HTTPStatus
 import logging
 
 import aiohttp
@@ -7,7 +8,7 @@ import async_timeout
 import voluptuous as vol
 
 from homeassistant.components.tts import CONF_LANG, PLATFORM_SCHEMA, Provider
-from homeassistant.const import CONF_API_KEY, HTTP_OK
+from homeassistant.const import CONF_API_KEY
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 import homeassistant.helpers.config_validation as cv
 
@@ -133,7 +134,7 @@ class YandexSpeechKitProvider(Provider):
 
                 request = await websession.get(YANDEX_API_URL, params=url_param)
 
-                if request.status != HTTP_OK:
+                if request.status != HTTPStatus.OK:
                     _LOGGER.error(
                         "Error %d on load URL %s", request.status, request.url
                     )

tests/components/wallbox/__init__.py

@@ -1,5 +1,6 @@
 """Tests for the Wallbox integration."""
 
+from http import HTTPStatus
 import json
 
 import requests_mock
@@ -33,12 +34,12 @@ async def setup_integration(hass):
         mock_request.get(
             "https://api.wall-box.com/auth/token/user",
             text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
         mock_request.get(
             "https://api.wall-box.com/chargers/status/12345",
             json=test_response,
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
         await hass.config_entries.async_setup(entry.entry_id)
         await hass.async_block_till_done()

tests/components/wallbox/test_config_flow.py

@@ -1,4 +1,5 @@
 """Test the Wallbox config flow."""
+from http import HTTPStatus
 import json
 
 import requests_mock
@@ -33,7 +34,7 @@ async def test_form_cannot_authenticate(hass):
         mock_request.get(
             "https://api.wall-box.com/auth/token/user",
             text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
-            status_code=403,
+            status_code=HTTPStatus.FORBIDDEN,
         )
         result2 = await hass.config_entries.flow.async_configure(
             result["flow_id"],
@@ -58,12 +59,12 @@ async def test_form_cannot_connect(hass):
         mock_request.get(
             "https://api.wall-box.com/auth/token/user",
             text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
         mock_request.get(
             "https://api.wall-box.com/chargers/status/12345",
             text='{"Temperature": 100, "Location": "Toronto", "Datetime": "2020-07-23", "Units": "Celsius"}',
-            status_code=404,
+            status_code=HTTPStatus.NOT_FOUND,
         )
         result2 = await hass.config_entries.flow.async_configure(
             result["flow_id"],
@@ -88,12 +89,12 @@ async def test_form_validate_input(hass):
         mock_request.get(
             "https://api.wall-box.com/auth/token/user",
             text='{"jwt":"fakekeyhere","user_id":12345,"ttl":145656758,"error":false,"status":200}',
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
         mock_request.get(
             "https://api.wall-box.com/chargers/status/12345",
             text='{"Temperature": 100, "Location": "Toronto", "Datetime": "2020-07-23", "Units": "Celsius"}',
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
         result2 = await hass.config_entries.flow.async_configure(
             result["flow_id"],

tests/components/webhook/test_init.py

@@ -1,4 +1,6 @@
 """Test the webhook component."""
+from http import HTTPStatus
+
 import pytest
 
 from homeassistant.config import async_process_ha_core_config
@@ -24,13 +26,13 @@ async def test_unregistering_webhook(hass, mock_client):
     hass.components.webhook.async_register("test", "Test hook", webhook_id, handle)
 
     resp = await mock_client.post(f"/api/webhook/{webhook_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
 
     hass.components.webhook.async_unregister(webhook_id)
 
     resp = await mock_client.post(f"/api/webhook/{webhook_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
 
 
@@ -54,14 +56,14 @@ async def test_async_generate_path(hass):
 async def test_posting_webhook_nonexisting(hass, mock_client):
     """Test posting to a nonexisting webhook."""
     resp = await mock_client.post("/api/webhook/non-existing")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
 
 
 async def test_posting_webhook_invalid_json(hass, mock_client):
     """Test posting to a nonexisting webhook."""
     hass.components.webhook.async_register("test", "Test hook", "hello", None)
     resp = await mock_client.post("/api/webhook/hello", data="not-json")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
 
 
 async def test_posting_webhook_json(hass, mock_client):
@@ -76,7 +78,7 @@ async def test_posting_webhook_json(hass, mock_client):
     hass.components.webhook.async_register("test", "Test hook", webhook_id, handle)
 
     resp = await mock_client.post(f"/api/webhook/{webhook_id}", json={"data": True})
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
     assert hooks[0][0] is hass
     assert hooks[0][1] == webhook_id
@@ -95,7 +97,7 @@ async def test_posting_webhook_no_data(hass, mock_client):
     hass.components.webhook.async_register("test", "Test hook", webhook_id, handle)
 
     resp = await mock_client.post(f"/api/webhook/{webhook_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
     assert hooks[0][0] is hass
     assert hooks[0][1] == webhook_id
@@ -115,7 +117,7 @@ async def test_webhook_put(hass, mock_client):
     hass.components.webhook.async_register("test", "Test hook", webhook_id, handle)
 
     resp = await mock_client.put(f"/api/webhook/{webhook_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
     assert hooks[0][0] is hass
     assert hooks[0][1] == webhook_id
@@ -134,7 +136,7 @@ async def test_webhook_head(hass, mock_client):
     hass.components.webhook.async_register("test", "Test hook", webhook_id, handle)
 
     resp = await mock_client.head(f"/api/webhook/{webhook_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert len(hooks) == 1
     assert hooks[0][0] is hass
     assert hooks[0][1] == webhook_id

tests/components/withings/common.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from http import HTTPStatus
 from unittest.mock import MagicMock
 from urllib.parse import urlparse
 
@@ -210,7 +211,7 @@ class ComponentFactory:
         # Simulate user being redirected from withings site.
         client: TestClient = await self._hass_client()
         resp = await client.get(f"{AUTH_CALLBACK_PATH}?code=abcd&state={state}")
-        assert resp.status == 200
+        assert resp.status == HTTPStatus.OK
         assert resp.headers["content-type"] == "text/html; charset=utf-8"
 
         self._aioclient_mock.clear_requests()

tests/components/withings/test_common.py

@@ -1,5 +1,6 @@
 """Tests for the Withings component."""
 import datetime
+from http import HTTPStatus
 import re
 from typing import Any
 from unittest.mock import MagicMock
@@ -43,7 +44,7 @@ async def test_config_entry_withings_api(hass: HomeAssistant) -> None:
     with requests_mock.mock() as rqmck:
         rqmck.get(
             re.compile(".*"),
-            status_code=200,
+            status_code=HTTPStatus.OK,
             json={"status": 0, "body": {"message": "success"}},
         )
 
@@ -119,7 +120,7 @@ async def test_webhook_head(
 
     client: TestClient = await aiohttp_client(hass.http.app)
     resp = await client.head(urlparse(data_manager.webhook_config.url).path)
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
 
 
 async def test_webhook_put(
@@ -141,7 +142,7 @@ async def test_webhook_put(
     # Wait for remaining tasks to complete.
     await hass.async_block_till_done()
 
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     data = await resp.json()
     assert data
     assert data["code"] == 2
@@ -196,7 +197,7 @@ async def test_data_manager_webhook_subscription(
     aioclient_mock.request(
         "HEAD",
         data_manager.webhook_config.url,
-        status=200,
+        status=HTTPStatus.OK,
     )
 
     # Test subscribing

tests/components/withings/test_config_flow.py

@@ -1,4 +1,6 @@
 """Tests for config flow."""
+from http import HTTPStatus
+
 from aiohttp.test_utils import TestClient
 
 from homeassistant import config_entries
@@ -83,7 +85,7 @@ async def test_config_reauth_profile(
 
     client: TestClient = await hass_client_no_auth()
     resp = await client.get(f"{AUTH_CALLBACK_PATH}?code=abcd&state={state}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert resp.headers["content-type"] == "text/html; charset=utf-8"
 
     aioclient_mock.clear_requests()

tests/components/xbox/test_config_flow.py

@@ -1,4 +1,5 @@
 """Test the xbox config flow."""
+from http import HTTPStatus
 from unittest.mock import patch
 
 from homeassistant import config_entries, data_entry_flow, setup
@@ -56,7 +57,7 @@ async def test_full_flow(
 
     client = await hass_client_no_auth()
     resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert resp.headers["content-type"] == "text/html; charset=utf-8"
 
     aioclient_mock.post(

tests/components/xiaomi/test_device_tracker.py

@@ -1,4 +1,5 @@
 """The tests for the Xiaomi router device tracker platform."""
+from http import HTTPStatus
 import logging
 from unittest.mock import MagicMock, call, patch
 
@@ -40,8 +41,8 @@ def mocked_requests(*args, **kwargs):
             return self.json()
 
         def raise_for_status(self):
-            """Raise an HTTPError if status is not 200."""
-            if self.status_code != 200:
+            """Raise an HTTPError if status is not OK."""
+            if self.status_code != HTTPStatus.OK:
                 raise requests.HTTPError(self.status_code)
 
     data = kwargs.get("data")

tests/components/yandextts/test_tts.py

@@ -69,7 +69,7 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
@@ -101,7 +101,7 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {
@@ -139,7 +139,7 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
@@ -175,7 +175,10 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, exc=asyncio.TimeoutError(), params=url_param
+            self._base_url,
+            status=HTTPStatus.OK,
+            exc=asyncio.TimeoutError(),
+            params=url_param,
         )
 
         config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
@@ -241,7 +244,7 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {
@@ -279,7 +282,7 @@ class TestTTSYandexPlatform:
             "speed": 1,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {
@@ -317,7 +320,7 @@ class TestTTSYandexPlatform:
             "speed": "0.1",
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {
@@ -351,7 +354,7 @@ class TestTTSYandexPlatform:
             "speed": 2,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
 
         config = {
@@ -385,7 +388,7 @@ class TestTTSYandexPlatform:
             "speed": 2,
         }
         aioclient_mock.get(
-            self._base_url, status=200, content=b"test", params=url_param
+            self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
         )
         config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
 

tests/components/zwave_js/test_api.py

@@ -1,5 +1,6 @@
 """Test the Z-Wave JS Websocket API."""
 from copy import deepcopy
+from http import HTTPStatus
 import json
 from unittest.mock import patch
 
@@ -1976,7 +1977,7 @@ async def test_dump_view(integration, hass_client):
         return_value=[{"hello": "world"}, {"second": "msg"}],
     ):
         resp = await client.get(f"/api/zwave_js/dump/{integration.entry_id}")
-    assert resp.status == 200
+    assert resp.status == HTTPStatus.OK
     assert json.loads(await resp.text()) == [{"hello": "world"}, {"second": "msg"}]
 
 
@@ -2060,7 +2061,7 @@ async def test_firmware_upload_view_failed_command(
             f"/api/zwave_js/firmware/upload/{integration.entry_id}/{multisensor_6.node_id}",
             data={"file": firmware_file},
         )
-        assert resp.status == 400
+        assert resp.status == HTTPStatus.BAD_REQUEST
 
 
 async def test_firmware_upload_view_invalid_payload(
@@ -2072,7 +2073,7 @@ async def test_firmware_upload_view_invalid_payload(
         f"/api/zwave_js/firmware/upload/{integration.entry_id}/{multisensor_6.node_id}",
         data={"wrong_key": bytes(10)},
     )
-    assert resp.status == 400
+    assert resp.status == HTTPStatus.BAD_REQUEST
 
 
 @pytest.mark.parametrize(
@@ -2087,7 +2088,7 @@ async def test_view_non_admin_user(
     # Verify we require admin user
     hass_admin_user.groups = []
     resp = await client.request(method, url.format(integration.entry_id))
-    assert resp.status == 401
+    assert resp.status == HTTPStatus.UNAUTHORIZED
 
 
 @pytest.mark.parametrize(
@@ -2104,7 +2105,7 @@ async def test_node_view_non_admin_user(
     resp = await client.request(
         method, url.format(integration.entry_id, multisensor_6.node_id)
     )
-    assert resp.status == 401
+    assert resp.status == HTTPStatus.UNAUTHORIZED
 
 
 @pytest.mark.parametrize(
@@ -2118,7 +2119,7 @@ async def test_view_invalid_entry_id(integration, hass_client, method, url):
     """Test an invalid config entry id parameter."""
     client = await hass_client()
     resp = await client.request(method, url)
-    assert resp.status == 400
+    assert resp.status == HTTPStatus.BAD_REQUEST
 
 
 @pytest.mark.parametrize(
@@ -2129,7 +2130,7 @@ async def test_view_invalid_node_id(integration, hass_client, method, url):
     """Test an invalid config entry id parameter."""
     client = await hass_client()
     resp = await client.request(method, url.format(integration.entry_id))
-    assert resp.status == 404
+    assert resp.status == HTTPStatus.NOT_FOUND
 
 
 async def test_subscribe_log_updates(hass, integration, client, hass_ws_client):