cybercinch/directdnsonly
1
0
Fork 0
Code Issues 3 Pull Requests Packages Projects Releases Wiki Activity

style: apply black formatting across codebase 🎨

Browse Source 
No logic changes — pure reformatting of line lengths, dict literals,
method-chain line breaks, and trailing newlines to satisfy black's style.
This commit is contained in:
Aaron Guise
2026-02-18 22:53:09 +13:00
parent 807d6271f1
commit 74c5f4012e
12 changed files with 291 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/conftest.py
View File

@@ -1,10 +1,14 @@
"""Shared test fixtures for directdnsonly test suite."""
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from directdnsonly.app.db import Base
from directdnsonly.app.db.models import Domain, Key # noqa: F401 — registers models with Base
from directdnsonly.app.db.models import (
Domain,
Key,
) # noqa: F401 — registers models with Base
@pytest.fixture

91
tests/test_admin_api.py
View File

@@ -1,4 +1,5 @@
"""Tests for directdnsonly.app.api.admin — DNSAdminAPI handler methods."""
import pytest
from unittest.mock import MagicMock, patch
from urllib.parse import parse_qs
@@ -60,8 +61,12 @@ def test_handle_exists_unsupported_action_returns_error(api):
def test_handle_exists_domain_not_found(api):
with patch("directdnsonly.app.api.admin.check_zone_exists", return_value=False), \
patch("directdnsonly.app.api.admin.check_parent_domain_owner", return_value=False):
with (
patch("directdnsonly.app.api.admin.check_zone_exists", return_value=False),
patch(
"directdnsonly.app.api.admin.check_parent_domain_owner", return_value=False
),
):
result = api._handle_exists({"action": "exists", "domain": "example.com"})
parsed = parse_qs(result)
@@ -73,8 +78,10 @@ def test_handle_exists_domain_found(api):
record = MagicMock()
record.hostname = "da1.example.com"
with patch("directdnsonly.app.api.admin.check_zone_exists", return_value=True), \
patch("directdnsonly.app.api.admin.get_domain_record", return_value=record):
with (
patch("directdnsonly.app.api.admin.check_zone_exists", return_value=True),
patch("directdnsonly.app.api.admin.get_domain_record", return_value=record),
):
result = api._handle_exists({"action": "exists", "domain": "example.com"})
parsed = parse_qs(result)
@@ -87,14 +94,22 @@ def test_handle_exists_parent_found(api):
parent = MagicMock()
parent.hostname = "da2.example.com"
with patch("directdnsonly.app.api.admin.check_zone_exists", return_value=False), \
patch("directdnsonly.app.api.admin.check_parent_domain_owner", return_value=True), \
patch("directdnsonly.app.api.admin.get_parent_domain_record", return_value=parent):
result = api._handle_exists({
"action": "exists",
"domain": "sub.example.com",
"check_for_parent_domain": "1",
})
with (
patch("directdnsonly.app.api.admin.check_zone_exists", return_value=False),
patch(
"directdnsonly.app.api.admin.check_parent_domain_owner", return_value=True
),
patch(
"directdnsonly.app.api.admin.get_parent_domain_record", return_value=parent
),
):
result = api._handle_exists(
{
"action": "exists",
"domain": "sub.example.com",
"check_for_parent_domain": "1",
}
)
parsed = parse_qs(result)
assert parsed["error"] == ["0"]
@@ -107,9 +122,11 @@ def test_handle_exists_no_parent_check_when_flag_absent(api):
record = MagicMock()
record.hostname = "da1.example.com"
with patch("directdnsonly.app.api.admin.check_zone_exists", return_value=True), \
patch("directdnsonly.app.api.admin.check_parent_domain_owner") as mock_parent, \
patch("directdnsonly.app.api.admin.get_domain_record", return_value=record):
with (
patch("directdnsonly.app.api.admin.check_zone_exists", return_value=True),
patch("directdnsonly.app.api.admin.check_parent_domain_owner") as mock_parent,
patch("directdnsonly.app.api.admin.get_domain_record", return_value=record),
):
api._handle_exists({"action": "exists", "domain": "example.com"})
mock_parent.assert_not_called()
@@ -123,14 +140,21 @@ SAMPLE_ZONE = "$ORIGIN example.com.\n$TTL 300\nexample.com. 300 IN A 1.2.3.4\n"
def test_rawsave_enqueues_item(api, save_queue):
with patch("directdnsonly.app.api.admin.validate_and_normalize_zone",
return_value=SAMPLE_ZONE), \
patch.object(cherrypy, "request", MagicMock(remote=MagicMock(ip="127.0.0.1"))):
result = api._handle_rawsave("example.com", {
"zone_file": SAMPLE_ZONE,
"hostname": "da1.example.com",
"username": "admin",
})
with (
patch(
"directdnsonly.app.api.admin.validate_and_normalize_zone",
return_value=SAMPLE_ZONE,
),
patch.object(cherrypy, "request", MagicMock(remote=MagicMock(ip="127.0.0.1"))),
):
result = api._handle_rawsave(
"example.com",
{
"zone_file": SAMPLE_ZONE,
"hostname": "da1.example.com",
"username": "admin",
},
)
save_queue.put.assert_called_once()
item = save_queue.put.call_args[0][0]
@@ -150,9 +174,13 @@ def test_rawsave_missing_zone_file_raises(api):
def test_rawsave_invalid_zone_raises(api):
with patch("directdnsonly.app.api.admin.validate_and_normalize_zone",
side_effect=ValueError("Invalid zone data: bad record")), \
patch.object(cherrypy, "request", MagicMock(remote=MagicMock(ip="127.0.0.1"))):
with (
patch(
"directdnsonly.app.api.admin.validate_and_normalize_zone",
side_effect=ValueError("Invalid zone data: bad record"),
),
patch.object(cherrypy, "request", MagicMock(remote=MagicMock(ip="127.0.0.1"))),
):
with pytest.raises(ValueError, match="Invalid zone data"):
api._handle_rawsave("example.com", {"zone_file": "garbage"})
@@ -164,10 +192,13 @@ def test_rawsave_invalid_zone_raises(api):
def test_delete_enqueues_item(api, delete_queue):
with patch.object(cherrypy, "request", MagicMock(remote=MagicMock(ip="10.0.0.1"))):
result = api._handle_delete("example.com", {
"hostname": "da1.example.com",
"username": "admin",
})
result = api._handle_delete(
"example.com",
{
"hostname": "da1.example.com",
"username": "admin",
},
)
delete_queue.put.assert_called_once()
item = delete_queue.put.call_args[0][0]

