export ATO_BASE="https://your-tenant.qatools.cloud"   # or self-hosted URL
export ATO_TOKEN="<api-token>"

import requests, os

BASE = os.environ["ATO_BASE"]
HEADERS = {
    "Authorization": f"Bearer {os.environ['ATO_TOKEN']}",
    "Content-Type": "application/json",
}

def create_case(project_id, name, description=None, precondition=None,
                scenario_steps=None, status="Active", layer_id=None,
                feature_id=None):
    """
    scenario_steps: list of {"keyword": "Given|When|Then|And",
                              "name": "...", "expectedResult": "..."}
    status: Active / Outdated / Archived
    layer_id: testing layer (UI / API / Component / Unit) - discover via
              /api/rs/testlayer
    """
    body = {
        "projectId": project_id,
        "name": name,
        "description": description,
        "precondition": precondition,
        "status": status,
        "layerId": layer_id,
        "featureId": feature_id,
    }
    r = requests.post(f"{BASE}/api/rs/testcase", json=body, headers=HEADERS)
    r.raise_for_status()
    created = r.json()
    if scenario_steps:
        attach_scenario(created["id"], scenario_steps)
    return created

def attach_scenario(case_id, steps):
    """
    steps: nested tree, e.g.,
      [
        {"keyword": "Given", "name": "User is logged in",
         "steps": [
           {"name": "Navigate to /login"},
           {"name": "Enter credentials"},
           {"name": "Submit form"},
         ]},
        {"keyword": "When", "name": "User clicks Checkout"},
        {"keyword": "Then", "name": "Order is placed",
         "expectedResult": "Confirmation page shows order ID"},
      ]
    """
    r = requests.post(
        f"{BASE}/api/rs/testcase/{case_id}/scenario",
        json={"steps": steps},
        headers=HEADERS,
    )
    r.raise_for_status()

# Discover layers in a project
layers = requests.get(f"{BASE}/api/rs/testlayer",
                      params={"projectId": project_id},
                      headers=HEADERS).json()
# [{"id": 1, "name": "API"}, {"id": 2, "name": "UI"}, ...]

# Discover features
features = requests.get(f"{BASE}/api/rs/feature",
                        params={"projectId": project_id},
                        headers=HEADERS).json()

def link_to_issue(case_id, integration_id, issue_key):
    r = requests.post(
        f"{BASE}/api/rs/testcase/{case_id}/issue",
        json={"integrationId": integration_id, "key": issue_key},
        headers=HEADERS,
    )
    r.raise_for_status()

def set_tags(case_id, tag_names):
    r = requests.post(
        f"{BASE}/api/rs/testcase/{case_id}/tag",
        json={"tags": [{"name": t} for t in tag_names]},
        headers=HEADERS,
    )
    r.raise_for_status()

case = requests.get(f"{BASE}/api/rs/testcase/{case_id}",
                    headers=HEADERS).json()

# Paginated list
def list_cases(project_id, page_size=100):
    out = []
    page = 0
    while True:
        r = requests.get(f"{BASE}/api/rs/testcase",
                         params={"projectId": project_id,
                                 "page": page, "size": page_size},
                         headers=HEADERS)
        r.raise_for_status()
        data = r.json()
        out.extend(data.get("content", []))
        if data.get("last", False):
            break
        page += 1
    return out

# In your automated test (using allure-pytest, allure-junit, etc.):
import allure

@allure.testcase("ATO-1234")  # links to Allure TestOps case ID 1234
def test_checkout_flow():
    ...

url = f"{BASE}/project/{project_id}/test-cases/{case_id}"

- name: Run tests with Allure
  run: pytest --alluredir=allure-results

- name: Upload to Allure TestOps
  env:
    ATO_TOKEN: ${{ secrets.ALLURE_TESTOPS_TOKEN }}
    ATO_BASE: ${{ vars.ALLURE_TESTOPS_BASE }}
    ATO_PROJECT_ID: ${{ vars.ATO_PROJECT_ID }}
  run: |
    allurectl upload allure-results \
      --endpoint $ATO_BASE \
      --token $ATO_TOKEN \
      --project-id $ATO_PROJECT_ID \
      --launch-name "CI run ${GITHUB_RUN_NUMBER}"

export QASE_TOKEN="<api-token-from-qase.io-settings>"

import requests, os

