format and lint changes

Author: rclarke0

abr-testing/abr_testing/automation/jira_tool.py

@@ -5,9 +5,10 @@
 import json
 import webbrowser
 import argparse
-from typing import Dict, List, Tuple
+from typing import Dict, List, Tuple, Any
 from abr_testing.data_collection import read_robot_logs, abr_google_drive, get_run_logs
 
+
 def get_error_runs_from_robot(ip: str) -> List[str]:
     """Get runs that have errors from robot."""
     error_run_ids = []
@@ -22,23 +23,28 @@ def get_error_runs_from_robot(ip: str) -> List[str]:
         if not run["current"] and num_of_errors > 0:
             error_run_ids.append(run_id)
     return error_run_ids
-    
-def get_error_info_from_robot(ip: str, one_run: str, storage_directory: str)-> Tuple[str, str, str, List[str], Dict[str, str]] :
+
+
+def get_error_info_from_robot(
+    ip: str, one_run: str, storage_directory: str
+) -> Tuple[str, str, str, List[str], str, str]:
     """Get error information from robot to fill out ticket."""
     description = dict()
     # get run information
     results = get_run_logs.get_run_data(one_run, ip)
     # save run information to local directory as .json file
-    saved_file_path = read_robot_logs.save_run_log_to_json(ip, results, storage_directory)
-    
+    saved_file_path = read_robot_logs.save_run_log_to_json(
+        ip, results, storage_directory
+    )
+
     # Error Printout
     (
-    num_of_errors,
-    error_type,
-    error_code,
-    error_instrument,
-    error_level,
-        ) = read_robot_logs.get_error_info(results)
+        num_of_errors,
+        error_type,
+        error_code,
+        error_instrument,
+        error_level,
+    ) = read_robot_logs.get_error_info(results)
     # JIRA Ticket Fields
     failure_level = "Level " + str(error_level) + " Failure"
     components = [failure_level, "Flex-RABR"]
@@ -47,35 +53,54 @@ def get_error_info_from_robot(ip: str, one_run: str, storage_directory: str)-> T
     print(parent)
     summary = parent + "_" + str(one_run) + "_" + str(error_code) + "_" + error_type
     # Description of error
-    description["protocol_name"] = results["protocol"]["metadata"].get("protocolName", "")
+    description["protocol_name"] = results["protocol"]["metadata"].get(
+        "protocolName", ""
+    )
     description["error"] = " ".join([error_code, error_type, error_instrument])
     description["protocol_step"] = list(results["commands"])[-1]
     description["right_mount"] = results.get("right", "No attachment")
     description["left_mount"] = results.get("left", "No attachment")
     description["gripper"] = results.get("extension", "No attachment")
     all_modules = abr_google_drive.get_modules(results)
     whole_description = {**description, **all_modules}
-    whole_description_str = ("{" + "\n".join("{!r}: {!r},".format(k, v) for k, v in whole_description.items()) + "}")
-    
-    return summary, parent, affects_version, components, whole_description_str, saved_file_path
+    whole_description_str = (
+        "{"
+        + "\n".join("{!r}: {!r},".format(k, v) for k, v in whole_description.items())
+        + "}"
+    )
+
+    return (
+        summary,
+        parent,
+        affects_version,
+        components,
+        whole_description_str,
+        saved_file_path,
+    )
+
 
 class JiraTicket:
     """Connects to JIRA ticket site."""
+
     def __init__(self, url: str, api_token: str, email: str) -> None:
         """Connect to jira."""
-        
+
         self.url = url
         self.api_token = api_token
         self.email = email
         self.auth = HTTPBasicAuth(email, api_token)
         self.headers = {
             "Accept": "application/json",
-            "Content-Type": "application/json"
-            }
-    
-    def issues_on_board(self, board_id) -> Dict[str, any]:
+            "Content-Type": "application/json",
+        }
+
+    def issues_on_board(self, board_id) -> List[str]:
         """Print Issues on board."""
-        response = requests.get(f"{self.url}/rest/agile/1.0/board/{board_id}/issue", headers=self.headers, auth=self.auth)
+        response = requests.get(
+            f"{self.url}/rest/agile/1.0/board/{board_id}/issue",
+            headers=self.headers,
+            auth=self.auth,
+        )
         response.raise_for_status()
         try:
             board_data = response.json()
@@ -87,41 +112,47 @@ def issues_on_board(self, board_id) -> Dict[str, any]:
             issue_id = i.get("id")
             issue_ids.append(issue_id)
         return issue_ids
