karl/karldns.de
Archived
1
0 Fork 0
Code Issues Packages Projects Releases Wiki Activity
This repository has been archived on 2026-10-28. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
ca92dc211aeca22cc7e6781d8444fd1e89f8f034
karldns.de/tests/test_updates.py
T

322 lines
12 KiB
Python
Raw Blame History

import logging
from pathlib import Path
from urllib.parse import parse_qs, urlparse
import pytest
from fastapi.testclient import TestClient
import router_dyndns.ddns_service as ddns_service
from router_dyndns.ddns import DdnsSettings, DdnsStore, DnsProviderError, make_app
def test_update_accepts_fritzbox_placeholders(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
shared_secret="secret",
admin_password="admin",
allowed_hosts={"home.example.net"},
)
)
client = TestClient(app)
response = client.get(
"/nic/update",
params={
"hostname": "home.example.net",
"myip": "203.0.113.8",
"myipv6": "2001:db8::8",
"password": "secret",
},
)
assert response.status_code == 200
assert response.text == "good 203.0.113.8"
records = client.get("/records", auth=("admin", "admin")).json()
assert records["total"] == 1
assert records["items"][0]["hostname"] == "home.example.net"
assert records["items"][0]["ipv4"] == "203.0.113.8"
assert records["items"][0]["ipv6"] == "2001:db8::8"
def test_update_rejects_wrong_secret(tmp_path: Path) -> None:
app = make_app(DdnsSettings(database_path=tmp_path / "ddns.sqlite3", shared_secret="secret"))
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": "home.example.net", "myip": "203.0.113.8", "password": "wrong"},
)
assert response.status_code == 401
def test_update_rejects_unlisted_host(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
shared_secret="secret",
allowed_hosts={"home.example.net"},
)
)
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": "other.example.net", "myip": "203.0.113.8", "password": "secret"},
)
assert response.status_code == 403
def test_generated_account_can_update_own_hostname(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)
response = client.get(
"/nic/update",
params={
"hostname": "home.example.net",
"myip": "203.0.113.9",
"username": account["username"],
"password": account["password"],
},
)
assert response.status_code == 200
assert response.text == "good 203.0.113.9"
slug_response = client.get(
f"/u/{account['update_slug']}",
params={"myip": "203.0.113.10"},
)
assert slug_response.status_code == 200
assert slug_response.text == "good 203.0.113.10"
def test_one_box_update_url_can_authenticate_with_query_credentials(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
account = store.create_account("home.example.net", "alice")
settings = DdnsSettings(
database_path=database_path,
admin_password="admin",
public_base_url="http://ddns.example.net",
)
app = make_app(settings)
client = TestClient(app)
one_box_url = ddns_service.DdnsService(settings, store).credentialed_update_url(account)
parsed = urlparse(one_box_url)
params = parse_qs(parsed.query)
params["myip"] = ["203.0.113.11"]
params["myipv6"] = [""]
response = client.get(parsed.path, params={key: value[0] for key, value in params.items()})
assert response.status_code == 200
assert response.text == "good 203.0.113.11"
assert response.headers["cache-control"] == "no-store"
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="router_dyndns.ddns_service")
monkeypatch.setattr(ddns_service, "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(f"/u/{account['update_slug']}", 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_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_service, "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(f"/u/{account['update_slug']}", 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_service, "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(f"/u/{account['update_slug']}", 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"
def test_unchanged_slug_update_is_read_only_for_history_and_record(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "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(f"/u/{account['update_slug']}", params={"myip": "203.0.113.9"})
second_response = client.get(f"/u/{account['update_slug']}", 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_shared_secret_still_updates_explicitly_allowed_hostname(tmp_path: Path) -> None:
app = make_app(
DdnsSettings(
database_path=tmp_path / "ddns.sqlite3",
shared_secret="secret",
admin_password="admin",
allowed_hosts={"home.example.net"},
)
)
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": "home.example.net", "myip": "203.0.113.9", "password": "secret"},
)
assert response.status_code == 200
assert response.text == "good 203.0.113.9"
def test_omitted_ip_family_is_preserved(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "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(f"/u/{account['update_slug']}", 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(
f"/u/{account['update_slug']}",
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_legacy_ip_is_badip_and_does_not_change_record(tmp_path: Path, monkeypatch) -> None:
published = []
monkeypatch.setattr(ddns_service, "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(f"/u/{account['update_slug']}", 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_family_legacy_ip_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(f"/u/{account['update_slug']}", 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"
def test_nic_update_invalid_ip_returns_badip(tmp_path: Path) -> None:
database_path = tmp_path / "ddns.sqlite3"
store = DdnsStore(database_path)
app = make_app(
DdnsSettings(
database_path=database_path,
shared_secret="secret",
admin_password="admin",
allowed_hosts={"home.example.net"},
)
)
client = TestClient(app)
response = client.get(
"/nic/update",
params={"hostname": "home.example.net", "myipv6": "203.0.113.9", "password": "secret"},
)
assert response.status_code == 400
assert response.text == "badip"
assert store.list_records() == []
Reference in new issue View Git Blame Copy Permalink