511 lines
20 KiB
Python
511 lines
20 KiB
Python
"""Tests for DDNSManager and DDNS provider classes."""
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch, call
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))
|
|
|
|
from ddns_manager import (
|
|
DDNSManager,
|
|
DDNSProvider,
|
|
DDNSError,
|
|
PicNgoDDNS,
|
|
CloudflareDDNS,
|
|
DuckDNSDDNS,
|
|
NoIPDDNS,
|
|
FreeDNSDDNS,
|
|
_get_public_ip,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_response(status_code=200, json_data=None, text=''):
|
|
"""Build a minimal requests.Response-like mock."""
|
|
resp = MagicMock()
|
|
resp.ok = (200 <= status_code < 300)
|
|
resp.status_code = status_code
|
|
resp.json.return_value = json_data or {}
|
|
resp.text = text
|
|
return resp
|
|
|
|
|
|
def _make_config_manager(ddns_cfg=None, domain_cfg=None):
|
|
"""Return a mock config_manager whose get_identity() returns a useful dict."""
|
|
cm = MagicMock()
|
|
if ddns_cfg is not None:
|
|
identity = {'domain': {'ddns': ddns_cfg}}
|
|
elif domain_cfg is not None:
|
|
identity = {'domain': domain_cfg}
|
|
else:
|
|
identity = {}
|
|
cm.get_identity.return_value = identity
|
|
return cm
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PicNgoDDNS tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPicNgoDDNSRegister(unittest.TestCase):
|
|
"""PicNgoDDNS.register() calls the correct URL with the correct body."""
|
|
|
|
def test_register_calls_correct_url(self):
|
|
provider = PicNgoDDNS(api_base_url='https://ddns.example.com')
|
|
mock_resp = _make_response(200, json_data={'token': 'tok123', 'subdomain': 'alpha'})
|
|
with patch('requests.post', return_value=mock_resp) as mock_post:
|
|
result = provider.register('alpha', '1.2.3.4')
|
|
mock_post.assert_called_once()
|
|
args, kwargs = mock_post.call_args
|
|
self.assertEqual(args[0], 'https://ddns.example.com/api/v1/register')
|
|
self.assertEqual(kwargs['json'], {'name': 'alpha', 'ip': '1.2.3.4'})
|
|
self.assertEqual(result, {'token': 'tok123', 'subdomain': 'alpha'})
|
|
|
|
def test_register_raises_ddns_error_on_http_error(self):
|
|
provider = PicNgoDDNS()
|
|
mock_resp = _make_response(500, text='Internal Server Error')
|
|
with patch('requests.post', return_value=mock_resp):
|
|
with self.assertRaises(DDNSError):
|
|
provider.register('alpha', '1.2.3.4')
|
|
|
|
def test_register_no_token_in_header(self):
|
|
"""register() must NOT send an Authorization header (no token yet)."""
|
|
provider = PicNgoDDNS()
|
|
mock_resp = _make_response(200, json_data={'token': 't', 'subdomain': 'x'})
|
|
with patch('requests.post', return_value=mock_resp) as mock_post:
|
|
provider.register('x', '1.2.3.4')
|
|
_, kwargs = mock_post.call_args
|
|
self.assertNotIn('Authorization', kwargs.get('headers', {}))
|
|
|
|
|
|
class TestPicNgoDDNSUpdate(unittest.TestCase):
|
|
"""PicNgoDDNS.update() calls the correct URL with Authorization header."""
|
|
|
|
def test_update_uses_bearer_token(self):
|
|
provider = PicNgoDDNS(api_base_url='https://ddns.example.com')
|
|
mock_resp = _make_response(200)
|
|
with patch('requests.put', return_value=mock_resp) as mock_put:
|
|
result = provider.update('mytoken', '5.6.7.8')
|
|
mock_put.assert_called_once()
|
|
args, kwargs = mock_put.call_args
|
|
self.assertEqual(args[0], 'https://ddns.example.com/api/v1/update')
|
|
self.assertIn('Authorization', kwargs['headers'])
|
|
self.assertEqual(kwargs['headers']['Authorization'], 'Bearer mytoken')
|
|
self.assertEqual(kwargs['json'], {'ip': '5.6.7.8'})
|
|
self.assertTrue(result)
|
|
|
|
def test_update_raises_ddns_error_on_failure(self):
|
|
provider = PicNgoDDNS()
|
|
mock_resp = _make_response(403, text='Forbidden')
|
|
with patch('requests.put', return_value=mock_resp):
|
|
with self.assertRaises(DDNSError):
|
|
provider.update('badtoken', '1.2.3.4')
|
|
|
|
|
|
class TestPicNgoDDNSChallenges(unittest.TestCase):
|
|
"""PicNgoDDNS.dns_challenge_create/delete call correct endpoints."""
|
|
|
|
def test_dns_challenge_create_calls_post(self):
|
|
provider = PicNgoDDNS(api_base_url='https://ddns.example.com')
|
|
mock_resp = _make_response(200)
|
|
with patch('requests.post', return_value=mock_resp) as mock_post:
|
|
result = provider.dns_challenge_create('tok', '_acme.alpha.pic.ngo', 'abc123')
|
|
mock_post.assert_called_once()
|
|
args, kwargs = mock_post.call_args
|
|
self.assertEqual(args[0], 'https://ddns.example.com/api/v1/dns-challenge')
|
|
self.assertEqual(kwargs['json'], {'fqdn': '_acme.alpha.pic.ngo', 'value': 'abc123'})
|
|
self.assertEqual(kwargs['headers']['Authorization'], 'Bearer tok')
|
|
self.assertTrue(result)
|
|
|
|
def test_dns_challenge_delete_calls_delete(self):
|
|
provider = PicNgoDDNS(api_base_url='https://ddns.example.com')
|
|
mock_resp = _make_response(200)
|
|
with patch('requests.delete', return_value=mock_resp) as mock_del:
|
|
result = provider.dns_challenge_delete('tok', '_acme.alpha.pic.ngo')
|
|
mock_del.assert_called_once()
|
|
args, kwargs = mock_del.call_args
|
|
self.assertEqual(args[0], 'https://ddns.example.com/api/v1/dns-challenge')
|
|
self.assertEqual(kwargs['json'], {'fqdn': '_acme.alpha.pic.ngo'})
|
|
self.assertEqual(kwargs['headers']['Authorization'], 'Bearer tok')
|
|
self.assertTrue(result)
|
|
|
|
def test_dns_challenge_create_raises_on_error(self):
|
|
provider = PicNgoDDNS()
|
|
mock_resp = _make_response(500, text='error')
|
|
with patch('requests.post', return_value=mock_resp):
|
|
with self.assertRaises(DDNSError):
|
|
provider.dns_challenge_create('tok', 'fqdn', 'val')
|
|
|
|
def test_dns_challenge_delete_raises_on_error(self):
|
|
provider = PicNgoDDNS()
|
|
mock_resp = _make_response(404, text='not found')
|
|
with patch('requests.delete', return_value=mock_resp):
|
|
with self.assertRaises(DDNSError):
|
|
provider.dns_challenge_delete('tok', 'fqdn')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DDNSManager.get_provider() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetProvider(unittest.TestCase):
|
|
"""DDNSManager.get_provider() returns the correct provider class."""
|
|
|
|
def test_returns_pic_ngo_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, PicNgoDDNS)
|
|
|
|
def test_returns_none_when_no_ddns_config(self):
|
|
cm = _make_config_manager() # empty identity
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsNone(provider)
|
|
|
|
def test_returns_none_when_no_provider_key(self):
|
|
cm = _make_config_manager(ddns_cfg={})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsNone(provider)
|
|
|
|
def test_returns_cloudflare_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={
|
|
'provider': 'cloudflare',
|
|
'api_token': 'cf_tok',
|
|
'zone_id': 'zid',
|
|
})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, CloudflareDDNS)
|
|
|
|
def test_returns_duckdns_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={
|
|
'provider': 'duckdns',
|
|
'token': 'duck_tok',
|
|
'domain': 'mypic',
|
|
})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, DuckDNSDDNS)
|
|
|
|
def test_returns_noip_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'noip'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, NoIPDDNS)
|
|
|
|
def test_returns_freedns_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'freedns'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, FreeDNSDDNS)
|
|
|
|
def test_returns_none_for_unknown_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'nonexistent'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsNone(provider)
|
|
|
|
def test_uses_custom_api_base_url(self):
|
|
cm = _make_config_manager(ddns_cfg={
|
|
'provider': 'pic_ngo',
|
|
'api_base_url': 'https://custom.example.com',
|
|
})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
provider = mgr.get_provider()
|
|
self.assertIsInstance(provider, PicNgoDDNS)
|
|
self.assertEqual(provider.api_base_url, 'https://custom.example.com')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DDNSManager.update_ip() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUpdateIp(unittest.TestCase):
|
|
"""DDNSManager.update_ip() calls provider.update() only when IP changed."""
|
|
|
|
def _make_manager_with_mock_provider(self, token='tok', last_ip=None):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': token})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mgr._last_ip = last_ip
|
|
mock_provider = MagicMock()
|
|
mock_provider.update.return_value = True
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
return mgr, mock_provider
|
|
|
|
def test_update_when_ip_changed(self):
|
|
mgr, mock_provider = self._make_manager_with_mock_provider(last_ip='1.1.1.1')
|
|
with patch('ddns_manager._get_public_ip', return_value='2.2.2.2'):
|
|
mgr.update_ip()
|
|
mock_provider.update.assert_called_once_with('tok', '2.2.2.2')
|
|
self.assertEqual(mgr._last_ip, '2.2.2.2')
|
|
|
|
def test_skips_update_when_ip_unchanged(self):
|
|
mgr, mock_provider = self._make_manager_with_mock_provider(last_ip='3.3.3.3')
|
|
with patch('ddns_manager._get_public_ip', return_value='3.3.3.3'):
|
|
mgr.update_ip()
|
|
mock_provider.update.assert_not_called()
|
|
self.assertEqual(mgr._last_ip, '3.3.3.3')
|
|
|
|
def test_skips_update_when_no_provider(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mgr._last_ip = None
|
|
# Should not raise, just silently skip
|
|
with patch('ddns_manager._get_public_ip', return_value='1.2.3.4'):
|
|
mgr.update_ip()
|
|
|
|
def test_skips_update_when_ip_unreachable(self):
|
|
mgr, mock_provider = self._make_manager_with_mock_provider(last_ip=None)
|
|
with patch('ddns_manager._get_public_ip', return_value=None):
|
|
mgr.update_ip()
|
|
mock_provider.update.assert_not_called()
|
|
|
|
def test_last_ip_not_updated_when_provider_returns_false(self):
|
|
mgr, mock_provider = self._make_manager_with_mock_provider(last_ip='1.1.1.1')
|
|
mock_provider.update.return_value = False
|
|
with patch('ddns_manager._get_public_ip', return_value='9.9.9.9'):
|
|
mgr.update_ip()
|
|
# IP should not be cached when provider says False
|
|
self.assertEqual(mgr._last_ip, '1.1.1.1')
|
|
|
|
def test_ddns_error_is_caught_not_propagated(self):
|
|
mgr, mock_provider = self._make_manager_with_mock_provider(last_ip='1.1.1.1')
|
|
mock_provider.update.side_effect = DDNSError("server error")
|
|
with patch('ddns_manager._get_public_ip', return_value='5.5.5.5'):
|
|
# Should not raise
|
|
mgr.update_ip()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DDNSManager.register() tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRegister(unittest.TestCase):
|
|
def test_register_stores_token_in_config(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 'new_tok', 'subdomain': 'alpha'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
result = mgr.register('alpha', '1.2.3.4')
|
|
self.assertEqual(result['token'], 'new_tok')
|
|
|
|
# set_identity_field('domain', ...) should have been called
|
|
cm.set_identity_field.assert_called_once()
|
|
field_name, field_value = cm.set_identity_field.call_args[0]
|
|
self.assertEqual(field_name, 'domain')
|
|
self.assertEqual(field_value['ddns']['token'], 'new_tok')
|
|
|
|
def test_register_raises_when_no_provider(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
with self.assertRaises(DDNSError):
|
|
mgr.register('alpha', '1.2.3.4')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DDNSManager.dns_challenge_create/delete delegation tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDnsChallenges(unittest.TestCase):
|
|
def _make_manager(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mock_provider = MagicMock()
|
|
mock_provider.dns_challenge_create.return_value = True
|
|
mock_provider.dns_challenge_delete.return_value = True
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
return mgr, mock_provider
|
|
|
|
def test_dns_challenge_create_delegates(self):
|
|
mgr, mock_provider = self._make_manager()
|
|
mgr.dns_challenge_create('_acme.alpha.pic.ngo', 'val123')
|
|
mock_provider.dns_challenge_create.assert_called_once_with(
|
|
'tok', '_acme.alpha.pic.ngo', 'val123'
|
|
)
|
|
|
|
def test_dns_challenge_delete_delegates(self):
|
|
mgr, mock_provider = self._make_manager()
|
|
mgr.dns_challenge_delete('_acme.alpha.pic.ngo')
|
|
mock_provider.dns_challenge_delete.assert_called_once_with(
|
|
'tok', '_acme.alpha.pic.ngo'
|
|
)
|
|
|
|
def test_dns_challenge_create_raises_when_no_provider(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
with self.assertRaises(DDNSError):
|
|
mgr.dns_challenge_create('fqdn', 'val')
|
|
|
|
def test_dns_challenge_delete_raises_when_no_provider(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
with self.assertRaises(DDNSError):
|
|
mgr.dns_challenge_delete('fqdn')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background heartbeat thread tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHeartbeat(unittest.TestCase):
|
|
"""Background heartbeat thread starts, runs, and can be stopped cleanly."""
|
|
|
|
def test_heartbeat_starts(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mgr.update_ip = MagicMock() # avoid real network
|
|
|
|
mgr.start_heartbeat()
|
|
try:
|
|
self.assertIsNotNone(mgr._heartbeat_thread)
|
|
self.assertTrue(mgr._heartbeat_thread.is_alive())
|
|
finally:
|
|
mgr.stop_heartbeat()
|
|
|
|
def test_heartbeat_can_be_stopped(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mgr.update_ip = MagicMock()
|
|
|
|
mgr.start_heartbeat()
|
|
mgr.stop_heartbeat()
|
|
# Thread should be dead after stop
|
|
if mgr._heartbeat_thread is not None:
|
|
self.assertFalse(mgr._heartbeat_thread.is_alive())
|
|
|
|
def test_start_heartbeat_is_idempotent(self):
|
|
"""Calling start_heartbeat() twice should not create a second thread."""
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
mgr.update_ip = MagicMock()
|
|
|
|
mgr.start_heartbeat()
|
|
thread1 = mgr._heartbeat_thread
|
|
mgr.start_heartbeat()
|
|
thread2 = mgr._heartbeat_thread
|
|
try:
|
|
self.assertIs(thread1, thread2)
|
|
finally:
|
|
mgr.stop_heartbeat()
|
|
|
|
def test_heartbeat_calls_update_ip(self):
|
|
"""Heartbeat loop must invoke update_ip() at least once."""
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
called_event = threading.Event()
|
|
|
|
def _fake_update_ip():
|
|
called_event.set()
|
|
|
|
mgr.update_ip = _fake_update_ip
|
|
|
|
mgr.start_heartbeat()
|
|
called = called_event.wait(timeout=3)
|
|
mgr.stop_heartbeat()
|
|
self.assertTrue(called, "update_ip() was not called within 3 seconds")
|
|
|
|
def test_heartbeat_survives_exception_in_update_ip(self):
|
|
"""An exception in update_ip() must not crash the heartbeat thread."""
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
call_count = [0]
|
|
survived_event = threading.Event()
|
|
|
|
def _flaky_update_ip():
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise RuntimeError("transient failure")
|
|
survived_event.set()
|
|
|
|
mgr.update_ip = _flaky_update_ip
|
|
|
|
# Patch the interval to be 0 so the loop spins immediately
|
|
import ddns_manager as _dm
|
|
original_interval = _dm._HEARTBEAT_INTERVAL
|
|
_dm._HEARTBEAT_INTERVAL = 0
|
|
try:
|
|
mgr.start_heartbeat()
|
|
survived = survived_event.wait(timeout=5)
|
|
mgr.stop_heartbeat()
|
|
self.assertTrue(survived, "Thread did not survive exception in update_ip()")
|
|
finally:
|
|
_dm._HEARTBEAT_INTERVAL = original_interval
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_status() and test_connectivity() smoke tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetStatus(unittest.TestCase):
|
|
def test_get_status_returns_dict(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
status = mgr.get_status()
|
|
self.assertIn('service', status)
|
|
self.assertEqual(status['service'], 'ddns')
|
|
self.assertIn('provider', status)
|
|
self.assertEqual(status['provider'], 'pic_ngo')
|
|
|
|
def test_get_status_no_config(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
status = mgr.get_status()
|
|
self.assertIsNone(status['provider'])
|
|
|
|
def test_test_connectivity_no_provider(self):
|
|
cm = _make_config_manager()
|
|
mgr = DDNSManager(config_manager=cm)
|
|
result = mgr.test_connectivity()
|
|
self.assertFalse(result['success'])
|
|
|
|
def test_test_connectivity_with_provider(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
with patch('ddns_manager._get_public_ip', return_value='1.2.3.4'):
|
|
result = mgr.test_connectivity()
|
|
self.assertTrue(result['success'])
|
|
self.assertEqual(result['public_ip'], '1.2.3.4')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_public_ip helper tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetPublicIp(unittest.TestCase):
|
|
def test_returns_ip_on_success(self):
|
|
mock_resp = MagicMock()
|
|
mock_resp.ok = True
|
|
mock_resp.text = ' 1.2.3.4 '
|
|
with patch('requests.get', return_value=mock_resp):
|
|
result = _get_public_ip()
|
|
self.assertEqual(result, '1.2.3.4')
|
|
|
|
def test_returns_none_on_failure(self):
|
|
with patch('requests.get', side_effect=Exception('network error')):
|
|
result = _get_public_ip()
|
|
self.assertIsNone(result)
|
|
|
|
def test_returns_none_on_non_ok_response(self):
|
|
mock_resp = MagicMock()
|
|
mock_resp.ok = False
|
|
with patch('requests.get', return_value=mock_resp):
|
|
result = _get_public_ip()
|
|
self.assertIsNone(result)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|