karldns.de/tests/test_updates.py

import logging
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

import karldns.services.ddns as ddns_module
from karldns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, make_app


def _update_path(account: dict[str, str]) -> str:
    return f"/api/v1/router/update/{account['update_slug']}"


def test_router_update_get_accepts_fritzbox_placeholders(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.ddns.example.net", "router")
    app = make_app(
        DdnsSettings(database_path=database_path, admin_password="admin", hostname_suffix="ddns.example.net")
    )
    client = TestClient(app)

    response = client.get(
        _update_path(account),
        params={
            "myip": "203.0.113.8",
            "myipv6": "2001:db8::8",
        },
    )

    assert response.status_code == 200
    assert response.text == "good 203.0.113.8"
    records = client.get("/api/v1/admin/records", auth=("admin", "admin")).json()
    assert records["total"] == 1
    assert records["items"][0]["hostname"] == "home.ddns.example.net"
    assert records["items"][0]["ipv4"] == "203.0.113.8"
    assert records["items"][0]["ipv6"] == "2001:db8::8"


def test_router_update_post_accepts_form_and_json(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    form_response = client.post(_update_path(account), data={"myip": "203.0.113.9"})
    json_response = client.post(_update_path(account), json={"myip": "203.0.113.10"})

    assert form_response.status_code == 200
    assert form_response.text == "good 203.0.113.9"
    assert json_response.status_code == 200
    assert json_response.text == "good 203.0.113.10"
    assert store.list_records()[0]["ipv4"] == "203.0.113.10"


def test_router_post_json_parser_keeps_only_string_values() -> None:
    from karldns.routers.updates import _json_post_values

    values = _json_post_values(b'{"myip":"203.0.113.10","count":1,"empty":"","nested":{"x":1}}')

    assert values == {"myip": "203.0.113.10", "empty": ""}


def test_router_post_form_parser_handles_invalid_utf8_as_empty() -> None:
    from karldns.routers.updates import _form_post_values

    assert _form_post_values(b"\xff") == {}


def test_router_ip_params_merge_body_without_overriding_query_values() -> None:
    from karldns.routers.updates import _IpParams

    params = _IpParams(myip="203.0.113.8", ipaddr=None, myipv6="", ip6addr=None)

    merged = params.with_body_values(
        {
            "myip": "203.0.113.9",
            "ipaddr": "198.51.100.9",
            "myipv6": "2001:db8::9",
        }
    )

    assert merged == _IpParams(
        myip="203.0.113.8",
        ipaddr="198.51.100.9",
        myipv6="2001:db8::9",
        ip6addr=None,
    )


def test_router_ip_params_from_query_builds_domain_params() -> None:
    from karldns.routers.updates import _ip_params_from_query
    from karldns.services.ddns import UpdateIpParams

    assert _ip_params_from_query(myip="203.0.113.8", ipaddr=None, myipv6=None, ip6addr="2001:db8::8") == (
        UpdateIpParams(myip="203.0.113.8", ipaddr=None, myipv6=None, ip6addr="2001:db8::8")
    )


def test_registration_hostname_rejects_disabled_or_unmanaged_accounts() -> None:
    from fastapi import HTTPException

    from karldns.routers.updates import _registration_hostname

    class Service:
        def hostname_is_managed(self, hostname: str) -> bool:
            return hostname.endswith(".managed.example")

    assert _registration_hostname({"hostname": "home.managed.example", "disabled": False}, Service()) == (
        "home.managed.example"
    )
    with pytest.raises(HTTPException):
        _registration_hostname({"hostname": "home.managed.example", "disabled": True}, Service())
    with pytest.raises(HTTPException):
        _registration_hostname({"hostname": "home.other.example", "disabled": False}, Service())


def test_router_update_requires_registered_hostname(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice", accepted=False)
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    before = client.get(_update_path(account), params={"myip": "203.0.113.9"})
    registered = client.get(f"/api/v1/router/register/{account['management_slug']}", params={"myip": "203.0.113.9"})
    after = client.get(_update_path(account), params={"myip": "203.0.113.10"})

    assert before.status_code == 404
    assert registered.status_code == 200
    assert registered.text == "good registered"
    assert after.status_code == 200
    assert after.text == "good 203.0.113.10"


def test_dns_publish_failure_does_not_update_record(tmp_path: Path, monkeypatch, caplog) -> None:
    def fail_publish(result, settings):
        raise DnsProviderError("dns down")

    caplog.set_level(logging.WARNING, logger="karldns.services.ddns")
    monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 500
    assert store.list_records() == []
    events = store.list_update_events()
    assert events[0]["status"] == "dns_failed"
    assert "DNS publish failed for hostname=home.example.net" in caplog.text
    assert account["update_slug"] not in caplog.text
    assert "dns down" not in caplog.text


def test_dns_reconciliation_failure_is_logged_separately(tmp_path: Path, monkeypatch, caplog) -> None:
    def fail_publish(result, settings):
        raise DnsReconciliationError("provider state mismatch")

    caplog.set_level(logging.WARNING, logger="karldns.services.ddns")
    monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 500
    assert store.list_records() == []
    assert store.list_update_events()[0]["status"] == "dns_reconcile_failed"
    assert store.get_service_metrics()["dns_reconcile_failures_total"] == 1
    assert "DNS reconciliation failed for hostname=home.example.net" in caplog.text


def test_programming_error_during_publish_is_not_mapped_to_dns_failure(tmp_path: Path, monkeypatch) -> None:
    def fail_publish(result, settings):
        raise AssertionError("publish bug")

    monkeypatch.setattr(ddns_module, "publish_dns", fail_publish)
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    with pytest.raises(AssertionError, match="publish bug"):
        client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert store.list_records() == []
    assert store.list_update_events() == []


def test_dns_publish_success_updates_record(tmp_path: Path, monkeypatch) -> None:
    published = []
    monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 200
    assert response.text == "good 203.0.113.9"
    assert len(published) == 1
    assert store.list_records()[0]["ipv4"] == "203.0.113.9"
    metrics = store.get_service_metrics()
    assert metrics["update_requests_total"] == 1
    assert metrics["update_success_total"] == 1
    assert metrics["update_latency_samples_total"] == 1


def test_unchanged_slug_update_is_read_only_for_history_and_record(tmp_path: Path, monkeypatch) -> None:
    published = []
    monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    store.upsert("home.example.net", "203.0.113.9", None)
    before = store.list_records()[0]["updated_at"]
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})
    second_response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 200
    assert response.text == "nochg 203.0.113.9"
    assert second_response.status_code == 200
    assert store.list_records()[0]["updated_at"] == before
    events = store.list_update_events()
    assert events[0]["status"] == "unchanged"
    assert events[0]["count"] == 2
    assert published == []


def test_slug_update_rejects_hostname_outside_provider_suffix(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "router")
    app = make_app(
        DdnsSettings(database_path=database_path, admin_password="admin", hostname_suffix="ddns.example.net")
    )
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 404
    assert store.list_records() == []


def test_omitted_ip_family_is_preserved(tmp_path: Path, monkeypatch) -> None:
    published = []
    monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    store.upsert("home.example.net", "203.0.113.8", "2001:db8::8")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "203.0.113.9"})

    assert response.status_code == 200
    record = store.list_records()[0]
    assert record["ipv4"] == "203.0.113.9"
    assert record["ipv6"] == "2001:db8::8"
    assert published[0].update_ipv4 is True
    assert published[0].update_ipv6 is False


