File multiple-fixes-to-test_hetzner.patch of Package python-dns-lexicon

Overview Repositories Revisions Requests Users Attributes Meta

File multiple-fixes-to-test_hetzner.patch of Package python-dns-lexicon (Revision 60526ede25657258648faa83fce3642e)

Currently displaying revision 60526ede25657258648faa83fce3642e , Show latest

From d6e3c1f4901baaca3b31489298b5bc598dc42bac Mon Sep 17 00:00:00 2001
From: Adrien Ferrand <ferrand.ad@gmail.com>
Date: Thu, 27 Dec 2018 21:46:00 +0100
Subject: [PATCH] Multiple fixes to test_hetzner

* Fixes #332 with raw string regex
 * Added fallback handling for resolving DNS when no domain found via
   local nameservers. Solves issue #333
 * Added mock for _get_dns_cname method
 * Removed tldextract package
 * Specified test domains for DNS resolution test
 * In favor for adferrand ;)
---
 lexicon/providers/gehirn.py     | 14 +++----
 lexicon/providers/hetzner.py    | 15 ++++---
 tests/providers/test_hetzner.py | 74 +++++++++++++++++++++++++--------
 tests/pylint_quality_gate.py    | 73 +++++---------------------------
 tox.ini                         |  1 +
 5 files changed, 84 insertions(+), 93 deletions(-)

diff --git a/lexicon/providers/gehirn.py b/lexicon/providers/gehirn.py
index a8e42e9..46b99a0 100644
--- a/lexicon/providers/gehirn.py
+++ b/lexicon/providers/gehirn.py
@@ -33,13 +33,13 @@ BUILD_FORMATS = {
 }
 
 FORMAT_RE = {
-    "A": re.compile("(?P<address>.+)"),
-    "AAAA": re.compile("(?P<address>.+)"),
-    "CNAME": re.compile("(?P<cname>.+)"),
-    "TXT": re.compile("(?P<data>.+)"),
-    "NS": re.compile("(?P<nsdname>.+)"),
-    "MX": re.compile("(?P<prio>\d+)\s+(?P<exchange>.+)"),
-    "SRV": re.compile("(?P<prio>\d+)\s+(?P<weight>\d+)\s+(?P<port>\d+)\s+(?P<target>.+)"),
+    "A": re.compile(r"(?P<address>.+)"),
+    "AAAA": re.compile(r"(?P<address>.+)"),
+    "CNAME": re.compile(r"(?P<cname>.+)"),
+    "TXT": re.compile(r"(?P<data>.+)"),
+    "NS": re.compile(r"(?P<nsdname>.+)"),
+    "MX": re.compile(r"(?P<prio>\d+)\s+(?P<exchange>.+)"),
+    "SRV": re.compile(r"(?P<prio>\d+)\s+(?P<weight>\d+)\s+(?P<port>\d+)\s+(?P<target>.+)"),
 }
 
 
diff --git a/lexicon/providers/hetzner.py b/lexicon/providers/hetzner.py
index 6d3ea58..d92b990 100644
--- a/lexicon/providers/hetzner.py
+++ b/lexicon/providers/hetzner.py
@@ -421,6 +421,7 @@ class Provider(BaseProvider):
         rrset = dns.rrset.from_text(name, 0, 1, rdtype)
         try:
             resolver = dns.resolver.Resolver()
+            resolver.lifetime = 1
             if nameservers:
                 resolver.nameservers = nameservers
             rrset = resolver.query(name, rdtype)
