Merge 5b22084 into 3a37808

.travis.yml

@@ -6,5 +6,10 @@ python:
 install:
   - pip3 install -r requirements.txt
   - pip3 install -e .
+  - pip3 install pytest
+  - pip3 install pytest-cov
+  - pip3 install coveralls
 script:
-  - pytest
+  - pytest --cov=./
+after_success:
+  - coveralls

src/ce/certcreate.py

@@ -1,24 +1,27 @@
 from src.common.functions import Color as Color
 from ibm_cloud_networking_services import SslCertificateApiV1
 
+
 class CertificateCreator:
 
     def __init__(self, crn, zone_id, endpoint, domain):
         self.crn = crn
         self.zone_id = zone_id
         self.endpoint = endpoint
-        self.hostNames=[domain,"*."+domain]
+        self.hostNames = [domain, "*."+domain]
 
-    def create_certificate(self):   
-        #setting tls mode to strict
+    def create_certificate(self):
+        # setting tls mode to strict
         cert = SslCertificateApiV1.new_instance(
             crn=self.crn, zone_identifier=self.zone_id, service_name="cis_services")
         cert.set_service_url(self.endpoint)
         try:
             resp = cert.change_ssl_setting(value="strict")
-            print(Color.GREEN+"SUCCESS: Set TLS mode to End-to-end CA Signed (strict)"+Color.END)
+            print(
+                Color.GREEN+"SUCCESS: Set TLS mode to End-to-end CA Signed (strict)"+Color.END)
         except:
