Merge pull request #428 from cjmayo/flake8-tests

Resolve flake8 violations in tests/

setup.cfg

@@ -40,7 +40,8 @@ per-file-ignores =
     linkcheck/htmlutil/htmlsoup.py: E402
     linkcheck/parser/__init__.py: E402
     tests/__init__.py: F401
-
+# E501: line too long
+    tests/test_ftpparse.py: E501
 extend-ignore =
 # https://pep8.readthedocs.org/en/latest/intro.html#error-codes
 # these are ignored by default:

tests/__init__.py

@@ -46,10 +46,9 @@ def run_checked(cmd, ret_ok=(0,), **kwargs):
     return retcode
 
 
-
 def run_silent(cmd):
     """Run given command without output."""
-    null = open(os.name == 'nt' and ':NUL' or "/dev/null", 'w')
+    null = open(os.name == "nt" and ":NUL" or "/dev/null", "w")
     try:
         return run(cmd, stdout=null, stderr=subprocess.STDOUT)
     finally:
@@ -58,13 +57,16 @@ def run_silent(cmd):
 
 def _need_func(testfunc, name):
     """Decorator skipping test if given testfunc fails."""
+
     def check_func(func):
         @wraps(func)
         def newfunc(*args, **kwargs):
             if not testfunc():
                 pytest.skip("%s is not available" % name)
             return func(*args, **kwargs)
+
         return newfunc
+
     return check_func
 
 
@@ -80,6 +82,7 @@ def has_network():
         pass
     return False
 
+
 need_network = _need_func(has_network, "network")
 
 
@@ -88,6 +91,7 @@ def has_msgfmt():
     """Test if msgfmt is available."""
     return run_silent(["msgfmt", "-V"]) == 0
 
+
 need_msgfmt = _need_func(has_msgfmt, "msgfmt")
 
 
@@ -96,6 +100,7 @@ def has_posix():
     """Test if this is a POSIX system."""
     return os.name == "posix"
 
+
 need_posix = _need_func(has_posix, "POSIX system")
 
 
@@ -104,6 +109,7 @@ def has_windows():
     """Test if this is a Windows system."""
     return os.name == "nt"
 
+
 need_windows = _need_func(has_windows, "Windows system")
 
 
@@ -112,6 +118,7 @@ def has_linux():
     """Test if this is a Linux system."""
     return sys.platform.startswith("linux")
 
+
 need_linux = _need_func(has_linux, "Linux system")
 
 
@@ -130,6 +137,7 @@ def has_clamav():
         pass
     return False
 
+
 need_clamav = _need_func(has_clamav, "ClamAV")
 
 
@@ -144,6 +152,7 @@ def has_proxy():
     except Exception:
         return False
 
+
 need_proxy = _need_func(has_proxy, "proxy")
 
 
@@ -152,10 +161,12 @@ def has_pyftpdlib():
     """Test if pyftpdlib is available."""
     try:
         import pyftpdlib
+
         return True
     except ImportError:
         return False
 
+
 need_pyftpdlib = _need_func(has_pyftpdlib, "pyftpdlib")
 
 
@@ -164,16 +175,19 @@ def has_biplist():
     """Test if biplist is available."""
     try:
         import biplist
+
         return True
     except ImportError:
         return False
 
+
 need_biplist = _need_func(has_biplist, "biplist")
 
 
 @lru_cache(1)
 def has_newsserver(server):
     import nntplib
+
     try:
         nntp = nntplib.NNTP(server, usenetrc=False)
         nntp.quit()
@@ -184,60 +198,70 @@ def has_newsserver(server):
 
 def need_newsserver(server):
     """Decorator skipping test if newsserver is not available."""
+
     def check_func(func):
         def newfunc(*args, **kwargs):
             if not has_newsserver(server):
                 pytest.skip("Newsserver `%s' is not available" % server)
             return func(*args, **kwargs)
+
         newfunc.__name__ = func.__name__
         return newfunc
-    return check_func
 
+    return check_func
 
 
 @lru_cache(1)
 def has_x11():
     """Test if DISPLAY variable is set."""
-    return os.getenv('DISPLAY') is not None
+    return os.getenv("DISPLAY") is not None
+
 
-need_x11 = _need_func(has_x11, 'X11')
+need_x11 = _need_func(has_x11, "X11")
 
 
 @lru_cache(1)
 def has_word():
     """Test if Word is available."""
     from linkcheck.plugins import parseword
+
     return parseword.has_word()
 
-need_word = _need_func(has_word, 'Word')
+
+need_word = _need_func(has_word, "Word")
 
 
 @lru_cache(1)
 def has_pdflib():
     from linkcheck.plugins import parsepdf
+
     return parsepdf.has_pdflib
 
