Merge branch 'edge' into EXEC-343-lld-add-cap-move

abr-testing/abr_testing/google_automation/google_drive_tool.py → abr-testing/abr_testing/automation/google_drive_tool.py

@@ -1,6 +1,8 @@
 """Google Drive Tool."""
 import os
-from typing import Set, Any
+from typing import Set, Any, Optional
+import webbrowser
+import mimetypes
 from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
 from googleapiclient.discovery import build
 from googleapiclient.http import MediaFileUpload
@@ -14,15 +16,15 @@
 class google_drive:
     """Google Drive Tool."""
 
-    def __init__(self, credentials: Any, folder_name: str, parent_folder: Any) -> None:
+    def __init__(self, credentials: Any, folder_name: str, email: str) -> None:
         """Connects to google drive via credentials file."""
         self.scope = ["https://www.googleapis.com/auth/drive"]
         self.credentials = ServiceAccountCredentials.from_json_keyfile_name(
             credentials, self.scope
         )
         self.drive_service = build("drive", "v3", credentials=self.credentials)
-        self.folder_name = folder_name
-        self.parent_folder = parent_folder
+        self.parent_folder = folder_name
+        self.email = email
 
     def list_folder(self, delete: Any = False) -> Set[str]:
         """List folders and files in Google Drive."""
@@ -72,26 +74,37 @@ def upload_file(self, file_path: str) -> str:
         """Upload file to Google Drive."""
         file_metadata = {
             "name": os.path.basename(file_path),
-            "mimeType": "application/vnd.google-apps.folder",
-            "parents": [self.parent_folder] if self.parent_folder else "",
+            "mimeType": str(mimetypes.guess_type(file_path)[0]),
+            "parents": [self.parent_folder],
         }
-
         media = MediaFileUpload(file_path, resumable=True)
 
         uploaded_file = (
             self.drive_service.files()
             .create(body=file_metadata, media_body=media, fields="id")  # type: ignore
             .execute()
         )
-
         return uploaded_file["id"]
 
-    def upload_missing_files(self, storage_directory: str, missing_files: set) -> None:
+    def upload_missing_files(self, storage_directory: str) -> None:
         """Upload missing files to Google Drive."""
+        # Read Google Drive .json files.
+        google_drive_files = self.list_folder()
+        google_drive_files_json = [
+            file for file in google_drive_files if file.endswith(".json")
+        ]
+        # Read local directory.
+        local_files_json = set(
+            file for file in os.listdir(storage_directory) if file.endswith(".json")
+        )
+        missing_files = local_files_json - set(google_drive_files_json)
+        print(f"Missing files: {len(missing_files)}")
+        # Upload missing files.
         uploaded_files = []
         for file in missing_files:
             file_path = os.path.join(storage_directory, file)
             uploaded_file_id = google_drive.upload_file(self, file_path)
+            self.share_permissions(uploaded_file_id)
             uploaded_files.append(
                 {"name": os.path.basename(file_path), "id": uploaded_file_id}
             )
@@ -108,3 +121,31 @@ def upload_missing_files(self, storage_directory: str, missing_files: set) -> No
                 print(
                     f"File '{this_name}' was not found in the list of files after uploading."
                 )
+
+    def open_folder(self) -> Optional[str]:
+        """Open folder in web browser."""
+        folder_metadata = (
+            self.drive_service.files()
+            .get(fileId=self.parent_folder, fields="webViewLink")
+            .execute()
+        )
+        folder_link = folder_metadata.get("webViewLink")
+        if folder_link:
+            print(f"Folder link: {folder_link}")
+            webbrowser.open(
+                folder_link
+            )  # Open the folder link in the default web browser
+        else:
+            print("Folder link not found.")
+        return folder_link
+
+    def share_permissions(self, file_id: str) -> None:
+        """Share permissions with self."""
+        new_permission = {
+            "type": "user",
+            "role": "writer",
+            "emailAddress": self.email,
+        }
+        self.drive_service.permissions().create(
+            fileId=file_id, body=new_permission, transferOwnership=False  # type: ignore
+        ).execute()

abr-testing/abr_testing/google_automation/google_sheets_tool.py → abr-testing/abr_testing/automation/google_sheets_tool.py

@@ -2,6 +2,7 @@
 import gspread  # type: ignore[import]
 import socket
 import httplib2
