From 62deda0ec73b6f46d4f54278e1f020a37900e158 Mon Sep 17 00:00:00 2001 From: Sunil Mohan Adapa Date: Fri, 26 Jun 2026 07:39:06 -0700 Subject: [PATCH] dynamicdns: Allow updating both IPv4 and IPv6 records together - Change the IPv6 checkbox into an option to select from IPv4 only, IPv6 only, or IPv4 and IPv6. Update functional tests. - Add type hints everywhere. - Separate out the generic URL update method to its own module. Add unit tests. - Migrate old configuration to new configuration on first read. - Store multiple IP addresses in the status. Support showing old status format of single IP address as well. Show multiple IP addresses in the status table. - Handle errors during update using exceptions for clearer code. - For generic URL based updates, capture return code, stdout and stdin in error message. Tests: - Adding a new GnuDIP domain leads to configuration setting ip_type = ipv4 - When adding a new dynamic domain, when generic option is selected IP Address Type is shown. When GnuDIP is selected, the field is hidden. - IP Address Type field is as expected. Options and description are as expected. - When a new domain is added for generic domain, the value of ip_type is set as expected. Changing the value to other values works. Configuration is updated as expected. Option values and description are as expected. - The column for IP Address is now 'IP Addresses'. When multiple IP addresses are present in the status, they are shown in multiple lines in the tables as
. When no IP addresses are present, a '-' is shown in the column.

- When generic domain update fails, subprocess failure code, stderr and stdout
  are shown in the error message.

- Unit tests work.

- Functional tests work.

- After 5 minute period, attempts are made to update the domain

- For GnuDIP and generic method:

  - IP address lookup works for IPv4 and IPv6.

  - IP address updates works for IPv4 (tested) and IPv6 (untested).

  - In 'both' configuration, when IPv6 update fails, IPv4 still succeeds.

  - Updates are skipped if the record is up-to-date.

  - Successful updates are performed otherwise.

  - Wrong password leads to proper error being shown.

- When the old status is stored, it is updated during read and proper status is
  shown.

- When configuration has old use_ipv6 key, then it is migrated when newer
  version of the service is started. True value is converted as 'ipv6', False
  value is converted as 'ipv4'. 'null' key is properly removed.

Signed-off-by: Sunil Mohan Adapa 
Reviewed-by: James Valleroy 
---
 plinth/modules/dynamicdns/__init__.py         | 203 ++++++++++--------
 plinth/modules/dynamicdns/forms.py            |  16 +-
 plinth/modules/dynamicdns/generic.py          |  45 ++++
 plinth/modules/dynamicdns/gnudip.py           |  13 +-
 .../modules/dynamicdns/static/dynamicdns.js   |   4 +-
 .../dynamicdns/templates/dynamicdns.html      |   6 +-
 .../dynamicdns/tests/test_functional.py       |   8 +-
 .../modules/dynamicdns/tests/test_generic.py  | 112 ++++++++++
 .../modules/dynamicdns/tests/test_gnudip.py   |  24 ++-
 9 files changed, 321 insertions(+), 110 deletions(-)
 create mode 100644 plinth/modules/dynamicdns/generic.py
 create mode 100644 plinth/modules/dynamicdns/tests/test_generic.py

diff --git a/plinth/modules/dynamicdns/__init__.py b/plinth/modules/dynamicdns/__init__.py
index 5dfaa13bf..a99a67665 100644
--- a/plinth/modules/dynamicdns/__init__.py
+++ b/plinth/modules/dynamicdns/__init__.py
@@ -7,7 +7,7 @@ import json
 import logging
 import subprocess
 import time
-import urllib
+from typing import Any, Literal, Tuple
 
 from django.utils.translation import gettext_lazy as _
 
@@ -20,7 +20,7 @@ from plinth.modules.users.components import UsersAndGroups
 from plinth.signals import domain_added, domain_removed
 from plinth.utils import format_lazy
 
-from . import gnudip, manifest
+from . import generic, gnudip, manifest
 
 logger = logging.getLogger(__name__)
 
@@ -125,99 +125,99 @@ class DynamicDNSApp(app_module.App):
             self.enable()
 
 
-def _lookup_public_address(domain):
+def _lookup_public_address(ip_type: Literal['ipv4', 'ipv6']):
     """Return the IP address by querying an external server."""
     try:
-        ip_type = 'ipv6' if domain['use_ipv6'] else 'ipv4'
         return lookup_public_address(ip_type)
     except Exception:
         return None
 
 
