ad2eaca273
Unit Tests / test (push) Successful in 15m45s
Adds DELETE /api/v1/registration to the DDNS server (token-authenticated, owner-only) and PicNgoDDNS.release() on the client. DDNSManager.register() now automatically releases the old subdomain before claiming the new one, so stale names are freed for others to use. Release failures are logged as warnings and do not block the new registration. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
611 lines
25 KiB
Python
611 lines
25 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 with a real configs dict."""
|
|
cm = MagicMock()
|
|
configs = {}
|
|
if ddns_cfg is not None:
|
|
configs['ddns'] = ddns_cfg
|
|
cm.configs = configs
|
|
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', {}))
|
|
|
|
def test_register_sends_otp_header_when_secret_configured(self):
|
|
"""register() sends X-Register-OTP when totp_secret is set."""
|
|
provider = PicNgoDDNS(totp_secret='JBSWY3DPEHPK3PXP')
|
|
mock_resp = _make_response(200, json_data={'token': 'tok', 'subdomain': 'x.pic.ngo'})
|
|
with patch('requests.post', return_value=mock_resp) as mock_post:
|
|
provider.register('x', '1.2.3.4')
|
|
_, kwargs = mock_post.call_args
|
|
self.assertIn('X-Register-OTP', kwargs.get('headers', {}))
|
|
otp = kwargs['headers']['X-Register-OTP']
|
|
self.assertEqual(len(otp), 6)
|
|
self.assertTrue(otp.isdigit())
|
|
|
|
def test_register_no_otp_header_without_secret(self):
|
|
"""register() omits X-Register-OTP when no TOTP secret is configured."""
|
|
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('X-Register-OTP', 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_ddns_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.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
result = mgr.register('alpha', '1.2.3.4')
|
|
self.assertEqual(result['token'], 'new_tok')
|
|
|
|
# Token saved to top-level ddns config so update_ip() can find it
|
|
cm.set_ddns_config.assert_called_once()
|
|
saved_ddns = cm.set_ddns_config.call_args[0][0]
|
|
self.assertEqual(saved_ddns['token'], 'new_tok')
|
|
|
|
# Subdomain saved to _identity.domain_name
|
|
cm.set_identity_field.assert_called_once_with('domain_name', 'alpha.pic.ngo')
|
|
|
|
def test_register_fetches_public_ip_when_empty(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 't', 'subdomain': 'alpha.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
with patch('ddns_manager._get_public_ip', return_value='5.6.7.8') as mock_ip:
|
|
mgr.register('alpha', '')
|
|
mock_ip.assert_called_once()
|
|
mock_provider.register.assert_called_once_with('alpha', '5.6.7.8')
|
|
|
|
def test_register_uses_provided_ip_without_fetching(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 't', 'subdomain': 'alpha.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
with patch('ddns_manager._get_public_ip') as mock_ip:
|
|
mgr.register('alpha', '1.2.3.4')
|
|
mock_ip.assert_not_called()
|
|
mock_provider.register.assert_called_once_with('alpha', '1.2.3.4')
|
|
|
|
def test_register_releases_old_name_when_changing(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'old_tok'})
|
|
cm.get_identity.return_value = {'domain_name': 'oldname.pic.ngo'}
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 'new_tok', 'subdomain': 'newname.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
mgr.register('newname', '1.2.3.4')
|
|
|
|
mock_provider.release.assert_called_once_with('old_tok')
|
|
mock_provider.register.assert_called_once_with('newname', '1.2.3.4')
|
|
|
|
def test_register_skips_release_when_name_unchanged(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'tok'})
|
|
cm.get_identity.return_value = {'domain_name': 'alpha.pic.ngo'}
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 'tok2', 'subdomain': 'alpha.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
mgr.register('alpha', '1.2.3.4')
|
|
|
|
mock_provider.release.assert_not_called()
|
|
|
|
def test_register_skips_release_when_no_old_token(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
|
|
cm.get_identity.return_value = {'domain_name': 'oldname.pic.ngo'}
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.register.return_value = {'token': 'new_tok', 'subdomain': 'newname.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
mgr.register('newname', '1.2.3.4')
|
|
|
|
mock_provider.release.assert_not_called()
|
|
|
|
def test_register_continues_if_release_fails(self):
|
|
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo', 'token': 'old_tok'})
|
|
cm.get_identity.return_value = {'domain_name': 'oldname.pic.ngo'}
|
|
mgr = DDNSManager(config_manager=cm)
|
|
|
|
mock_provider = MagicMock()
|
|
mock_provider.release.side_effect = DDNSError("server down")
|
|
mock_provider.register.return_value = {'token': 'new_tok', 'subdomain': 'newname.pic.ngo'}
|
|
mgr.get_provider = MagicMock(return_value=mock_provider)
|
|
|
|
result = mgr.register('newname', '1.2.3.4')
|
|
self.assertEqual(result['token'], 'new_tok')
|
|
mock_provider.register.assert_called_once()
|
|
|
|
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()
|