fix: restore Python 2.7 compatibility for EL7

- Replace `from builtins import chr` with `six.unichr()` (Country.py)
- Remove `exist_ok=True` from os.makedirs (Python 3.2+) (WebClient.py)
- Replace `FileNotFoundError` with `OSError` + errno check (FirewallWrapper.py)
- Replace `text=True` with `universal_newlines=True` (test_unblock_region.py)
- Add Python 2.7 import tests (test_py2_compat.py)
- Enable CI workflow with Python 2.7 check using CentOS 7 container

Author: dvershinin

.github/workflows/test.yml

@@ -4,9 +4,32 @@ name: Super-Linter
 on:
   push:
     branches:
-      - none
+      - master
 
 jobs:
+  # Python 2.7 syntax/import check using CentOS 7 container
+  test-py27:
+    name: Python 2.7 compatibility check
+    runs-on: ubuntu-latest
+    container:
+      image: centos:7
+    steps:
+      - name: Install git
+        run: yum -y install git
+
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Install dependencies
+        run: |
+          yum -y install python2 python2-pip python-six
+          pip2 install netaddr
+
+      - name: Verify Python 2.7 imports
+        run: |
+          echo '__version__ = "0.0.0"' > fds/__about__.py
+          python2 -c "import fds; import fds.Country; import fds.Countries; import fds.WebClient; import fds.FirewallWrapper"
+
   # Set the job key. The key is displayed as the job name
   # when a job name is not provided
   test:
@@ -26,7 +49,7 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Fake version
-        run: | 
+        run: |
           echo "ref-names: HEAD -> master, tag: v0" > ".git_archival.txt"
 
       - name: Ensure venv

fds/Country.py

@@ -1,6 +1,6 @@
 from __future__ import unicode_literals
 
-from builtins import chr
+import six
 
 
 class Country:
@@ -9,7 +9,7 @@ class Country:
     def getFlag(self):
         code = self.code
         if code:
-            return chr(ord(code[0]) + Country.OFFSET) + chr(ord(code[1]) + Country.OFFSET)
+            return six.unichr(ord(code[0]) + Country.OFFSET) + six.unichr(ord(code[1]) + Country.OFFSET)
         return False
 
     def __init__(self, name, data={}):

fds/FirewallWrapper.py

@@ -183,12 +183,16 @@ def block_ip(self, ip, ipset_name=None, reload=True):
             log.info('Reloading FirewallD to apply permanent configuration')
             self.fw.reload()
         log.info('Breaking connection with {}'.format(ip))
+        import errno
         from subprocess import CalledProcessError, check_output, STDOUT
         try:
             check_output(["/sbin/conntrack", "-D", "-s", str(ip)], stderr=STDOUT)
-        except FileNotFoundError:
-            log.warning('conntrack not found, skipping connection drop')
-        except CalledProcessError as e:
+        except OSError as e:
+            if e.errno == errno.ENOENT:
+                log.warning('conntrack not found, skipping connection drop')
+            else:
+                raise
+        except CalledProcessError:
             pass
 
     def get_blocked_ips4(self, name=None):

fds/WebClient.py

@@ -88,7 +88,7 @@ def download_file(self, url, local_filename=None, display_name=None,
             # Ensure destination directory exists
             dest_dir = os.path.dirname(local_filename)
             if dest_dir and not os.path.isdir(dest_dir):
-                os.makedirs(dest_dir, exist_ok=True)
+                os.makedirs(dest_dir)
             with open(local_filename, 'wb') as f:
                 for chunk in r.iter_content(chunk_size=chunk_size):
                     if chunk:  # filter out keep-alive new chunks

tests/test_py2_compat.py

@@ -0,0 +1,21 @@
+"""Verify all fds modules can be imported (syntax/compatibility check)."""
+from __future__ import unicode_literals
+
+
+def test_all_modules_import():
+    import fds
+    import fds.Country
+    import fds.Countries
+    import fds.Config
+    import fds.fds
+    import fds.FirewallWrapper
+    import fds.utils
+    import fds.WebClient
+
+
+def test_country_flag_unicode():
+    from fds.Country import Country
+    country = Country('Germany', {'cca2': 'DE', 'demonym': 'German', 'tld': '.de'})
+    flag = country.getFlag()
+    assert flag is not False
+    assert len(flag) == 2

tests/test_unblock_region.py

@@ -6,12 +6,12 @@
 def exec_in_container(cmd):
     return subprocess.check_output([
         'docker', 'exec', 'firewalld-container', '/bin/bash', '-lc', cmd
-    ], stderr=subprocess.STDOUT, text=True)
+    ], stderr=subprocess.STDOUT, universal_newlines=True)
 
 
 def test_unblock_region_smoke():
     # Ensure container is up
-    out = subprocess.check_output(['docker', 'ps', '--filter', 'name=firewalld-container', '--format', '{{.Names}}'], text=True)
+    out = subprocess.check_output(['docker', 'ps', '--filter', 'name=firewalld-container', '--format', '{{.Names}}'], universal_newlines=True)
     assert 'firewalld-container' in out.strip()
 
     # Start firewalld if needed