Merge pull request #11 from mietzen/implement-unit-tests

Implement unit tests and fixed some bugs from the findings

.github/workflows/lint_and_test.yml

@@ -1,13 +1,12 @@
-name: Python package
+name: Python Lint and Test
 
 on:
   pull_request:
     branches:
       - 'main'
 
 jobs:
-  build:
-
+  Test:
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -34,4 +33,18 @@ jobs:
         run: |
           # stop the build if there are Python syntax errors or undefined names
           PY_VER=$(echo py${PY_VER} | tr -d '.')
-          ruff --format=github --select=E9,F63,F7,F82 --ignore=E501 --target-version=${PY_VER} .
+          ruff --format=github --ignore=E501 --exclude=__init__.py --target-version=${PY_VER} ./porkbun_ddns
+
+  Check-Test:
+    if: ${{ always() }}
+    runs-on: ubuntu-latest
+    needs:
+      - Test
+    steps:
+      - run: |
+          result="${{ needs.Test.result }}"
+          if [[ $result == "success" || $result == "skipped" ]]; then
+            exit 0
+          else
+            exit 1
+          fi

.gitignore

@@ -10,3 +10,4 @@ dist
 *.egg-info
 docker-compose.yml
 local_test.py
+.env

Docker/entrypoint.py

@@ -49,9 +49,9 @@
                            fritzbox_ip=fritzbox, ipv4=ipv4, ipv6=ipv6)
 
 while True:
-    subdomains = os.getenv('SUBDOMAINS').replace(' ', '').split(',')
+    subdomains = os.getenv('SUBDOMAINS', '')
     if subdomains:
-        for subdomain in subdomains:
+        for subdomain in subdomains.replace(' ', '').split(','):
             porkbun_ddns.set_subdomain(subdomain)
             porkbun_ddns.update_records()
     else:

porkbun_ddns/__init__.py

@@ -1 +1,2 @@
-from .porkbun_ddns import *
+from .porkbun_ddns import PorkbunDDNS
+from .porkbun_ddns import PorkbunDDNS_Error

porkbun_ddns/cli.py

@@ -1,9 +1,19 @@
 import argparse
 import sys
 import traceback
+import logging
 from .porkbun_ddns import PorkbunDDNS, PorkbunDDNS_Error
 
 
