# -*- coding: utf-8 -*- """ Created on 2024-04-12 :author: Philipp Schneider """ __copyright__ = "Copyright © by tracetronic GmbH, Dresden" __license__ = "This file is distributed as an integral part of tracetronic's software products " \ "and may only be used in connection with and pursuant to the terms and conditions " \ "of a valid tracetronic software product license." from datetime import datetime import json import logging import os import requests import time logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO) logging.basicConfig(level=logging.INFO) class TGReportArchivist: TIME_ZULU_PATTERN = '%Y-%m-%dT%H:%M:%SZ' TIME_UTC_OFFSET_PATTERN = '%Y-%m-%dT%H:%M:%S%z' REQUESTS_VERIFY = True def __init__(self, url: str, authKey: str, projectId: int, archivFolderPath: str, useExportStorage: bool): """ Constructor :param url: test.guide URL e.g.: https://your.test-guide.instance:8443 :param authKey: test.guide upload authentication key :param projectId: Id of the test.guide project :param archivFolderPath: Directory in which the exported reports are to be stored, structured by date. :param useExportStorage: For larger resulting export files, using the configured export/import storage is necessary. """ self.__url = url.strip("/") self.__defaultAuthPayLoad = {'projectId': projectId, 'useExportStorage': useExportStorage} self.__archivFolderPath = archivFolderPath self.__authKey = authKey def StartArchiving(self, startDate: datetime, endDate: datetime): """ Starts archiving in the specified date range, i.e. all reports with all attachments are exported and saved in the specified archive directory. If a report has been successfully exported to the archive, it is also deleted from test.guide. :param startDate: Start date of the filter range to be archived. :param endDate: End date of the filter range to be archived. """ startDateZuluTime = startDate.strftime(self.TIME_ZULU_PATTERN) endDateZuluTime = endDate.strftime(self.TIME_ZULU_PATTERN) logging.info(f'Start archiving from {startDateZuluTime} to {endDateZuluTime}') result = self.__FilterTimeRange(startDateZuluTime, endDateZuluTime) logging.info(f'Start export of {len(result)} reports') for eachReportId in result: startLabel = f"### Archive report id {eachReportId} ###" logging.info(startLabel) logging.info(f'\tStart export of report id {eachReportId}') self.__ExportReportToArchive(eachReportId) logging.info(f'\tStart Deletion of report id {eachReportId}') self.__RemoveReport(eachReportId) endLabel = ''.rjust(len(startLabel), '#') logging.info(endLabel) def __FilterTimeRange(self, startDate: str, endDate: str) -> set[int]: resultSetOfReportIds = set() filterUrl = f"{self.__url}/api/report/testCaseExecutions/filter" logging.info(f'\tFilter time range on {filterUrl}') headers = { 'Content-type': 'application/json', 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } filterContent = { "dateFrom": f"{startDate}", "dateTo": f"{endDate}", } requestLimit = 100 payload = self.__defaultAuthPayLoad.copy() payload['limit'] = requestLimit numberOfItemCounter = 0 while True: payload['offset'] = numberOfItemCounter response = requests.post(filterUrl, params=payload, data=json.dumps(filterContent), headers=headers, verify=self.REQUESTS_VERIFY) response.raise_for_status() resultJson = response.json() if not resultJson: # No more results break for each in resultJson: resultSetOfReportIds.add(each.get("reportId")) logging.debug(f'\t\tFiltering {len(resultSetOfReportIds)} reports...') numberOfItemCounter += requestLimit logging.info(f'\tFilter found {len(resultSetOfReportIds)} reports') return resultSetOfReportIds def __ExportReportToArchive(self, reportId: int): headers = { 'Content-type': 'application/json', 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } exportUrl = f"{self.__url}/api/report/export/atx/{reportId}" response = requests.post(exportUrl, params=self.__defaultAuthPayLoad, headers=headers, verify=self.REQUESTS_VERIFY) response.raise_for_status() taskId = response.json().get("taskId") logging.info(f'\tExport of report id {reportId} is being processed under task id {taskId}.') reportExceInfo = self.__GetExecutionInfo(reportId) self.__WaitUntilExportIsFinished(taskId) self.__DownloadExportReport(reportId, reportExceInfo, taskId) def __GetExecutionInfo(self, reportId: int) -> tuple[str, datetime]: headers = { 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } endpoint = f"{self.__url}/api/report/reports/{reportId}" response = requests.get(endpoint, headers=headers, params=self.__defaultAuthPayLoad, verify=self.REQUESTS_VERIFY) response.raise_for_status() firstTestCaseExcutionUrl = response.json()[0]['href'] response = requests.get(firstTestCaseExcutionUrl, headers=headers, params=self.__defaultAuthPayLoad, verify=self.REQUESTS_VERIFY) response.raise_for_status() firstTestCaseExecDate = response.json()['executionTimestamp'] testSuiteName = response.json()['testSuiteName'] return (testSuiteName, datetime.strptime(firstTestCaseExecDate, self.TIME_UTC_OFFSET_PATTERN)) def __WaitUntilExportIsFinished(self, taskId): headers = { 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } endpoint = f"{self.__url}/api/report/export/atx/{taskId}/status" while True: response = requests.get(endpoint, headers=headers, params=self.__defaultAuthPayLoad, verify=self.REQUESTS_VERIFY) response.raise_for_status() jsonResponse = response.json() exportStatus = jsonResponse.get("done") if exportStatus: logging.info(f'\tExport task {taskId} is finished') return True else: logging.debug(f'\tExport task is not done yet....') time.sleep(5) def __DownloadExportReport(self, reportId: int, reportExceInfo: tuple[str, datetime], taskId: str): headers = { 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } testSuiteName = reportExceInfo[0] testSuiteExecDate = reportExceInfo[1] archiveFolder = self.__MakeArchiveDir(testSuiteExecDate) archiveFileName = os.path.join(archiveFolder, f"{datetime.timestamp(testSuiteExecDate)}-{reportId}-{testSuiteName}.zip") logging.info(f'\tDownload report {archiveFileName}...') endpoint = f"{self.__url}/api/report/export/atx/{taskId}/download" with requests.get(endpoint, stream=True, headers=headers, params=self.__defaultAuthPayLoad, verify=self.REQUESTS_VERIFY) as response: response.raise_for_status() with open(archiveFileName, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) logging.info(f'\tDownload report {archiveFileName} finished') def __MakeArchiveDir(self, testSuiteExecDate: datetime) -> str: archiveDir = os.path.join(self.__archivFolderPath, f"{testSuiteExecDate.year}", f"{testSuiteExecDate.month}", f"{testSuiteExecDate.day}") os.makedirs(archiveDir, exist_ok=True) return archiveDir def __RemoveReport(self, reportId: int): headers = { 'Content-type': 'application/json', 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } deletetUrl = f"{self.__url}/api/report/reports/{reportId}" response = requests.delete(deletetUrl, params=self.__defaultAuthPayLoad, headers=headers, verify=self.REQUESTS_VERIFY) response.raise_for_status() taskId = response.json().get("taskId") logging.info(f'\tDeletion of report id {reportId} is being processed under task id {taskId}.') self.__WaitUntilDeletionIsFinished(taskId) def __WaitUntilDeletionIsFinished(self, taskId: str): headers = { 'Accept': 'application/json', 'TestGuide-AuthKey': self.__authKey } endpoint = f"{self.__url}/api/report/reports/deletestatus/{taskId}" while True: response = requests.get(endpoint, headers=headers, params=self.__defaultAuthPayLoad, verify=self.REQUESTS_VERIFY) response.raise_for_status() jsonResponse = response.json() deleteStatus = jsonResponse.get("status") msg = jsonResponse.get("detailedMessage") if deleteStatus == 'success': logging.info(f'\tDeletion of task id {taskId} is finished: {msg}') return True if deleteStatus == 'error': if "test reports belonging to a locked release" in msg: logging.warning(f'\tDeletion task aborted because: {msg}') return True else: raise RuntimeError(f'\tDeletion of task id {taskId} failed: {msg}') else: logging.debug(f'\tDeletion task is not done yet.... {deleteStatus}') time.sleep(5) if __name__ == '__main__': # test.guide access parameter # Permissions to view and delete reports are required for archiving. AUTHKEY = "jRx1le8=.HanQ3VS7ZlqAgq1YI86P_7sS_5t5WxFKfUMyT1trZMQ=" # <1> URL = "https://localhost:8443" # <2> PROJECT_ID = 1 # <3> # Export Parameter ARCHIVE_FOLDER = os.path.dirname(__file__) # <4> ARCHIVE_START_DATE = datetime(2023, 1, 1, 0, 0) # <5> ARCHIVE_END_DATE = datetime(2024, 3, 31, 0, 0) # <6> USE_EXPORT_STORAGE = False # <7> ARCHIVAR = TGReportArchivist(URL, AUTHKEY, PROJECT_ID, ARCHIVE_FOLDER, USE_EXPORT_STORAGE) ARCHIVAR.StartArchiving(ARCHIVE_START_DATE, ARCHIVE_END_DATE)