@@ -459,7 +460,9 @@ class Provider(BaseProvider):
         more linked record name was found for the given fully qualified record name or
         the CNAME lookup was disabled, and then returns the parameters as a tuple.
         """
-        domain = dns.resolver.zone_for_name(name).to_text(True)
+        resolver = dns.resolver.Resolver()
+        resolver.lifetime = 1
+        domain = dns.resolver.zone_for_name(name, resolver=resolver).to_text(True)
         nameservers = Provider._get_nameservers(domain)
         cname = None
         links, max_links = 0, 5
@@ -474,9 +477,9 @@ class Provider(BaseProvider):
             if rrset:
                 links += 1
                 cname = rrset[0].to_text()
-                qdomain = dns.resolver.zone_for_name(cname)
-                if domain != qdomain.to_text(True):
-                    domain = qdomain.to_text(True)
+                qdomain = dns.resolver.zone_for_name(cname, resolver=resolver).to_text(True)
+                if domain != qdomain:
+                    domain = qdomain
                     nameservers = Provider._get_nameservers(qdomain)
             else:
                 link = False
@@ -504,10 +507,10 @@ class Provider(BaseProvider):
             if action != 'update' or name == qname or not qname:
                 LOGGER.info('Hetzner => Enable CNAME lookup '
                             '(see --linked parameter)')
-                return qname, True
+                return name, True
         LOGGER.info('Hetzner => Disable CNAME lookup '
                     '(see --linked parameter)')
-        return qname, False
+        return name, False
 
     def _propagated_record(self, rdtype, name, content, nameservers=None):
         """
diff --git a/tests/providers/test_hetzner.py b/tests/providers/test_hetzner.py
index a7fdd6c..d0a7baa 100644
--- a/tests/providers/test_hetzner.py
+++ b/tests/providers/test_hetzner.py
@@ -1,16 +1,50 @@
-# Test for one implementation of the interface
+from unittest import TestCase
+import os
+import mock
+import pytest
+from bs4 import BeautifulSoup
+import dns.resolver
 from lexicon.providers.hetzner import Provider
 from integration_tests import IntegrationTests
-from unittest import TestCase
-import pytest
 
-import os
-from bs4 import BeautifulSoup
+def _no_dns_lookup():
+    _domains = ['rimek.info', 'bettilaila.com']
+    _resolver = dns.resolver.Resolver()
+    _resolver.lifetime = 1
+    try:
+        for _domain in _domains:
+            _ = dns.resolver.zone_for_name(_domain, resolver=_resolver)
+        return False
+    except dns.exception.DNSException:
+        pass
+    return True
 
-# Hook into testing framework by inheriting unittest.TestCase and reuse
-# the tests which *each and every* implementation of the interface must
-# pass, by inheritance from integration_tests.IntegrationTests
-class HetznerRobotProviderTests(TestCase, IntegrationTests):
+class HetznerIntegrationTests(IntegrationTests):
+
+    @pytest.fixture(autouse=True)
+    def dns_cname_mock(self, request):
+        _ignore_mock = request.node.get_marker('ignore_dns_cname_mock')
+        _domain_mock = self.domain
+        if request.node.name == 'test_Provider_authenticate_with_unmanaged_domain_should_fail':
+            _domain_mock = 'thisisadomainidonotown.com'
+        if _ignore_mock:
+            yield
+        else:
+            with mock.patch('lexicon.providers.hetzner.Provider._get_dns_cname',
+                            return_value=(_domain_mock, [], None)) as fixture:
+                yield fixture
+
+    @pytest.mark.skipif(_no_dns_lookup(), reason='No DNS resolution possible.')
+    @pytest.mark.ignore_dns_cname_mock
+    def test_get_dns_cname(self):
+        """Ensure that zone for name can be resolved through dns.resolver call."""
+        _domain, _nameservers, _cname = Provider._get_dns_cname(('_acme-challenge.fqdn.{}.'
+                                                                 .format(self.domain)), False)
+        assert _domain == self.domain
+        assert _nameservers
+        assert not _cname
+
+class HetznerRobotProviderTests(TestCase, HetznerIntegrationTests):
 
     Provider = Provider
     provider_name = 'hetzner'
@@ -18,7 +52,7 @@ class HetznerRobotProviderTests(TestCase, IntegrationTests):
     domain = 'rimek.info'
 
     def _filter_post_data_parameters(self):
-        return ['_username','_password', '_csrf_token']
+        return ['_username', '_password', '_csrf_token']
 
     def _filter_headers(self):
         return ['Cookie']