-def _query_dns_address(domain):
+def _query_dns_address(domain: str, ip_type: Literal['ipv4',
+                                                     'ipv6']) -> str | None:
     """Return the IP address in the DNS records."""
-    ip_option = 'AAAA' if domain['use_ipv6'] else 'A'
+    ip_option = 'AAAA' if ip_type == 'ipv6' else 'A'
     try:
-        output = subprocess.check_output(
-            ['host', '-t', ip_option, domain['domain']])
+        output = subprocess.check_output(['host', '-t', ip_option, domain])
         return output.decode().split(' ')[-1].strip().lower()
     except subprocess.CalledProcessError as exception:
-        logger.warning('Unable to lookup DNS for host %s: %s',
-                       domain['domain'], exception)
+        logger.warning('Unable to lookup DNS for host %s: %s', domain,
+                       exception)
         return None
 
 
-def _update_using_url(domain, external_address):
-    """Update DNS entry using an update URL."""
-    update_url = domain['update_url']
-    quote = urllib.parse.quote
-    if external_address:
-        update_url = update_url.replace('', quote(external_address))
+def _check_uptodate_for_ip_type(
+        domain: dict[str, Any],
+        ip_type: Literal['ipv4', 'ipv6']) -> Tuple[bool, str | None, str]:
+    """Return whether a domain is up-to-date for a given IP address type."""
+    uptodate = False
+    dns_address = _query_dns_address(domain['domain'], ip_type)
+    external_address = _lookup_public_address(ip_type)
+    if dns_address == external_address and dns_address is not None:
+        logger.info('Dynamic domain %s is up-to-date: %s (%s)',
+                    domain['domain'], dns_address, ip_type)
+        uptodate = True
 
-    if domain['domain']:
-        update_url = update_url.replace('', quote(domain['domain']))
-
-    if domain['username']:
-        update_url = update_url.replace('', quote(domain['username']))
-
-    if domain['password']:
-        update_url = update_url.replace('', quote(domain['password']))
-
-    options = ['-o', '/dev/null', '-t', '3', '-T', '3']
-    if domain['use_http_basic_auth']:
-        options += [
-            '--user', domain['username'], '--password', domain['password']
-        ]
-
-    if domain['disable_ssl_cert_check']:
-        options += ['--no-check-certificate']
-
-    if domain['use_ipv6']:
-        options += ['-6']
-    else:
-        options += ['-4']
-
-    command = ['wget', '-O', '/dev/null'] + options + [update_url]
-    process = subprocess.run(command, check=False)
-    return process.returncode == 0, external_address
+    return uptodate, dns_address, external_address
 
 
-def _update_dns_for_domain(domain):
+def _update_dns_for_domain(domain: dict[str, Any], ip_type: Literal['ipv4',
+                                                                    'ipv6'],
+                           dns_address: str | None,
+                           external_address: str) -> str | None:
     """Update DNS records for a single domain."""
-    result = False
-    ip_address = None
-    error = None
+    logger.info(
+        'Updating dynamic domain %s, DNS address %s, looked up '
+        'external address %s', domain['domain'], dns_address, external_address)
+    if domain['service_type'] == 'gnudip':
+        return gnudip.update(domain['server'], ip_type, domain['domain'],
+                             domain['username'], domain['password'])
+    else:
+        return generic.update(domain, ip_type, external_address)
+
+
+def _check_and_update_dns_for_domain(domain: dict[str, Any]):
+    """Update DNS records for a single domain only when it is out-of-date."""
+    result = True
+    ip_addresses: list[str] = []
+    error_code: str | None = None
+    error_message: str | None = None
+
+    ip_types: list[Literal['ipv4', 'ipv6']]
+    if domain['ip_type'] == 'both':
+        ip_types = ['ipv4', 'ipv6']
+    else:
+        ip_types = [domain['ip_type']]
+
+    for ip_type in ip_types:
+        try:
+            uptodate, dns_address, external_address = \
+                _check_uptodate_for_ip_type(domain, ip_type)
 
-    try:
-        dns_address = _query_dns_address(domain)
-        external_address = _lookup_public_address(domain)
-        if dns_address == external_address and dns_address is not None:
-            logger.info('Dynamic domain %s is up-to-date: %s',
-                        domain['domain'], dns_address)
-            result = True
             ip_address = dns_address