+logger = logging.getLogger('porkbun_ddns')
+logger.setLevel(logging.DEBUG)
+handler = logging.StreamHandler(sys.stdout)
+handler.setLevel(logging.DEBUG)
+formatter = logging.Formatter('%(message)s')
+handler.setFormatter(formatter)
+logger.addHandler(handler)
+
+
 def main(argv=sys.argv[1:]):
     parser = argparse.ArgumentParser(
         description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
@@ -29,8 +39,12 @@ def main(argv=sys.argv[1:]):
     ip.add_argument('-6', '--ipv6-only', action='store_true',
                     help="Only set/update IPv6 AAAA Records")
 
-    if len(sys.argv) == 1:
+    if argv and len(argv) == 1:
+        parser.print_help()
+        exit(1)
+    if not argv:
         parser.print_help()
+        exit()
 
     args = parser.parse_args(argv)
 
@@ -43,16 +57,15 @@ def main(argv=sys.argv[1:]):
         porkbun_ddns = PorkbunDDNS(config=args.config, domain=args.domain,
                                    public_ips=args.public_ips, fritzbox_ip=args.fritzbox,
                                    ipv4=ipv4, ipv6=ipv6)
-        porkbun_ddns.set_subdomain(args.subdomain)
+        if args.subdomain:
+            porkbun_ddns.set_subdomain(args.subdomain)
         porkbun_ddns.update_records()
     except PorkbunDDNS_Error as e:
-        sys.stderr.write("Error: " + str(e))
-        sys.exit(1)
+        logger.error("Error: " + str(e))
     except Exception as e:
-        sys.stderr.write("This shouldn't have happened!")
-        sys.stderr.write("Error: " + str(e))
-        sys.stderr.write(traceback.format_exc())
-        sys.exit(1)
+        logger.error("This shouldn't have happened!")
+        logger.error("Error: " + str(e))
+        logger.error(traceback.format_exc())
 
 
 if __name__ == '__main__':

porkbun_ddns/porkbun_ddns.py

@@ -1,8 +1,12 @@
+from __future__ import annotations
+
+import logging
 import json
 import ipaddress
 import urllib.request
 from .helpers import get_ips_from_fritzbox
 
+logger = logging.getLogger('porkbun_ddns')
 
 class PorkbunDDNS_Error(Exception):
     pass
@@ -24,7 +28,15 @@ def __init__(
         if isinstance(config, dict):
             self.config = config
         else:
-            self._load_config(config)
+            if isinstance(config, str):
+                try:
+                    self._load_config(config)
+                except FileNotFoundError as err:
+                    raise FileNotFoundError("Config path is invalid!\nPath:\n{}".format(config)) from err
+            else:
+                raise TypeError("Invalid config! Config should be a str (filepath) or dict!\nYour config:\n{}\nType: {}".format(
+                config, type(config)))
+
         self._check_config()
         self.static_ips = public_ips
         self.domain = domain.lower()
@@ -46,8 +58,8 @@ def _check_config(self) -> None:
         """
         required_keys = ["secretapikey", "apikey"]
         if all(x not in self.config for x in required_keys):
-            raise PorkbunDDNS_Error("all of the following are required in '{}': {}".format(
-                self.config, required_keys))
+            raise PorkbunDDNS_Error("Missing keys! All of the following are required: '{}'\nYour config:\n{}".format(
+                required_keys, self.config))
         if 'endpoint' not in self.config.keys():
             self.config["endpoint"] = "https://porkbun.com/api/json/v3"
 
@@ -128,11 +140,11 @@ def update_records(self):
                             self._delete_record(i['id'])
                             self._create_records(ip, record_type)
                         # Create missing A or AAAA entry
-                        if i["type"] != record_type and record_type not in [x['type'] for x in self.records if x['name'] == self.fqdn]:
+                        if i["type"] not in ["ALIAS", "CNAME", record_type] and record_type not in [x['type'] for x in self.records if x['name'] == self.fqdn]:
                             self._create_records(ip, record_type)
                         # Everything is up to date
                         if i["type"] == record_type and i['content'] == ip.exploded:
-                            print('{}-Record of {} is up to date!'.format(
+                            logger.info('{}-Record of {} is up to date!'.format(
                                 i["type"], i["name"]))
             else:
                 # Create new record
@@ -145,16 +157,16 @@ def _delete_record(self, domain_id: str):
         type, name, content = [(x['type'], x['name'], x['content'])
                                for x in self.records if x['id'] == domain_id][0]
         status = self._api("/dns/delete/" + self.domain + "/" + domain_id)
-        print('Deleting {}-Record for {} with content: {}, Status: {}'.format(type,
+        logger.info('Deleting {}-Record for {} with content: {}, Status: {}'.format(type,
               name, content, status["status"]))
 
     def _create_records(self, ip: ipaddress, record_type: str):
         """Create DNS records for the subdomain with the given IP address and type.
         """
 
-        obj = self.config.copy()
-        obj.update({"name": self.subdomain, "type": record_type,
+        data = self.config.copy()
+        data.update({"name": self.subdomain, "type": record_type,
                     "content": ip.exploded, "ttl": 600})
-        status = self._api("/dns/create/" + self.domain, obj)
-        print('Creating {}-Record for {} with content: {}, Status: {}'.format(record_type,
+        status = self._api("/dns/create/" + self.domain, data)
+        logger.info('Creating {}-Record for {} with content: {}, Status: {}'.format(record_type,
               self.fqdn, ip.exploded, status["status"]))

porkbun_ddns/test/test_porkbun_ddns.py

@@ -0,0 +1,159 @@
+import unittest
+from unittest.mock import patch
+from ..porkbun_ddns import PorkbunDDNS, PorkbunDDNS_Error
+
+valid_config = {"endpoint": "https://porkbun.com/api/json/v3",
+                "apikey": "pk1_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+                "secretapikey": "sk1_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+                }
+
+domain = 'my-domain.local'
+ips = ['127.0.0.1', '::1']
+
+
+def mock_api(status="SUCCESS", mock_records=None):
+    records = list()
+    if mock_records:
+        mock_id = 1111111111
+        for record in mock_records:
+            records.append(
+                {
+                    "id": str(mock_id),
+                    "name": record["name"],
+                    "type": record["type"],
+                    "content": record["content"],
+                    "ttl": "600",
+                    "prio": "0",
+                    "notes": ""
+                }
+            )
+            mock_id += 1
+    return {"status": status, "records": records}
+
+
+class TestPorkbunDDNS(unittest.TestCase):
+    maxDiff = None
+
+    def test_check_valid_config(self):
+        porkbun_ddns = PorkbunDDNS(valid_config, domain, ips)
+        self.assertEqual(porkbun_ddns.config, valid_config)
+        valid_config_wo_endpoint = valid_config.copy()
+        valid_config_wo_endpoint.pop('endpoint')
+        porkbun_ddns = PorkbunDDNS(valid_config_wo_endpoint, domain, ips)
+        self.assertEqual(porkbun_ddns.config, valid_config)
+
+    def test_check_invalid_config(self):
+        self.assertRaises(PorkbunDDNS_Error, PorkbunDDNS,
+                          {'invalid': 000}, domain, ips)
+        self.assertRaises(FileNotFoundError, PorkbunDDNS,
+                          'invalid', domain, ips)
+        self.assertRaises(TypeError, PorkbunDDNS, None, domain, ips)
+        self.assertRaises(TypeError, PorkbunDDNS, 000, domain, ips)
+
+    @patch.object(PorkbunDDNS,
+                  "_api",
+                  return_value=mock_api(
+                      status='SUCCESS',
+                      mock_records=[
+                          {
+                              "name": "my-domain.local",
+                              "type": "A",
+                              "content": "127.0.0.1"},
+                          {
+                              "name": "my-domain.local",
+                              "type": "AAAA",
+                              "content": "0000:0000:0000:0000:0000:0000:0000:0001"}
+                      ]))
+    def test_record_exists_and_up_to_date(self, mocker=None):
+        porkbun_ddns = PorkbunDDNS(valid_config, domain, ips)
+        with self.assertLogs('porkbun_ddns', level='INFO') as cm:
+            porkbun_ddns.set_subdomain('@')
+            porkbun_ddns.update_records()
+            self.assertEqual(cm.output,
+                             ['INFO:porkbun_ddns:A-Record of my-domain.local is up to date!',
+                              'INFO:porkbun_ddns:AAAA-Record of my-domain.local is up to date!'])
+
+    @patch.object(PorkbunDDNS,
+                  "_api",
+                  return_value=mock_api(
+                      status='SUCCESS',
+                      mock_records=[
+                          {
+                              "name": "my-domain.local",
+                              "type": "A",
+                              "content": "127.0.0.2"},
+                          {
+                              "name": "my-domain.local",
+                              "type": "AAAA",
+                              "content": "0000:0000:0000:0000:0000:0000:0000:0002"}
+                      ]))
+    def test_record_exists_and_out_dated(self, mocker=None):
+        porkbun_ddns = PorkbunDDNS(valid_config, domain, ips)
+        with self.assertLogs('porkbun_ddns', level='INFO') as cm:
+            porkbun_ddns.set_subdomain('@')
+            porkbun_ddns.update_records()
+            self.assertEqual(cm.output,
+                             ['INFO:porkbun_ddns:Deleting A-Record for my-domain.local with content: '
+                              '127.0.0.2, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating A-Record for my-domain.local with content: '
+                              '127.0.0.1, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Deleting AAAA-Record for my-domain.local with content: '
+                              '0000:0000:0000:0000:0000:0000:0000:0002, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating AAAA-Record for my-domain.local with content: '
+                              '0000:0000:0000:0000:0000:0000:0000:0001, Status: SUCCESS'])
+
+    @patch.object(PorkbunDDNS,
+                  "_api",
+                  return_value=mock_api())
+    def test_record_do_not_exists(self, mocker=None):
+        porkbun_ddns = PorkbunDDNS(valid_config, domain, ips)
+        with self.assertLogs('porkbun_ddns', level='INFO') as cm:
+            porkbun_ddns.set_subdomain('@')
+            porkbun_ddns.update_records()
+            self.assertEqual(cm.output,
+                             ['INFO:porkbun_ddns:Creating A-Record for my-domain.local with content: '
+                              '127.0.0.1, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating AAAA-Record for my-domain.local with content: '
+                              '0000:0000:0000:0000:0000:0000:0000:0001, Status: SUCCESS'])
+
+    @patch.object(PorkbunDDNS,
+                  "_api",
+                  return_value=mock_api(
+                      status='SUCCESS',
+                      mock_records=[
+                          {
+                              "name": "my-domain.local",
+                              "type": "ALIAS",
+                              "content": "my-domain.lan"
+                          },
+                          {
+                              "name": "my-domain.local",
+                              "type": "CNAME",
+                              "content": "my-domain.lan"}
+                      ]))
+    def test_record_overwrite_alias_and_cname(self, mocker=None):
+        porkbun_ddns = PorkbunDDNS(valid_config, domain, ips)
+        with self.assertLogs('porkbun_ddns', level='INFO') as cm:
+            porkbun_ddns.set_subdomain('@')
+            porkbun_ddns.update_records()
+            self.assertEqual(cm.output,
+                             ['INFO:porkbun_ddns:Deleting ALIAS-Record for my-domain.local with content: '
+                              'my-domain.lan, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating A-Record for my-domain.local with content: '
+                              '127.0.0.1, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Deleting CNAME-Record for my-domain.local with content: '
+                              'my-domain.lan, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating A-Record for my-domain.local with content: '
+                              '127.0.0.1, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Deleting ALIAS-Record for my-domain.local with content: '
+                              'my-domain.lan, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating AAAA-Record for my-domain.local with content: '
+                              '0000:0000:0000:0000:0000:0000:0000:0001, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Deleting CNAME-Record for my-domain.local with content: '
+                              'my-domain.lan, Status: SUCCESS',
+                              'INFO:porkbun_ddns:Creating AAAA-Record for my-domain.local with content: '
+                              '0000:0000:0000:0000:0000:0000:0000:0001, Status: SUCCESS'])
+
+
+if __name__ == '__main__':
+    unittest.main()