BASE = "https://api.qase.io/v1"
HEADERS = {
    "Token": os.environ["QASE_TOKEN"],
    "Content-Type": "application/json",
}

def create_case(project_code, title, description=None, preconditions=None,
                postconditions=None, steps=None, suite_id=None,
                severity=4, priority=2, type=1, automation=0,
                status=1, params=None):
    """
    severity: 1=Blocker, 2=Critical, 3=Major, 4=Normal, 5=Minor, 6=Trivial
    priority: 1=High, 2=Medium, 3=Low
    type:     1=Functional, 2=Smoke, 3=Regression, 4=Security, etc. (per project enum)
    automation: 0=Manual, 1=Automated, 2=To-be-automated
    status:   0=Actual, 1=Draft, 2=Deprecated
    steps: list of {"action": "...", "expected_result": "...", "data": "..."}
    """
    body = {
        "title": title,
        "description": description,
        "preconditions": preconditions,
        "postconditions": postconditions,
        "severity": severity,
        "priority": priority,
        "type": type,
        "automation": automation,
        "status": status,
        "suite_id": suite_id,
        "steps": steps or [],
        "params": params or {},
    }
    r = requests.post(f"{BASE}/case/{project_code}",
                      json=body, headers=HEADERS)
    r.raise_for_status()
    return r.json()

steps = [
    {"action": "Navigate to /login",
     "expected_result": "Login form rendered",
     "data": ""},
    {"action": "Enter alice@example.com + correct password",
     "expected_result": "Submit button enabled",
     "data": "alice@example.com / pw123"},
    {"action": "Click Submit",
     "expected_result": "Redirected to /dashboard within 2 s",
     "data": ""},
]
case = create_case("AUTH", "Login redirects to dashboard",
                   steps=steps, suite_id=42,
                   severity=3, priority=1)

def create_suite(project_code, title, description=None, parent_id=None):
    body = {"title": title, "description": description,
            "parent_id": parent_id}
    r = requests.post(f"{BASE}/suite/{project_code}",
                      json=body, headers=HEADERS)
    r.raise_for_status()
    return r.json()

def create_shared_step(project_code, title, action, expected_result, data=None):
    r = requests.post(
        f"{BASE}/shared_step/{project_code}",
        json={"title": title, "action": action,
              "expected_result": expected_result, "data": data},
        headers=HEADERS,
    )
    r.raise_for_status()
    return r.json()

# Reference shared step in a case
steps = [
    {"shared_step_hash": shared_step_hash},
    {"action": "...", "expected_result": "..."},
]

def update_case(project_code, case_id, **fields):
    r = requests.patch(f"{BASE}/case/{project_code}/{case_id}",
                       json=fields, headers=HEADERS)
    r.raise_for_status()
    return r.json()

case = requests.get(f"{BASE}/case/{project_code}/{case_id}",
                    headers=HEADERS).json()

def list_cases(project_code, limit=100):
    cases = []
    offset = 0
    while True:
        r = requests.get(f"{BASE}/case/{project_code}",
                         params={"limit": limit, "offset": offset},
                         headers=HEADERS)
        r.raise_for_status()
        data = r.json().get("result", {})
        cases.extend(data.get("entities", []))
        if len(data.get("entities", [])) < limit:
            break
        offset += limit
    return cases

import csv, time

rows = list(csv.DictReader(open("legacy.csv")))
created, failed = [], []
for row in rows:
    steps = [
        {"action": s, "expected_result": e, "data": d}
        for s, e, d in zip(
            row["steps"].split("|"),
            row["expected"].split("|"),
            (row.get("data") or "").split("|"),
        )
    ]
    try:
        r = create_case(
            project_code=row["project"],
            title=row["title"],
            preconditions=row.get("preconditions"),
            steps=steps,
            severity=int(row.get("severity", 4)),
            priority=int(row.get("priority", 2)),
            suite_id=int(row["suite_id"]),
        )
        created.append(r["result"]["id"])
    except Exception as e:
        failed.append((row["title"], str(e)))
    time.sleep(1)  # ~60 req/min; avoids 429s

update_case(project_code, case_id, tags=["jira:ENG-123"])

- name: Sync Qase cases
  env:
    QASE_TOKEN: ${{ secrets.QASE_TOKEN }}
  run: python scripts/sync-qase.py

