roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2085f77733da3089d7446099501497164f09f62f
pic/tests/test_ddns_manager.py
T
roof 3d750ed1e8
Unit Tests / test (push) Successful in 7m23s
Details
Fix DDNS security and reliability gaps (#2, #3, #5, #6, #7) 
- Fix #2: Move DDNS bearer token from cell_config.json to data/api/ddns_token.
  Token is now in the secrets store (data/) rather than the config store (config/).
  Auto-migrates existing installs on first access. ConfigManager.get/set_ddns_token()
  added. set_ddns_config() now strips 'token' key to prevent it leaking back.

- Fix #3: Set Caddyfile permissions to 0o600 after write so the token embedded
  in the Caddyfile is not world-readable on the host filesystem.

- Fix #5: Heartbeat now fires IDENTITY_CHANGED after re-registration so Caddy
  regenerates its config with the new token automatically — users no longer need
  to click Re-register in Settings after a wizard registration failure.
  Also: heartbeat skips the 401-cycle when no token exists and goes straight to
  registration instead. DDNSManager now accepts service_bus= and is wired up.

- Fix #6: Settings page starts polling GET /api/caddy/cert-status every 15s
  after a successful DDNS re-registration and shows "Acquiring certificate…"
  feedback until Let's Encrypt issues the cert (up to 5 minutes).

- Fix #7: regenerate_with_installed() is debounced (5 s window) so two rapid
  IDENTITY_CHANGED events (e.g. wizard + heartbeat) can't start simultaneous
  ACME orders that interfere with each other.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-09 03:37:48 -04:00

648 lines
27 KiB
Python
Raw Blame History

"""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,
DDNSTokenExpired,
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'] = {k: v for k, v in ddns_cfg.items() if k != 'token'}
cm.configs = configs
# Token is stored outside cell_config.json via get/set_ddns_token
cm.get_ddns_token.return_value = (ddns_cfg or {}).get('token', '')
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() sends token in the request body (DDNS server validates it there)."""
def test_update_sends_token_in_body(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')
# Token must be in the JSON body — server validates it there, not in Authorization
self.assertEqual(kwargs['json'], {'ip': '5.6.7.8', 'token': 'mytoken'})
self.assertTrue(result)
def test_update_does_not_use_bearer_header(self):
"""Token must NOT be sent as Authorization: Bearer — server ignores it and returns 422."""
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:
provider.update('mytoken', '1.2.3.4')
_, kwargs = mock_put.call_args
self.assertNotIn('Authorization', kwargs.get('headers', {}))
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')
def test_update_raises_ddns_token_expired_on_401(self):
provider = PicNgoDDNS()
mock_resp = _make_response(401, text='Unauthorized')
with patch('requests.put', return_value=mock_resp):
with self.assertRaises(DDNSTokenExpired):
provider.update('expiredtoken', '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, cm
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()
def test_no_token_triggers_registration_and_fires_identity_changed(self):
"""When no token exists, update_ip() registers immediately and fires IDENTITY_CHANGED."""
cm = _make_config_manager(ddns_cfg={'provider': 'pic_ngo'})
cm.get_ddns_token.return_value = ''
cm.get_identity.return_value = {'cell_name': 'mytest'}
mock_sbus = MagicMock()
mgr = DDNSManager(config_manager=cm, service_bus=mock_sbus)
mgr._last_ip = None
mock_provider = MagicMock()
mock_provider.register.return_value = {'token': 'new_tok', 'subdomain': 'mytest.pic.ngo'}
mgr.get_provider = MagicMock(return_value=mock_provider)
with patch('ddns_manager._get_public_ip', return_value='1.2.3.4'):
mgr.update_ip()
mock_provider.register.assert_called_once_with('mytest', '1.2.3.4')
mock_provider.update.assert_not_called()
self.assertEqual(mgr._last_ip, '1.2.3.4')
mock_sbus.publish_event.assert_called_once()
# ---------------------------------------------------------------------------
# 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 stored via set_ddns_token (not embedded in cell_config.json)
cm.set_ddns_token.assert_called_once_with('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()
Reference in New Issue View Git Blame Copy Permalink