13
tests/test_coredns_mysql.py
View File

@@ -1,4 +1,5 @@
"""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
@@ -141,8 +142,16 @@ 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()
session.add(Record(zone_id=zone.id, hostname="phantom", type="A",
data="10.0.0.99", ttl=300, online=True))
session.add(
Record(
zone_id=zone.id,
hostname="phantom",
type="A",
data="10.0.0.99",
ttl=300,
online=True,
)
)
session.commit()
session.close()

71
tests/test_reconciler.py
View File

@@ -1,4 +1,5 @@
"""Tests for directdnsonly.app.reconciler — ReconciliationWorker."""
import pytest
import requests.exceptions
from queue import Queue
@@ -12,7 +13,13 @@ from directdnsonly.app.db.models import Domain
# Fixtures
# ---------------------------------------------------------------------------
SERVER = {"hostname": "da1.example.com", "port": 2222, "username": "admin", "password": "secret", "ssl": True}
SERVER = {
"hostname": "da1.example.com",
"port": 2222,
"username": "admin",
"password": "secret",
"ssl": True,
}
BASE_CONFIG = {
"enabled": True,
@@ -46,7 +53,9 @@ def dry_run_worker(delete_queue):
def test_orphan_queued_when_domain_missing_from_da(worker, delete_queue, patch_connect):
patch_connect.add(Domain(domain="orphan.com", hostname="da1.example.com", username="admin"))
patch_connect.add(
Domain(domain="orphan.com", hostname="da1.example.com", username="admin")
)
patch_connect.commit()
with patch.object(worker, "_fetch_da_domains", return_value=set()):
@@ -59,7 +68,9 @@ def test_orphan_queued_when_domain_missing_from_da(worker, delete_queue, patch_c
def test_orphan_not_queued_in_dry_run(dry_run_worker, delete_queue, patch_connect):
patch_connect.add(Domain(domain="orphan.com", hostname="da1.example.com", username="admin"))
patch_connect.add(
Domain(domain="orphan.com", hostname="da1.example.com", username="admin")
)
patch_connect.commit()
with patch.object(dry_run_worker, "_fetch_da_domains", return_value=set()):
@@ -70,7 +81,9 @@ def test_orphan_not_queued_in_dry_run(dry_run_worker, delete_queue, patch_connec
def test_orphan_not_queued_for_unknown_server(worker, delete_queue, patch_connect):
"""Domains whose recorded master is NOT in our configured servers are skipped."""
patch_connect.add(Domain(domain="other.com", hostname="da99.unknown.com", username="admin"))
patch_connect.add(
Domain(domain="other.com", hostname="da99.unknown.com", username="admin")
)
patch_connect.commit()
with patch.object(worker, "_fetch_da_domains", return_value=set()):
@@ -80,7 +93,9 @@ def test_orphan_not_queued_for_unknown_server(worker, delete_queue, patch_connec
def test_active_domain_not_queued(worker, delete_queue, patch_connect):
patch_connect.add(Domain(domain="good.com", hostname="da1.example.com", username="admin"))
patch_connect.add(
Domain(domain="good.com", hostname="da1.example.com", username="admin")
)
patch_connect.commit()
with patch.object(worker, "_fetch_da_domains", return_value={"good.com"}):
@@ -106,7 +121,9 @@ def test_backfill_null_hostname(worker, patch_connect):
def test_migration_updates_hostname(worker, patch_connect):
patch_connect.add(Domain(domain="moved.com", hostname="da-old.example.com", username="admin"))
patch_connect.add(
Domain(domain="moved.com", hostname="da-old.example.com", username="admin")
)
patch_connect.commit()
with patch.object(worker, "_fetch_da_domains", return_value={"moved.com"}):
@@ -150,7 +167,9 @@ def test_fetch_returns_domains_from_json(worker):
mock_resp = _make_json_response(["example.com", "test.com"])
with patch("requests.get", return_value=mock_resp):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result == {"example.com", "test.com"}
@@ -160,7 +179,9 @@ def test_fetch_paginates(worker):
page2 = _make_json_response(["b.com"], total_pages=2)
with patch("requests.get", side_effect=[page1, page2]):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result == {"a.com", "b.com"}
@@ -170,9 +191,13 @@ def test_fetch_redirect_triggers_session_login(worker):
redirect_resp.status_code = 302
redirect_resp.is_redirect = True
with patch("requests.get", return_value=redirect_resp), \
patch.object(worker, "_da_session_login", return_value=None):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
with (
patch("requests.get", return_value=redirect_resp),
patch.object(worker, "_da_session_login", return_value=None),
):
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result is None
@@ -185,28 +210,40 @@ def test_fetch_html_response_returns_none(worker):
mock_resp.raise_for_status = MagicMock()
with patch("requests.get", return_value=mock_resp):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result is None
def test_fetch_connection_error_returns_none(worker):
with patch("requests.get", side_effect=requests.exceptions.ConnectionError("refused")):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
with patch(
"requests.get", side_effect=requests.exceptions.ConnectionError("refused")
):
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result is None
def test_fetch_timeout_returns_none(worker):
with patch("requests.get", side_effect=requests.exceptions.Timeout()):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result is None
def test_fetch_ssl_error_returns_none(worker):
with patch("requests.get", side_effect=requests.exceptions.SSLError("cert verify failed")):
result = worker._fetch_da_domains("da1.example.com", 2222, "admin", "secret", True)
with patch(
"requests.get", side_effect=requests.exceptions.SSLError("cert verify failed")
):
result = worker._fetch_da_domains(
"da1.example.com", 2222, "admin", "secret", True
)
assert result is None

1
tests/test_utils.py
View File

@@ -1,4 +1,5 @@
"""Tests for directdnsonly.app.utils — zone index helper functions."""
import pytest
from directdnsonly.app.db.models import Domain

1
tests/test_zone_parser.py
View File

@@ -1,4 +1,5 @@
"""Tests for directdnsonly.app.utils.zone_parser."""
import pytest
from dns.exception import DNSException