-            print(Color.RED+"ERROR: Unable to set mode TLS mode to End-to-end CA Signed (strict)"+Color.END)
+            print(
+                Color.RED+"ERROR: Unable to set mode TLS mode to End-to-end CA Signed (strict)"+Color.END)
 
         # checking for duplicated hostnames
         cert = SslCertificateApiV1.new_instance(
@@ -27,17 +30,19 @@ def create_certificate(self):
         resp = cert.list_certificates()
         for cert in resp.result['result']:
             if set(self.hostNames) == set(cert['hosts']):
-                print(Color.YELLOW+"WARNING: certificate already made with host names: "+" ".join(self.hostNames)+Color.END)
+                print(Color.YELLOW+"WARNING: certificate already made with host names: " +
+                      " ".join(self.hostNames)+Color.END)
             return
         # end
-    
+
         # creating certificate
         cert = SslCertificateApiV1.new_instance(
             crn=self.crn, zone_identifier=self.zone_id, service_name="cis_services")
         cert.set_service_url(self.endpoint)
 
         try:
-            resp = cert.order_certificate(x_correlation_id="1864", type="dedicated", hosts=self.hostNames)
+            resp = cert.order_certificate(
+                x_correlation_id="1864", type="dedicated", hosts=self.hostNames)
             print(Color.GREEN+"SUCCESS: Created certificate"+Color.END)
         except:
             print(Color.RED+"ERROR: Unable to create certificate. Make sure hostname(s) match custom domain name"+Color.END)

src/ce/delete_edge.py

@@ -2,6 +2,9 @@
 import requests
 from src.common.functions import Color
 
+def get_input(text):
+    return input(text)
+
 class DeleteEdge:
     def __init__(self, crn: str, zone_id: str, cis_domain: str, apikey: str, token: str) -> None:
         self.crn = crn
@@ -11,7 +14,7 @@ def __init__(self, crn: str, zone_id: str, cis_domain: str, apikey: str, token:
         self.token = token
 
     def delete_edge(self):
-        execute = input("Delete edge function? Input 'y' or 'yes' to execute: ").lower()
+        execute = get_input("Delete edge function? Input 'y' or 'yes' to execute: ").lower()
         if execute == 'y' or execute == 'yes':
             action_name = self.cis_domain.replace('.','-')
             #token = self.request_token(self.apikey)

src/ce/delete_glb.py

@@ -4,16 +4,19 @@
 from ibm_cloud_sdk_core.api_exception import ApiException
 from src.common.functions import Color
 
+def get_input(text):
+    return input(text)
+
 class DeleteGLB:
     def __init__(self, crn: str, zone_id: str, endpoint: str, cis_domain: str) -> None:
         self.crn = crn
         self.zone_id = zone_id
         self.endpoint = endpoint
-        self.cis_domain = cis_domain
+        self.cis_domain = cis_domain    
 
     def delete_glb(self):
         # delete the glb
-        execute_glb = input("Delete global load balancer? Input 'y' or 'yes' to execute: ").lower()
+        execute_glb = get_input("Delete global load balancer? Input 'y' or 'yes' to execute: ").lower()
         if execute_glb == 'y' or execute_glb == 'yes':
             globalLoadBalancer = GlobalLoadBalancerV1.new_instance(
                 crn=self.crn, zone_identifier=self.zone_id, service_name="cis_services")
@@ -40,7 +43,7 @@ def delete_glb(self):
 
             if keepgoing:
                 # delete the origin pool
-                execute_origin = input("Delete origin pool? Input 'y' or 'yes' to execute: ").lower()
+                execute_origin = get_input("Delete origin pool? Input 'y' or 'yes' to execute: ").lower()
                 if execute_origin == 'y' or execute_origin == 'yes':
                     globalLoadBalancerPools = GlobalLoadBalancerPoolsV0.new_instance(
                         crn=self.crn, service_name="cis_services")
@@ -64,7 +67,7 @@ def delete_glb(self):
 
                     if keepgoing:
                         # delete the linked health check
-                        execute_monitor = input("Delete health check monitor? Input 'y' or 'yes' to execute: ").lower()
+                        execute_monitor = get_input("Delete health check monitor? Input 'y' or 'yes' to execute: ").lower()
                         if execute_monitor == 'y' or execute_monitor == 'yes':
 
                             cert = GlobalLoadBalancerMonitorV1.new_instance(

src/common/delete_dns.py

@@ -4,6 +4,9 @@
 from ibm_cloud_sdk_core.api_exception import ApiException
 from src.common.functions import Color
 
+def get_input(text):
+    return input(text)
+
 class DeleteDNS:
     def __init__(self, crn: str, zone_id: str, endpoint: str, cis_domain: str) -> None:
         self.crn = crn
@@ -12,7 +15,7 @@ def __init__(self, crn: str, zone_id: str, endpoint: str, cis_domain: str) -> No
         self.cis_domain = cis_domain
 
     def delete_dns(self):
-        execute = input("Delete DNS Records? Input 'y' or 'yes' to execute: ").lower()
+        execute = get_input("Delete DNS Records? Input 'y' or 'yes' to execute: ").lower()
         if execute == 'y' or execute == 'yes':
             # create instance
             record = DnsRecordsV1.new_instance(

src/iks/certcreate_iks.py

@@ -1,6 +1,9 @@
-import requests, json, time
+import requests
+import json
+import time
 from src.common.functions import Color as Color
 
+
 class SecretCertificateCreator:
 
     def __init__(self, cis_crn, cluster_id, cis_domain, cert_manager_crn, token, cert_name):
@@ -15,7 +18,7 @@ def __init__(self, cis_crn, cluster_id, cis_domain, cert_manager_crn, token, cer
     def create_secret(self):
         cert_crn = self.check_certificate()
         cert_url = "https://containers.cloud.ibm.com/global/ingress/v2/secret/createSecret"
-        
+
         # Creating the data required for the request
         cert_data = json.dumps({
             "cluster": self.cluster_id,
@@ -36,10 +39,12 @@ def create_secret(self):
         try_counter = 0
 
         while keepgoing:
-            cert_response = requests.request("POST", url=cert_url, headers=cert_headers, data=cert_data)
-
+            cert_response = requests.request(
+                "POST", url=cert_url, headers=cert_headers, data=cert_data)
+
             if try_counter == 10:
-                print(Color.RED+"ERROR: Timed out while waiting for certificate. Make sure you haven't been rate limited"+Color.END)
+                print(
+                    Color.RED+"ERROR: Timed out while waiting for certificate. Make sure you haven't been rate limited"+Color.END)
                 break
 
             if cert_response.status_code == 200:
@@ -50,11 +55,12 @@ def create_secret(self):
                 print(cert_response.json())
                 time.sleep(2)
             else:
-                print(Color.RED+"ERROR: Failed to create secret for IKS with error code " + str(cert_response.status_code) + Color.END)
+                print(Color.RED+"ERROR: Failed to create secret for IKS with error code " +
+                      str(cert_response.status_code) + Color.END)
                 keepgoing = False
-                
+
             try_counter += 1
-            
+
         return cert_response
 
     # Creates a certificate (if necessary) and returns a CRN
@@ -73,15 +79,19 @@ def check_certificate(self):
         }
 
         # Gets all certificates previously present in the certificate manager
-        cert_check_response = requests.request("GET", url=cert_check_url, headers=cert_check_headers)
-        #print(cert_check_response.text)
+        cert_check_response = requests.request(
+            "GET", url=cert_check_url, headers=cert_check_headers)
+        print(cert_check_response)
+        print(Color.RED+"cert_check_response.text\n" +
+              Color.END + json.dumps(cert_check_response.json(), indent=2))
         # If a valid certificate exists, it returns the CRN of that certificate
         if cert_check_response.status_code == 200:
             for cert in cert_check_response.json()["certificates"]:
                 if self.cis_domain in cert["domains"] and ("*." + self.cis_domain) in cert["domains"]:
-                    print("Certificate with domain already exists in certificate manager")
+                    print(
+                        "Certificate with domain already exists in certificate manager")
                     return cert["_id"]
-        
+
         print("Ordering a certificate for the certificate manager...")
 
         cert_create_url = f"https://{region}.certificate-manager.cloud.ibm.com/api/v1/{url_cert_man_crn}/certificates/order"
@@ -99,14 +109,15 @@ def check_certificate(self):
         }
 
         # Orders a new certificate through the certificate manager
-        cert_create_response = requests.request("POST", url=cert_create_url, headers=cert_create_headers, data=cert_create_data)
+        cert_create_response = requests.request(
+            "POST", url=cert_create_url, headers=cert_create_headers, data=cert_create_data)
         print(cert_create_response)
-        print(cert_create_response.text)
+        print(type(cert_create_response))
         # Returns the CRN of the new certificate
         print(Color.GREEN+"SUCCESS: Ordered a certificate for the certificate manager!"+Color.END)
         return cert_create_response.json()["_id"]
-        
+
     # Converts the certificate manager CRN into a URL-encoded CRN
     def URLify(self, replacement_str):
         new_string = replacement_str.replace(":", "%3A")
-        return new_string.replace("/", "%2F")
+        return new_string.replace("/", "%2F")

test_var.env

@@ -0,0 +1,8 @@
+CRN="test_crn"
+ZONE_ID="test_zone_id"
+API_ENDPOINT="www.test_api_endpoint.com"
+CIS_SERVICES_APIKEY="test_api_key"
+CIS_NAME="test_instance_name"
+RESOURCE_GROUP="test-resource-group"
+APP_DOMAIN="test_app_url.com"
+CIS_DOMAIN="gcat-interns-test.com"