api("add_suite/123", method="POST", body={"name": "Authentication"})
api("add_section/123", method="POST",
    body={"suite_id": 7, "name": "Login flows", "parent_id": None})

fields = api("get_case_fields")
for f in fields:
    print(f["system_name"], f["label"], f["type_id"])
# custom_preconds Preconditions 3
# custom_severity Severity 6
# custom_automation_type Automation Type 6

types = api("get_case_types")   # -> [{"id", "name", "is_default"}, ...]
prios = api("get_priorities")   # -> [{"id", "name", "priority", ...}, ...]

# From Xray Global Settings → API Keys, create a client
export XRAY_CLIENT_ID="..."
export XRAY_CLIENT_SECRET="..."
export JIRA_BASE="https://your-tenant.atlassian.net"

import requests, os

# Exchange client credentials for a JWT
def get_token():
    r = requests.post("https://xray.cloud.getxray.app/api/v2/authenticate",
                      json={"client_id": os.environ["XRAY_CLIENT_ID"],
                            "client_secret": os.environ["XRAY_CLIENT_SECRET"]})
    r.raise_for_status()
    return r.text.strip('"')   # response is a quoted string

XRAY_BASE = "https://xray.cloud.getxray.app/api/v2"
GRAPHQL = "https://xray.cloud.getxray.app/api/v2/graphql"

def xray_headers(token):
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

CREATE_TEST_MUTATION = """
mutation CreateTest($input: CreateTestInput!) {
  createTest(input: $input) {
    test { issueId jira(fields: ["key", "summary"]) }
    warnings
  }
}
"""

def create_manual_test(token, project_key, summary, steps,
                       preconditions_issue_id=None):
    """
    steps: list of {"action": "...", "data": "...", "result": "..."}
    """
    variables = {
        "input": {
            "testType": {"name": "Manual"},
            "jira": {
                "fields": {
                    "summary": summary,
                    "project": {"key": project_key},
                    "issuetype": {"name": "Test"},
                }
            },
            "steps": steps,
            "preconditionIssueIds": [preconditions_issue_id] if preconditions_issue_id else [],
        }
    }
    r = requests.post(GRAPHQL,
                      json={"query": CREATE_TEST_MUTATION, "variables": variables},
                      headers=xray_headers(token))
    r.raise_for_status()
    return r.json()

def bulk_import(token, project_key, tests):
    """
    tests: list of {"testtype": "Manual", "fields": {...}, "steps": [...]}
    """
    r = requests.post(
        f"{XRAY_BASE}/import/test/bulk",
        json={"projectKey": project_key, "tests": tests},
        headers=xray_headers(token),
    )
    r.raise_for_status()
    return r.json()  # returns job id + status poll URL

# Upload .feature file via /api/v2/import/feature
with open("checkout.feature", "rb") as f:
    r = requests.post(
        f"{XRAY_BASE}/import/feature",
        files={"file": f},
        params={"projectKey": "ENG"},
        headers={"Authorization": f"Bearer {token}"},
    )

def link_test_to_requirement(jira_base, jira_email, jira_token,
                              test_key, req_key, link_type="Tests"):
    r = requests.post(
        f"{jira_base}/rest/api/3/issueLink",
        json={
            "type": {"name": link_type},
            "inwardIssue": {"key": req_key},
            "outwardIssue": {"key": test_key},
        },
        auth=(jira_email, jira_token),
    )
    r.raise_for_status()

UPDATE_STEPS_MUTATION = """
mutation UpdateTestSteps($issueId: String!, $steps: [UpdateStepInput!]!) {
  updateTestSteps(issueId: $issueId, steps: $steps) {
    test { issueId }
    warnings
  }
}
"""

url = f"{os.environ['JIRA_BASE']}/browse/{jira_key}"