-    
-    def open_issue(self, issue_key)-> None:
+
+    def open_issue(self, issue_key) -> None:
         """Open issue on web browser."""
         url = f"{self.url}/browse/{issue_key}"
         webbrowser.open(url)
-        
-            
-    def create_ticket(self, summary: str, description: str, project_key: str, issue_type: str, priority: str, components: list, affects_versions: str, robot: str) -> str:
+
+    def create_ticket(
+        self,
+        summary: str,
+        description: str,
+        project_key: str,
+        issue_type: str,
+        priority: str,
+        components: list,
+        affects_versions: str,
+        robot: str,
+    ) -> Tuple[str, str]:
         """Create ticket."""
         data = {
-        "fields": {
-            "project": {"id": "10273", "key": project_key},
-            "issuetype": {"name": issue_type},
-            "summary": summary,
-            "reporter": {"accountId": "712020:f32d03f8-f91a-465d-871b-45135b7b955f"},
-            "parent": {"key": robot},
-            "priority": {"name": priority},
-            "components": [{"name": component} for component in components],
-            "versions": [{"name": affects_versions}],
-            "description": {
-                "content": [
-                    {
+            "fields": {
+                "project": {"id": "10273", "key": project_key},
+                "issuetype": {"name": issue_type},
+                "summary": summary,
+                "reporter": {
+                    "accountId": "712020:f32d03f8-f91a-465d-871b-45135b7b955f"
+                },
+                "parent": {"key": robot},
+                "priority": {"name": priority},
+                "components": [{"name": component} for component in components],
+                "versions": [{"name": affects_versions}],
+                "description": {
                     "content": [
                         {
-                        "text": description,
-                        "type": "text"
+                            "content": [{"text": description, "type": "text"}],
+                            "type": "paragraph",
                         }
                     ],
-                    "type": "paragraph"
-                    }
-                ],
-                "type": "doc",
-                "version": 1
+                    "type": "doc",
+                    "version": 1,
                 }
-            # Include other required fields as needed
+                # Include other required fields as needed
             }
         }
         try:
@@ -132,36 +163,35 @@ def create_ticket(self, summary: str, description: str, project_key: str, issue_
                 json=data,
             )
             response.raise_for_status()
+            response_str = str(response.content)
             issue_url = response.json().get("self")
             issue_key = response.json().get("key")
             if issue_key is None:
                 print("Error: Could not create issue. No key returned.")
-        except requests.exceptions.HTTPError as e:
-            print(f"HTTP error occurred: {e}")
-            print(f"Response content: {response.content}")
-        except json.JSONDecodeError as e:
-            print(f"JSON decoding error occurred: {e}")
-            print(f"Response content: {response.content}")
-        except Exception as e:
-            print(f"An unexpected error occurred: {e}")
+        except requests.exceptions.HTTPError:
+            print(f"HTTP error occurred. Response content: {response_str}")
+        except json.JSONDecodeError:
+            print(f"JSON decoding error occurred. Response content: {response_str}")
         return issue_url, issue_key
-    
+
     def post_attachment_to_ticket(self, issue_id: str, attachment_path: str):
         """Adds attachments to ticket."""
-        file = {"file": open(attachment_path, 'rb')}
+        # TODO: Ensure that file is actually uploaded.
+        file = {"file": open(attachment_path, "rb")}
         JSON_headers = {"Accept": "application/json"}
         try:
             response = requests.post(
                 f"{self.url}/rest/api/3/issue/{issue_id}/attachments",
-                headers = JSON_headers,
-                auth = self.auth,
-                files = file
-                )
+                headers=JSON_headers,
+                auth=self.auth,
+                files=file,
+            )
             print(response)
-        except json.JSONDecodeError as e:
-            print(f"JSON decoding error occurred: {e}")
-            print(f"Response content: {response.content}")
-        
+        except json.JSONDecodeError:
+            error_message = str(response.content)
+            print(f"JSON decoding error occurred. Response content: {error_message}.")
+
+
 if __name__ == "__main__":
     """Create ticket for specified robot."""
     parser = argparse.ArgumentParser(description="Pulls run logs from ABR robots.")
@@ -173,11 +203,11 @@ def post_attachment_to_ticket(self, issue_id: str, attachment_path: str):
         help="Path to long term storage directory for run logs.",
     )
     parser.add_argument(
-        "robot_ip", 
+        "robot_ip",
         metavar="ROBOT_IP",
-        type = str,
-        nargs = 1,
-        help = "IP address of robot as string."
+        type=str,
+        nargs=1,
+        help="IP address of robot as string.",
     )
     args = parser.parse_args()
     storage_directory = args.storage_directory[0]
