Use http.HTTPStatus in components/[tuv]* (#58325)

homeassistant/components/tado/device_tracker.py

@@ -2,6 +2,7 @@
 import asyncio
 from collections import namedtuple
 from datetime import timedelta
+from http import HTTPStatus
 import logging
 
 import aiohttp
@@ -13,7 +14,7 @@ from homeassistant.components.device_tracker import (
     PLATFORM_SCHEMA,
     DeviceScanner,
 )
-from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, HTTP_OK
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
 import homeassistant.helpers.config_validation as cv
 from homeassistant.util import Throttle
@@ -114,7 +115,7 @@ class TadoDeviceScanner(DeviceScanner):
 
                 response = await self.websession.get(url)
 
-                if response.status != HTTP_OK:
+                if response.status != HTTPStatus.OK:
                     _LOGGER.warning("Error %d on %s", response.status, self.tadoapiurl)
                     return False
 

homeassistant/components/thethingsnetwork/sensor.py

@@ -1,5 +1,6 @@
 """Support for The Things Network's Data storage integration."""
 import asyncio
+from http import HTTPStatus
 import logging
 
 import aiohttp
@@ -13,8 +14,6 @@ from homeassistant.const import (
     ATTR_TIME,
     CONF_DEVICE_ID,
     CONTENT_TYPE_JSON,
-    HTTP_NOT_FOUND,
-    HTTP_UNAUTHORIZED,
 )
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 import homeassistant.helpers.config_validation as cv
@@ -134,15 +133,15 @@ class TtnDataStorage:
 
         status = response.status
 
-        if status == 204:
+        if status == HTTPStatus.NO_CONTENT:
             _LOGGER.error("The device is not available: %s", self._device_id)
             return None
 
-        if status == HTTP_UNAUTHORIZED:
+        if status == HTTPStatus.UNAUTHORIZED:
             _LOGGER.error("Not authorized for Application ID: %s", self._app_id)
             return None
 
-        if status == HTTP_NOT_FOUND:
+        if status == HTTPStatus.NOT_FOUND:
             _LOGGER.error("Application ID is not available: %s", self._app_id)
             return None
 

homeassistant/components/tomato/device_tracker.py

@@ -1,4 +1,5 @@
 """Support for Tomato routers."""
+from http import HTTPStatus
 import json
 import logging
 import re
@@ -18,8 +19,6 @@ from homeassistant.const import (
     CONF_SSL,
     CONF_USERNAME,
     CONF_VERIFY_SSL,
-    HTTP_OK,
-    HTTP_UNAUTHORIZED,
 )
 import homeassistant.helpers.config_validation as cv
 
@@ -104,7 +103,7 @@ class TomatoDeviceScanner(DeviceScanner):
 
             # Calling and parsing the Tomato api here. We only need the
             # wldev and dhcpd_lease values.
-            if response.status_code == HTTP_OK:
+            if response.status_code == HTTPStatus.OK:
 
                 for param, value in self.parse_api_pattern.findall(response.text):
 
@@ -112,7 +111,7 @@ class TomatoDeviceScanner(DeviceScanner):
                         self.last_results[param] = json.loads(value.replace("'", '"'))
                 return True
 
-            if response.status_code == HTTP_UNAUTHORIZED:
+            if response.status_code == HTTPStatus.UNAUTHORIZED:
                 # Authentication error
                 _LOGGER.exception(
                     "Failed to authenticate, please check your username and password"

homeassistant/components/twitter/notify.py

@@ -1,6 +1,7 @@
 """Twitter platform for notify component."""
 from datetime import datetime, timedelta
 from functools import partial
+from http import HTTPStatus
 import json
 import logging
 import mimetypes
@@ -14,7 +15,7 @@ from homeassistant.components.notify import (
     PLATFORM_SCHEMA,
     BaseNotificationService,
 )
-from homeassistant.const import CONF_ACCESS_TOKEN, CONF_USERNAME, HTTP_OK
+from homeassistant.const import CONF_ACCESS_TOKEN, CONF_USERNAME
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.event import async_track_point_in_time
 
@@ -88,7 +89,7 @@ class TwitterNotificationService(BaseNotificationService):
         if self.user:
             user_resp = self.api.request("users/lookup", {"screen_name": self.user})
             user_id = user_resp.json()[0]["id"]
-            if user_resp.status_code != HTTP_OK:
+            if user_resp.status_code != HTTPStatus.OK:
                 self.log_error_resp(user_resp)
             else:
                 _LOGGER.debug("Message posted: %s", user_resp.json())
@@ -108,7 +109,7 @@ class TwitterNotificationService(BaseNotificationService):
                 "statuses/update", {"status": message, "media_ids": media_id}
             )
 
-        if resp.status_code != HTTP_OK:
+        if resp.status_code != HTTPStatus.OK:
             self.log_error_resp(resp)
         else:
             _LOGGER.debug("Message posted: %s", resp.json())
@@ -171,7 +172,7 @@ class TwitterNotificationService(BaseNotificationService):
         while bytes_sent < total_bytes:
             chunk = file.read(4 * 1024 * 1024)
             resp = self.upload_media_append(chunk, media_id, segment_id)
-            if resp.status_code not in range(HTTP_OK, 299):
+            if not HTTPStatus.OK <= resp.status_code < HTTPStatus.MULTIPLE_CHOICES:
                 self.log_error_resp_append(resp)
                 return None
             segment_id = segment_id + 1
@@ -200,7 +201,7 @@ class TwitterNotificationService(BaseNotificationService):
             {"command": "STATUS", "media_id": media_id},
             method_override="GET",
         )