def test_blank_ip_params_fall_back_to_source_ip(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app, client=("198.51.100.44", 50000))

    response = client.get(
        _update_path(account),
        params={"myip": "", "myipv6": ""},
    )

    assert response.status_code == 200
    assert response.text == "good 198.51.100.44"
    assert store.list_records()[0]["ipv4"] == "198.51.100.44"


def test_invalid_ip_is_badip_and_does_not_change_record(tmp_path: Path, monkeypatch) -> None:
    published = []
    monkeypatch.setattr(ddns_module, "publish_dns", lambda result, settings: published.append(result))
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    store.upsert("home.example.net", "203.0.113.8", "2001:db8::8")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "not-an-ip"})

    assert response.status_code == 400
    assert response.text == "badip"
    assert published == []
    record = store.list_records()[0]
    assert record["ipv4"] == "203.0.113.8"
    assert record["ipv6"] == "2001:db8::8"


def test_wrong_ip_family_is_badip_and_preserves_records(tmp_path: Path) -> None:
    database_path = tmp_path / "ddns.sqlite3"
    store = DdnsStore(database_path)
    account = store.create_account("home.example.net", "alice")
    store.upsert("home.example.net", "203.0.113.8", "2001:db8::8")
    app = make_app(DdnsSettings(database_path=database_path, admin_password="admin"))
    client = TestClient(app)

    response = client.get(_update_path(account), params={"myip": "2001:db8::9"})

    assert response.status_code == 400
    assert response.text == "badip"
    record = store.list_records()[0]
    assert record["ipv4"] == "203.0.113.8"
    assert record["ipv6"] == "2001:db8::8"