-need_pdflib = _need_func(has_pdflib, 'pdflib')
+
+need_pdflib = _need_func(has_pdflib, "pdflib")
 
 
 @contextmanager
 def _limit_time(seconds):
     """Raises LinkCheckerInterrupt if given number of seconds have passed."""
-    if os.name == 'posix':
+    if os.name == "posix":
+
         def signal_handler(signum, frame):
             raise LinkCheckerInterrupt("timed out")
+
         old_handler = signal.getsignal(signal.SIGALRM)
         signal.signal(signal.SIGALRM, signal_handler)
         signal.alarm(seconds)
     yield
-    if os.name == 'posix':
+    if os.name == "posix":
         signal.alarm(0)
         if old_handler is not None:
             signal.signal(signal.SIGALRM, old_handler)
 
 
 def limit_time(seconds, skip=False):
     """Limit test time to the given number of seconds, else fail or skip."""
+
     def run_limited(func):
         def new_func(*args, **kwargs):
             try:
@@ -247,8 +271,10 @@ def new_func(*args, **kwargs):
                 if skip:
                     pytest.skip("time limit of %d seconds exceeded" % seconds)
                 assert False, msg
+
         new_func.__name__ = func.__name__
         return new_func
+
     return run_limited
 
 
@@ -262,7 +288,7 @@ def get_file(filename=None):
     return directory
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     print("has clamav", has_clamav())
     print("has network", has_network())
     print("has msgfmt", has_msgfmt())

tests/cache/test_urlqueue.py

@@ -20,21 +20,18 @@
 from linkcheck.cache.results import ResultCache
 from linkcheck.cache.urlqueue import Empty, NUM_PUTS_CLEANUP, UrlQueue
 
-UrlData = namedtuple('UrlData', 'url cache_url aggregate has_result')
-Aggregate = namedtuple('Aggregate', 'result_cache')
+UrlData = namedtuple("UrlData", "url cache_url aggregate has_result")
+Aggregate = namedtuple("Aggregate", "result_cache")
 
 
 class TestUrlQueue(unittest.TestCase):
-
     def setUp(self):
         self.result_cache = ResultCache()
         self.urlqueue = UrlQueue()
         self.urldata1 = UrlData(
             url="Foo",
             cache_url="Foo",
-            aggregate=Aggregate(
-                result_cache=self.result_cache,
-            ),
+            aggregate=Aggregate(result_cache=self.result_cache),
             has_result=True,
         )
 
@@ -72,9 +69,7 @@ def test_put_get(self):
         the item and it can be get only once
         """
         self.urlqueue.put(self.urldata1)
-        cached_item = (
-            self.result_cache.get_result(self.urldata1)
-        )
+        cached_item = self.result_cache.get_result(self.urldata1)
         self.assertEqual(cached_item, None)
         self.assertEqual(self.urlqueue.get(), self.urldata1)
         with self.assertRaises(Empty):
@@ -89,9 +84,7 @@ def test_put_has_result_false(self):
         urldata = UrlData(
             url="Bar",
             cache_url="Bar",
-            aggregate=Aggregate(
-                result_cache=self.result_cache,
-            ),
+            aggregate=Aggregate(result_cache=self.result_cache),
             has_result=False,
         )
         self.urlqueue.put(urldata)
@@ -109,9 +102,7 @@ def test_put_has_result_true(self):
         urldata = UrlData(
             url="Bar",
             cache_url="Bar",
-            aggregate=Aggregate(
-                result_cache=self.result_cache,
-            ),
+            aggregate=Aggregate(result_cache=self.result_cache),
             has_result=True,
         )
         self.urlqueue.put(urldata)
@@ -129,9 +120,7 @@ def test_put_cache(self):
         urldata = UrlData(
             url="Bar",
             cache_url="Foo",
-            aggregate=Aggregate(
-                result_cache=self.result_cache,
-            ),
+            aggregate=Aggregate(result_cache=self.result_cache),
             has_result=True,
         )
         self.urlqueue.put(urldata)
@@ -152,19 +141,15 @@ def test_cleanup(self):
                 UrlData(
                     url="Bar",
                     cache_url="Bar address %s" % i,
-                    aggregate=Aggregate(
-                        result_cache=self.result_cache,
-                    ),
+                    aggregate=Aggregate(result_cache=self.result_cache),
                     has_result=False,
                 ),
             )
         self.assertEqual(self.urlqueue.qsize(), NUM_PUTS_CLEANUP - 1)
         urldata = UrlData(
             url="Bar",
             cache_url="Bar address",
-            aggregate=Aggregate(
-                result_cache=self.result_cache,
-            ),
+            aggregate=Aggregate(result_cache=self.result_cache),
             has_result=False,
         )
         self.result_cache.add_result("Bar address 2", "asdf")