@@ -28,9 +62,11 @@ class HetznerRobotProviderTests(TestCase, IntegrationTests):
             if cookie in response['headers']:
                 del response['headers'][cookie]
         if os.environ.get('LEXICON_LIVE_TESTS', 'false') == 'true':
-            filter_body = BeautifulSoup(response['body']['string'], 'html.parser').find(id='center_col')
+            filter_body = (BeautifulSoup(response['body']['string'], 'html.parser')
+                           .find(id='center_col'))
             if not filter_body:
-                filter_body = BeautifulSoup(response['body']['string'], 'html.parser').find(id='login-form')
+                filter_body = (BeautifulSoup(response['body']['string'], 'html.parser')
+                               .find(id='login-form'))
             response['body']['string'] = str(filter_body).encode('UTF-8')
         return response
 
@@ -41,7 +77,7 @@ class HetznerRobotProviderTests(TestCase, IntegrationTests):
                    'latency': 1}
         return options
 
-class HetznerKonsoleHProviderTests(TestCase, IntegrationTests):
+class HetznerKonsoleHProviderTests(TestCase, HetznerIntegrationTests):
 
     Provider = Provider
     provider_name = 'hetzner'
@@ -49,7 +85,7 @@ class HetznerKonsoleHProviderTests(TestCase, IntegrationTests):
     domain = 'bettilaila.com'
 
     def _filter_post_data_parameters(self):
-        return ['login_user_inputbox','login_pass_inputbox', '_csrf_name', '_csrf_token']
+        return ['login_user_inputbox', 'login_pass_inputbox', '_csrf_name', '_csrf_token']
 
     def _filter_headers(self):
         return ['Cookie']
@@ -59,15 +95,17 @@ class HetznerKonsoleHProviderTests(TestCase, IntegrationTests):
             if cookie in response['headers']:
                 del response['headers'][cookie]
         if os.environ.get('LEXICON_LIVE_TESTS', 'false') == 'true':
-            filter_body = BeautifulSoup(response['body']['string'], 'html.parser').find(id='content')
+            filter_body = (BeautifulSoup(response['body']['string'], 'html.parser')
+                           .find(id='content'))
             if not filter_body:
-                filter_body = BeautifulSoup(response['body']['string'], 'html.parser').find(id='loginform')
+                filter_body = (BeautifulSoup(response['body']['string'], 'html.parser')
+                               .find(id='loginform'))
             response['body']['string'] = str(filter_body).encode('UTF-8')
         return response
 
     def _test_parameters_overrides(self):
