- Updated strings with f-strings on test suite.

- Removed future imports.

Author: rvelaVenafi

tests/test_env.py

@@ -18,25 +18,25 @@
 
 from six import text_type
 
-FAKE = environ.get("FAKE")
-RANDOM_DOMAIN = environ.get("RANDOM_DOMAIN")
-TPP_URL = environ.get("TPP_URL")
-TPP_USER = environ.get("TPP_USER")
-TPP_PASSWORD = environ.get("TPP_PASSWORD")
-TPP_ZONE = environ.get("TPP_ZONE")
-TPP_ZONE_ECDSA = environ.get("TPP_ZONE_ECDSA")
-TPP_TOKEN_URL = environ.get("TPP_TOKEN_URL")
-TPP_ACCESS_TOKEN = environ.get("TPP_ACCESS_TOKEN")
-CLOUD_URL = environ.get("CLOUD_URL")
-CLOUD_APIKEY = environ.get("CLOUD_APIKEY")
-CLOUD_ZONE = environ.get("CLOUD_ZONE")
+FAKE = environ.get('FAKE')
+RANDOM_DOMAIN = environ.get('RANDOM_DOMAIN')
+TPP_URL = environ.get('TPP_URL')
+TPP_USER = environ.get('TPP_USER')
+TPP_PASSWORD = environ.get('TPP_PASSWORD')
+TPP_ZONE = environ.get('TPP_ZONE')
+TPP_ZONE_ECDSA = environ.get('TPP_ZONE_ECDSA')
+TPP_TOKEN_URL = environ.get('TPP_TOKEN_URL')
+TPP_ACCESS_TOKEN = environ.get('TPP_ACCESS_TOKEN')
+CLOUD_URL = environ.get('CLOUD_URL')
+CLOUD_APIKEY = environ.get('CLOUD_APIKEY')
+CLOUD_ZONE = environ.get('CLOUD_ZONE')
 
-TPP_PM_ROOT = environ.get("TPP_PM_ROOT")
-TPP_CA_NAME = environ.get("TPP_CA_NAME")
-CLOUD_ENTRUST_CA_NAME = environ.get("CLOUD_ENTRUST_CA_NAME")
-CLOUD_DIGICERT_CA_NAME = environ.get("CLOUD_DIGICERT_CA_NAME")
+TPP_PM_ROOT = environ.get('TPP_PM_ROOT')
+TPP_CA_NAME = environ.get('TPP_CA_NAME')
+CLOUD_ENTRUST_CA_NAME = environ.get('CLOUD_ENTRUST_CA_NAME')
+CLOUD_DIGICERT_CA_NAME = environ.get('CLOUD_DIGICERT_CA_NAME')
 
-TPP_SSH_CADN = environ.get("TPP_SSH_CADN")
+TPP_SSH_CADN = environ.get('TPP_SSH_CADN')
 
 if RANDOM_DOMAIN and not isinstance(RANDOM_DOMAIN, text_type):
     RANDOM_DOMAIN = RANDOM_DOMAIN.decode()

tests/test_local_methods.py

@@ -21,8 +21,8 @@
 from cryptography.hazmat.backends import default_backend
 
 from assets import POLICY_CLOUD1, POLICY_TPP1, EXAMPLE_CSR, EXAMPLE_CHAIN
