Handle guest logs

Fetch log content,save it to provided location or current dir

Author: skycastlelily

tmt/steps/finish/__init__.py

@@ -189,6 +189,7 @@ def go(self, force: bool = False) -> None:
 
         # Stop and remove provisioned guests
         for guest in self.plan.provision.guests():
+            guest.handle_guest_logs()
             guest.stop()
             guest.remove()
 

tmt/steps/provision/__init__.py

@@ -1103,6 +1103,22 @@ def requires(cls) -> list['tmt.base.Dependency']:
         """ All requirements of the guest implementation """
         return []
 
+    def acquire_log(self, log_name: str) -> Optional[str]:
+        """fetch and return content of a requested log"""
+        raise NotImplementedError
+
+    def store_log(
+            self,
+            log_name: Optional[str],
+            log_path: Optional[Path],
+            log_content: str) -> None:
+        """save log content and return the path"""
+        raise NotImplementedError
+
+    def handle_guest_logs(self, log_list: Optional[list[str]] = None) -> None:
+        """get log content and save it to the workdir/provision/,
+        list the log dir in guests.yaml"""
+
 
 @dataclasses.dataclass
 class GuestSshData(GuestData):

tmt/steps/provision/local.py

@@ -133,6 +133,43 @@ def pull(
             extend_options: Optional[list[str]] = None) -> None:
         """ Nothing to be done to pull workdir """
 
+    def acquire_log(self, log_name: str) -> Optional[str]:
+        """fetch and return content of a requested log"""
+        if log_name == 'dmesg':
+            return self.execute(Command('dmesg')).stdout
+        return None
+
+    def store_log(
+            self,
+            log_name: Optional[str],
+            log_path: Optional[Path],
+            log_content: str) -> None:
+        """save log content and return the path"""
+        if log_path and log_name:
+            # if log_path contains log file name
+            if log_path.name == log_name:
+                log_path.write_text(log_content)
+            else:
+                (log_path / log_name).write_text(log_content)
+        elif log_path and not log_name:
+            log_path.write_text(log_content)
+        elif log_name and not log_path:
+            self.info('no log_path provided,store log to current dir')
+            (Path.cwd() / log_name).write_text(log_content)
+        else:
+            self.warn('log_name and log_path cannot both be None')
+
+    def handle_guest_logs(self, logs: Optional[list[str]] = None) -> None:
+        """get log content and save it to the workdir/provision/,
+        list the log dir in guests.yaml"""
+        logs = logs or []
+        for log_name in logs:
+            log_content = self.acquire_log(log_name)
+            if log_content:
+                self.store_log(log_name, self.workdir, log_content)
+            else:
+                self.debug(f'no content in {log_name}')
+
 
 @tmt.steps.provides_method('local')
 class ProvisionLocal(tmt.steps.provision.ProvisionPlugin[ProvisionLocalData]):