- name: Sync feature files to Xray
  env:
    XRAY_CLIENT_ID: ${{ secrets.XRAY_CLIENT_ID }}
    XRAY_CLIENT_SECRET: ${{ secrets.XRAY_CLIENT_SECRET }}
  run: |
    TOKEN=$(curl -s -X POST -H "Content-Type: application/json" \
      -d '{"client_id":"'$XRAY_CLIENT_ID'","client_secret":"'$XRAY_CLIENT_SECRET'"}' \
      https://xray.cloud.getxray.app/api/v2/authenticate | tr -d '"')
    for f in features/*.feature; do
      curl -X POST -H "Authorization: Bearer $TOKEN" \
        -F "file=@$f" \
        "https://xray.cloud.getxray.app/api/v2/import/feature?projectKey=ENG"
    done

export ZS_TOKEN="<bearer-token-from-zephyr-scale-ui>"

import requests, os

BASE = "https://api.zephyrscale.smartbear.com/v2"
HEADERS = {
    "Authorization": f"Bearer {os.environ['ZS_TOKEN']}",
    "Content-Type": "application/json",
}

def create_test_case(project_key, name, objective=None, precondition=None,
                     steps=None, owner=None, folder_id=None,
                     labels=None, components=None, priority="Normal",
                     status="Approved"):
    """
    steps: list of {"inline": {"description": "...", "expectedResult": "..."}}
    priority: Highest / High / Normal / Low / Lowest (per project config)
    status: Approved / Draft / Deprecated
    """
    body = {
        "projectKey": project_key,
        "name": name,
        "objective": objective,
        "precondition": precondition,
        "ownerId": owner,
        "folderId": folder_id,
        "labels": labels or [],
        "componentId": components,
        "priorityName": priority,
        "statusName": status,
    }
    r = requests.post(f"{BASE}/testcases", json=body, headers=HEADERS)
    r.raise_for_status()
    created = r.json()
    if steps:
        attach_steps(created["key"], steps)
    return created

def attach_steps(test_case_key, steps):
    """
    steps: list of {"inline": {"description": str, "expectedResult": str,
                                "testData": str}}
    """
    body = {"mode": "OVERWRITE", "items": [{"inline": s} for s in steps]}
    r = requests.post(f"{BASE}/testcases/{test_case_key}/teststeps",
                      json=body, headers=HEADERS)
    r.raise_for_status()
    return r.json()

steps = [
    {"description": "Navigate to /login",
     "expectedResult": "Login form rendered", "testData": ""},
    {"description": "Enter credentials and click Submit",
     "expectedResult": "Redirected to /dashboard",
     "testData": "alice@example.com / pw123"},
]
attach_steps("PROJ-T123", steps)

def create_folder(project_key, name, folder_type="TEST_CASE", parent_id=None):
    r = requests.post(f"{BASE}/folders", json={
        "projectKey": project_key, "name": name,
        "folderType": folder_type,  # TEST_CASE / TEST_PLAN / TEST_CYCLE
        "parentId": parent_id,
    }, headers=HEADERS)
    r.raise_for_status()
    return r.json()

def link_to_jira(test_case_key, issue_key):
    r = requests.post(f"{BASE}/testcases/{test_case_key}/links/issues",
                      json={"issueId": resolve_jira_issue_id(issue_key)},
                      headers=HEADERS)
    r.raise_for_status()

case = requests.get(f"{BASE}/testcases/PROJ-T123", headers=HEADERS).json()

# Paginated list
def list_cases(project_key, max_results=100):
    cases = []
    start_at = 0
    while True:
        r = requests.get(f"{BASE}/testcases",
                         params={"projectKey": project_key,
                                 "startAt": start_at,
                                 "maxResults": max_results},
                         headers=HEADERS)
        r.raise_for_status()
        data = r.json()
        cases.extend(data["values"])
        if data["isLast"]:
            break
        start_at += max_results
    return cases

import csv, time

rows = list(csv.DictReader(open("legacy.csv")))
created, failed = [], []
for row in rows:
    try:
        case = create_test_case(
            project_key=row["project"],
            name=row["title"],
            objective=row.get("objective"),
            precondition=row.get("precondition"),
            priority=row.get("priority", "Normal"),
            labels=row.get("labels", "").split(",") if row.get("labels") else None,
        )
        steps = [
            {"description": s, "expectedResult": e, "testData": d}
            for s, e, d in zip(
                row["steps"].split("|"),
                row["expected"].split("|"),
                (row.get("data") or "").split("|"),
            )
        ]
        attach_steps(case["key"], steps)
        created.append(case["key"])
    except Exception as e:
        failed.append((row["title"], str(e)))
    time.sleep(1)  # ~60 req/min; a 429 means back off further

print(f"created {len(created)}, failed {len(failed)}")

- name: Sync to Zephyr Scale
  env:
    ZS_TOKEN: ${{ secrets.ZEPHYR_SCALE_TOKEN }}
  run: python scripts/sync-zephyr-scale.py specs/