diff --git a/api/ddns_manager.py b/api/ddns_manager.py index 8b508bc..d5e3d76 100644 --- a/api/ddns_manager.py +++ b/api/ddns_manager.py @@ -112,6 +112,14 @@ class PicNgoDDNS(DDNSProvider): # 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: """POST /api/v1/register — register subdomain, returns token + subdomain.""" url = f'{self.api_base_url}/api/v1/register' @@ -398,6 +406,18 @@ class DDNSManager(BaseServiceManager): if not ip: 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) if self.config_manager is not None: diff --git a/tests/test_ddns_manager.py b/tests/test_ddns_manager.py index 43c8d08..9eb4744 100644 --- a/tests/test_ddns_manager.py +++ b/tests/test_ddns_manager.py @@ -352,6 +352,60 @@ class TestRegister(unittest.TestCase): 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)