fix single quotes usage

Signed-off-by: Javan Lacerda <javanlacerda@google.com>

Author: javanlacerda

src/clusterfuzz/_internal/batch/service.py

@@ -209,7 +209,7 @@ def _get_config_names(batch_tasks: List[types.RemoteTask]):
   config_map = {}
   for task in batch_tasks:
     if task.job_type not in job_map:
-      logs.error(f"{task.job_type} doesn't exist.")
+      logs.error(f'{task.job_type} doesn\'t exist.')
       continue
     if task.command == 'fuzz':
       suffix = '-PREEMPTIBLE-UNPRIVILEGED'

src/clusterfuzz/_internal/datastore/feature_flags.py

@@ -1,4 +1,4 @@
-# Copyright 2024 Google LLC
+# Copyright 2026 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -58,13 +58,13 @@ def description(self):
     """Get a feature flag value."""
     flag = self.flag
     if not flag or flag.description is None:
-      return ""
+      return ''
     return flag.description
 
   @property
   def string_value(self):
     """Get a feature flag value."""
     flag = self.flag
     if not flag or flag.string_value is None:
-      return ""
+      return ''
     return flag.string_value

src/clusterfuzz/_internal/k8s/__init__.py

@@ -1,4 +1,4 @@
-# Copyright 2025 Google LLC
+# Copyright 2026 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.

src/clusterfuzz/_internal/k8s/service.py

@@ -62,7 +62,7 @@ def _get_config_names(remote_tasks: typing.List[types.RemoteTask]):
   config_map = {}
   for task in remote_tasks:
     if task.job_type not in job_map:
-      logs.error(f"{task.job_type} doesn't exist.")
+      logs.error(f'{task.job_type} doesn\'t exist.')
       continue
     if task.command == 'fuzz':
       suffix = '-PREEMPTIBLE-UNPRIVILEGED'
@@ -176,7 +176,7 @@ def _load_gke_credentials(self):
     credentials, _ = google.auth.default()
     project = utils.get_application_id()
     service = discovery.build('container', 'v1', credentials=credentials)
-    parent = f"projects/{project}/locations/-"
+    parent = f'projects/{project}/locations/-'
 
     try:
       # pylint: disable=no-member
@@ -186,12 +186,12 @@ def _load_gke_credentials(self):
       cluster = next((c for c in clusters if c['name'] == CLUSTER_NAME), None)
 
       if not cluster:
-        logs.error(f"Cluster {CLUSTER_NAME} not found in project {project}.")
-        print(f"DEBUG: Cluster {CLUSTER_NAME} not found in project {project}.")
+        logs.error(f'Cluster {CLUSTER_NAME} not found in project {project}.')
+        print(f'DEBUG: Cluster {CLUSTER_NAME} not found in project {project}.')
         return
 
     except Exception as e:
-      logs.error(f"Failed to list clusters in {project}: {e}")
+      logs.error(f'Failed to list clusters in {project}: {e}')
       return
 
     endpoint = cluster['endpoint']
@@ -212,13 +212,13 @@ def get_token(creds):
       request = google_requests.Request()
       if not creds.valid or creds.expired:
         creds.refresh(request)
-      return {"authorization": "Bearer " + creds.token}
+      return {'authorization': 'Bearer ' + creds.token}
 
     configuration.refresh_api_key_hook = lambda _: get_token(credentials)
     configuration.api_key = get_token(credentials)
 
     k8s_client.Configuration.set_default(configuration)
-    logs.info("GKE credentials loaded successfully.")
+    logs.info('GKE credentials loaded successfully.')
 
   def _create_service_account_if_needed(self,
                                         service_account_email: str) -> str:
@@ -264,10 +264,10 @@ def _get_pending_jobs_count(self) -> int:
           namespace='default',
           label_selector='app.kubernetes.io/name=clusterfuzz-kata-job',
           field_selector='status.phase=Pending')
-      logs.info(f"Found {len(pods.items)} pending jobs.")
+      logs.info(f'Found {len(pods.items)} pending jobs.')
       return len(pods.items)
     except Exception as e:
