fix(quality): resolve CodeQL unused local variables and add Python & Shell test suites

.github/workflows/ci.yml

@@ -8,16 +8,20 @@ on:
 
 permissions:
   contents: read
-  pull-requests: write
-  security-events: write
 
 jobs:
   security-scan:
     name: Security Scans
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: write
+      security-events: write
     steps:
       - name: Checkout Repository
         uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+        with:
+          persist-credentials: false
 
       - name: Run tfsec
         uses: aquasecurity/tfsec-pr-commenter-action@1015a3975c7f1400ee4d9f423a7786a3df9fcbec
@@ -38,3 +42,49 @@ jobs:
         uses: github/codeql-action/upload-sarif@c4dd10e44af883a891fe31ced449bcb4a6728b9b
         with:
           sarif_file: 'trivy-results.sarif'
+
+  shell-tests:
+    name: Shell Script Tests
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+        with:
+          persist-credentials: false
+
+      - name: Run Shell Script Test Suites
+        run: |
+          bash tests/scripts/run_shell_tests.sh
+
+  python-tests:
+    name: Python Unit Tests & Code Quality
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@42375524e23c412d93fb67b49958b491fce71c38
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r tools/requirements.txt
+          pip install -r blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/requirements.txt
+          pip install pytest pyflakes
+
+      - name: CodeQL py/unused-local-variable Check (pyflakes)
+        run: |
+          pyflakes tools/ blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/ tests/
+
+      - name: Run Python Unit Tests
+        run: |
+          pytest

blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/auth.py

@@ -40,7 +40,7 @@ def get_credentials():
         subprocess.run(['gcloud', 'auth', 'login'])
         subprocess.run(['gcloud', 'auth', 'application-default', 'login'])
 
-    credentials, project = google.auth.default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
+    credentials, _ = google.auth.default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
     return credentials
 
 def force_reauthentication():

blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/data_stores.py

@@ -19,7 +19,6 @@
 from google.api_core.client_options import ClientOptions
 import random
 import string
-import json
 
 def generate_id(prefix):
     """Generates a random 6-character alphanumeric string."""
@@ -307,7 +306,7 @@ def create_data_store_schema(credentials, project_id, data_store_id, schema):
     )
 
     try:
-        response = request.execute()
+        request.execute()
         click.echo(f"Default schema for data store {data_store_id} successfully patched")
     except Exception as e:
         click.echo(click.style(f"An error occurred while patching the default schema: {parse_http_error(e)}", fg=(255,165,0)))

blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/gem4gov.py

@@ -14,7 +14,6 @@
 
 import sys
 import click
-import google.auth
 from googleapiclient.discovery import build
 from googleapiclient.errors import HttpError
 from google.api_core.client_options import ClientOptions
@@ -78,7 +77,7 @@ def init():
         click.echo("Could not set the gcloud project configuration. Please ensure gcloud is installed and configured correctly.")
         click.echo(click.style("Exiting Onboarding process...", fg="red"))
         sys.exit(1)
-    credentials = get_credentials()
+    get_credentials()
     click.echo(f"Successfully set project ID to: {project_id}")
 
 ##############################################################
@@ -294,7 +293,7 @@ def onboard():
                             click.echo(f"Using path prefix: {path_prefix}")
                             create_gcs_data_store(credentials, project_id, data_store_id, display_name)
                             import_gcs_documents(credentials, project_id, data_store_id, gcs_bucket, path_prefix)
-                            click.echo(f"Google Cloud Storage data store created and indexing operation started successfully...")
+                            click.echo("Google Cloud Storage data store created and indexing operation started successfully...")
                             data_store_list.append(data_store_id)
 
                     # Create BigQuery data store
@@ -328,7 +327,7 @@ def onboard():
                                         click.echo(f'Using schema field "{id_field_name}" as the unique document ID.')
                                     else:
                                         id_property['id'] = 'auto'