@@ -186,14 +216,30 @@ def post_attachment_to_ticket(self, issue_id: str, attachment_path: str):
     api_token = "ATATT3xFfGF0SZgHzGog6J_bTK3j6HFRnbUC-tE_gB5Sn_56d-NSdlRb5C-ywGxNgV4fHfSGbhgENlGKI1aSPc5dy8ql0EiVYW4dHifJhyKUueu5yq6BrL9Xcsfwz_p1W6xUmSsyecvZ0lGPFLb5sreqRg5kE3FRs6SqcKaQxH6_EGFGS0Obb5c=BA8756FE"
     email = "[email protected]"
     board_id = "217"
-    
+
     ticket = JiraTicket(url, api_token, email)
     error_runs = get_error_runs_from_robot(ip)
     one_run = error_runs[0]
     print(one_run)
-    summary, robot, affects_version, components, whole_description_str, saved_file_path = get_error_info_from_robot(ip, one_run, storage_directory)
+    (
+        summary,
+        robot,
+        affects_version,
+        components,
+        whole_description_str,
+        saved_file_path,
+    ) = get_error_info_from_robot(ip, one_run, storage_directory)
     project_key = "RABR"
     parent_key = project_key + "-" + robot[-1]
-    issue_url, issue_key = ticket.create_ticket(summary, whole_description_str, project_key, "Bug", "Medium", components, affects_version, parent_key)
+    issue_url, issue_key = ticket.create_ticket(
+        summary,
+        whole_description_str,
+        project_key,
+        "Bug",
+        "Medium",
+        components,
+        affects_version,
+        parent_key,
+    )
     ticket.open_issue(issue_key)
     ticket.post_attachment_to_ticket(issue_key, saved_file_path)

abr-testing/abr_testing/data_collection/get_run_logs.py

@@ -23,6 +23,7 @@ def get_run_ids_from_robot(ip: str) -> Set[str]:
             run_ids.add(run_id)
     return run_ids
 
+
 def get_run_data(one_run: Any, ip: str) -> Dict[str, Any]:
     """Use http requests to get command, health, and protocol data from robot."""
     response = requests.get(
@@ -79,9 +80,9 @@ def save_runs(runs_to_save: Set[str], ip: str, storage_directory: str) -> Set[st
     saved_file_paths = set()
     for a_run in runs_to_save:
         data = get_run_data(a_run, ip)
-        data_file_name = ip + "_" + data["run_id"] + ".json"
-        saved_file_path = os.path.join(storage_directory, data_file_name)
-        json.dump(data, open(saved_file_path, mode="w"))
+        saved_file_path = read_robot_logs.save_run_log_to_json(
+            ip, data, storage_directory
+        )
         saved_file_paths.add(saved_file_path)
     print(f"Saved {len(runs_to_save)} run(s) from robot with IP address {ip}.")
     return saved_file_paths

abr-testing/abr_testing/data_collection/read_robot_logs.py

@@ -137,13 +137,17 @@ def get_unseen_run_ids(runs: Set[str], runs_from_storage: Set[str]) -> Set[str]:
     print(f"There are {str(len(runs_to_save))} new run(s) to save.")
     return runs_to_save
 
-def save_run_log_to_json(ip:str, results: Dict[str, Any], storage_directory: str)-> str:
+
+def save_run_log_to_json(
+    ip: str, results: Dict[str, Any], storage_directory: str
+) -> str:
     """Save run log to local json file."""
     data_file_name = ip + "_" + results["run_id"] + ".json"
     saved_file_path = os.path.join(storage_directory, data_file_name)
-    json.dump(results, open(saved_file_path, mode = "w"))
+    json.dump(results, open(saved_file_path, mode="w"))
     return saved_file_path
 
+
 def get_run_ids_from_google_drive(google_drive: Any) -> Set[str]:
     """Get run ids in google drive folder."""
     # Run ids in google_drive_folder

abr-testing/abr_testing/tools/abr_asair_sensor.py

@@ -6,7 +6,7 @@
 import time as t
 from typing import List
 import argparse
-from abr_testing.google_automation import google_sheets_tool
+from abr_testing.automation import google_sheets_tool
 
 
 class _ABRAsairSensor: