resolve unsoundness caused by pytype.

PiperOrigin-RevId: 840944874

Author: p3rf Team

perfkitbenchmarker/archive.py

@@ -69,9 +69,10 @@ def ArchiveRun(
   )
   p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
 
-  with p.stdin:
-    with tarfile.open(mode='w:gz', fileobj=p.stdin) as tar:
-      tar.add(run_temp_directory, os.path.basename(run_temp_directory))
+  if p.stdin is not None:
+    with p.stdin:
+      with tarfile.open(mode='w:gz', fileobj=p.stdin) as tar:
+        tar.add(run_temp_directory, os.path.basename(run_temp_directory))
 
   status = p.wait()
   if status:

perfkitbenchmarker/dpb_service.py

@@ -549,6 +549,10 @@ def _InitializeMetadata(self) -> None:
   def _CreateDependencies(self):
     """Creates a bucket to use with the cluster."""
     if self.manage_bucket:
+      if self.storage_service is None:
+        raise ValueError(
+            'storage_service is None. Initialize before use.'
+        )
       self.storage_service.MakeBucket(self.bucket)
 
   def _Create(self):
@@ -558,6 +562,10 @@ def _Create(self):
   def _DeleteDependencies(self):
     """Deletes the bucket used with the cluster."""
     if self.manage_bucket:
+      if self.storage_service is None:
+        raise ValueError(
+            'storage_service is None. Initialize before use.'
+        )
       self.storage_service.DeleteBucket(self.bucket)
 
   def _Delete(self):
@@ -932,6 +940,10 @@ def SubmitJob(
       cmd_list += job_arguments
     cmd_string = ' '.join(cmd_list)
 
+    if self.leader is None:
+      raise JobSubmissionError(
+          'Cannot submit job as leader VM is not initialized.'
+      )
     start_time = datetime.datetime.now()
     try:
       stdout, stderr = self.leader.RobustRemoteCommand(cmd_string)
@@ -1027,6 +1039,10 @@ def SubmitJob(
         job_type=job_type,
         properties=properties,
     )
+    if self.leader is None:
+      raise JobSubmissionError(
+          'Cannot submit job as leader VM is not initialized.'
+      )
     start_time = datetime.datetime.now()
     try:
       stdout, _ = self.leader.RobustRemoteCommand(' '.join(cmd))

perfkitbenchmarker/providers/alicloud/ali_network.py

@@ -346,16 +346,22 @@ def Create(self):
         self.security_group = AliSecurityGroup(
             self.name, self.region, use_vpc=True, vpc_id=self.vpc.id
         )
-      self.security_group.Create()
+      if self.security_group is not None:
+        self.security_group.Create()
     else:
-      self.security_group.Create()
+      if self.security_group is not None:
+        self.security_group.Create()
 
   def Delete(self):
     """Deletes the network."""
     if self.use_vpc:
-      self.security_group.Delete()
-      self.vswitch.Delete()
-      self.security_group.Delete()
+      if self.security_group is not None:
+        self.security_group.Delete()
+      if self.vswitch is not None:
+        self.vswitch.Delete()
+      if self.security_group is not None:
+        self.security_group.Delete()
       self.vpc.Delete()
     else:
-      self.security_group.Delete()
+      if self.security_group is not None:
+        self.security_group.Delete()

perfkitbenchmarker/scripts/messaging_service_scripts/common/e2e/latency_runner.py

@@ -143,6 +143,9 @@ def run_phase(self, number_of_messages: int, message_size: int):
     return asyncio.run(self._async_run_phase(number_of_messages))
 
   async def _wait_until_received(self, seq: int) -> protocol.ReceptionReport:
+    assert self._receiver is not None, (
+        f'Receiver is not initialized. Message sequence: {seq}.'
+    )
     while True:
       report = await self._receiver.receive()
       self._record_message_reception(report)

perfkitbenchmarker/scripts/messaging_service_scripts/common/e2e/main_process.py

@@ -62,6 +62,7 @@ async def stop(self, timeout: Optional[float] = None):
     Args:
       timeout: Timeout in seconds. Defaults to BaseWorker.DEFAULT_TIMEOUT.
     """
+    assert self.subprocess is not None, 'stop() called before start()'
     try:
       self.subprocess.terminate()
       await self._join_subprocess(timeout)
@@ -89,7 +90,7 @@ async def _read_subprocess_output(
   async def _join_subprocess(self, timeout=None):
     timeout = self.DEFAULT_TIMEOUT if timeout is None else timeout
     deadline = time.time() + timeout
-    while self.subprocess.exitcode is None:
+    while self.subprocess and self.subprocess.exitcode is None:
       await asyncio.sleep(self.SLEEP_TIME)
       if time.time() > deadline:
         raise errors.EndToEnd.SubprocessTimeoutError