-            error = ValueError('up-to-date')
-        else:
-            logger.info(
-                'Updating dynamic domain %s, DNS address %s, looked up '
-                'external address %s', domain['domain'], dns_address,
-                external_address)
-            if domain['service_type'] == 'gnudip':
-                result, ip_address = gnudip.update(domain['server'],
-                                                   domain['domain'],
-                                                   domain['username'],
-                                                   domain['password'])
-            else:
-                result, ip_address = _update_using_url(domain,
-                                                       external_address)
-    except Exception as exception:
-        logger.exception('Failed to be update Dynamic DNS - %s', exception)
-        error = exception
+            if not uptodate:
+                ip_address = _update_dns_for_domain(domain, ip_type,
+                                                    dns_address,
+                                                    external_address)
 
-    set_status(domain, result, ip_address, error)
+            ip_addresses.append(ip_address)  # type: ignore
+        except subprocess.CalledProcessError as exception:
+            logger.exception('Failed to be update Dynamic DNS - %s', exception)
+            result = False
+            error_code = str(exception.__class__.__name__)
+            error_message = f'Command failed: code={exception.returncode}, ' \
+                f'stderr={exception.stderr.decode()}, ' \
+                f'stdout={exception.stdout.decode()}'
+        except Exception as exception:
+            logger.exception('Failed to be update Dynamic DNS - %s', exception)
+            result = False
+            error_code = str(exception.__class__.__name__)
+            error_message = str(exception.args[0])
+
+    set_status(domain, result, ip_addresses, error_code, error_message)
 
 
-def update_dns(_data):
+def update_dns(_data) -> None:
     """For all configured domains, check and up to date DNS records."""
     config = get_config()
     app = app_module.App.get('dynamicdns')
@@ -226,10 +226,10 @@ def update_dns(_data):
 
     # Update for each domain
     for domain in config['domains'].values():
-        _update_dns_for_domain(domain)
+        _check_and_update_dns_for_domain(domain)
 
 
-def get_status():
+def get_status() -> dict[str, Any]:
     """Return the status of recent update for each domain."""
     status = kvstore.get_default('dynamicdns_status', '{}')
     status = json.loads(status)
@@ -242,16 +242,25 @@ def get_status():
             status['domains'][domain] = {
                 'domain': domain,
                 'result': False,
-                'ip_address': None,
+                'ip_addresses': [],
                 'error_code': None,
                 'error_message': None,
                 'timestamp': 0,
             }
 
+    for domain_name, domain in status['domains'].items():
+        domain.setdefault('ip_addresses', [])
+        if 'ip_address' in domain:
+            if domain['ip_address']:
+                domain['ip_addresses'].append(domain['ip_address'])
+
+            del domain['ip_address']
+
     return status
 
 
-def set_status(domain, result, ip_address, error=None):
+def set_status(domain: dict[str, Any], result: bool, ip_addresses: list[str],
+               error_code: str | None, error_message: str | None):
     """Set the status of most recent update."""
     status = kvstore.get_default('dynamicdns_status', '{}')
     status = json.loads(status)
@@ -259,38 +268,50 @@ def set_status(domain, result, ip_address, error=None):
     domains[domain['domain']] = {
         'domain': domain['domain'],
         'result': result,
-        'ip_address': ip_address,
-        'error_code': str(error.__class__.__name__) if error else None,
-        'error_message': str(error.args[0]) if error and error.args else None,
+        'ip_addresses': ip_addresses,
+        'error_code': error_code,
+        'error_message': error_message,
         'timestamp': int(time.time()),
     }
     kvstore.set('dynamicdns_status', json.dumps(status))
 
 
-def get_config():
+def get_config() -> dict[str, Any]:
     """Return the current configuration."""
-    default_config = {'domains': {}}
+    default_config: dict[str, Any] = {'domains': {}}
     config = kvstore.get_default('dynamicdns_config', '{}')
     config = json.loads(config) or default_config
-    return _fix_corrupt_config(config)
+    return _migrate_old_config(config)
 
 
-def _fix_corrupt_config(config):
-    """Fix malformed configuration result of bug in older version."""
-    if 'null' not in config['domains']:
-        return config
+def _migrate_old_config(config: dict[str, Any]) -> dict[str, Any]:
+    """Upgrade the old configuration to newer format."""
+    updated = False
+
+    # Fix malformed configuration result of bug in older version.
+    if 'null' in config['domains']:
+        del config['domains']['null']
+        updated = True
+
+    # Upgrade from 'use_ipv6' to 'ip_type' key in domain configuration.
+    for domain_name, domain in config['domains'].items():
+        if 'use_ipv6' in domain:
+            domain['ip_type'] = 'ipv6' if domain['use_ipv6'] else 'ipv4'
+            del domain['use_ipv6']
+            updated = True
+
+    if updated:
+        set_config(config)
 