-        env_username = os.environ.get('LEXICON_HETZNER_KONSOLEH_USERNAME')
-        env_password = os.environ.get('LEXICON_HETZNER_KONSOLEH_PASSWORD')
+        env_username = os.environ.get('LEXICON_HETZNER_KONSOLEH_USERNAME', 'placeholder_username')
+        env_password = os.environ.get('LEXICON_HETZNER_KONSOLEH_PASSWORD', 'placeholder_password')
         options = {'auth_account': 'konsoleh',
                    'auth_username': env_username,
                    'auth_password': env_password,
diff --git a/tests/pylint_quality_gate.py b/tests/pylint_quality_gate.py
index ddc5bef..085263b 100644
--- a/tests/pylint_quality_gate.py
+++ b/tests/pylint_quality_gate.py
@@ -1,64 +1,16 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 import os
-import shutil
-import tempfile
-import contextlib
-import stat
 import sys
-import subprocess
-from io import StringIO
 
 from pylint import lint
 
 
 REPO_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+GLOBAL_NOTE_THRESHOLD = 8.12
 
 
-@contextlib.contextmanager
-def capture():
-    oldout, olderr = sys.stdout, sys.stderr
-    try:
-        out = [StringIO(), StringIO()]
-        sys.stdout, sys.stderr = out
-        yield out
-    finally:
-        sys.stdout, sys.stderr = oldout, olderr
-        out[0] = out[0].getvalue()
-        out[1] = out[1].getvalue()
-
-
-def get_pylint_upstream_master_note():
-    """
-    Get the pylint global note of lexicon on upstream master branch
-    """
-    sys.stdout.write(
-        '===> Preparing a temporary local repository for upstream ... <===\n')
-    worktree_dir = tempfile.mkdtemp()
-
-    try:
-        sys.stdout.write('===> Executing pylint on upstream master '
-                         'to calculate pylint global note diff ... <===\n')
-
-        subprocess.check_output([
-            'git', 'clone', '--depth=1', 'https://github.com/AnalogJ/lexicon.git',
-            worktree_dir], stderr=subprocess.STDOUT)
-        subprocess.check_output(['pip', 'install', '-e', worktree_dir], stderr=subprocess.STDOUT)
-        with capture():
-            results = lint.Run([
-                os.path.join(worktree_dir, 'lexicon'), os.path.join(worktree_dir, 'tests'),
-                os.path.join(worktree_dir, 'tests', 'providers'), '--persistent=n'],
-                do_exit=False)
-
-        return results.linter.stats['global_note']
-    finally:
-        def del_rw(_, name, __):
-            os.chmod(name, stat.S_IWRITE)
-            os.remove(name)
-        shutil.rmtree(worktree_dir, onerror=del_rw)
-
-
-def quality_gate(stats, upstream_master_note):
+def quality_gate(stats):
     """
     Trigger various performance metrics on code quality.
     Raise if these metrics do not match expectations.
@@ -83,34 +35,31 @@ def quality_gate(stats, upstream_master_note):
     else:
         sys.stdout.write('2) OK. No "error" issues have been found.\n')
 
-    if stats['global_note'] < upstream_master_note:
-        sys.stderr.write('3) Failure: pylint global note is '
-                         'decreasing compared to master: {0} => {1}\n'
-                         .format(upstream_master_note, stats['global_note']))
+    if stats['global_note'] < GLOBAL_NOTE_THRESHOLD:
+        sys.stderr.write('3) Failure: pylint global note is below threshold: {0} < {1}\n'
+                         .format(stats['global_note'], GLOBAL_NOTE_THRESHOLD))
         quality_errors = True
     else:
-        sys.stdout.write('3) OK: pylint global is increasing or stable compared to master: '
-                         '{0} => {1}\n'.format(upstream_master_note, stats['global_note']))
+        sys.stdout.write('3) OK: pylint global note is beyond threshold: {0} >= {1}\n'
+                         .format(stats['global_note'], GLOBAL_NOTE_THRESHOLD))
 
     return 0 if not quality_errors else 1
 
 
 def main():
     """Main process"""
-    upstream_master_note = get_pylint_upstream_master_note()
-
     # Script is located two levels deep in the repository root (./tests/pylint_quality_gate.py)
     repo_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
-    sys.stdout.write('===> Executing pylint on current branch ... <===\n')
-    subprocess.check_output(['pip', 'install', '-e', repo_dir], stderr=subprocess.STDOUT)
+    sys.stdout.write('===> Executing pylint ... <===\n')
     results = lint.Run([
-        os.path.join(repo_dir, 'lexicon'), os.path.join(repo_dir, 'tests'),
+        os.path.join(repo_dir, 'lexicon'),
+        os.path.join(repo_dir, 'tests'),
         os.path.join(repo_dir, 'tests', 'providers'), '--persistent=n'],
         do_exit=False)
 
     stats = results.linter.stats
-    sys.exit(quality_gate(stats, upstream_master_note))
+    sys.exit(quality_gate(stats))
 
 
 if __name__ == '__main__':
diff --git a/tox.ini b/tox.ini
index 44a4b1d..191717d 100644
--- a/tox.ini
+++ b/tox.ini
@@ -33,6 +33,7 @@ deps =
 commands =
     python tests/pylint_quality_gate.py
 deps =
+    -r requirements.txt
     -r test-requirements.txt
     -r optional-requirements.txt
     pylint==2.1.1
-- 
2.20.1

Places

File multiple-fixes-to-test_hetzner.patch of Package python-dns-lexicon (Revision 60526ede25657258648faa83fce3642e)

Places