-from vcert import CloudConnection, KeyType, TPPConnection, CertificateRequest, ZoneConfig, CertField, FakeConnection, \
-    logger
+from vcert import (CloudConnection, KeyType, TPPConnection, CertificateRequest, ZoneConfig, CertField, FakeConnection,
+                   logger)
 from vcert.pem import parse_pem, Certificate
 
 pkcs12_enc_cert = """-----BEGIN CERTIFICATE-----
@@ -171,9 +171,8 @@ def test_parse_cloud_zone1(self):
         p = conn._parse_policy_response_to_object(json.loads(POLICY_CLOUD1))
         self.assertEqual(p.id, "3da4ba30-c370-11e9-9e69-99559a9ae32a")
         self.assertEqual(p.SubjectCNRegexes[-1], ".*.test")
-        self.assertTrue(
-            p.SubjectCRegexes == p.SubjectLRegexes == p.SubjectORegexes == p.SubjectOURegexes == p.SubjectSTRegexes == [
-                ".*"])
+        self.assertTrue(p.SubjectCRegexes == p.SubjectLRegexes == p.SubjectORegexes == p.SubjectOURegexes
+                        == p.SubjectSTRegexes == [".*"])
         self.assertEqual(p.key_types[0].key_type, KeyType.RSA)
         self.assertEqual(p.key_types[0].option, 2048)
         self.assertEqual(p.key_types[1].key_type, KeyType.RSA)
@@ -341,9 +340,9 @@ def test_parse_pem_chain(self):
     def test_pkcs12_enc_pk(self):
         certificate = Certificate(cert=pkcs12_enc_cert, chain=chain, key=pkcs12_enc_pk)
         output = certificate.as_pkcs12(passphrase=pkcs12_enc_pk_pass)
-        log.info("PKCS12 created successfully:\n%s" % output)
+        log.info(f"PKCS12 created successfully:\n{output}")
 
     def test_pkcs12_plain_pk(self):
         cert = Certificate(cert=pkcs12_plain_cert, chain=chain, key=pkcs12_plain_pk)
         output = cert.as_pkcs12()
-        log.info("PKCS12 created successfully:\n%s" % output)
+        log.info(f"PKCS12 created successfully:\n{output}")

tests/test_pm.py

@@ -14,23 +14,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
 import os
 import unittest
 from pprint import pformat
 
-from test_env import TPP_TOKEN_URL, CLOUD_APIKEY, CLOUD_URL, TPP_PM_ROOT, CLOUD_ENTRUST_CA_NAME, \
-    CLOUD_DIGICERT_CA_NAME, TPP_CA_NAME, TPP_USER, TPP_PASSWORD
+from test_env import (TPP_TOKEN_URL, CLOUD_APIKEY, CLOUD_URL, TPP_PM_ROOT, CLOUD_ENTRUST_CA_NAME,CLOUD_DIGICERT_CA_NAME,
+                      TPP_CA_NAME, TPP_USER, TPP_PASSWORD)
 from test_utils import timestamp
 from vcert import TPPTokenConnection, CloudConnection, Authentication, SCOPE_PM, logger
 from vcert.parser import json_parser, yaml_parser
 from vcert.parser.utils import parse_policy_spec
-from vcert.policy import Policy, Subject, KeyPair, SubjectAltNames, Defaults, DefaultSubject, DefaultKeyPair, \
-    PolicySpecification
+from vcert.policy import (Policy, Subject, KeyPair, SubjectAltNames, Defaults, DefaultSubject, DefaultKeyPair,
+                          PolicySpecification)
 from vcert.policy.pm_cloud import CA_TYPE_DIGICERT, CA_TYPE_ENTRUST
 
-POLICY_SPEC_JSON = '/resources/policy_specification.json'
-POLICY_SPEC_YAML = '/resources/policy_specification.yaml'
+POLICY_SPEC_JSON = 'resources/policy_specification.json'
+POLICY_SPEC_YAML = 'resources/policy_specification.yaml'
 CA_TYPE_TPP = 'TPP'
 
 log = logger.get_child("test-pm")
@@ -68,7 +67,7 @@ def test_yaml_serialization(self):
 
 class TestTPPPolicyManagement(unittest.TestCase):
     def __init__(self, *args, **kwargs):
-        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={"verify": "/tmp/chain.pem"})
+        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={'verify': "/tmp/chain.pem"})
         auth = Authentication(user=TPP_USER, password=TPP_PASSWORD, scope=SCOPE_PM)
         self.tpp_conn.get_access_token(auth)
         self.json_file = _resolve_resources_path(POLICY_SPEC_JSON)
@@ -102,7 +101,7 @@ def test_create_policy_no_defaults(self):
         self._create_policy_tpp(policy=policy)
 
     def _create_policy_tpp(self, policy_spec=None, policy=None, defaults=None):
-        zone = '%s\\%s' % (TPP_PM_ROOT, _get_tpp_policy_name())
+        zone = f"{TPP_PM_ROOT}\\{_get_tpp_policy_name()}"
         create_policy(self.tpp_conn, zone, policy_spec, policy, defaults)
 
 
@@ -166,7 +165,7 @@ def create_policy(connector, zone, policy_spec=None, policy=None, defaults=None)
     connector.set_policy(zone, policy_spec)
     resp = connector.get_policy(zone)
     data = parse_policy_spec(resp)
-    log.debug('Created Policy at %s' % zone)
+    log.debug(f"Created Policy at {zone}")
     log.debug(pformat(data))
     return resp
 
@@ -228,29 +227,29 @@ def _get_defaults_obj():
 
 
 def _get_app_name():
-    name = 'vcert-python-app-%s'
+    name = 'vcert-python-app-{}'
     return name
 
 
 def _get_cit_name():
-    cit_name = 'vcert-python-cit-%s'
+    cit_name = 'vcert-python-cit-{}'
     return cit_name
 
 
 def _get_zone():
     time = timestamp()
-    zone = (_get_app_name() + '\\' + _get_cit_name()) % (time, time)
+    zone = f"{_get_app_name().format(time)}\\{_get_cit_name().format(time)}"
     return zone
 
 
 def _get_tpp_policy_name():
     time = timestamp()
-    return _get_app_name() % time
+    return f"{_get_app_name().format(time)}"
 
 
 def _resolve_resources_path(path):
     resources_dir = os.path.dirname(__file__)
-    log.debug('Testing root folder: [%s]' % resources_dir)
-    resolved_path = ('.%s' % path) if resources_dir.endswith('tests') else ('./tests%s' % path)
-    log.debug('resolved path: [%s]' % resolved_path)
+    log.debug(f"Testing root folder: [{resources_dir}]")
+    resolved_path = f"./{path}" if resources_dir.endswith('tests') else f"./tests/{path}"
+    log.debug(f"resolved path: [{resolved_path}]")
     return resolved_path

tests/test_ssh.py

@@ -14,27 +14,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import logging
 import platform
 import re
 import unittest
 
 from assets import SSH_CERT_DATA, SSH_PRIVATE_KEY, SSH_PUBLIC_KEY
 from test_env import TPP_TOKEN_URL, TPP_USER, TPP_PASSWORD, TPP_SSH_CADN
 from test_utils import timestamp
-from vcert import CommonConnection, SSHCertRequest, TPPTokenConnection, Authentication, \
-    SCOPE_SSH, write_ssh_files, logger, venafi_connection, VenafiPlatform
+from vcert import (CommonConnection, SSHCertRequest, TPPTokenConnection, Authentication,
+                   SCOPE_SSH, write_ssh_files, logger, venafi_connection, VenafiPlatform)
 from vcert.ssh_utils import SSHRetrieveResponse, SSHKeyPair, SSHCATemplateRequest
 
 log = logger.get_child("test-ssh")
 
-SERVICE_GENERATED_NO_KEY_ERROR = "%s key data is %s empty for Certificate %s"  # type: str
-SSH_CERT_DATA_ERROR = "Certificate data is empty for Certificate %s"  # type: str
+SERVICE_GENERATED_NO_KEY_ERROR = "{} key data is {} empty for Certificate {}"  # type: str
+SSH_CERT_DATA_ERROR = "Certificate data is empty for Certificate {}"  # type: str
 
 
 class TestTPPSSHCertificate(unittest.TestCase):
     def __init__(self, *args, **kwargs):
-        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={"verify": "/tmp/chain.pem"})
+        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={'verify': "/tmp/chain.pem"})
         auth = Authentication(user=TPP_USER, password=TPP_PASSWORD, scope=SCOPE_SSH)
         self.tpp_conn.get_access_token(auth)
         super(TestTPPSSHCertificate, self).__init__(*args, **kwargs)
@@ -49,64 +48,65 @@ def test_enroll_local_generated_keypair(self):
         request.set_public_key_data(keypair.public_key())
         response = _enroll_ssh_cert(self.tpp_conn, request)
         self.assertTrue(response.private_key_data is None,
-                        SERVICE_GENERATED_NO_KEY_ERROR % ("Private", "not", request.key_id))
-        self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR % ("Public", "", request.key_id))
+                        SERVICE_GENERATED_NO_KEY_ERROR.format("Private", "not", request.key_id))
+        self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR.format("Public", "", request.key_id))
         self.assertTrue(response.public_key_data == request.get_public_key_data(),
-                        "Public key on response does not match request.\nExpected: %s\nGot: %s"
-                        % (request.get_public_key_data(), response.public_key_data))
-        self.assertTrue(response.certificate_data, SSH_CERT_DATA_ERROR % request.key_id)
+                        f"Public key on response does not match request."
+                        f"\nExpected: {request.get_public_key_data()}"
+                        f"\nGot: {response.public_key_data}")
+        self.assertTrue(response.certificate_data, SSH_CERT_DATA_ERROR.format(request.key_id))
 
     def test_enroll_service_generated_keypair(self):
         request = SSHCertRequest(cadn=TPP_SSH_CADN, key_id=_random_key_id())
         request.validity_period = "4h"
         request.source_addresses = ["test.com"]
         response = _enroll_ssh_cert(self.tpp_conn, request)
-        self.assertTrue(response.private_key_data, SERVICE_GENERATED_NO_KEY_ERROR % ("Private", "", request.key_id))
-        self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR % ("Public", "", request.key_id))
-        self.assertTrue(response.certificate_data, SSH_CERT_DATA_ERROR % request.key_id)
+        self.assertTrue(response.private_key_data, SERVICE_GENERATED_NO_KEY_ERROR.format("Private", "", request.key_id))
+        self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR.format("Public", "", request.key_id))
+        self.assertTrue(response.certificate_data, SSH_CERT_DATA_ERROR.format(request.key_id))
 
     def test_retrieve_ca_public_key(self):
         tpp_connector = venafi_connection(platform=VenafiPlatform.TPP, url=TPP_TOKEN_URL,
-                                          http_request_kwargs={"verify": "/tmp/chain.pem"})
+                                          http_request_kwargs={'verify': "/tmp/chain.pem"})
         request = SSHCATemplateRequest(ca_template=TPP_SSH_CADN)
         ssh_config = tpp_connector.retrieve_ssh_config(ca_request=request)
-        self.assertIsNotNone(ssh_config.ca_public_key, "%s Public Key data is empty" % TPP_SSH_CADN)
-        self.assertIsNone(ssh_config.ca_principals, "%s default principals is not empty" % TPP_SSH_CADN)
-        log.debug("%s Public Key data:\n%s" % (TPP_SSH_CADN, ssh_config.ca_public_key))
+        self.assertIsNotNone(ssh_config.ca_public_key, f"{TPP_SSH_CADN} Public Key data is empty")
+        self.assertIsNone(ssh_config.ca_principals, f"{TPP_SSH_CADN} default principals is not empty")
+        log.debug(f"{TPP_SSH_CADN} Public Key data:\n{ssh_config.ca_public_key}")
 
     def test_retrieve_ca_public_key_and_principals(self):
         request = SSHCATemplateRequest(ca_template=TPP_SSH_CADN)
         ssh_config = self.tpp_conn.retrieve_ssh_config(ca_request=request)
-        self.assertIsNotNone(ssh_config.ca_public_key, "%s Public Key data is empty" % TPP_SSH_CADN)
-        self.assertIsNotNone(ssh_config.ca_principals, "%s default principals is empty" % TPP_SSH_CADN)
-        log.debug("%s Public Key data: %s" % (TPP_SSH_CADN, ssh_config.ca_public_key))
-        log.debug("%s default principals: %s" % (TPP_SSH_CADN, ssh_config.ca_principals))
+        self.assertIsNotNone(ssh_config.ca_public_key, f"{TPP_SSH_CADN} Public Key data is empty")
+        self.assertIsNotNone(ssh_config.ca_principals, f"{TPP_SSH_CADN} default principals is empty")
+        log.debug(f"{TPP_SSH_CADN} Public Key data: {ssh_config.ca_public_key}")
+        log.debug(f"{TPP_SSH_CADN} default principals: {ssh_config.ca_principals}")
 
 
 class TestSSHUtils(unittest.TestCase):
     def test_write_ssh_files(self):
         key_id = _random_key_id()
         normalized_name = re.sub(r"[^A-Za-z0-9]+", "_", key_id)
-        full_path = "./" + normalized_name
+        full_path = f"./{normalized_name}"
         write_ssh_files("./", key_id, SSH_CERT_DATA, SSH_PRIVATE_KEY, SSH_PUBLIC_KEY)
 
-        err_msg = "%s serialization does not match expected value"
+        err_msg = "{} serialization does not match expected value"
 
-        with open(full_path + "-cert.pub", "r") as cert_file:
+        with open(f"{full_path}-cert.pub", "r") as cert_file:
             s_cert = cert_file.read()
-            self.assertTrue(SSH_CERT_DATA == s_cert, err_msg % "SSH Certificate")
+            self.assertTrue(SSH_CERT_DATA == s_cert, err_msg.format("SSH Certificate"))
 
         with open(full_path, "r") as priv_key_file:
             s_priv_key = priv_key_file.read()
             expected_priv_key = SSH_PRIVATE_KEY
             if platform.system() != "Windows":
                 expected_priv_key = expected_priv_key.replace("\r\n", "\n")
 
-            self.assertTrue(expected_priv_key == s_priv_key, err_msg % "SSH Private Key")
+            self.assertTrue(expected_priv_key == s_priv_key, err_msg.format("SSH Private Key"))
 
-        with open(full_path + ".pub", "r") as pub_key_file:
+        with open(f"{full_path}.pub", "r") as pub_key_file:
             s_pub_key = pub_key_file.read()
-            self.assertTrue(SSH_PUBLIC_KEY == s_pub_key, err_msg % "SSH Public Key")
+            self.assertTrue(SSH_PUBLIC_KEY == s_pub_key, err_msg.format("SSH Public Key"))
 
 
 def _enroll_ssh_cert(connector, request):
@@ -123,4 +123,4 @@ def _enroll_ssh_cert(connector, request):
 
 
 def _random_key_id():
-    return "vcert-python-ssh-%s" % timestamp()
+    return f"vcert-python-ssh-{timestamp()}"

tests/test_e2e.py tests/test_tpp.pytests/test_e2e.py renamed to tests/test_tpp.py

@@ -14,10 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
-from __future__ import absolute_import, division, generators, unicode_literals, print_function, nested_scopes, \
-    with_statement
-
 import binascii
 import logging
 import time
@@ -41,22 +37,22 @@ class TestFakeMethods(unittest.TestCase):
     def test_fake_enroll(self):
         conn = FakeConnection()
         zone = "Default"
-        cn = random_word(10) + ".venafi.example.com"
+        cn = f"{random_word(10)}.venafi.example.com"
         enroll(conn, zone, cn)
 
 
 class TestTPPMethods(unittest.TestCase):
     def __init__(self, *args, **kwargs):
         self.tpp_zone = TPP_ZONE
         self.tpp_zone_ecdsa = TPP_ZONE_ECDSA
-        self.tpp_conn = TPPConnection(TPP_USER, TPP_PASSWORD, TPP_URL, http_request_kwargs={"verify": "/tmp/chain.pem"})
+        self.tpp_conn = TPPConnection(TPP_USER, TPP_PASSWORD, TPP_URL, http_request_kwargs={'verify': "/tmp/chain.pem"})
         super(TestTPPMethods, self).__init__(*args, **kwargs)
 
     def test_tpp_enroll(self):
-        cn = random_word(10) + ".venafi.example.com"
+        cn = f"{random_word(10)}.venafi.example.com"
         _, pkey, cert, _, cert_guid = enroll(self.tpp_conn, self.tpp_zone, cn)
         cert_config = self.tpp_conn._get_certificate_details(cert_guid)
-        self.assertEqual(cert_config["Origin"], "Venafi VCert-Python")
+        self.assertEqual(cert_config['Origin'], "Venafi VCert-Python")
 
     def test_tpp_enroll_with_custom_fields(self):
         cn = random_word(10) + ".venafi.example.com"