-    del config['domains']['null']
-    set_config(config)
     return config
 
 
-def set_config(config):
+def set_config(config: dict[str, Any]):
     """Set a new configuration."""
     kvstore.set('dynamicdns_config', json.dumps(config))
 
 
-def notify_domain_added(domain_name):
+def notify_domain_added(domain_name: str):
     """Send a signal that domain has been added."""
     if app_module.App.get('dynamicdns').is_enabled():
         domain_added.send_robust(sender='dynamicdns',
@@ -298,7 +319,7 @@ def notify_domain_added(domain_name):
                                  name=domain_name, services='__all__')
 
 
-def notify_domain_removed(domain_name):
+def notify_domain_removed(domain_name: str):
     """Send a signal that domain has been removed."""
     if app_module.App.get('dynamicdns').is_enabled():
         domain_removed.send_robust(sender='dynamicdns',
diff --git a/plinth/modules/dynamicdns/forms.py b/plinth/modules/dynamicdns/forms.py
index a870ed28a..0d9f7a7a5 100644
--- a/plinth/modules/dynamicdns/forms.py
+++ b/plinth/modules/dynamicdns/forms.py
@@ -45,11 +45,19 @@ class DomainForm(forms.Form):
         gettext_lazy('The username that was used when the account was '
                      'created.')
 
+    help_ip_type = \
+        gettext_lazy('Select based on what your ISP provides: a public IPv4 '
+                     'address, a public IPv6 address, or both.')
+
     provider_choices = (('gnudip', gettext_lazy('GnuDIP')),
                         ('noip.com', 'noip.com'), ('freedns.afraid.org',
                                                    'freedns.afraid.org'),
                         ('other', gettext_lazy('Other update URL')))
 
+    ip_type_choices = (('ipv4', gettext_lazy('IPv4 address only')),
+                       ('ipv6', gettext_lazy('IPv6 address only')),
+                       ('both', gettext_lazy('Both IPv4 and IPv6 addresses')))
+
     service_type = forms.ChoiceField(label=gettext_lazy('Service Type'),
                                      help_text=help_service_type,
                                      choices=provider_choices)
@@ -89,8 +97,9 @@ class DomainForm(forms.Form):
     show_password = forms.BooleanField(label=gettext_lazy('Show password'),
                                        required=False)
 
-    use_ipv6 = forms.BooleanField(
-        label=gettext_lazy('Use IPv6 instead of IPv4'), required=False)
+    ip_type = forms.ChoiceField(label=gettext_lazy('IP Address Type'),
+                                help_text=help_ip_type,
+                                choices=ip_type_choices)
 
     def clean(self):
         """Further validate and transform field data."""
@@ -125,6 +134,9 @@ class DomainForm(forms.Form):
                     if not cleaned_data.get(field_name):
                         self.add_error(field_name, message)
 
+        if service_type == 'gnudip':
+            cleaned_data['ip_type'] = 'ipv4'
+
         del cleaned_data['show_password']
         return cleaned_data
 
diff --git a/plinth/modules/dynamicdns/generic.py b/plinth/modules/dynamicdns/generic.py
new file mode 100644
index 000000000..da0952984
--- /dev/null
+++ b/plinth/modules/dynamicdns/generic.py
@@ -0,0 +1,45 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+Generic HTTP request client for updating Dynamic DNS records.
+"""
+
+import subprocess
+import urllib.parse
+from typing import Any, Literal
+
+
+def update(domain: dict[str, Any], ip_type: Literal['ipv4', 'ipv6'],
+           external_address: str | None) -> str | None:
+    """Update DNS entry using an update URL."""
+    update_url = domain['update_url']
+    quote = urllib.parse.quote
+    if external_address:
+        update_url = update_url.replace('', quote(external_address))
+
+    if domain['domain']:
+        update_url = update_url.replace('', quote(domain['domain']))
+
+    if domain['username']:
+        update_url = update_url.replace('', quote(domain['username']))
+
+    if domain['password']:
+        update_url = update_url.replace('', quote(domain['password']))
+
+    options = ['-t', '3', '-T', '3']
+    if domain['use_http_basic_auth']:
+        options += [
+            '--user', domain['username'], '--password', domain['password']
+        ]
+
+    if domain['disable_ssl_cert_check']:
+        options += ['--no-check-certificate']
+
+    if ip_type == 'ipv6':
+        options += ['-6']
+    else:
+        options += ['-4']
+
+    command = ['wget', '-O', '-'] + options + [update_url]
+    subprocess.run(command, check=True, stdout=subprocess.PIPE,
+                   stderr=subprocess.PIPE)
+    return external_address
diff --git a/plinth/modules/dynamicdns/gnudip.py b/plinth/modules/dynamicdns/gnudip.py
index 6f6b93b6a..99397c48e 100644
--- a/plinth/modules/dynamicdns/gnudip.py
+++ b/plinth/modules/dynamicdns/gnudip.py
@@ -7,6 +7,7 @@ import hashlib
 import logging
 import socket
 from html.parser import HTMLParser
+from typing import Literal
 
 import requests
 
@@ -67,8 +68,8 @@ def _request_get_ipv4(*args, **kwargs):
         socket.getaddrinfo = original
 
 
-def update(server: str, domain: str, username: str,
-           password: str) -> tuple[bool, str | None]:
+def update(server: str, ip_type: Literal['ipv4', 'ipv6'], domain: str,
+           username: str, password: str) -> str | None:
     """Update Dynamic DNS record using GnuDIP protocol.
 
     Protocol documentation:
@@ -78,6 +79,9 @@ def update(server: str, domain: str, username: str,
     support IPv6 (it does have any code to update AAAA records). So, make a
     request only using IPv4 stack.
     """
+    if ip_type == 'ipv6':
+        raise NotImplementedError
+
     domain = domain.removeprefix(username + '.')
     password_digest = hashlib.md5(password.encode()).hexdigest()
 
@@ -105,4 +109,7 @@ def update(server: str, domain: str, username: str,
     update_result = _extract_content_from_meta_tags(update_response.text)
     _check_required_keys(update_result, ['retc'])
     result = (int(update_result['retc']) == 0)
-    return result, update_result.get('addr')
+    if not result:
+        raise Exception('Server responded with an error')
+
+    return update_result.get('addr')
diff --git a/plinth/modules/dynamicdns/static/dynamicdns.js b/plinth/modules/dynamicdns/static/dynamicdns.js
index acc18d840..b467b8aaa 100644
--- a/plinth/modules/dynamicdns/static/dynamicdns.js
+++ b/plinth/modules/dynamicdns/static/dynamicdns.js
@@ -65,7 +65,7 @@ document.addEventListener('DOMContentLoaded', () => {
         document.getElementById('id_domain-update_url').closest('.form-group').style.display = 'none';
         document.getElementById('id_domain-disable_ssl_cert_check').closest('.form-group').style.display = 'none';
         document.getElementById('id_domain-use_http_basic_auth').closest('.form-group').style.display = 'none';
-        document.getElementById('id_domain-use_ipv6').closest('.form-group').style.display = 'none';
+        document.getElementById('id_domain-ip_type').closest('.form-group').style.display = 'none';
         document.getElementById('id_domain-server').closest('.form-group').style.display = 'block';
     }
 
@@ -73,7 +73,7 @@ document.addEventListener('DOMContentLoaded', () => {
         document.getElementById('id_domain-update_url').closest('.form-group').style.display = 'block';
         document.getElementById('id_domain-disable_ssl_cert_check').closest('.form-group').style.display = 'block';
         document.getElementById('id_domain-use_http_basic_auth').closest('.form-group').style.display = 'block';
-        document.getElementById('id_domain-use_ipv6').closest('.form-group').style.display = 'block';
+        document.getElementById('id_domain-ip_type').closest('.form-group').style.display = 'block';
         document.getElementById('id_domain-server').closest('.form-group').style.display = 'none';
     }
 
diff --git a/plinth/modules/dynamicdns/templates/dynamicdns.html b/plinth/modules/dynamicdns/templates/dynamicdns.html
index 7eaf9c74e..c8bf8ace0 100644
--- a/plinth/modules/dynamicdns/templates/dynamicdns.html
+++ b/plinth/modules/dynamicdns/templates/dynamicdns.html
@@ -26,7 +26,7 @@
             {% trans "Domain" %}
             {% trans "Last update" %}
             {% trans "Result" %}
-            {% trans "IP Address" %}
+            {% trans "IP Addresses" %}
             {% trans "Actions" %}
           
         
@@ -66,7 +66,9 @@
                   ({{ domain.error_code }})
                 {% endif %}
               
-              {{ domain.ip_address|default_if_none:'-' }}
+              
{% for ip_address in domain.ip_addresses %}{{ ip_address }}
+{% empty %}-
+{% endfor %}
:@example.com' + '/update?hostname=&ip='), + 'username': 'tester', + 'password': 'testingtesting', + 'use_http_basic_auth': False, + 'disable_ssl_cert_check': True, + } + + +@patch('subprocess.run') +def test_update_ip_types(run, domain): + """Test that various IP types are handled as expected.""" + # IPv4 + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '-4' in run.mock_calls[0].args[0] + + # IPv6 + run.reset_mock() + generic.update(domain, 'ipv6', external_address='1.1.1.1') + assert '-6' in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_ssl_cert_check(run, domain): + """Test that SSL certificate check can be disabled.""" + # Check certificate + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--no-check-certificate' in run.mock_calls[0].args[0] + + # Don't check certificate + run.reset_mock() + domain['disable_ssl_cert_check'] = False + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--no-check-certificate' not in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_basic_auth(run, domain): + """Test that using basic authentication works.""" + # Check certificate + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert '--username' not in run.mock_calls[0].args[0] + assert '--password' not in run.mock_calls[0].args[0] + + # Don't check certificate + run.reset_mock() + domain['use_http_basic_auth'] = True + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert ['--user', 'tester', '--password', + 'testingtesting'] == run.mock_calls[0].args[0][7:11] + + +@patch('subprocess.run') +def test_update_url_parameters(run, domain): + """Test that replacing parameters in the URL works as expected.""" + generic.update(domain, 'ipv4', external_address='1.1.1.1') + expected_url = ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=1.1.1.1') + assert expected_url in run.mock_calls[0].args[0] + + # No substitution for unavailable values + run.reset_mock() + generic.update(domain, 'ipv4', external_address=None) + expected_url = ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=') + assert expected_url in run.mock_calls[0].args[0] + + +@patch('subprocess.run') +def test_update_call(run, domain): + """Test that calling the wget command is as expected.""" + generic.update(domain, 'ipv4', external_address='1.1.1.1') + assert run.mock_calls == [ + call([ + 'wget', '-O', '-', '-t', '3', '-T', '3', '--no-check-certificate', + '-4', + ('https://tester:testingtesting@example.com/update?' + 'hostname=example.org&ip=1.1.1.1') + ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ] + + +@patch('subprocess.run') +def test_update_return_value(run, domain): + """Test that return value or raising exception works.""" + assert '1.1.1.1' == generic.update(domain, 'ipv4', + external_address='1.1.1.1') + + run.side_effect = subprocess.CalledProcessError(1, ['foo']) + with pytest.raises(subprocess.CalledProcessError) as exception_info: + generic.update(domain, 'ipv4', external_address='1.1.1.1') + + assert exception_info.value.returncode == 1 + assert exception_info.value.cmd == ['foo'] diff --git a/plinth/modules/dynamicdns/tests/test_gnudip.py b/plinth/modules/dynamicdns/tests/test_gnudip.py index c1ff29f20..1546096e1 100644 --- a/plinth/modules/dynamicdns/tests/test_gnudip.py +++ b/plinth/modules/dynamicdns/tests/test_gnudip.py @@ -1,3 +1,8 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +""" +Tests for GnuDIP based Dynamic DNS updates. +""" + from unittest.mock import Mock, patch import pytest @@ -66,11 +71,18 @@ def test_update_success(): update_resp = Mock() update_resp.text = response_to_update_request - with patch("plinth.modules.dynamicdns.gnudip.requests.get", + with patch('plinth.modules.dynamicdns.gnudip.requests.get', side_effect=[salt_resp, update_resp]) as mock_get: - result, addr = gnudip.update(server="http://www.2mbit.com:80", - domain="gnudip.dyn.mpis.net", - username="gnudip", password="password") - assert result - assert addr == "24.81.172.128" + addr = gnudip.update(server='http://www.2mbit.com:80', ip_type='ipv4', + domain='gnudip.dyn.mpis.net', username='gnudip', + password='password') + assert addr == '24.81.172.128' assert mock_get.call_count == 2 + + +def test_update_ipv6(): + """Test that updating IPv6 raises and exception.""" + with pytest.raises(NotImplementedError): + gnudip.update(server='http://www.2mbit.com:80', ip_type='ipv6', + domain='gnudip.dyn.mpis.net', username='gnudip', + password='password')