Avoid use of datetime.utc*
methods deprecated in Python 3.12 (#93684)
Co-authored-by: Marc Mueller <30130371+cdce8p@users.noreply.github.com>
This commit is contained in:
parent
61c17291fb
commit
9da192c752
18 changed files with 57 additions and 54 deletions
|
@ -17,7 +17,7 @@ import homeassistant.helpers.config_validation as cv
|
||||||
from homeassistant.helpers.event import async_track_time_interval
|
from homeassistant.helpers.event import async_track_time_interval
|
||||||
from homeassistant.helpers.storage import Store
|
from homeassistant.helpers.storage import Store
|
||||||
from homeassistant.helpers.typing import ConfigType
|
from homeassistant.helpers.typing import ConfigType
|
||||||
from homeassistant.util.dt import utc_from_timestamp
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
_LOGGER = getLogger(__name__)
|
_LOGGER = getLogger(__name__)
|
||||||
|
|
||||||
|
@ -207,7 +207,7 @@ class FeedManager:
|
||||||
self._firstrun = False
|
self._firstrun = False
|
||||||
else:
|
else:
|
||||||
# Set last entry timestamp as epoch time if not available
|
# Set last entry timestamp as epoch time if not available
|
||||||
self._last_entry_timestamp = datetime.utcfromtimestamp(0).timetuple()
|
self._last_entry_timestamp = dt_util.utc_from_timestamp(0).timetuple()
|
||||||
for entry in self._feed.entries:
|
for entry in self._feed.entries:
|
||||||
if (
|
if (
|
||||||
self._firstrun
|
self._firstrun
|
||||||
|
@ -286,6 +286,6 @@ class StoredData:
|
||||||
def _async_save_data(self) -> dict[str, str]:
|
def _async_save_data(self) -> dict[str, str]:
|
||||||
"""Save feed data to storage."""
|
"""Save feed data to storage."""
|
||||||
return {
|
return {
|
||||||
feed_id: utc_from_timestamp(timegm(struct_utc)).isoformat()
|
feed_id: dt_util.utc_from_timestamp(timegm(struct_utc)).isoformat()
|
||||||
for feed_id, struct_utc in self._data.items()
|
for feed_id, struct_utc in self._data.items()
|
||||||
}
|
}
|
||||||
|
|
|
@ -102,9 +102,7 @@ class FileSizeCoordinator(DataUpdateCoordinator):
|
||||||
raise UpdateFailed(f"Can not retrieve file statistics {error}") from error
|
raise UpdateFailed(f"Can not retrieve file statistics {error}") from error
|
||||||
|
|
||||||
size = statinfo.st_size
|
size = statinfo.st_size
|
||||||
last_updated = datetime.utcfromtimestamp(statinfo.st_mtime).replace(
|
last_updated = dt_util.utc_from_timestamp(statinfo.st_mtime)
|
||||||
tzinfo=dt_util.UTC
|
|
||||||
)
|
|
||||||
|
|
||||||
_LOGGER.debug("size %s, last updated %s", size, last_updated)
|
_LOGGER.debug("size %s, last updated %s", size, last_updated)
|
||||||
data: dict[str, int | float | datetime] = {
|
data: dict[str, int | float | datetime] = {
|
||||||
|
|
|
@ -40,7 +40,7 @@ from homeassistant.helpers.network import NoURLAvailableError, get_url
|
||||||
from homeassistant.helpers.typing import ConfigType
|
from homeassistant.helpers.typing import ConfigType
|
||||||
from homeassistant.loader import bind_hass
|
from homeassistant.loader import bind_hass
|
||||||
from homeassistant.setup import async_start_setup, async_when_setup_or_start
|
from homeassistant.setup import async_start_setup, async_when_setup_or_start
|
||||||
from homeassistant.util import ssl as ssl_util
|
from homeassistant.util import dt as dt_util, ssl as ssl_util
|
||||||
from homeassistant.util.json import json_loads
|
from homeassistant.util.json import json_loads
|
||||||
|
|
||||||
from .auth import async_setup_auth
|
from .auth import async_setup_auth
|
||||||
|
@ -503,14 +503,15 @@ class HomeAssistantHTTP:
|
||||||
x509.NameAttribute(NameOID.COMMON_NAME, host),
|
x509.NameAttribute(NameOID.COMMON_NAME, host),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
now = dt_util.utcnow()
|
||||||
cert = (
|
cert = (
|
||||||
x509.CertificateBuilder()
|
x509.CertificateBuilder()
|
||||||
.subject_name(subject)
|
.subject_name(subject)
|
||||||
.issuer_name(issuer)
|
.issuer_name(issuer)
|
||||||
.public_key(key.public_key())
|
.public_key(key.public_key())
|
||||||
.serial_number(x509.random_serial_number())
|
.serial_number(x509.random_serial_number())
|
||||||
.not_valid_before(datetime.datetime.utcnow())
|
.not_valid_before(now)
|
||||||
.not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=30))
|
.not_valid_after(now + datetime.timedelta(days=30))
|
||||||
.add_extension(
|
.add_extension(
|
||||||
x509.SubjectAlternativeName([x509.DNSName(host)]),
|
x509.SubjectAlternativeName([x509.DNSName(host)]),
|
||||||
critical=False,
|
critical=False,
|
||||||
|
|
|
@ -19,7 +19,7 @@ async def async_get_config_entry_diagnostics(
|
||||||
payload: dict[str, Any] = {
|
payload: dict[str, Any] = {
|
||||||
"now": dt_util.now().isoformat(),
|
"now": dt_util.now().isoformat(),
|
||||||
"timezone": str(dt_util.DEFAULT_TIME_ZONE),
|
"timezone": str(dt_util.DEFAULT_TIME_ZONE),
|
||||||
"system_timezone": str(datetime.datetime.utcnow().astimezone().tzinfo),
|
"system_timezone": str(datetime.datetime.now().astimezone().tzinfo),
|
||||||
}
|
}
|
||||||
store = hass.data[DOMAIN][config_entry.entry_id]
|
store = hass.data[DOMAIN][config_entry.entry_id]
|
||||||
ics = await store.async_load()
|
ics = await store.async_load()
|
||||||
|
|
|
@ -2,7 +2,7 @@
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from datetime import datetime, timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
@ -19,6 +19,7 @@ from homeassistant.helpers.update_coordinator import (
|
||||||
DataUpdateCoordinator,
|
DataUpdateCoordinator,
|
||||||
UpdateFailed,
|
UpdateFailed,
|
||||||
)
|
)
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from .const import CONF_ACCOUNT, DATA_CLIENT, DATA_COORDINATOR, DOMAIN
|
from .const import CONF_ACCOUNT, DATA_CLIENT, DATA_COORDINATOR, DOMAIN
|
||||||
|
|
||||||
|
@ -58,7 +59,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
raise UpdateFailed(exception) from exception
|
raise UpdateFailed(exception) from exception
|
||||||
if not authenticated:
|
if not authenticated:
|
||||||
raise ConfigEntryAuthFailed("Not authenticated with OVO Energy")
|
raise ConfigEntryAuthFailed("Not authenticated with OVO Energy")
|
||||||
return await client.get_daily_usage(datetime.utcnow().strftime("%Y-%m"))
|
return await client.get_daily_usage(dt_util.utcnow().strftime("%Y-%m"))
|
||||||
|
|
||||||
coordinator = DataUpdateCoordinator[OVODailyUsage](
|
coordinator = DataUpdateCoordinator[OVODailyUsage](
|
||||||
hass,
|
hass,
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
"""Support for monitoring Repetier Server Sensors."""
|
"""Support for monitoring Repetier Server Sensors."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import datetime
|
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
|
|
||||||
|
@ -10,6 +9,7 @@ from homeassistant.core import HomeAssistant, callback
|
||||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.typing import UNDEFINED, ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import UNDEFINED, ConfigType, DiscoveryInfoType
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from . import REPETIER_API, SENSOR_TYPES, UPDATE_SIGNAL, RepetierSensorEntityDescription
|
from . import REPETIER_API, SENSOR_TYPES, UPDATE_SIGNAL, RepetierSensorEntityDescription
|
||||||
|
|
||||||
|
@ -170,7 +170,7 @@ class RepetierJobEndSensor(RepetierSensor):
|
||||||
print_time = data["print_time"]
|
print_time = data["print_time"]
|
||||||
from_start = data["from_start"]
|
from_start = data["from_start"]
|
||||||
time_end = start + round(print_time, 0)
|
time_end = start + round(print_time, 0)
|
||||||
self._state = datetime.utcfromtimestamp(time_end)
|
self._state = dt_util.utc_from_timestamp(time_end)
|
||||||
remaining = print_time - from_start
|
remaining = print_time - from_start
|
||||||
remaining_secs = int(round(remaining, 0))
|
remaining_secs = int(round(remaining, 0))
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
|
@ -192,7 +192,7 @@ class RepetierJobStartSensor(RepetierSensor):
|
||||||
job_name = data["job_name"]
|
job_name = data["job_name"]
|
||||||
start = data["start"]
|
start = data["start"]
|
||||||
from_start = data["from_start"]
|
from_start = data["from_start"]
|
||||||
self._state = datetime.utcfromtimestamp(start)
|
self._state = dt_util.utc_from_timestamp(start)
|
||||||
elapsed_secs = int(round(from_start, 0))
|
elapsed_secs = int(round(from_start, 0))
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Job %s elapsed %s",
|
"Job %s elapsed %s",
|
||||||
|
|
|
@ -11,6 +11,7 @@ from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.device_registry import DeviceInfo
|
from homeassistant.helpers.device_registry import DeviceInfo
|
||||||
from homeassistant.helpers.event import async_track_time_interval
|
from homeassistant.helpers.event import async_track_time_interval
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
_LOGGER,
|
_LOGGER,
|
||||||
|
@ -141,7 +142,9 @@ class StarlineAccount:
|
||||||
def gps_attrs(device: StarlineDevice) -> dict[str, Any]:
|
def gps_attrs(device: StarlineDevice) -> dict[str, Any]:
|
||||||
"""Attributes for device tracker."""
|
"""Attributes for device tracker."""
|
||||||
return {
|
return {
|
||||||
"updated": datetime.utcfromtimestamp(device.position["ts"]).isoformat(),
|
"updated": dt_util.utc_from_timestamp(device.position["ts"])
|
||||||
|
.replace(tzinfo=None)
|
||||||
|
.isoformat(),
|
||||||
"online": device.online,
|
"online": device.online,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -14,6 +14,7 @@ import attr
|
||||||
import av
|
import av
|
||||||
|
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from . import redact_credentials
|
from . import redact_credentials
|
||||||
from .const import (
|
from .const import (
|
||||||
|
@ -140,7 +141,7 @@ class StreamMuxer:
|
||||||
self._part_has_keyframe = False
|
self._part_has_keyframe = False
|
||||||
self._stream_settings = stream_settings
|
self._stream_settings = stream_settings
|
||||||
self._stream_state = stream_state
|
self._stream_state = stream_state
|
||||||
self._start_time = datetime.datetime.utcnow()
|
self._start_time = dt_util.utcnow()
|
||||||
|
|
||||||
def make_new_av(
|
def make_new_av(
|
||||||
self,
|
self,
|
||||||
|
|
|
@ -2,7 +2,7 @@
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from datetime import datetime, timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from pytraccar import (
|
from pytraccar import (
|
||||||
|
@ -44,7 +44,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.event import async_track_time_interval
|
from homeassistant.helpers.event import async_track_time_interval
|
||||||
from homeassistant.helpers.restore_state import RestoreEntity
|
from homeassistant.helpers.restore_state import RestoreEntity
|
||||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||||
from homeassistant.util import slugify
|
from homeassistant.util import dt as dt_util, slugify
|
||||||
|
|
||||||
from . import DOMAIN, TRACKER_UPDATE
|
from . import DOMAIN, TRACKER_UPDATE
|
||||||
from .const import (
|
from .const import (
|
||||||
|
@ -334,7 +334,8 @@ class TraccarScanner:
|
||||||
|
|
||||||
async def import_events(self):
|
async def import_events(self):
|
||||||
"""Import events from Traccar."""
|
"""Import events from Traccar."""
|
||||||
start_intervel = datetime.utcnow()
|
# get_reports_events requires naive UTC datetimes as of 1.0.0
|
||||||
|
start_intervel = dt_util.utcnow().replace(tzinfo=None)
|
||||||
events = await self._api.get_reports_events(
|
events = await self._api.get_reports_events(
|
||||||
devices=[device.id for device in self._devices],
|
devices=[device.id for device in self._devices],
|
||||||
start_time=start_intervel,
|
start_time=start_intervel,
|
||||||
|
|
|
@ -1,13 +1,13 @@
|
||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""Helper script to bump the current version."""
|
"""Helper script to bump the current version."""
|
||||||
import argparse
|
import argparse
|
||||||
from datetime import datetime
|
|
||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
from packaging.version import Version
|
from packaging.version import Version
|
||||||
|
|
||||||
from homeassistant import const
|
from homeassistant import const
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
|
|
||||||
def _bump_release(release, bump_type):
|
def _bump_release(release, bump_type):
|
||||||
|
@ -86,10 +86,7 @@ def bump_version(version, bump_type):
|
||||||
if not version.is_devrelease:
|
if not version.is_devrelease:
|
||||||
raise ValueError("Can only be run on dev release")
|
raise ValueError("Can only be run on dev release")
|
||||||
|
|
||||||
to_change["dev"] = (
|
to_change["dev"] = ("dev", dt_util.utcnow().strftime("%Y%m%d"))
|
||||||
"dev",
|
|
||||||
datetime.utcnow().date().isoformat().replace("-", ""),
|
|
||||||
)
|
|
||||||
|
|
||||||
else:
|
else:
|
||||||
assert False, f"Unsupported type: {bump_type}"
|
assert False, f"Unsupported type: {bump_type}"
|
||||||
|
@ -206,7 +203,7 @@ def test_bump_version():
|
||||||
assert bump_version(Version("0.56.0.dev0"), "minor") == Version("0.56.0")
|
assert bump_version(Version("0.56.0.dev0"), "minor") == Version("0.56.0")
|
||||||
assert bump_version(Version("0.56.2.dev0"), "minor") == Version("0.57.0")
|
assert bump_version(Version("0.56.2.dev0"), "minor") == Version("0.57.0")
|
||||||
|
|
||||||
today = datetime.utcnow().date().isoformat().replace("-", "")
|
today = dt_util.utcnow().strftime("%Y%m%d")
|
||||||
assert bump_version(Version("0.56.0.dev0"), "nightly") == Version(
|
assert bump_version(Version("0.56.0.dev0"), "nightly") == Version(
|
||||||
f"0.56.0.dev{today}"
|
f"0.56.0.dev{today}"
|
||||||
)
|
)
|
||||||
|
|
|
@ -20,6 +20,7 @@ from homeassistant.components.application_credentials import (
|
||||||
from homeassistant.components.google import DOMAIN
|
from homeassistant.components.google import DOMAIN
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||||
|
@ -138,9 +139,7 @@ def token_scopes() -> list[str]:
|
||||||
def token_expiry() -> datetime.datetime:
|
def token_expiry() -> datetime.datetime:
|
||||||
"""Expiration time for credentials used in the test."""
|
"""Expiration time for credentials used in the test."""
|
||||||
# OAuth library returns an offset-naive timestamp
|
# OAuth library returns an offset-naive timestamp
|
||||||
return datetime.datetime.fromtimestamp(
|
return dt_util.utcnow().replace(tzinfo=None) + datetime.timedelta(hours=1)
|
||||||
datetime.datetime.utcnow().timestamp()
|
|
||||||
) + datetime.timedelta(hours=1)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
|
|
@ -92,7 +92,7 @@ async def test_new_token(hass: HomeAssistant) -> None:
|
||||||
assert len(entries) == 1
|
assert len(entries) == 1
|
||||||
assert entries[0].state == ConfigEntryState.LOADED
|
assert entries[0].state == ConfigEntryState.LOADED
|
||||||
|
|
||||||
one_hour_after = datetime.utcnow() + timedelta(hours=1)
|
one_hour_after = datetime.now() + timedelta(hours=1)
|
||||||
|
|
||||||
with patch("lacrosse_view.LaCrosse.login", return_value=True) as login, patch(
|
with patch("lacrosse_view.LaCrosse.login", return_value=True) as login, patch(
|
||||||
"lacrosse_view.LaCrosse.get_sensors",
|
"lacrosse_view.LaCrosse.get_sensors",
|
||||||
|
@ -122,7 +122,7 @@ async def test_failed_token(hass: HomeAssistant) -> None:
|
||||||
assert len(entries) == 1
|
assert len(entries) == 1
|
||||||
assert entries[0].state == ConfigEntryState.LOADED
|
assert entries[0].state == ConfigEntryState.LOADED
|
||||||
|
|
||||||
one_hour_after = datetime.utcnow() + timedelta(hours=1)
|
one_hour_after = datetime.now() + timedelta(hours=1)
|
||||||
|
|
||||||
with patch(
|
with patch(
|
||||||
"lacrosse_view.LaCrosse.login", side_effect=LoginError("Test")
|
"lacrosse_view.LaCrosse.login", side_effect=LoginError("Test")
|
||||||
|
|
|
@ -4,7 +4,6 @@ This file contains the original models definitions before schema tracking was
|
||||||
implemented. It is used to test the schema migration logic.
|
implemented. It is used to test the schema migration logic.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from datetime import datetime
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
@ -40,7 +39,7 @@ class Events(Base): # type: ignore
|
||||||
event_data = Column(Text)
|
event_data = Column(Text)
|
||||||
origin = Column(String(32))
|
origin = Column(String(32))
|
||||||
time_fired = Column(DateTime(timezone=True))
|
time_fired = Column(DateTime(timezone=True))
|
||||||
created = Column(DateTime(timezone=True), default=datetime.utcnow)
|
created = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def from_event(event):
|
def from_event(event):
|
||||||
|
@ -77,9 +76,9 @@ class States(Base): # type: ignore
|
||||||
state = Column(String(255))
|
state = Column(String(255))
|
||||||
attributes = Column(Text)
|
attributes = Column(Text)
|
||||||
event_id = Column(Integer, ForeignKey("events.event_id"))
|
event_id = Column(Integer, ForeignKey("events.event_id"))
|
||||||
last_changed = Column(DateTime(timezone=True), default=datetime.utcnow)
|
last_changed = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
last_updated = Column(DateTime(timezone=True), default=datetime.utcnow)
|
last_updated = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
created = Column(DateTime(timezone=True), default=datetime.utcnow)
|
created = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
|
|
||||||
__table_args__ = (
|
__table_args__ = (
|
||||||
Index("states__state_changes", "last_changed", "last_updated", "entity_id"),
|
Index("states__state_changes", "last_changed", "last_updated", "entity_id"),
|
||||||
|
@ -131,10 +130,10 @@ class RecorderRuns(Base): # type: ignore
|
||||||
|
|
||||||
__tablename__ = "recorder_runs"
|
__tablename__ = "recorder_runs"
|
||||||
run_id = Column(Integer, primary_key=True)
|
run_id = Column(Integer, primary_key=True)
|
||||||
start = Column(DateTime(timezone=True), default=datetime.utcnow)
|
start = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
end = Column(DateTime(timezone=True))
|
end = Column(DateTime(timezone=True))
|
||||||
closed_incorrect = Column(Boolean, default=False)
|
closed_incorrect = Column(Boolean, default=False)
|
||||||
created = Column(DateTime(timezone=True), default=datetime.utcnow)
|
created = Column(DateTime(timezone=True), default=dt_util.utcnow)
|
||||||
|
|
||||||
def entity_ids(self, point_in_time=None):
|
def entity_ids(self, point_in_time=None):
|
||||||
"""Return the entity ids that existed in this run.
|
"""Return the entity ids that existed in this run.
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
"""Test Scrape component setup process."""
|
"""Test Scrape component setup process."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import datetime
|
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
@ -11,6 +10,7 @@ from homeassistant.components.scrape.const import DEFAULT_SCAN_INTERVAL, DOMAIN
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers import entity_registry as er
|
from homeassistant.helpers import entity_registry as er
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from . import MockRestData, return_integration_config
|
from . import MockRestData, return_integration_config
|
||||||
|
|
||||||
|
@ -67,7 +67,7 @@ async def test_setup_no_data_fails_with_recovery(
|
||||||
assert "Platform scrape not ready yet" in caplog.text
|
assert "Platform scrape not ready yet" in caplog.text
|
||||||
|
|
||||||
mocker.payload = "test_scrape_sensor"
|
mocker.payload = "test_scrape_sensor"
|
||||||
async_fire_time_changed(hass, datetime.utcnow() + DEFAULT_SCAN_INTERVAL)
|
async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
state = hass.states.get("sensor.ha_version")
|
state = hass.states.get("sensor.ha_version")
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
"""The tests for the Scrape sensor platform."""
|
"""The tests for the Scrape sensor platform."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import datetime, timedelta
|
from datetime import timedelta
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
@ -247,7 +247,7 @@ async def test_scrape_sensor_no_data_refresh(hass: HomeAssistant) -> None:
|
||||||
assert state.state == "Current Version: 2021.12.10"
|
assert state.state == "Current Version: 2021.12.10"
|
||||||
|
|
||||||
mocker.payload = "test_scrape_sensor_no_data"
|
mocker.payload = "test_scrape_sensor_no_data"
|
||||||
async_fire_time_changed(hass, datetime.utcnow() + DEFAULT_SCAN_INTERVAL)
|
async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
state = hass.states.get("sensor.ha_version")
|
state = hass.states.get("sensor.ha_version")
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
"""Collection of test helpers."""
|
"""Collection of test helpers."""
|
||||||
from datetime import datetime
|
|
||||||
from fractions import Fraction
|
from fractions import Fraction
|
||||||
import functools
|
import functools
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
@ -15,8 +14,9 @@ from homeassistant.components.stream.fmp4utils import (
|
||||||
XYW_ROW,
|
XYW_ROW,
|
||||||
find_box,
|
find_box,
|
||||||
)
|
)
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
FAKE_TIME = datetime.utcnow()
|
FAKE_TIME = dt_util.utcnow()
|
||||||
|
|
||||||
# Segment with defaults filled in for use in tests
|
# Segment with defaults filled in for use in tests
|
||||||
DefaultSegment = partial(
|
DefaultSegment = partial(
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
"""The tests for the Traccar device tracker platform."""
|
"""The tests for the Traccar device tracker platform."""
|
||||||
from datetime import datetime
|
|
||||||
from unittest.mock import AsyncMock, patch
|
from unittest.mock import AsyncMock, patch
|
||||||
|
|
||||||
from pytraccar import ReportsEventeModel
|
from pytraccar import ReportsEventeModel
|
||||||
|
@ -17,6 +16,7 @@ from homeassistant.const import (
|
||||||
)
|
)
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from tests.common import async_capture_events
|
from tests.common import async_capture_events
|
||||||
|
|
||||||
|
@ -47,7 +47,7 @@ async def test_import_events_catch_all(hass: HomeAssistant) -> None:
|
||||||
"maintenanceId": 1,
|
"maintenanceId": 1,
|
||||||
"deviceId": device["id"],
|
"deviceId": device["id"],
|
||||||
"type": "ignitionOn",
|
"type": "ignitionOn",
|
||||||
"eventTime": datetime.utcnow().isoformat(),
|
"eventTime": dt_util.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||||
"attributes": {},
|
"attributes": {},
|
||||||
}
|
}
|
||||||
),
|
),
|
||||||
|
@ -59,7 +59,7 @@ async def test_import_events_catch_all(hass: HomeAssistant) -> None:
|
||||||
"maintenanceId": 1,
|
"maintenanceId": 1,
|
||||||
"deviceId": device["id"],
|
"deviceId": device["id"],
|
||||||
"type": "ignitionOff",
|
"type": "ignitionOff",
|
||||||
"eventTime": datetime.utcnow().isoformat(),
|
"eventTime": dt_util.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||||
"attributes": {},
|
"attributes": {},
|
||||||
}
|
}
|
||||||
),
|
),
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
"""Test Home Assistant date util methods."""
|
"""Test Home Assistant date util methods."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import datetime, timedelta
|
from datetime import UTC, datetime, timedelta
|
||||||
import time
|
import time
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
@ -41,9 +41,9 @@ def test_set_default_time_zone() -> None:
|
||||||
|
|
||||||
def test_utcnow() -> None:
|
def test_utcnow() -> None:
|
||||||
"""Test the UTC now method."""
|
"""Test the UTC now method."""
|
||||||
assert abs(dt_util.utcnow().replace(tzinfo=None) - datetime.utcnow()) < timedelta(
|
assert abs(
|
||||||
seconds=1
|
dt_util.utcnow().replace(tzinfo=None) - datetime.now(UTC).replace(tzinfo=None)
|
||||||
)
|
) < timedelta(seconds=1)
|
||||||
|
|
||||||
|
|
||||||
def test_now() -> None:
|
def test_now() -> None:
|
||||||
|
@ -51,13 +51,14 @@ def test_now() -> None:
|
||||||
dt_util.set_default_time_zone(dt_util.get_time_zone(TEST_TIME_ZONE))
|
dt_util.set_default_time_zone(dt_util.get_time_zone(TEST_TIME_ZONE))
|
||||||
|
|
||||||
assert abs(
|
assert abs(
|
||||||
dt_util.as_utc(dt_util.now()).replace(tzinfo=None) - datetime.utcnow()
|
dt_util.as_utc(dt_util.now()).replace(tzinfo=None)
|
||||||
|
- datetime.now(UTC).replace(tzinfo=None)
|
||||||
) < timedelta(seconds=1)
|
) < timedelta(seconds=1)
|
||||||
|
|
||||||
|
|
||||||
def test_as_utc_with_naive_object() -> None:
|
def test_as_utc_with_naive_object() -> None:
|
||||||
"""Test the now method."""
|
"""Test the now method."""
|
||||||
utcnow = datetime.utcnow()
|
utcnow = datetime.now(UTC).replace(tzinfo=None)
|
||||||
|
|
||||||
assert utcnow == dt_util.as_utc(utcnow).replace(tzinfo=None)
|
assert utcnow == dt_util.as_utc(utcnow).replace(tzinfo=None)
|
||||||
|
|
||||||
|
@ -82,7 +83,9 @@ def test_as_utc_with_local_object() -> None:
|
||||||
def test_as_local_with_naive_object() -> None:
|
def test_as_local_with_naive_object() -> None:
|
||||||
"""Test local time with native object."""
|
"""Test local time with native object."""
|
||||||
now = dt_util.now()
|
now = dt_util.now()
|
||||||
assert abs(now - dt_util.as_local(datetime.utcnow())) < timedelta(seconds=1)
|
assert abs(
|
||||||
|
now - dt_util.as_local(datetime.now(UTC).replace(tzinfo=None))
|
||||||
|
) < timedelta(seconds=1)
|
||||||
|
|
||||||
|
|
||||||
def test_as_local_with_local_object() -> None:
|
def test_as_local_with_local_object() -> None:
|
||||||
|
|
Loading…
Add table
Reference in a new issue