443 lines
17 KiB
Python
443 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from ddns.unifi_update import (
|
|
_get_policy_key,
|
|
_get_policy_map,
|
|
_get_session,
|
|
list_dns_policies,
|
|
update_records,
|
|
)
|
|
|
|
|
|
class TestGetPolicyKey:
|
|
def test_a_record(self) -> None:
|
|
policy = {"domain": "test.example.com", "type": "A_RECORD"}
|
|
assert _get_policy_key(policy) == "test.example.com:A_RECORD"
|
|
|
|
def test_aaaa_record(self) -> None:
|
|
policy = {"domain": "test.example.com", "type": "AAAA_RECORD"}
|
|
assert _get_policy_key(policy) == "test.example.com:AAAA_RECORD"
|
|
|
|
def test_empty_domain(self) -> None:
|
|
policy = {"type": "A_RECORD"}
|
|
assert _get_policy_key(policy) == ":A_RECORD"
|
|
|
|
def test_empty_type(self) -> None:
|
|
policy = {"domain": "test.example.com"}
|
|
assert _get_policy_key(policy) == "test.example.com:"
|
|
|
|
|
|
class TestGetPolicyMap:
|
|
def test_empty_policies(self) -> None:
|
|
assert _get_policy_map([]) == {}
|
|
|
|
def test_multiple_policies(self) -> None:
|
|
policies = [
|
|
{"domain": "a.example.com", "type": "A_RECORD", "id": "1"},
|
|
{"domain": "a.example.com", "type": "AAAA_RECORD", "id": "2"},
|
|
{"domain": "b.example.com", "type": "A_RECORD", "id": "3"},
|
|
]
|
|
policy_map = _get_policy_map(policies)
|
|
assert len(policy_map) == 3
|
|
assert policy_map["a.example.com:A_RECORD"]["id"] == "1"
|
|
assert policy_map["a.example.com:AAAA_RECORD"]["id"] == "2"
|
|
assert policy_map["b.example.com:A_RECORD"]["id"] == "3"
|
|
|
|
def test_duplicate_keys_overwrite(self) -> None:
|
|
policies = [
|
|
{"domain": "test.com", "type": "A_RECORD", "id": "1"},
|
|
{"domain": "test.com", "type": "A_RECORD", "id": "2"},
|
|
]
|
|
policy_map = _get_policy_map(policies)
|
|
assert len(policy_map) == 1
|
|
assert policy_map["test.com:A_RECORD"]["id"] == "2"
|
|
|
|
|
|
class TestListDnsPolicies:
|
|
def test_list_dns_policies_success(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"data": [
|
|
{
|
|
"id": "policy-1",
|
|
"type": "A_RECORD",
|
|
"domain": "test.example.com",
|
|
"ipv4Address": "1.2.3.4",
|
|
"ttlSeconds": 14400,
|
|
"enabled": True,
|
|
}
|
|
]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.verify = False
|
|
|
|
policies = list_dns_policies(mock_session, "https://unifi.example.com/api", "site123")
|
|
|
|
assert len(policies) == 1
|
|
assert policies[0]["id"] == "policy-1"
|
|
assert policies[0]["domain"] == "test.example.com"
|
|
assert policies[0]["ipv4Address"] == "1.2.3.4"
|
|
|
|
mock_session.get.assert_called_once_with(
|
|
"https://unifi.example.com/api/sites/site123/dns/policies",
|
|
verify=False,
|
|
)
|
|
|
|
def test_list_dns_policies_empty(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
|
|
policies = list_dns_policies(mock_session, "https://unifi.example.com/api", "site123")
|
|
assert policies == []
|
|
|
|
def test_list_dns_policies_raises_on_error(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = Exception("HTTP error")
|
|
mock_session.get.return_value = mock_response
|
|
|
|
with pytest.raises(Exception, match="HTTP error"):
|
|
list_dns_policies(mock_session, "https://unifi.example.com/api", "site123")
|
|
|
|
|
|
class TestGetSession:
|
|
def test_get_session_sets_csrf_token_header(self) -> None:
|
|
with patch("ddns.unifi_update.requests.Session") as mock_session_cls:
|
|
mock_session = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
session = _get_session("https://unifi.example.com", "my-api-token", False)
|
|
|
|
mock_session_cls.assert_called_once()
|
|
mock_session.headers.update.assert_called_once_with(
|
|
{"X-API-Key": "my-api-token"}
|
|
)
|
|
assert session is mock_session
|
|
|
|
def test_get_session_verify_ssl(self) -> None:
|
|
with patch("ddns.unifi_update.requests.Session") as mock_session_cls:
|
|
mock_session = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
_get_session("https://unifi.example.com", "my-api-token", True)
|
|
mock_session.verify = True
|
|
|
|
def test_get_session_default_verify_false(self) -> None:
|
|
with patch("ddns.unifi_update.requests.Session") as mock_session_cls:
|
|
mock_session = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
_get_session("https://unifi.example.com", "my-api-token")
|
|
mock_session.verify = False
|
|
|
|
|
|
class TestUpdateRecords:
|
|
def test_update_records_with_ipv4(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
mock_session.post.assert_called_once()
|
|
call_args = mock_session.post.call_args
|
|
assert "dns/policies" in call_args[0][0]
|
|
payload = call_args[1]["json"]
|
|
assert payload["type"] == "A_RECORD"
|
|
assert payload["domain"] == "test.example.com"
|
|
assert payload["ipv4Address"] == "1.2.3.4"
|
|
assert payload["ttlSeconds"] == 7200
|
|
|
|
def test_update_records_with_ipv6(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4=None, ipv6="2001:db8::1") # type: ignore[arg-type]
|
|
|
|
mock_session.post.assert_called_once()
|
|
call_args = mock_session.post.call_args
|
|
payload = call_args[1]["json"]
|
|
assert payload["type"] == "AAAA_RECORD"
|
|
assert payload["ipv6Address"] == "2001:db8::1"
|
|
|
|
def test_update_records_updates_existing_policy(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"data": [
|
|
{
|
|
"id": "existing-policy-id",
|
|
"type": "A_RECORD",
|
|
"domain": "test.example.com",
|
|
"ipv4Address": "5.6.7.8",
|
|
"ttlSeconds": 14400,
|
|
"enabled": True,
|
|
}
|
|
]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.put.return_value = MagicMock()
|
|
mock_session.put.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
mock_session.put.assert_called_once()
|
|
call_args = mock_session.put.call_args
|
|
assert "existing-policy-id" in call_args[0][0]
|
|
payload = call_args[1]["json"]
|
|
assert payload["ipv4Address"] == "1.2.3.4"
|
|
assert payload["ttlSeconds"] == 7200
|
|
|
|
def test_update_records_multiple_records(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [
|
|
{"record": "a.example.com", "ttl_seconds": 7200},
|
|
{"record": "b.example.com", "ttl_seconds": 3600},
|
|
],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
assert mock_session.post.call_count == 2
|
|
|
|
def test_update_records_default_ttl(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com"}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
call_args = mock_session.post.call_args
|
|
payload = call_args[1]["json"]
|
|
assert payload["ttlSeconds"] == 14400
|
|
|
|
def test_update_records_handles_api_base_with_trailing_slash(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com/",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
mock_session.get.assert_called_once()
|
|
call_url = mock_session.get.call_args[0][0]
|
|
assert "https://unifi.example.com/proxy/network/integration/v1/sites/site123/dns/policies" == call_url
|
|
|
|
def test_update_records_raises_on_post_failure(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status.side_effect = Exception("Create failed")
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
with pytest.raises(Exception, match="Create failed"):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
def test_update_records_raises_on_put_failure(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"data": [
|
|
{
|
|
"id": "policy-1",
|
|
"type": "A_RECORD",
|
|
"domain": "test.example.com",
|
|
"ipv4Address": "5.6.7.8",
|
|
"ttlSeconds": 14400,
|
|
"enabled": True,
|
|
}
|
|
]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.put.return_value = MagicMock()
|
|
mock_session.put.return_value.raise_for_status.side_effect = Exception("Update failed")
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
with pytest.raises(Exception, match="Update failed"):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6=None) # type: ignore[arg-type]
|
|
|
|
def test_update_records_skip_ipv4(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv4": True}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1") # type: ignore[arg-type]
|
|
|
|
mock_session.post.assert_called_once()
|
|
payload = mock_session.post.call_args[1]["json"]
|
|
assert payload["type"] == "AAAA_RECORD"
|
|
assert payload["ipv6Address"] == "2001:db8::1"
|
|
|
|
def test_update_records_skip_ipv6(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.post.return_value = MagicMock()
|
|
mock_session.post.return_value.raise_for_status = MagicMock()
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv6": True}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1") # type: ignore[arg-type]
|
|
|
|
mock_session.post.assert_called_once()
|
|
payload = mock_session.post.call_args[1]["json"]
|
|
assert payload["type"] == "A_RECORD"
|
|
assert payload["ipv4Address"] == "1.2.3.4"
|
|
|
|
def test_update_records_skip_both(self) -> None:
|
|
mock_session = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"data": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session.verify = False
|
|
|
|
unifi_config = {
|
|
"host": "https://unifi.example.com",
|
|
"site_id": "site123",
|
|
"api_token": "my-token",
|
|
"verify_ssl": False,
|
|
"records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv4": True, "skip_ipv6": True}],
|
|
}
|
|
|
|
with patch("ddns.unifi_update._get_session", return_value=mock_session):
|
|
update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1") # type: ignore[arg-type]
|
|
|
|
mock_session.post.assert_not_called()
|