feat: release old pic.ngo subdomain when cell name changes
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>
This commit is contained in:
2026-05-26 17:07:13 -04:00
parent de43f4a9a0
commit ad2eaca273
2 changed files with 74 additions and 0 deletions
+20
View File
@@ -112,6 +112,14 @@ class PicNgoDDNS(DDNSProvider):
# Public interface # Public interface
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def release(self, token: str) -> bool:
"""DELETE /api/v1/registration — release the subdomain owned by token."""
url = f'{self.api_base_url}/api/v1/registration'
resp = requests.delete(url, json={'token': token},
headers=self._headers(), timeout=self.TIMEOUT)
self._raise_for_status(resp, 'release')
return True
def register(self, name: str, ip: str) -> dict: def register(self, name: str, ip: str) -> dict:
"""POST /api/v1/register — register subdomain, returns token + subdomain.""" """POST /api/v1/register — register subdomain, returns token + subdomain."""
url = f'{self.api_base_url}/api/v1/register' url = f'{self.api_base_url}/api/v1/register'
@@ -398,6 +406,18 @@ class DDNSManager(BaseServiceManager):
if not ip: if not ip:
ip = _get_public_ip() or '' ip = _get_public_ip() or ''
# Release the old subdomain if the name is changing and we hold a token
if self.config_manager is not None and hasattr(provider, 'release'):
old_token = self._ddns_cfg().get('token', '')
old_domain = self._identity().get('domain_name', '')
old_name = old_domain.replace('.pic.ngo', '') if old_domain else ''
if old_token and old_name and old_name != name:
try:
provider.release(old_token)
logger.info("DDNS released old subdomain %r before registering %r", old_name, name)
except Exception as exc:
logger.warning("DDNS could not release old subdomain %r: %s", old_name, exc)
result = provider.register(name, ip) result = provider.register(name, ip)
if self.config_manager is not None: if self.config_manager is not None:
+54
View File
@@ -352,6 +352,60 @@ class TestRegister(unittest.TestCase):
mock_ip.assert_not_called() mock_ip.assert_not_called()
mock_provider.register.assert_called_once_with('alpha', '1.2.3.4') 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): def test_register_raises_when_no_provider(self):
cm = _make_config_manager() cm = _make_config_manager()
mgr = DDNSManager(config_manager=cm) mgr = DDNSManager(config_manager=cm)