Initial test

ip_listener/cli.py

@@ -2,17 +2,20 @@ import click
 
 from ip_listener.main import detect
 
-DEFAULT_RESOLVERS = ["@resolver1.opendns.com", "@resolver2.opendns.com"]
+DEFAULT_DNS = "myip.opendns.com"
+DEFAULT_DNS_NAME = "@resolver1.opendns.com"
 DEFAULT_API_URL = "https://api.transip.nl/v6"
 
 
 @click.command()
-@click.argument("domains", envvar="DOMAIN", nargs=-1)
+@click.argument("domains", envvar="DOMAINS", nargs=-1)
 @click.argument("token", envvar="TOKEN")
-@click.option(
-    "--resolvers", envvar="RESOLVERS", default=DEFAULT_RESOLVERS, multiple=True
-)
+@click.option("--dns", envvar="DNS", default=DEFAULT_DNS)
+@click.option("--dns-name", envvar="DNS_NAME", default=DEFAULT_DNS_NAME)
 @click.option("--api-url", envvar="API_URL", default=DEFAULT_API_URL)
 @click.option("--read-only/--write", envvar="READ_ONLY", default=False)
-def run(*args):
-    detect(*args)
+def run(domains, token, dns, dns_name, api_url, read_only):
+    if not domains:
+        raise ValueError("No domain(s) specified")
+
+    detect(domains, (dns, dns_name), api_url, token, read_only)

ip_listener/main.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import subprocess
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -16,6 +17,8 @@ def _get_ip(resolvers):
     except subprocess.CalledProcessError as e:
         raise OSError("Unable to retrieve current IP") from e
 
+    return output.decode("utf-8").strip()
+
 
 def _get_domain(domain, token, api_url):
     headers = {"Authorization": f"Bearer {token}"}
@@ -43,7 +46,7 @@ def _get_domain_data(domains, token, api_url):
             yield {"domain": domain, **response.json()}
 
 
-def _update_domain(domain, payload, token, api_url):
+def _update_domain(domain, payload, api_url, token):
     headers = {"Authorization": f"Bearer {token}"}
 
     return requests.put(
@@ -51,13 +54,13 @@ def _update_domain(domain, payload, token, api_url):
     )
 
 
-def _update_domains(updated_domains, token, read_only):
+def _update_domains(updated_domains, api_url, token, read_only):
     if read_only:
         return
 
     with ThreadPoolExecutor(max_workers=10) as executor:
         futures = {
-            executor.submit(_update_domain, domain, entries, token): domain
+            executor.submit(_update_domain, domain, entries, api_url, token): domain
             for domain, entries in updated_domains.items()
         }
 
@@ -74,7 +77,7 @@ def _update_domains(updated_domains, token, read_only):
             logger.info(f"Updated domain {domain}")
 
 
-def detect(domains, resolvers, token, api_url, read_only):
+def detect(domains, resolvers, api_url, token, read_only):
     ip = _get_ip(resolvers)
     domain_data = _get_domain_data(domains, token, api_url)
     updated_domains = {}
@@ -97,4 +100,4 @@ def detect(domains, resolvers, token, api_url, read_only):
 
         updated_domains[domain] = {"dnsEntries": updated_entries}
 
-    _update_domains(updated_domains, token, api_url, read_only)
+    _update_domains(updated_domains, api_url, token, read_only)

ip_listener/tests.py

@@ -0,0 +1,76 @@
+import json
+from unittest import TestCase
+from unittest.mock import patch
+
+from click.testing import CliRunner
+
+from ip_listener.cli import DEFAULT_API_URL, run
+
+
+class RunTestCase(TestCase):
+    def setUp(self):
+        patcher = patch("ip_listener.main.subprocess.check_output")
+        self.mocked_dns = patcher.start()
+
+        patcher = patch("ip_listener.main.requests.get")
+        self.mocked_get = patcher.start()
+
+        patcher = patch("ip_listener.main.requests.put")
+        self.mocked_put = patcher.start()
+
+    def test_simple(self):
+        self.mocked_dns.return_value = b"111.420\n"
+        self.mocked_get.return_value.json.return_value = {
+            "dnsEntries": [
+                {
+                    "name": "@",
+                    "expire": 60,
+                    "type": "A",
+                    "content": "111.421",
+                }
+            ],
+            "_links": [
+                {
+                    "rel": "self",
+                    "link": "https://api.transip.nl/v6/domains/foobar.com/dns",
+                },
+                {
+                    "rel": "domain",
+                    "link": "https://api.transip.nl/v6/domains/foobar.com",
+                },
+            ],
+        }
+
+        with self.assertLogs("ip_listener.main", level="INFO") as logger:
+            runner = CliRunner()
+            result = runner.invoke(run, ["foobar.com", "TOKEN"])
+
+        self.assertEqual(
+            logger.output, ["INFO:ip_listener.main:Updated domain foobar.com"]
+        )
+
+        self.assertEqual(result.exit_code, 0)
+
+        self.mocked_get.assert_called_with(
+            f"{DEFAULT_API_URL}/domains/foobar.com/dns",
+            headers={"Authorization": "Bearer TOKEN"},
+        )
+
+        expected_json = json.dumps(
+            {
+                "dnsEntries": [
+                    {
+                        "name": "@",
+                        "expire": 60,
+                        "type": "A",
+                        "content": "111.420",
+                    }
+                ]
+            }
+        )
+
+        self.mocked_put.assert_called_with(
+            f"{DEFAULT_API_URL}/domains/foobar.com/dns",
+            data=expected_json,
+            headers={"Authorization": "Bearer TOKEN"},
+        )