Fix SQLAlchemy test warnings (#54116)

This commit is contained in:
Paulus Schoutsen 2021-08-08 20:33:47 -07:00 committed by GitHub
parent 5d56ce67f5
commit 557cc792e9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 13 additions and 6 deletions

View File

@ -393,7 +393,7 @@ class Filters:
if includes and not excludes:
return or_(*includes)
if not excludes and includes:
if not includes and excludes:
return not_(or_(*excludes))
return or_(*includes) & not_(or_(*excludes))

View File

@ -20,8 +20,7 @@ from sqlalchemy import (
distinct,
)
from sqlalchemy.dialects import mysql
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.orm.session import Session
from homeassistant.const import (

View File

@ -10,6 +10,7 @@ import os
import time
from typing import TYPE_CHECKING, Callable
from sqlalchemy import text
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy.orm.session import Session
@ -332,4 +333,5 @@ def perodic_db_cleanups(instance: Recorder):
if instance.engine.dialect.name == "sqlite":
# Execute sqlite to create a wal checkpoint and free up disk space
_LOGGER.debug("WAL checkpoint")
instance.engine.execute("PRAGMA wal_checkpoint(TRUNCATE);")
with instance.engine.connect() as connection:
connection.execute(text("PRAGMA wal_checkpoint(TRUNCATE);"))

View File

@ -6,6 +6,7 @@ from unittest.mock import MagicMock, patch
import pytest
from sqlalchemy import text
from sqlalchemy.sql.elements import TextClause
from homeassistant.components.recorder import run_information_with_session, util
from homeassistant.components.recorder.const import DATA_INSTANCE, SQLITE_URL_PREFIX
@ -253,6 +254,11 @@ def test_end_incomplete_runs(hass_recorder, caplog):
def test_perodic_db_cleanups(hass_recorder):
"""Test perodic db cleanups."""
hass = hass_recorder()
with patch.object(hass.data[DATA_INSTANCE].engine, "execute") as execute_mock:
with patch.object(hass.data[DATA_INSTANCE].engine, "connect") as connect_mock:
util.perodic_db_cleanups(hass.data[DATA_INSTANCE])
assert execute_mock.call_args[0][0] == "PRAGMA wal_checkpoint(TRUNCATE);"
text_obj = connect_mock.return_value.__enter__.return_value.execute.mock_calls[0][
1
][0]
assert isinstance(text_obj, TextClause)
assert str(text_obj) == "PRAGMA wal_checkpoint(TRUNCATE);"