Merge pull request #3 from KevinMidboe/feat/test

Feat: Tests 🧪

.drone.yml

@@ -1,3 +1,21 @@
+---
+kind: pipeline
+type: docker
+name: Test
+
+
+platform:
+  os: linux
+  arch: amd64
+
+steps:
+  - name: Run unit tests
+    image: python:3.10-alpine
+    commands:
+      - pip install -r requirements.txt
+      - pip install -r requirements-test.txt
+      - python3 -m unittest tests/test_*.py -v
+
 ---
 kind: pipeline
 type: docker
@@ -31,6 +49,9 @@ trigger:
   branch:
     - main
 
+depends_on:
+  - Test
+
 ---
 kind: pipeline
 type: docker
@@ -91,6 +112,7 @@ trigger:
     - main
 
 depends_on:
+  - Test
   - Publish
 
 volumes:
@@ -99,6 +121,6 @@ volumes:
 
 ---
 kind: signature
-hmac: d3088aaf784f4eaac3223f43a86a19bfccff416fd854351c527d785002ae2c26
+hmac: 2fb50ffa037eb368bcf6d596ced4c0ef42cfde413781ee39dd42b5f695396132
 
 ...

.env.example

@@ -0,0 +1,5 @@
+API_KEY=
+DDNS_ZONE=
+
+SMS_API_KEY=
+SMS_RECIPIENT=

Dockerfile

@@ -1,6 +1,7 @@
 FROM python:3.10-alpine
 
