Jira Cloud 인스턴스 간 Xray 데이터 마이그레이션

import requests
import logging
import json
import base64

# 로깅 설정
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def get_access_token(client_id, client_secret):
    """
    대상 환경에서 액세스 토큰을 발급받는 함수
    :param client_id: Xray 환경의 Client ID
    :param client_secret: Xray 환경의 Client Secret
    :return: 발급받은 액세스 토큰
    """
    url = "https://xray.cloud.getxray.app/api/v2/authenticate"
    payload = {
        "client_id": client_id,
        "client_secret": client_secret
    }
    response = requests.post(url, json=payload)
    if response.status_code == 200:
        return json.loads(response.text)  # 토큰 반환
    else:
        raise Exception(f"Failed to get access token: {response.status_code}, {response.text}")


def fetch_test_execution_data(token):
    """
    GraphQL API를 통해 테스트 실행 데이터를 가져오는 함수
    :param token: Xray 소스 환경의 액세스 토큰
    :return: GraphQL API 응답 데이터 (JSON)
    """
    url = "https://xray.cloud.getxray.app/api/v2/graphql"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    
    query = {
        "query": """
        {
            getTestExecutions(jql: "project = 'ADON'", limit: 10) {
                total
                start
                limit
                results {
                    issueId
                    tests(limit: 10) {
                        total
                        start
                        limit
                        results {
                            status {
                                name
                            }
                            issueId
                            testType {
                                name
                            }
                        }
                    }
                    jira(fields: ["assignee", "reporter"])
                }
            }
        }
        """
    }

    try:
        logging.info("Sending request to GraphQL API.")
        response = requests.post(url, json=query, headers=headers)
        logging.info(f"Response received: Status code {response.status_code}")
        response.raise_for_status()  # HTTP 상태 코드가 200이 아니면 예외 발생
        return response.json()
    except requests.exceptions.Timeout:
        logging.error("Request timed out. Please try again later.")
        raise Exception("Request timed out. Please try again later.")
    except requests.exceptions.RequestException as e:
        logging.error(f"Failed to fetch test execution data. Error: {e}")
        raise Exception(f"Failed to fetch test execution data: {e}")

def get_issue_key(issue_id, jira_base_url, jira_token, jira_user_name):
    """
    Jira REST API를 사용하여 issueId로부터 issueKey 조회
    :param issue_id: Xray에서 반환된 issueId
    :param jira_base_url: Jira 인스턴스 URL
    :param jira_token: Jira API 인증 토큰 (Bearer Token)
    :return: 해당 issueId의 issueKey
    """
    url = f"{jira_base_url}/rest/api/2/issue/{issue_id}"
    # Basic Authentication 문자열 생성
    auth_string = f"{jira_user_name}:{jira_token}"
    authBase64 = base64.b64encode(auth_string.encode()).decode()  # Base64로 인코딩
    
    headers = {
        "Authorization": f"Basic {authBase64}",
        "Content-Type": "application/json"
    }
    
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json().get("key")  # JSON 응답에서 issueKey 추출
    else:
        raise Exception(f"Failed to fetch issueKey for issueId {issue_id}: "
                        f"{response.status_code}, {response.text}")

def transform_data_for_upload(source_data, jira_base_url, jira_token):
    """
    GraphQL API 응답 데이터를 REST API로 업로드할 형식으로 변환
    :param source_data: GraphQL API 응답 데이터 (JSON)
    :return: 변환된 데이터 리스트
    """
    transformed_data = []
    executions = source_data.get("data", {}).get("getTestExecutions", {}).get("results", [])
    
    for execution in executions:
        # Jira REST API를 통해 issueId -> issueKey로 변환
        try:
            print("issueId : " +execution.get("issueId"))
            test_execution_key = get_issue_key(execution.get("issueId"), jira_base_url, jira_token, jira_user_name)
        except Exception as e:
            logging.warning(f"Skipping test execution with issueId {execution.get('issueId')}: {e}")
            continue

        # 빈 tests 배열 건너뛰기
        tests = execution.get("tests", {}).get("results", [])
        if not tests:
            continue

        execution_data = {
            "testExecutionKey": test_execution_key,  # 변환된 issueKey 사용
            "info": {
                "summary": f"Test Execution {execution.get('issueId')}",
                "description": "Migrated from source environment"
            },
            "tests": []
        }
         
        for test in tests:
            try:
                test_key = get_issue_key(test.get("issueId"), jira_base_url, jira_token, jira_user_name)  # issueId -> issueKey 변환
            except Exception as e:
                print(f"Skipping test with issueId {test.get('issueId')}: {e}")
                continue

            test_data = {
                "testKey": test_key,  # 변환된 issueKey 사용
                "testInfo": {
                    "type": test.get("testType", {}).get("name")
                },
                "status": test.get("status", {}).get("name", "TO DO")
            }
            execution_data["tests"].append(test_data)

        transformed_data.append(execution_data)

    return transformed_data


def upload_to_target_environment(transformed_data, target_token):
    """
    변환된 데이터를 Xray 대상 환경으로 업로드
    :param transformed_data: 업로드할 데이터 리스트
    :param target_token: 대상 환경의 액세스 토큰
    """
    url = "https://xray.cloud.getxray.app/api/v2/import/execution"
    headers = {"Authorization": f"Bearer {target_token}", "Content-Type": "application/json"}
    
    for execution in transformed_data:
        response = requests.post(url, json=execution, headers=headers)
        if response.status_code == 200:
            print(f"Successfully uploaded execution: {execution['testExecutionKey']}")
        else:
            print(f"Failed to upload execution: {execution['testExecutionKey']}, "
                  f"Status: {response.status_code}, Response: {response.text}")


if __name__ == "__main__":
    # 1. 소스 환경의 API 인증 정보
    source_client_id = "source_client_id"
    source_client_secret = "source_client_secret"

    # 2. 대상 환경의 API 인증 정보
    target_client_id = "target_client_id"
    target_client_secret = "target_client_secret"

    # 3. 소스 환경에서 데이터 가져오기
    try:
        source_token = get_access_token(source_client_id, source_client_secret)
        logging.info("Source token successfully obtained.")
        source_data = fetch_test_execution_data(source_token)
        logging.info("Test execution data fetched successfully.")
    except Exception as e:
        print(f"Error in source environment: {e}")
        exit(1)

    #Jira API 인증 정보
    jira_base_url = "jira_base_url"
    jira_token = "jira_token"
    jira_user_name = "jira_user_name"
    # 4. 데이터 변환
    try:
        transformed_data = transform_data_for_upload(source_data, jira_base_url, jira_token)
        logging.info("Data transformed successfully.")
    except Exception as e:
        print(f"Error in data transformation: {e}")
        exit(1)

    # 5. 대상 환경으로 데이터 업로드
    try:
        target_token = get_access_token(target_client_id, target_client_secret)
        print("Target token successfully obtained.")
        upload_to_target_environment(transformed_data, target_token)
        print("Data uploaded successfully.")
    except Exception as e:
        print(f"Error in target environment: {e}")