-        if resp.status_code != HTTP_OK:
+        if resp.status_code != HTTPStatus.OK:
             _LOGGER.error("Media processing error: %s", resp.json())
         processing_info = resp.json()["processing_info"]
 

homeassistant/components/uk_transport/sensor.py

@@ -1,5 +1,6 @@
 """Support for UK public transport data provided by transportapi.com."""
 from datetime import datetime, timedelta
+from http import HTTPStatus
 import logging
 import re
 
@@ -7,7 +8,7 @@ import requests
 import voluptuous as vol
 
 from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
-from homeassistant.const import CONF_MODE, HTTP_OK, TIME_MINUTES
+from homeassistant.const import CONF_MODE, TIME_MINUTES
 import homeassistant.helpers.config_validation as cv
 from homeassistant.util import Throttle
 import homeassistant.util.dt as dt_util
@@ -121,7 +122,7 @@ class UkTransportSensor(SensorEntity):
         )
 
         response = requests.get(self._url, params=request_params)
-        if response.status_code != HTTP_OK:
+        if response.status_code != HTTPStatus.OK:
             _LOGGER.warning("Invalid response from API")
         elif "error" in response.json():
             if "exceeded" in response.json()["error"]:

homeassistant/components/verisure/coordinator.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from datetime import timedelta
+from http import HTTPStatus
 
 from verisure import (
     Error as VerisureError,
@@ -10,7 +11,7 @@ from verisure import (
 )
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, HTTP_SERVICE_UNAVAILABLE
+from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
 from homeassistant.core import Event, HomeAssistant
 from homeassistant.helpers.storage import STORAGE_DIR
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
@@ -68,7 +69,7 @@ class VerisureDataUpdateCoordinator(DataUpdateCoordinator):
             )
         except VerisureResponseError as ex:
             LOGGER.error("Could not read overview, %s", ex)
-            if ex.status_code == HTTP_SERVICE_UNAVAILABLE:  # Service unavailable
+            if ex.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
                 LOGGER.info("Trying to log in again")
                 await self.async_login()
                 return {}

homeassistant/components/viaggiatreno/sensor.py

@@ -1,5 +1,6 @@
 """Support for the Italian train system using ViaggiaTreno API."""
 import asyncio