-      logs.error(f"Failed to list pods: {e}")
+      logs.error(f'Failed to list pods: {e}')
       return 0
 
   def create_utask_main_job(self, module: str, job_type: str,
@@ -308,9 +308,9 @@ def create_utask_main_jobs(self, remote_tasks: typing.List[types.RemoteTask]):
       logs.info(f'Scheduling {remote_task.command}, {remote_task.job_type}.')
       config = configs[(remote_task.command, remote_task.job_type)]
       job_specs[config].append(remote_task.input_download_url)
-    logs.info('Creating batch jobs.')
+    logs.info('Creating Kubernetes jobs.')
     jobs = []
-    logs.info('Batching utask_mains.')
+    logs.info('Batching k8s utask_mains.')
     for config, input_urls in job_specs.items():
       # TODO(javanlacerda): Batch multiple tasks into a single job.
       for input_url in input_urls:

src/clusterfuzz/_internal/remote_task/__init__.py

@@ -94,7 +94,7 @@ def get_job_frequency(self):
 
       if total_weight >= 1.0 and weight > 0.0:
         logs.warning(
-            "Total weight for jobs frequency bigger than 1.0. Adapter starving",
+            'Total weight for jobs frequency bigger than 1.0. Adapter starving',
             adapter=adapter.id)
         break
 
@@ -106,7 +106,7 @@ def get_job_frequency(self):
       total_weight += weight
       frequencies[adapter.id] = weight if weight >= 0 else 0.0
 
-    logs.info("Job frequencies", frequencies=frequencies)
+    logs.info('Job frequencies', frequencies=frequencies)
     return frequencies
 
   def create_utask_main_job(self, module: str, job_type: str,

src/clusterfuzz/_internal/tests/core/k8s/k8s_integration_test.py

@@ -75,23 +75,23 @@ def test_load_credentials(self, mock_discovery_build, mock_auth_default,
 
         # Verify configuration
         config = client.Configuration.get_default_copy()
-        print(f"Loaded Host: {config.host}")
+        print(f'Loaded Host: {config.host}')
 
         # Check that we got a valid https endpoint
-        self.assertTrue(config.host.startswith("https://"))
+        self.assertTrue(config.host.startswith('https://'))
         self.assertTrue(config.verify_ssl)
         self.assertIsNotNone(config.ssl_ca_cert)
 
         # Verify API key fix is present (Crucial for manual path)
-        self.assertIn("authorization", config.api_key)
+        self.assertIn('authorization', config.api_key)
 
         # Verify hook is present
         self.assertIsNotNone(config.refresh_api_key_hook)
 
         # Verify actual connectivity and auth
-        print("Attempting to list jobs to verify authentication...")
+        print('Attempting to list jobs to verify authentication...')
         k8s_service._batch_api.list_namespaced_job(namespace='default', limit=1)
-        print("Successfully listed jobs.")
+        print('Successfully listed jobs.')
 
       finally:
         if old_kubeconfig:

src/clusterfuzz/_internal/tests/core/k8s/k8s_service_e2e_test.py

@@ -213,7 +213,7 @@ def test_create_job(self, mock_get_logging_config_dict):
 
     self.assertTrue(
         job_running,
-        f"Job {actual_job_name} did not start running. Status: {job.status}")
+        f'Job {actual_job_name} did not start running. Status: {job.status}')
 
     self.api_client.delete_namespaced_job(
         name=actual_job_name,
@@ -246,7 +246,7 @@ def test_create_kata_container_job(self, mock_get_logging_config_dict):
 
     self.assertTrue(
         job_running,
-        f"Kata Job {actual_job_name} did not start running. Status: {job.status}"
+        f'Kata Job {actual_job_name} did not start running. Status: {job.status}'
     )
 
     self.api_client.delete_namespaced_job(
@@ -294,7 +294,7 @@ def test_create_uworker_main_batch_job(self, mock_get_command_from_module,
 
     self.assertTrue(
         job_running,
-        f"Job {actual_job_name} did not start running. Status: {job.status}")
+        f'Job {actual_job_name} did not start running. Status: {job.status}')
 
     self.api_client.delete_namespaced_job(
         name=actual_job_name,
@@ -359,7 +359,7 @@ def test_create_uworker_main_batch_jobs(self, mock_get_command_from_module,
 
       self.assertTrue(
           job_running,
-          f"Job {job_name} did not start running. Status: {job.status}")
+          f'Job {job_name} did not start running. Status: {job.status}')
 
       self.api_client.delete_namespaced_job(
           name=job_name,