-                                        click.echo(f'Autogenerating unique document ID.')
+                                        click.echo('Autogenerating unique document ID.')
 
                                     click.echo(nl=True)
                                     click.echo(nl=True)
@@ -363,7 +362,7 @@ def onboard():
                                     create_bq_data_store(credentials, project_id, data_store_id, display_name, dataset, table)
                                     create_data_store_schema(credentials, project_id, data_store_id, discovery_engine_schema)
                                     import_bq_documents(credentials, project_id, data_store_id, dataset, table, id_property)
-                                    click.echo(f"BigQuery data store created and indexing operation started successfully...")
+                                    click.echo("BigQuery data store created and indexing operation started successfully...")
                                     data_store_list.append(data_store_id)
 
                                 else:
@@ -397,11 +396,11 @@ def onboard():
                 click.echo(click.style("The following data stores are invalid and cannot be connected to the Gemini Enterprise application:", fg="red"))
                 for ds in invalid_data_stores:
                     if ds.get('display_name', None) == None:
-                        click.echo(click.style(f"- {ds["id"]} (Does not exist)", fg="red"))
+                        click.echo(click.style(f"- {ds['id']} (Does not exist)", fg="red"))
                     elif ds.get('kms_key_name', None) == None:
-                        click.echo(click.style(f"- {ds["id"]} (Not CMEK encrypted)", fg="red"))
+                        click.echo(click.style(f"- {ds['id']} (Not CMEK encrypted)", fg="red"))
                     else:
-                        click.echo(click.style(f"- {ds["id"]} (Incompatible)", fg="red"))
+                        click.echo(click.style(f"- {ds['id']} (Incompatible)", fg="red"))
                 data_store_list = []
                 valid_data_stores = []
                 invalid_data_stores = []
@@ -411,7 +410,7 @@ def onboard():
                 # List valid data stores and prompt the user to confirm the list
                 click.echo(click.style("The following data stores have been validated and will be connected to the Gemini Enterprise application:", fg="yellow"))
                 for ds in valid_data_stores:
-                    click.echo(click.style(f"- {ds["id"]} ({ds["display_name"]})", fg="yellow"))
+                    click.echo(click.style(f"- {ds['id']} ({ds['display_name']})", fg="yellow"))
 
                 if click.confirm('Please confirm that you would like to connect the above list of data stores to the Gemini Enterprise application'):
                     break
