Fix suite data parsing

github_scripts/suite_data.py

@@ -8,14 +8,17 @@
 Class containing helper methods for gathering data needed for a SuiteReport object
 """
 
-import re
+import ast
 import os
+import re
 import sqlite3
 import subprocess
-import yaml
 from collections import defaultdict
 from pathlib import Path
-from typing import Dict, List, Optional, Set, Union
+from typing import Optional, Union
+
+import yaml
+
 from git_bdiff import GitBDiff, GitInfo
 
 
@@ -41,7 +44,7 @@ def __init__(self, suite_path: Path = None) -> None:
         self.source_root = suite_path / "share" / "source"
         self.task_states = {}
 
-    def get_um_failed_configs(self) -> Set[str]:
+    def get_um_failed_configs(self) -> set[str]:
         """
         Read through failed UM rose_ana tasks
         """
@@ -74,7 +77,7 @@ def read_um_section(self, change: str) -> str:
             section = "Unknown"
         return section
 
-    def get_changed_um_section(self) -> Set[str]:
+    def get_changed_um_section(self) -> set[str]:
         """
         Read through bdiff of UM source and find code owner section for each changed
         file
@@ -111,7 +114,7 @@ def get_changed_um_section(self) -> Set[str]:
 
         return changed_sections
 
-    def get_um_owners(self, filename: str) -> Dict:
+    def get_um_owners(self, filename: str) -> dict:
         """
         Read UM Code Owners file and write to a dictionary
         """
@@ -149,7 +152,7 @@ def get_um_owners(self, filename: str) -> Dict:
 
         return owners
 
-    def parse_tasks(self) -> Dict[str, List[str]]:
+    def parse_tasks(self) -> dict[str, list[str]]:
         """
         Read through the tasks run, sorting by state
         """
@@ -217,7 +220,7 @@ def determine_primary_source(self) -> str:
 
         return "unknown"
 
-    def read_rose_conf(self) -> Dict[str, str]:
+    def read_rose_conf(self) -> dict[str, str]:
         """
         Read the suite rose-suite.conf file into a dictionary
         """
@@ -271,7 +274,7 @@ def find_unknown_dependency(self, dependency: str) -> str:
                     return match.group(1).rstrip("/")
         raise RuntimeError(f"Unable to find source for dependency {dependency}")
 
-    def read_dependencies(self) -> Dict[str, Dict]:
+    def read_dependencies(self) -> dict[str, dict]:
         """
         Read the suite dependencies from the dependencies.yaml file - this is assumed to
         have been copied to the suite_path directory
@@ -328,7 +331,7 @@ def get_suite_starttime(self) -> str:
             starttime = row[0]
         return starttime.split("+")[0]
 
-    def read_groups_run(self) -> List[str]:
+    def read_groups_run(self) -> list[str]:
         """
         Read in groups run as part of suite from the cylc database file
         """
@@ -338,13 +341,15 @@ def read_groups_run(self) -> List[str]:
             self.suite_path / "log" / "db", ["key", "value"], "workflow_template_vars"
         ):
             if row[0] in ("g", "group"):
-                groups = row[1].strip("[]'\"").split(",")
+                groups = ast.literal_eval(row[1])
+                # optionally strip whitespaces from elements
+                groups = [item.strip() for item in groups]
                 break
         if groups is None:
             groups = ["suite_default"]
         return groups
 
-    def get_task_states(self) -> Dict[str, str]:
+    def get_task_states(self) -> dict[str, str]:
         """
         Query the database and return a dictionary of states. This is assumed to be in
         suite_path/log/db
@@ -358,8 +363,8 @@ def get_task_states(self) -> Dict[str, str]:
         return data
 
     def query_suite_database(
-        self, database: Path, selections: List[str], source: str
-    ) -> List[tuple]:
+        self, database: Path, selections: list[str], source: str
+    ) -> list[tuple]:
         """
         Create an sql statement and query provided database. Return the result
         """
@@ -376,7 +381,7 @@ def query_suite_database(
         return data
 
     def run_command(
-        self, command: Union[str, List[str]], shell: bool = False, rval: bool = False
+        self, command: Union[str, list[str]], shell: bool = False, rval: bool = False
     ) -> Optional[subprocess.CompletedProcess]:
         """
         Run a subprocess command and return the result object

github_scripts/suite_report_git.py

@@ -16,12 +16,11 @@
 from collections import defaultdict
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Dict, List, Set, Tuple
 
 from suite_data import SuiteData
 
 
-def create_markdown_row(*columns: str, header=False) -> List[str]:
+def create_markdown_row(*columns: str, header=False) -> list[str]:
     """
     Join any number of columns into a markdown formatted table row
     Will attempt to format column entries as str
@@ -73,16 +72,16 @@ def __init__(self, suite_path: Path) -> None:
         self.suite_starttime: str = self.get_suite_starttime()
         self.workflow_id: str = self.get_workflow_id()
         self.cylc_url: str = self.generate_cylc_url()
-        self.task_states: Dict[str, str] = self.get_task_states()
-        self.groups: List[str] = self.read_groups_run()
-        self.rose_data: Dict[str, str] = self.read_rose_conf()
-        self.dependencies: Dict[str, Dict] = self.read_dependencies()
+        self.task_states: dict[str, str] = self.get_task_states()
+        self.groups: list[str] = self.read_groups_run()
+        self.rose_data: dict[str, str] = self.read_rose_conf()
+        self.dependencies: dict[str, dict] = self.read_dependencies()
         self.primary_source: str = self.determine_primary_source()
         self.populate_gitinfo()
         self.populate_gitbdiff()
         self.trac_log = []
 
-    def parse_local_source(self, source: str) -> Tuple[str, str]:
+    def parse_local_source(self, source: str) -> tuple[str, str]:
         """
         Find the branch name or hash and remote reference for a given source
         """
@@ -162,7 +161,7 @@ def create_dependency_table(self) -> None:
 
         self.trac_log.append("")
 
-    def create_task_tables(self, parsed_tasks: Dict[str, List[str]]) -> None:
+    def create_task_tables(self, parsed_tasks: dict[str, list[str]]) -> None:
         """
         Create tables containing summary of task states and number of tasks won
         """
@@ -200,12 +199,12 @@ def create_task_tables(self, parsed_tasks: Dict[str, List[str]]) -> None:
                 self.trac_log.extend(create_markdown_row(task, state))
             self.trac_log.append(self.close_collapsed)
 
-    def create_um_code_owner_table(self, owners: Dict) -> None:
+    def create_um_code_owner_table(self, owners: dict) -> None:
         """
         Create a table of required UM code owner approvals
         """
 
-        changed_sections: Set[str] = self.get_changed_um_section()
+        changed_sections: set[str] = self.get_changed_um_section()
         if changed_sections:
             self.trac_log.extend(
                 create_markdown_row("Section", "Owner", "Deputy", "State", header=True)
@@ -218,12 +217,12 @@ def create_um_code_owner_table(self, owners: Dict) -> None:
         else:
             self.trac_log.append("* No UM Code Owners Required")
 
-    def create_um_config_owner_table(self, owners: Dict) -> None:
+    def create_um_config_owner_table(self, owners: dict) -> None:
         """
         Create a table of required UM config owner approvals
         """
 
-        failed_configs: Set[str] = self.get_um_failed_configs()
+        failed_configs: set[str] = self.get_um_failed_configs()
         if not failed_configs:
             self.trac_log.append("No UM Config Owners Required")
             return