+from datetime import datetime
 from oauth2client.service_account import ServiceAccountCredentials  # type: ignore[import]
 from typing import Dict, List, Any, Set, Tuple
 
@@ -57,6 +58,12 @@ def write_to_row(self, data: List) -> None:
         """Write data into a row in a List[] format."""
         try:
             self.row_index += 1
+            data = [
+                item.strftime("%Y/%m/%d %H:%M:%S")
+                if isinstance(item, datetime)
+                else item
+                for item in data
+            ]
             self.worksheet.insert_row(data, index=self.row_index)
         except socket.gaierror:
             pass

abr-testing/abr_testing/automation/jira_tool.py

@@ -0,0 +1,165 @@
+"""JIRA Ticket Creator."""
+
+import requests
+from requests.auth import HTTPBasicAuth
+import json
+import webbrowser
+import argparse
+from typing import List
+
+
+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: str) -> 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.raise_for_status()
+        try:
+            board_data = response.json()
+            all_issues = board_data["issues"]
+        except json.JSONDecodeError as e:
+            print("Error decoding json: ", e)
+        issue_ids = []
+        for i in all_issues:
+            issue_id = i.get("id")
+            issue_ids.append(issue_id)
+        return issue_ids
+
+    def open_issue(self, issue_key: str) -> str:
+        """Open issue on web browser."""
+        url = f"{self.url}/browse/{issue_key}"
+        print(f"Opening at {url}.")
+        webbrowser.open(url)
+        return url
+
+    def create_ticket(
+        self,
+        summary: str,
+        description: str,
+        project_key: str,
+        reporter_id: str,
+        issue_type: str,
+        priority: str,
+        components: list,
+        affects_versions: str,
+        robot: str,
+    ) -> str:
+        """Create ticket."""
+        data = {
+            "fields": {
+                "project": {"id": "10273", "key": project_key},
+                "issuetype": {"name": issue_type},
+                "summary": summary,
+                "reporter": {"id": reporter_id},
+                "parent": {"key": robot},
+                "priority": {"name": priority},
+                "components": [{"name": component} for component in components],
+                "versions": [{"name": affects_versions}],
+                "description": {
+                    "content": [
+                        {
+                            "content": [{"text": description, "type": "text"}],
+                            "type": "paragraph",
+                        }
+                    ],
+                    "type": "doc",
+                    "version": 1,
+                }
+                # Include other required fields as needed
+            }
+        }
+        try:
+            response = requests.post(
+                f"{self.url}/rest/api/3/issue/",
+                headers=self.headers,
+                auth=self.auth,
+                json=data,
+            )
+            response.raise_for_status()
+            response_str = str(response.content)
+            issue_url = response.json().get("self")
+            issue_key = response.json().get("key")
+            print(f"issue key {issue_key}")
+            print(f"issue url{issue_url}")
+            if issue_key is None:
+                print("Error: Could not create issue. No key returned.")
+        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_key
+
+    def post_attachment_to_ticket(self, issue_id: str, attachment_path: str) -> None:
+        """Adds attachments to ticket."""
+        # 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,
+            )
+            print(response)
+        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.")
+    parser.add_argument(
+        "jira_api_token",
+        metavar="JIRA_API_TOKEN",
+        type=str,
+        nargs=1,
+        help="JIRA API Token. Get from https://id.atlassian.com/manage-profile/security.",
+    )
+    parser.add_argument(
+        "email",
+        metavar="EMAIL",
+        type=str,
+        nargs=1,
+        help="Email connected to JIRA account.",
+    )
+    # TODO: write function to get reporter_id from email.
+    parser.add_argument(
+        "reporter_id",
+        metavar="REPORTER_ID",
+        type=str,
+        nargs=1,
+        help="JIRA Reporter ID.",
+    )
+    # TODO: improve help comment on jira board id.
+    parser.add_argument(
+        "board_id",
+        metavar="BOARD_ID",
+        type=str,
+        nargs=1,
+        help="JIRA Board ID. RABR is 217",
+    )
+    args = parser.parse_args()
+    url = "https://opentrons.atlassian.net"
+    api_token = args.jira_api_token[0]
+    email = args.email[0]
+    board_id = args.board_id[0]
+    reporter_id = args.reporter_id[0]
+    ticket = JiraTicket(url, api_token, email)