+from http import HTTPStatus
 import logging
 import time
 
@@ -8,7 +9,7 @@ import async_timeout
 import voluptuous as vol
 
 from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
-from homeassistant.const import ATTR_ATTRIBUTION, HTTP_OK, TIME_MINUTES
+from homeassistant.const import ATTR_ATTRIBUTION, TIME_MINUTES
 import homeassistant.helpers.config_validation as cv
 
 _LOGGER = logging.getLogger(__name__)
@@ -71,7 +72,7 @@ async def async_http_request(hass, uri):
         session = hass.helpers.aiohttp_client.async_get_clientsession(hass)
         with async_timeout.timeout(REQUEST_TIMEOUT):
             req = await session.get(uri)
-        if req.status != HTTP_OK:
+        if req.status != HTTPStatus.OK:
             return {"error": req.status}
         json_response = await req.json()
         return json_response

homeassistant/components/voicerss/tts.py

@@ -1,5 +1,6 @@
 """Support for the voicerss speech 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
 
@@ -198,7 +199,7 @@ class VoiceRSSProvider(Provider):
             with async_timeout.timeout(10):
                 request = await websession.post(VOICERSS_API_URL, data=form_data)
 
-                if request.status != HTTP_OK:
+                if request.status != HTTPStatus.OK:
                     _LOGGER.error(
                         "Error %d on load url %s", request.status, request.url
                     )

tests/components/tado/test_config_flow.py

@@ -1,4 +1,5 @@
 """Test the Tado config flow."""
+from http import HTTPStatus
 from unittest.mock import MagicMock, patch
 
 import requests
@@ -59,7 +60,7 @@ async def test_form_invalid_auth(hass):
     )
 
     response_mock = MagicMock()
-    type(response_mock).status_code = 401
+    type(response_mock).status_code = HTTPStatus.UNAUTHORIZED
     mock_tado_api = _get_mock_tado_api(getMe=requests.HTTPError(response=response_mock))
 
     with patch(
@@ -82,7 +83,7 @@ async def test_form_cannot_connect(hass):
     )
 
     response_mock = MagicMock()
-    type(response_mock).status_code = 500
+    type(response_mock).status_code = HTTPStatus.INTERNAL_SERVER_ERROR
     mock_tado_api = _get_mock_tado_api(getMe=requests.HTTPError(response=response_mock))
 
     with patch(

tests/components/toon/test_config_flow.py

@@ -1,4 +1,5 @@
 """Tests for the Toon config flow."""
+from http import HTTPStatus
 from unittest.mock import patch
 
 from toonapi import Agreement, ToonError
@@ -77,7 +78,7 @@ async def test_full_flow_implementation(
 
     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/traccar/test_init.py

@@ -1,4 +1,5 @@
 """The tests the for Traccar device tracker platform."""
+from http import HTTPStatus
 from unittest.mock import patch
 
 import pytest
@@ -8,12 +9,7 @@ from homeassistant.components import traccar, zone
 from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN
 from homeassistant.components.traccar import DOMAIN, TRACKER_UPDATE
 from homeassistant.config import async_process_ha_core_config
-from homeassistant.const import (
+from homeassistant.const import STATE_HOME, STATE_NOT_HOME
-    HTTP_OK,
-    HTTP_UNPROCESSABLE_ENTITY,
-    STATE_HOME,
-    STATE_NOT_HOME,
-)
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.dispatcher import DATA_DISPATCHER
 from homeassistant.setup import async_setup_component
@@ -84,21 +80,21 @@ async def test_missing_data(hass, client, webhook_id):
     # No data
     req = await client.post(url)
     await hass.async_block_till_done()
-    assert req.status == HTTP_UNPROCESSABLE_ENTITY
+    assert req.status == HTTPStatus.UNPROCESSABLE_ENTITY
 
     # No latitude
     copy = data.copy()
     del copy["lat"]
     req = await client.post(url, params=copy)
     await hass.async_block_till_done()
-    assert req.status == HTTP_UNPROCESSABLE_ENTITY
+    assert req.status == HTTPStatus.UNPROCESSABLE_ENTITY
 
     # No device
     copy = data.copy()
     del copy["id"]
     req = await client.post(url, params=copy)
     await hass.async_block_till_done()
-    assert req.status == HTTP_UNPROCESSABLE_ENTITY
+    assert req.status == HTTPStatus.UNPROCESSABLE_ENTITY
 
 
 async def test_enter_and_exit(hass, client, webhook_id):
@@ -109,7 +105,7 @@ async def test_enter_and_exit(hass, client, webhook_id):
     # Enter the Home
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state_name = hass.states.get(
         "{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
     ).state
@@ -118,7 +114,7 @@ async def test_enter_and_exit(hass, client, webhook_id):
     # Enter Home again
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state_name = hass.states.get(
         "{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
     ).state
@@ -130,7 +126,7 @@ async def test_enter_and_exit(hass, client, webhook_id):
     # Enter Somewhere else
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state_name = hass.states.get(
         "{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
     ).state
@@ -160,7 +156,7 @@ async def test_enter_with_attrs(hass, client, webhook_id):
 
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"]))
     assert state.state == STATE_NOT_HOME
     assert state.attributes["gps_accuracy"] == 10.5
@@ -182,7 +178,7 @@ async def test_enter_with_attrs(hass, client, webhook_id):
 
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"]))
     assert state.state == STATE_HOME
     assert state.attributes["gps_accuracy"] == 123
@@ -201,7 +197,7 @@ async def test_two_devices(hass, client, webhook_id):
     # Exit Home
     req = await client.post(url, params=data_device_1)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
 
     state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data_device_1["id"]))
     assert state.state == "not_home"
@@ -213,7 +209,7 @@ async def test_two_devices(hass, client, webhook_id):
     data_device_2["id"] = "device_2"
     req = await client.post(url, params=data_device_2)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
 
     state = hass.states.get("{}.{}".format(DEVICE_TRACKER_DOMAIN, data_device_2["id"]))
     assert state.state == "home"
@@ -232,7 +228,7 @@ async def test_load_unload_entry(hass, client, webhook_id):
     # Enter the Home
     req = await client.post(url, params=data)
     await hass.async_block_till_done()
-    assert req.status == HTTP_OK
+    assert req.status == HTTPStatus.OK
     state_name = hass.states.get(
         "{}.{}".format(DEVICE_TRACKER_DOMAIN, data["id"])
     ).state

tests/components/tts/test_init.py

@@ -1,4 +1,5 @@
 """The tests for the TTS component."""
+from http import HTTPStatus
 from unittest.mock import PropertyMock, patch
 
 import pytest
@@ -15,7 +16,6 @@ from homeassistant.components.media_player.const import (
 import homeassistant.components.tts as tts
 from homeassistant.components.tts import _get_cache_files
 from homeassistant.config import async_process_ha_core_config
-from homeassistant.const import HTTP_NOT_FOUND
 from homeassistant.setup import async_setup_component
 
 from tests.common import assert_setup_component, async_mock_service
@@ -486,7 +486,7 @@ async def test_setup_component_and_test_service_with_receive_voice(
         "en",
         None,
     )
-    assert req.status == 200
+    assert req.status == HTTPStatus.OK
     assert await req.read() == demo_data
 
 
@@ -523,7 +523,7 @@ async def test_setup_component_and_test_service_with_receive_voice_german(
         "de",
         None,
     )
-    assert req.status == 200
+    assert req.status == HTTPStatus.OK
     assert await req.read() == demo_data
 
 
@@ -539,7 +539,7 @@ async def test_setup_component_and_web_view_wrong_file(hass, hass_client):
     url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
 
     req = await client.get(url)
-    assert req.status == HTTP_NOT_FOUND
+    assert req.status == HTTPStatus.NOT_FOUND
 
 
 async def test_setup_component_and_web_view_wrong_filename(hass, hass_client):
@@ -554,7 +554,7 @@ async def test_setup_component_and_web_view_wrong_filename(hass, hass_client):
     url = "/api/tts_proxy/265944dsk32c1b2a621be5930510bb2cd_en_-_demo.mp3"
 
     req = await client.get(url)
-    assert req.status == HTTP_NOT_FOUND
+    assert req.status == HTTPStatus.NOT_FOUND
 
 
 async def test_setup_component_test_without_cache(hass, empty_cache_dir):
@@ -682,7 +682,7 @@ async def test_setup_component_load_cache_retrieve_without_mem_cache(
     url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
 
     req = await client.get(url)
-    assert req.status == 200
+    assert req.status == HTTPStatus.OK
     assert await req.read() == demo_data
 
 
@@ -698,7 +698,7 @@ async def test_setup_component_and_web_get_url(hass, hass_client):
     data = {"platform": "demo", "message": "There is someone at the door."}
 
     req = await client.post(url, json=data)
-    assert req.status == 200
+    assert req.status == HTTPStatus.OK
     response = await req.json()
     assert response == {
         "url": "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
@@ -718,7 +718,7 @@ async def test_setup_component_and_web_get_url_bad_config(hass, hass_client):
     data = {"message": "There is someone at the door."}
 
     req = await client.post(url, json=data)
-    assert req.status == 400
+    assert req.status == HTTPStatus.BAD_REQUEST
 
 
 async def test_tags_with_wave(hass, demo_provider):

tests/components/unifi/test_controller.py

@@ -3,6 +3,7 @@
 import asyncio
 from copy import deepcopy
 from datetime import timedelta
+from http import HTTPStatus
 from unittest.mock import Mock, patch
 
 import aiounifi
@@ -404,7 +405,9 @@ async def test_reconnect_mechanism(hass, aioclient_mock, mock_unifi_websocket):
     await setup_unifi_integration(hass, aioclient_mock)
 
     aioclient_mock.clear_requests()
-    aioclient_mock.post(f"https://{DEFAULT_HOST}:1234/api/login", status=502)
+    aioclient_mock.post(
+        f"https://{DEFAULT_HOST}:1234/api/login", status=HTTPStatus.BAD_GATEWAY
+    )
 
     mock_unifi_websocket(state=STATE_DISCONNECTED)
     await hass.async_block_till_done()

tests/components/voicerss/test_tts.py

@@ -1,5 +1,6 @@
 """The tests for the VoiceRSS speech platform."""
 import asyncio
+from http import HTTPStatus
 import os
 import shutil
 
@@ -65,7 +66,9 @@ class TestTTSVoiceRSSPlatform:
         """Test service call say."""
         calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
 
-        aioclient_mock.post(self.url, data=self.form_data, status=200, content=b"test")
+        aioclient_mock.post(
+            self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test"
+        )
 
         config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
 
@@ -92,7 +95,9 @@ class TestTTSVoiceRSSPlatform:
         calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
 
         self.form_data["hl"] = "de-de"
-        aioclient_mock.post(self.url, data=self.form_data, status=200, content=b"test")
+        aioclient_mock.post(
+            self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test"
+        )
 
         config = {
             tts.DOMAIN: {
@@ -124,7 +129,9 @@ class TestTTSVoiceRSSPlatform:
         calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
 
         self.form_data["hl"] = "de-de"
-        aioclient_mock.post(self.url, data=self.form_data, status=200, content=b"test")
+        aioclient_mock.post(
+            self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test"
+        )
 
         config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
 
@@ -203,7 +210,7 @@ class TestTTSVoiceRSSPlatform:
         aioclient_mock.post(
             self.url,
             data=self.form_data,
-            status=200,
+            status=HTTPStatus.OK,
             content=b"The subscription does not support SSML!",
         )