-COPY *.py ./
+COPY requirements.txt .
+COPY src/*.py ./
 
-RUN pip install requests python-dotenv
+RUN pip install -r requirements.txt
 CMD python3 main.py

README.md

@@ -29,7 +29,7 @@ virtualenv
 source -p $(which python3) env/bin/activate
 
 pip install -r requirements.txt
-python main.py
+python src/main.py
 ```
 
 ## Kubernetes
@@ -46,8 +46,14 @@ metadata:
   namespace: cloudflare-ddns
 type: Opaque
 data:
-  API_KEY: BASE_64_ENCODED_CLOUDFLARE_API_KEY
+  API_KEY: CLOUDFLARE_API_KEY
-  DDNS_ZONE: BASE64_ENCODED_CLOUDFLARE_ZONE_ID
+  DDNS_ZONE: CLOUDFLARE_ZONE_ID
+```
+
+(Optional: receive a SMS from gateway API by appending to secrets data)
+```yaml
+  SMS_API_KEY: GATEWAY_API_API_KEY
+  SMS_RECIPIENT: PHONE_NUMBER_TO_RECEIVE
 ```
 
 ```

main.py

@@ -1,65 +0,0 @@
-import os
-import re
-import requests
-from bulk_dns_update import updateAllZones, setAPIKey, getDDNSAddresszoneId
-from notify import notify
-from dotenv import load_dotenv
-from logger import logger
-
-load_dotenv()
-
-currentIP = None
-recordedIP = None
-DDNS_ZONE = os.getenv('DDNS_ZONE')
-
-
-def validIP(ipString):
-    ipRegex = '^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$'
-    return re.search(ipRegex, ipString)
-
-def publicAddress():
-    global currentIP
-    logger.info('Getting public IP from ifconfg.me...')
-
-    r = requests.get('https://ifconfig.me')
-    if r.status_code != 200 or not validIP(r.text):
-        return
-
-    currentIP = r.text
-    logger.info('Public IP: {}'.format(currentIP))
-
-
-def cloudflareDDNS():
-    global recordedIP
-    logger.info('Checking IP recorded in Cloudflare...')
-    ddnsRecord = getDDNSAddresszoneId(DDNS_ZONE)
-    recordedIP = ddnsRecord['content']
-    logger.info('Found ddns recorded IP: {}'.format(recordedIP))
-
-    if currentIP != recordedIP and validIP(recordedIP):
-        logger.info('Public IP has changed, updating all A records.')
-        return True
-    else:
-        logger.info('is same, exiting')
-        return False
-
-
-def main():
-    apiKey = os.getenv('API_KEY')
-    if apiKey is None:
-        raise Exception('In .env file or environment set Cloudflare variable: API_KEY')
-    if DDNS_ZONE is None:
-        raise Exception('In .env file or environment; set Cloudflare zone where addr. points to current IP.')
-
-    setAPIKey(apiKey)
-
-    publicAddress()
-    changed = cloudflareDDNS()
-
-    if changed is True:
-        notify("IP changed to: {}. Updating all cloudflare zones!".format(currentIP))
-        updateAllZones(recordedIP, currentIP)
-
-
-if __name__ == '__main__':
-    main()

requirements-test.txt

@@ -0,0 +1 @@
+responses==0.24.1

requirements.txt

@@ -0,0 +1,2 @@
+requests==2.31.0
+python-dotenv==1.0.1

src/__init__.py

@@ -0,0 +1,7 @@
+import os
+import sys
+PROJECT_PATH = os.getcwd()
+SOURCE_PATH = os.path.join(
+    PROJECT_PATH, "src"
+)
+sys.path.append(SOURCE_PATH)

src/bulk_dns_update.py

@@ -63,6 +63,11 @@ def getZones():
     url = 'https://api.cloudflare.com/client/v4/zones'
     data = cloudflareRequest(url)
 
+    if 'success' not in data and 'errors' not in data:
+        logger.info("Unexpected cloudflare error when getting zones, no response!")
+        logger.info("data:" + str(data))
+        raise Exception('Unexpected Cloudflare error, missing response! Check logs.')
+
     if data['success'] is False:
         logger.info('Request to cloudflare was unsuccessful, error:')
         logger.info(data['errors'])
@@ -77,9 +82,15 @@ def getZones():
 
 
 def getRecordsForZone(zoneId):
-    url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records?type=A'.format(zoneId)
+    url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records?type=A'.format(
+        zoneId)
     data = cloudflareRequest(url)
 
+    if 'success' not in data and 'errors' not in data:
+        logger.info("Unexpected cloudflare error when getting records, no response!")
+        logger.info("data:" + str(data))
+        raise Exception('Unexpected Cloudflare error, missing response! Check logs.')
+
     if data['success'] is False:
         logger.info('Request from cloudflare was unsuccessful, error:')
         logger.info(data['errors'])
@@ -105,7 +116,8 @@ def getDDNSAddresszoneId(ddnsZone):
 
 
 def updateRecord(zoneId, recordId, name, newIP, ttl, proxied):
-    url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records/{}'.format(zoneId, recordId)
+    url = 'https://api.cloudflare.com/client/v4/zones/{}/dns_records/{}'.format(
+        zoneId, recordId)
     data = {
         'type': 'A',
         'name': name,
@@ -115,7 +127,8 @@ def updateRecord(zoneId, recordId, name, newIP, ttl, proxied):
     }
 
     response = cloudflareUpdateRequest(url, data)
-    logger.info('\tRecord updated: {}'.format('✅' if response['success'] is True else '❌'))
+    logger.info('\tRecord updated: {}'.format(
+        '✅' if response['success'] is True else '❌'))
 
 
 def getMatchingRecordsForZone(zoneId, oldIP):
@@ -131,8 +144,10 @@ def updateZone(zone, oldIP, newIP):
         return
 
     for record in records:
-        logger.info('\tRecord {}: {} -> {}'.format(record['name'], record['content'], newIP))
+        logger.info(
-        updateRecord(zone['id'], record['id'], record['name'], newIP, record['ttl'], record['proxied'])
+            '\tRecord {}: {} -> {}'.format(record['name'], record['content'], newIP))
+        updateRecord(zone['id'], record['id'], record['name'],
+                     newIP, record['ttl'], record['proxied'])
 
 
 def updateAllZones(oldIP, newIP):

src/main.py

@@ -0,0 +1,70 @@
+import os
+import re
+import requests
+from bulk_dns_update import updateAllZones, setAPIKey, getDDNSAddresszoneId
+from notify import notify
+from dotenv import load_dotenv
+from logger import logger
+
+load_dotenv()
+
+DDNS_ZONE = os.getenv('DDNS_ZONE')
+
+
+def validIP(ipString):
+    ipRegex = '^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$'
+    return bool(re.search(ipRegex, str(ipString)))
+
+
+def publicAddress():
+    logger.info('Getting public IP from ifconfg.me...')
+
+    r = requests.get('https://ifconfig.me')
+    if r.status_code != 200:
+        return
+
+    ip = r.text
+    if not validIP(ip):
+        return
+
+    logger.info('Public IP: {}'.format(ip))
+    return ip
+
+
+def cloudflareDDNS():
+    logger.info('Checking IP recorded in Cloudflare...')
+    ddnsRecord = getDDNSAddresszoneId(DDNS_ZONE)
+    ip = ddnsRecord['content']
+    logger.info('Found ddns recorded IP: {}'.format(ip))
+
+    if not validIP(ip):
+        return
+
+    return ip
+
+
+def main():
+    apiKey = os.getenv('API_KEY')
+    if apiKey is None:
+        raise Exception(
+            'In .env file or environment set Cloudflare variable: API_KEY')
+    if DDNS_ZONE is None:
+        raise Exception(
+            'In .env file or environment; set Cloudflare zone where addr. points to current IP.')
+
+    setAPIKey(apiKey)
+
+    currentIP = publicAddress()
+    recordedIP = cloudflareDDNS()
+
+    if currentIP == recordedIP or None in [currentIP, recordedIP]:
+        logger.info('is same, exiting')
+        return
+
+    logger.info('Public IP has changed, updating all A records.')
+    notify("IP changed to: {}. Updating all cloudflare zones!".format(currentIP))
+    updateAllZones(recordedIP, currentIP)
+
+
+if __name__ == '__main__':
+    main()

tests/__init__.py

@@ -0,0 +1,2 @@
+from src.logger import logger
+logger.setLevel('WARNING')

tests/test_cloudflare_dns_record_ip.py

@@ -0,0 +1,80 @@
+import re
+import unittest
+import responses
+from src.main import cloudflareDDNS
+
+MOCK_IP = '44.208.147.61'
+CLOUDFLARE_GET_RECORDS_URL = re.compile(
+    r"https\:\/\/api.cloudflare.com\/client\/v4\/zones\/\w*\/dns_records\?type\=A")
+CLOUDFLARE_ADDR_RECORD_EXISTS_RESPONSE = {
+    'success': True,
+    'result': [{
+        'content': MOCK_IP,
+        'name': 'addr.',
+        'id': 'id',
+        'ttl': 86400,
+        'proxied': True
+    }]
+}
+CLOUDFLARE_ADDR_RECORD_NONEXISTANT_RESPONSE = {
+    'success': True,
+    'result': [{
+        'content': MOCK_IP,
+        'name': None,
+        'id': 'id',
+        'ttl': 86400,
+        'proxied': True
+    }]
+}
+CLOUDFLARE_500_RESPONSE = {
+    'success': False,
+    'errors': 'someerror'
+}
+
+
+class TestCloudflareDNSRecordIP(unittest.TestCase):
+
+    @responses.activate
+    def test_successfull_response(self):
+        responses.add(responses.GET, CLOUDFLARE_GET_RECORDS_URL,
+                      json=CLOUDFLARE_ADDR_RECORD_EXISTS_RESPONSE, status=200)
+
+        ip = cloudflareDDNS()
+
+        self.assertEqual(MOCK_IP, ip)
+
+    @responses.activate
+    def test_addr_record_exists(self):
+        responses.add(responses.GET, CLOUDFLARE_GET_RECORDS_URL,
+                      json=CLOUDFLARE_ADDR_RECORD_NONEXISTANT_RESPONSE,
+                      status=200)
+
+        self.assertRaises(Exception, cloudflareDDNS)
+
+    @responses.activate
+    def test_cloudflare_500_response(self):
+        responses.add(responses.GET, CLOUDFLARE_GET_RECORDS_URL,
+                      json=CLOUDFLARE_500_RESPONSE,
+                      status=500)
+
+        self.assertRaises(Exception, cloudflareDDNS)
+
+    @responses.activate
+    def test_cloudflare_empty_200_response(self):
+        responses.add(responses.GET, CLOUDFLARE_GET_RECORDS_URL,
+                      json={},
+                      status=500)
+
+        self.assertRaises(Exception, cloudflareDDNS)
+
+    @responses.activate
+    def test_cloudflare_empty_500_response(self):
+        responses.add(responses.GET, CLOUDFLARE_GET_RECORDS_URL,
+                      json={},
+                      status=500)
+
+        self.assertRaises(Exception, cloudflareDDNS)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_ip_address.py

@@ -0,0 +1,33 @@
+import unittest
+from src.main import validIP
+
+MOCK_IP = "44.208.147.61"
+
+
+class TestIPAddress(unittest.TestCase):
+    def test_valid_ip(self):
+        self.assertTrue(validIP(MOCK_IP))
+
+    def test_invalid_ip(self):
+        ip = "256.0.0.1"
+        self.assertFalse(validIP(ip))
+
+    def test_invalid_format(self):
+        ip = "192.168.1"
+        self.assertFalse(validIP(ip))
+
+    def test_empty_string(self):
+        ip = ""
+        self.assertFalse(validIP(ip))
+
+    def test_error_looking_string(self):
+        ip = "upstream connect error or disconnect/reset before headers. reset reason: connection timeout."
+        self.assertFalse(validIP(ip))
+
+    def test_none(self):
+        ip = None
+        self.assertFalse(validIP(ip))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_public_ip.py

@@ -0,0 +1,40 @@
+import unittest
+import responses
+from src.main import publicAddress
+
+MOCK_IP = '44.208.147.61'
+MOCK_TIMEOUT = 'upstream connect error or disconnect/reset before headers. reset reason: connection timeout.'
+
+
+class TestPublicAddress(unittest.TestCase):
+
+    @responses.activate
+    def test_successfull_response(self):
+        responses.add(responses.GET, 'https://ifconfig.me',
+                      body=MOCK_IP, status=200)
+
+        ip = publicAddress()
+
+        self.assertEqual(MOCK_IP, ip)
+
+    @responses.activate
+    def test_timeout_response(self):
+        responses.add(responses.GET, 'https://ifconfig.me',
+                      body=MOCK_TIMEOUT, status=500)
+
+        ip = publicAddress()
+
+        self.assertIsNone(ip)
+
+    @responses.activate
+    def test_mangled_response(self):
+        responses.add(responses.GET, 'https://ifconfig.me',
+                      body='123.22', status=200)
+
+        ip = publicAddress()
+
+        self.assertIsNone(ip)
+
+
+if __name__ == '__main__':
+    unittest.main()