You've already forked directdnsonly
feat: add test suite, fix backend bugs, remove legacy artifacts 🧪
- Add 73-test suite across conftest, utils, admin API, reconciler, zone parser, and CoreDNS MySQL backend (all green, ~0.5s) - Fix zone_exists filter using wrong column name (name → zone_name) - Fix delete_zone missing dot_fqdn normalization on lookup - Remove spurious unused `from config import config` in coredns_mysql.py - Fix config loader to search module-relative path so tests find app.yml without needing a root-level config/ directory - Remove legacy v1 Flask prototype (app.py), empty config.json, and duplicate root config/app.yml
This commit is contained in:
@@ -1,47 +1,158 @@
|
||||
"""Tests for the CoreDNS MySQL backend (run against in-memory SQLite)."""
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import scoped_session, sessionmaker
|
||||
|
||||
from directdnsonly.app.backends.coredns_mysql import CoreDNSMySQLBackend, CoreDNSRecord
|
||||
from loguru import logger
|
||||
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(tmp_path):
|
||||
# Setup in-memory SQLite for testing (replace with test MySQL in CI)
|
||||
def mysql_backend():
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
CoreDNSRecord.metadata.create_all(engine)
|
||||
Base.metadata.create_all(engine)
|
||||
|
||||
class TestBackend(CoreDNSMySQLBackend):
|
||||
class _TestBackend(CoreDNSMySQLBackend):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
# Manually initialise without triggering the MySQL create_engine call
|
||||
self.config = {}
|
||||
self.instance_name = "test"
|
||||
self.engine = engine
|
||||
self.Session = scoped_session(sessionmaker(bind=engine))
|
||||
|
||||
yield TestBackend()
|
||||
yield _TestBackend()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_zone_operations(mysql_backend):
|
||||
zone_data = """
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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
|
||||
"""
|
||||
# Test zone creation
|
||||
assert mysql_backend.write_zone("example.com", zone_data)
|
||||
|
||||
|
||||
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")
|
||||
|
||||
# Test record update
|
||||
updated_zone = """
|
||||
|
||||
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_zone)
|
||||
assert mysql_backend.write_zone("example.com", updated)
|
||||
|
||||
# Test record removal
|
||||
reduced_zone = "example.com. 300 IN A 192.0.2.1"
|
||||
assert mysql_backend.write_zone("example.com", reduced_zone)
|
||||
|
||||
# Test zone deletion
|
||||
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.query(Zone).filter_by(zone_name="example.com.").first()
|
||||
records = session.query(Record).filter_by(zone_id=zone.id, type="AAAA").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")
|
||||
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.query(Zone).filter_by(zone_name="example.com.").first()
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user