directdnsonly/tests/test_coredns_mysql.py

"""Tests for the CoreDNS MySQL backend (run against in-memory SQLite)."""

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import scoped_session, sessionmaker

from directdnsonly.app.backends.coredns_mysql import (
    Base,
    CoreDNSMySQLBackend,
    Record,
    Zone,
)


# ---------------------------------------------------------------------------
# Fixture — in-memory SQLite backend (bypasses real MySQL connection)
# ---------------------------------------------------------------------------


@pytest.fixture
def mysql_backend():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)

    class _TestBackend(CoreDNSMySQLBackend):
        def __init__(self):
            # Manually initialise without triggering the MySQL create_engine call
            self.config = {}
            self.instance_name = "test"
            self.engine = engine
            self.Session = scoped_session(sessionmaker(engine))

    yield _TestBackend()
    engine.dispose()


# ---------------------------------------------------------------------------
# write_zone / zone_exists / delete_zone
# ---------------------------------------------------------------------------


ZONE_DATA = """\
$ORIGIN example.com.
$TTL 300
example.com. 300 IN SOA ns.example.com. admin.example.com. (2023 3600 1800 604800 86400)
example.com. 300 IN A 192.0.2.1
"""


def test_write_zone_creates_zone(mysql_backend):
    assert mysql_backend.write_zone("example.com", ZONE_DATA)


def test_zone_exists_after_write(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    assert mysql_backend.zone_exists("example.com")


def test_zone_does_not_exist_before_write(mysql_backend):
    assert not mysql_backend.zone_exists("missing.com")


def test_write_zone_idempotent(mysql_backend):
    assert mysql_backend.write_zone("example.com", ZONE_DATA)
    assert mysql_backend.write_zone("example.com", ZONE_DATA)


def test_write_zone_updates_records(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)

    updated = """\
$ORIGIN example.com.
$TTL 300
example.com. 3600 IN A 192.0.2.1
example.com. 300 IN AAAA 2001:db8::1
"""
    assert mysql_backend.write_zone("example.com", updated)


def test_write_zone_removes_stale_records(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)

    reduced = "example.com. 300 IN A 192.0.2.1"
    mysql_backend.write_zone("example.com", reduced)

    session = mysql_backend.Session()
    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()


def test_delete_zone_removes_zone_and_records(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    assert mysql_backend.delete_zone("example.com")
    assert not mysql_backend.zone_exists("example.com")


def test_delete_nonexistent_zone_returns_false(mysql_backend):
    assert not mysql_backend.delete_zone("ghost.com")


def test_reload_zone_returns_true(mysql_backend):
    assert mysql_backend.reload_zone("example.com")
    assert mysql_backend.reload_zone()


# ---------------------------------------------------------------------------
# verify_zone_record_count
# ---------------------------------------------------------------------------


def test_verify_zone_record_count_match(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    # SOA + A = 2 records total
    matches, count = mysql_backend.verify_zone_record_count("example.com", 2)
    assert matches
    assert count == 2


def test_verify_zone_record_count_mismatch(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    matches, count = mysql_backend.verify_zone_record_count("example.com", 99)
    assert not matches
    assert count == 2


def test_verify_zone_record_count_missing_zone(mysql_backend):
    matches, count = mysql_backend.verify_zone_record_count("ghost.com", 0)
    assert not matches
    assert count == 0


# ---------------------------------------------------------------------------
# reconcile_zone_records
# ---------------------------------------------------------------------------


def test_reconcile_removes_extra_records(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)

    # Inject a phantom record directly into the DB
    session = mysql_backend.Session()
    zone = session.execute(select(Zone).filter_by(zone_name="example.com.")).scalar_one_or_none()
    session.add(
        Record(
            zone_id=zone.id,
            hostname="phantom",
            type="A",
            data="10.0.0.99",
            ttl=300,
            online=True,
        )
    )
    session.commit()
    session.close()

    success, removed = mysql_backend.reconcile_zone_records("example.com", ZONE_DATA)
    assert success
    assert removed == 1


def test_reconcile_no_changes_when_zone_matches(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    success, removed = mysql_backend.reconcile_zone_records("example.com", ZONE_DATA)
    assert success
    assert removed == 0


# ---------------------------------------------------------------------------
# managed_by field
# ---------------------------------------------------------------------------


def test_write_zone_sets_managed_by_directadmin(mysql_backend):
    mysql_backend.write_zone("example.com", ZONE_DATA)
    session = mysql_backend.Session()
    zone = session.execute(
        select(Zone).filter_by(zone_name="example.com.")
    ).scalar_one_or_none()
    assert zone.managed_by == "directadmin"
    session.close()


def test_write_zone_migrates_null_managed_by(mysql_backend):
    """Zones that pre-exist without managed_by get it set on next write."""
    session = mysql_backend.Session()
    zone = Zone(zone_name="example.com.", managed_by=None)
    session.add(zone)
    session.commit()
    session.close()

    mysql_backend.write_zone("example.com", ZONE_DATA)

    session = mysql_backend.Session()
    zone = session.execute(
        select(Zone).filter_by(zone_name="example.com.")
    ).scalar_one_or_none()
    assert zone.managed_by == "directadmin"
    session.close()