@@ -779,7 +778,7 @@ def import_documents_helper(credentials, project_id, source_type, data_store_id=
     """Helper to import documents into a selected data store."""
     if not data_store_id:
         click.echo(nl=True)
-        click.echo(click.style(f"Fetching available Gemini Enterprise data stores for import destination...", fg='yellow'))
+        click.echo(click.style("Fetching available Gemini Enterprise data stores for import destination...", fg='yellow'))
 
         # List and select data store
         data_store_id = list_data_stores(credentials, project_id)
@@ -846,8 +845,6 @@ def create_application_logic(credentials, project_id, data_store_list, workforce
 
     if not engine_id:
         click.echo(nl=True)
-        import random
-        import string
         engine_id = 'g4g-gem-ent-app-' + ''.join(random.choices(string.ascii_lowercase + string.digits, k=4))
 
     create_engine(credentials, project_id, engine_id, engine_display_name, company_name, data_store_list, enable_audit_logs)
@@ -1043,8 +1040,8 @@ def configure_identity_provider(credentials, project_id, idp_type, workforce_poo
     )
 
     try:
-        response = request.execute()
-        click.echo(f"Identity provider configured successfully.")
+        request.execute()
+        click.echo("Identity provider configured successfully.")
         if idp_type == '1':
             return "GSUITE"
         elif idp_type == '2':
@@ -1100,12 +1097,12 @@ def validate_kms_key(credentials, kms_key_name):
 
         # Validate the key's location
         if 'us' not in response['name']:
-            click.echo(f"KMS key is not in the 'us' multi-region.")
+            click.echo("KMS key is not in the 'us' multi-region.")
             return False
 
         # Validate that the key is symmetric
         if response['purpose'] != 'ENCRYPT_DECRYPT' or 'GOOGLE_SYMMETRIC_ENCRYPTION' not in response.get('versionTemplate', {}).get('algorithm', ''):
-            click.echo(f"KMS key is not a symmetric key.")
+            click.echo("KMS key is not a symmetric key.")
             return False
 
         click.echo("KMS key validated successfully.")
@@ -1120,7 +1117,7 @@ def grant_kms_permissions(credentials, kms_key_name, project_number):
     """Grants KMS permissions to the necessary service accounts."""
     try:
         service = build('cloudkms', 'v1', credentials=credentials)
-        click.echo(f'Granting Discovery Engine and Cloud Storage Service Accounts the "Cloud KMS CryptoKey Encrypter/Decrypter" IAM role on the provided key.')
+        click.echo('Granting Discovery Engine and Cloud Storage Service Accounts the "Cloud KMS CryptoKey Encrypter/Decrypter" IAM role on the provided key.')
         # Get the current IAM policy
         request = service.projects().locations().keyRings().cryptoKeys().getIamPolicy(resource=kms_key_name)
         policy = request.execute()
@@ -1152,7 +1149,7 @@ def grant_kms_permissions(credentials, kms_key_name, project_number):
         request = service.projects().locations().keyRings().cryptoKeys().setIamPolicy(resource=kms_key_name, body=body)
         request.execute()
 
-        click.echo(f"Successfully granted KMS permissions to Discovery Engine and Cloud Storage Service Accounts.")
+        click.echo("Successfully granted KMS permissions to Discovery Engine and Cloud Storage Service Accounts.")
         return True
 
     except Exception as e:
@@ -1176,8 +1173,8 @@ def configure_cmek(credentials, project_id, kms_key_name):
     )
 
     try:
-        response = request.execute()
-        click.echo(f"CMEK configured successfully.")
+        request.execute()
+        click.echo("CMEK configured successfully.")
         return True
 
     except Exception as e:
@@ -1245,15 +1242,15 @@ def create_engine(credentials, project_id, engine_id, display_name, company_name
 
         # Check if response is an Operation (LRO)
         if 'name' in response and 'operations' in response['name']:
-             click.echo(f"Engine creation initiated. Waiting for Engine to be ready...")
+             click.echo("Engine creation initiated. Waiting for Engine to be ready...")
 
              engine_full_name = f"projects/{project_id}/locations/us/collections/default_collection/engines/{engine_id}"
 
              while True:
                 try:
                     # Poll the Engine resource directly
                     eng_request = service.projects().locations().collections().engines().get(name=engine_full_name)
-                    eng_response = eng_request.execute()
+                    eng_request.execute()
 
                     # If we get here, the engine exists.
                     click.echo("Engine created successfully!")
@@ -1268,7 +1265,7 @@ def create_engine(credentials, project_id, engine_id, display_name, company_name
              click.echo(nl=True)
         else:
              # If it's not an operation or already done (unlikely for create)
-             click.echo(f"Engine created successfully!")
+             click.echo("Engine created successfully!")
 
     except Exception as e:
         click.echo("Received an API error during creation. Checking if engine was created asynchronously despite the error...")
@@ -1280,7 +1277,7 @@ def create_engine(credentials, project_id, engine_id, display_name, company_name
                 eng_request.execute()
                 click.echo("Engine verified successfully! Proceeding with configuration.")
                 return
-            except Exception as inner_e:
+            except Exception:
                 if attempt < max_retries - 1:
                     click.echo(".", nl=False)
                     time.sleep(5)
@@ -1304,12 +1301,6 @@ def configure_idp_for_widget(credentials, project_id, engine_id, workforce_pool_
             f"engines/{engine_id}/widgetConfigs/default_search_widget_config?updateMask=accessSettings"
         )
 
-        headers = {
-            "Authorization": f"Bearer {access_token}",
-            "x-goog-user-project": project_id,
-            "Content-Type": "application/json"
-        }
-
         data = {
             "accessSettings": {
                 "enableWebApp": True,
@@ -1367,12 +1358,6 @@ def disable_user_event_collection(credentials, project_id, engine_id):
             f"engines/{engine_id}/widgetConfigs/default_search_widget_config?updateMask=uiSettings.disableUserEventsCollection"
         )
 
-        headers = {
-            "Authorization": f"Bearer {access_token}",
-            "x-goog-user-project": project_id,
-            "Content-Type": "application/json"
-        }
-
         data = {
             "uiSettings": {
                 "disableUserEventsCollection": True
@@ -1444,7 +1429,7 @@ def configure_gemini_enterprise_for_fedramp_high(credentials, project_id, engine
     )
 
     try:
-        engine_response = engine_request.execute()
+        engine_request.execute()
         click.echo(f"Engine {engine_id} configured for FedRAMP High.")
     except Exception as e:
         click.echo(f"An error occurred while configuring the engine for FedRAMP High: {e}")
@@ -1497,7 +1482,7 @@ def configure_gemini_enterprise_for_fedramp_high(credentials, project_id, engine
             if result.returncode == 0 and "error" not in result.stdout.lower():
                  click.echo(f"Default assistant for engine {engine_id} configured for FedRAMP High.")
             else:
-                 click.echo(f"An error occurred while configuring the default assistant for FedRAMP High:")
+                 click.echo("An error occurred while configuring the default assistant for FedRAMP High:")
                  click.echo(result.stderr)
                  click.echo(result.stdout)
                  # Do not exit
@@ -1557,7 +1542,7 @@ def configure_gemini_enterprise_for_il4(credentials, project_id, engine_id):
     )
 
     try:
-        engine_response = engine_request.execute()
+        engine_request.execute()
         click.echo(f"Engine {engine_id} configured for IL4.")
     except Exception as e:
         click.echo(f"An error occurred while configuring the engine for IL4: {e}")
@@ -1607,7 +1592,7 @@ def configure_gemini_enterprise_for_il4(credentials, project_id, engine_id):
         if result.returncode == 0 and "error" not in result.stdout.lower():
              click.echo(f"Default assistant for engine {engine_id} configured for IL4.")
         else:
-             click.echo(f"An error occurred while configuring the default assistant for IL4:")
+             click.echo("An error occurred while configuring the default assistant for IL4:")
              click.echo(result.stderr)
              click.echo(result.stdout)
              click.echo(click.style("Exiting Onboarding process...", fg="red"))
@@ -1669,7 +1654,7 @@ def configure_gemini_enterprise_for_il5(credentials, project_id, engine_id):
     )
 
     try:
-        engine_response = engine_request.execute()
+        engine_request.execute()
         click.echo(f"Engine {engine_id} configured for IL5.")
     except Exception as e:
         click.echo(f"An error occurred while configuring the engine for IL5: {e}")
@@ -1719,7 +1704,7 @@ def configure_gemini_enterprise_for_il5(credentials, project_id, engine_id):
         if result.returncode == 0 and "error" not in result.stdout.lower():
              click.echo(f"Default assistant for engine {engine_id} configured for IL5.")
         else:
-             click.echo(f"An error occurred while configuring the default assistant for IL5:")
+             click.echo("An error occurred while configuring the default assistant for IL5:")
              click.echo(result.stderr)
              click.echo(result.stdout)
              click.echo(click.style("Exiting Onboarding process...", fg="red"))

blueprints/fedramp-high/gemini-enterprise/gem4gov-cli/setup.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from setuptools import setup, find_packages
+from setuptools import setup
 
 setup(
     name='gem4gov',

pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+addopts = -v