Close file descriptor for temp file before attempt to remove it (#40)

* Close file descriptor for temp file before attempt to remove it

* Bump version + add tests

* Minor refactor + additional tests

* Remove debug statements

apsconnectcli/apsconnect.py

@@ -29,6 +29,7 @@
 from kubernetes.client.rest import ApiException
 
 from apsconnectcli.action_logger import Logger
+from apsconnectcli.cluster import read_cluster_certificate
 
 if sys.version_info >= (3,):
     import tempfile
@@ -102,12 +103,7 @@ class APSConnectUtil:
 
     def init_cluster(self, cluster_endpoint, user, pwd, ca_cert):
         """ Connect your kubernetes (k8s) cluster"""
-        try:
-            with open(ca_cert) as _file:
-                ca_cert_data = base64.b64encode(_file.read().encode())
-        except Exception as e:
-            print("Unable to read ca_cert file, error: {}".format(e))
-            sys.exit(1)
+        ca_cert_data = read_cluster_certificate(ca_cert)
 
         auth_template = copy.deepcopy(AUTH_TEMPLATE)
         cluster = auth_template['clusters'][0]['cluster']
@@ -118,12 +114,12 @@ def init_cluster(self, cluster_endpoint, user, pwd, ca_cert):
         user_data['username'] = user
         user_data['password'] = pwd
 
-        _, temp_config = tempfile.mkstemp()
-        with open(temp_config, 'w') as fd:
-            yaml.safe_dump(auth_template, fd)
+        fd, tmp_path = tempfile.mkstemp()
+        with os.fdopen(fd, 'w') as tmp:
+            yaml.safe_dump(auth_template, tmp)
 
         try:
-            api_client = _get_k8s_api_client(temp_config)
+            api_client = _get_k8s_api_client(tmp_path)
             api = client.VersionApi(api_client)
             code = api.get_code()
             print("Connectivity with k8s cluster api [ok]")
@@ -132,8 +128,9 @@ def init_cluster(self, cluster_endpoint, user, pwd, ca_cert):
             print("Unable to communicate with k8s cluster {}, error: {}".format(
                 cluster_endpoint, e))
             sys.exit(1)
-
-        os.remove(temp_config)
+        finally:
+            os.close(fd)
+            os.remove(tmp_path)
 
         if not os.path.exists(KUBE_DIR_PATH):
             os.mkdir(KUBE_DIR_PATH)
@@ -617,7 +614,7 @@ def _get_hub_version(hub):
 def _assert_hub_version(hub_version):
     supported_version = False
 
-    match = re.match(r'^oa-(?P<major>\d)\.(?P<minor>\d)-', hub_version)
+    match = re.match(r'^oa-(?P<major>\d)\.(?P<minor>\d+)-', hub_version)
     if match:
         major = int(match.groupdict()['major'])
         minor = int(match.groupdict()['minor'])

apsconnectcli/cluster.py

@@ -0,0 +1,26 @@
+import base64
+import os
+import sys
+
+from apsconnectcli.action_logger import Logger
+
+LOG_DIR = os.path.expanduser('~/.apsconnect')
+
+if not os.path.exists(LOG_DIR):
+    os.makedirs(LOG_DIR)
+
+LOG_FILE = os.path.join(LOG_DIR, "apsconnect.log")
+
+sys.stdout = Logger(LOG_FILE, sys.stdout)
+sys.stderr = Logger(LOG_FILE, sys.stderr)
+
+
+def read_cluster_certificate(ca_cert):
+    try:
+        with open(ca_cert) as _file:
+            ca_cert_data = base64.b64encode(_file.read().encode())
+    except Exception as e:
+        print("Unable to read ca_cert file, error: {}".format(e))
+        sys.exit(1)
+    else:
+        return ca_cert_data

setup.py

@@ -11,7 +11,7 @@
 setup(
     name='apsconnectcli',
     author='Ingram Micro',
-    version='1.7.19',
+    version='1.7.20',
     keywords='aps apsconnect connector automation',
     extras_require={
         ':python_version<="2.7"': ['backports.tempfile==1.0']},

tests/test_apsconnect_internals.py

@@ -4,6 +4,7 @@
 
 from apsconnectcli.apsconnect import (
     KUBE_FILE_PATH,
+    _assert_hub_version,
     _osaapi_raise_for_status,
     _get_cfg,
     _get_k8s_api_client,
@@ -708,3 +709,17 @@ def test_ok(self):
             self.assertEqual(config, "Config data")
             self.assertFalse(print_mock.called)
             sys_mock.exit.assert_not_called()
+
+
+class AssertHubVersion(TestCase):
+    def test_supported_version(self):
+        with patch('apsconnectcli.apsconnect.sys') as sys_mock:
+            _assert_hub_version('oa-7.13-1216')
+
+            sys_mock.exit.assert_not_called()
+
+    def test_unsupported_version(self):
+        with patch('apsconnectcli.apsconnect.sys') as sys_mock:
+            _assert_hub_version('oa-7.0-1216')
+
+            sys_mock.exit.assert_called_with(1)

tests/test_cluster.py

@@ -0,0 +1,38 @@
+import base64
+import sys
+from unittest import TestCase
+
+from apsconnectcli.cluster import read_cluster_certificate
+
+if sys.version_info >= (3,):
+    from unittest.mock import patch, MagicMock
+
+    _BUILTINS_OPEN = 'builtins.open'
+    _BUILTINS_PRINT = 'builtins.print'
+else:
+    from mock import patch, MagicMock
+
+    _BUILTINS_OPEN = 'apsconnectcli.cluster.open'
+    _BUILTINS_PRINT = 'apsconnectcli.cluster.print'
+
+
+class TestClusterOperation(TestCase):
+    def test_read_cluster_certificate_file_not_found(self):
+        with patch(_BUILTINS_OPEN) as open_mock, \
+                patch('apsconnectcli.cluster.sys') as sys_mock:
+            open_mock.side_effect = Exception("All is lost")
+
+            read_cluster_certificate(None)
+
+        sys_mock.exit.assert_called_with(1)
+
+    def test_read_cluster_certificate_ok(self):
+        with patch(_BUILTINS_OPEN) as open_mock, \
+                patch('apsconnectcli.cluster.sys') as sys_mock:
+            _file = MagicMock()
+            _file.read.return_value = "Certificate data"
+            open_mock.return_value.__enter__.return_value = _file
+            data = read_cluster_certificate(None)
+
+        self.assertEqual(base64.b64decode(data).decode(), "Certificate data")
+        sys_mock.exit.assert_not_called()