chore: complete SQLAlchemy 2.0 migration in coredns_mysql backend and tests ⬆️

Migrate remaining session.query() calls in coredns_mysql.py to
select()/session.execute() style; update bulk delete to delete()
construct and count to func.count(); drop sessionmaker(bind=).
Update test fixtures and assertions to match.

Zero session.query() calls remaining across the entire codebase.
This commit is contained in:
2026-02-19 23:43:54 +13:00
parent d81ecd6bdd
commit f9907d2859
3 changed files with 41 additions and 38 deletions

View File

@@ -1,7 +1,7 @@
"""Tests for the CoreDNS MySQL backend (run against in-memory SQLite)."""
import pytest
from sqlalchemy import create_engine
from sqlalchemy import create_engine, select
from sqlalchemy.orm import scoped_session, sessionmaker
from directdnsonly.app.backends.coredns_mysql import (
@@ -28,7 +28,7 @@ def mysql_backend():
self.config = {}
self.instance_name = "test"
self.engine = engine
self.Session = scoped_session(sessionmaker(bind=engine))
self.Session = scoped_session(sessionmaker(engine))
yield _TestBackend()
engine.dispose()
@@ -84,8 +84,8 @@ def test_write_zone_removes_stale_records(mysql_backend):
mysql_backend.write_zone("example.com", reduced)
session = mysql_backend.Session()
zone = session.query(Zone).filter_by(zone_name="example.com.").first()
records = session.query(Record).filter_by(zone_id=zone.id, type="AAAA").all()
zone = session.execute(select(Zone).filter_by(zone_name="example.com.")).scalar_one_or_none()
records = session.execute(select(Record).filter_by(zone_id=zone.id, type="AAAA")).scalars().all()
assert records == []
session.close()
@@ -141,7 +141,7 @@ def test_reconcile_removes_extra_records(mysql_backend):
# Inject a phantom record directly into the DB
session = mysql_backend.Session()
zone = session.query(Zone).filter_by(zone_name="example.com.").first()
zone = session.execute(select(Zone).filter_by(zone_name="example.com.")).scalar_one_or_none()
session.add(
Record(
zone_id=zone.id,