MVP Create test runs based on change log

changelog-analysis/get_cases.py

@@ -0,0 +1,120 @@
+import os
+import requests
+from datetime import datetime
+from typing import List, Dict, Any, Set, Optional
+
+
+TESTRAIL_HOST = os.environ["TESTRAIL_HOST"]
+TESTRAIL_USERNAME = os.environ["TESTRAIL_USERNAME"]
+TESTRAIL_PASSWORD = os.environ["TESTRAIL_PASSWORD"]
+
+PROJECT_ID = 75
+SUITE_ID = 40281
+MILESTONE_ID = 6652
+
+PAGE_LIMIT = 250  # TestRail max page size
+
+API_BASE = f"{TESTRAIL_HOST}/index.php?/api/v2"
+
+
+# ===============================
+# HELPERS
+# ===============================
+
+def tr_get(url: str) -> Dict[str, Any]:
+    r = requests.get(url, auth=(TESTRAIL_USERNAME, TESTRAIL_PASSWORD))
+    if r.status_code >= 400:
+        try:
+            print("TestRail error:", r.json())
+        except Exception:
+            print("TestRail error text:", r.text[:2000])
+        r.raise_for_status()
+    return r.json()
+
+def tr_post(url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    r = requests.post(url, json=payload, auth=(TESTRAIL_USERNAME, TESTRAIL_PASSWORD))
+    if r.status_code >= 400:
+        try:
+            print("TestRail error:", r.json())
+        except Exception:
+            print("TestRail error text:", r.text[:2000])
+        r.raise_for_status()
+    return r.json()
+
+
+def case_label_titles(case):
+    labels = case.get("labels") or []
+    return {
+        l.get("title")
+        for l in labels
+        if isinstance(l, dict) and l.get("title")
+    }
+
+# ===============================
+# FETCH + FILTER CASES
+# ===============================
+
+def get_all_cases_in_suite(project_id: int, suite_id: int) -> List[Dict[str, Any]]:
+    all_cases = []
+    offset = 0
+
+    while True:
+        url = (
+            f"{API_BASE}/get_cases/{project_id}"
+            f"&suite_id={suite_id}"
+            f"&limit={PAGE_LIMIT}&offset={offset}"
+        )
+
+        data = tr_get(url)
+        cases = data.get("cases", [])
+        all_cases.extend(cases)
+
+        if len(cases) < PAGE_LIMIT:
+            break
+        offset += PAGE_LIMIT
+
+    return all_cases
+
+def get_cases_by_labels(labels: List[str],
+                        project_id: int = PROJECT_ID,
+                        suite_id: int = SUITE_ID) -> List[Dict[str, Any]]:
+    all_cases = get_all_cases_in_suite(project_id, suite_id)
+    return filter_cases_by_labels(all_cases, labels)
+
+
+def filter_cases_by_labels(cases: List[Dict[str, Any]], labels: List[str]) -> List[Dict[str, Any]]:
+    wanted = set(labels)
+    matched = []
+    for c in cases:
+        if case_label_titles(c) & wanted:
+            matched.append(c)
+    return matched
+
+# ===============================
+# CREATE RUN
+# ===============================
+
+def create_test_run(case_ids, name=None, description=None, milestone_id=None):
+    if not case_ids:
+        raise ValueError("case_ids is empty — refusing to create an empty run.")
+
+    if name is None:
+        from datetime import datetime
+        ts = datetime.now().strftime("%Y-%m-%d %H:%M")
+        name = f"RTS Smart Run ({ts})"
+
+    payload = {
+        "suite_id": SUITE_ID,
+        "name": name,
+        "include_all": False,
+        "case_ids": sorted(set(case_ids)),
+    }
+
+    if description:
+        payload["description"] = description
+
+    if milestone_id:
+        payload["milestone_id"] = milestone_id
+
+    url = f"{API_BASE}/add_run/{PROJECT_ID}"
+    return tr_post(url, payload)

changelog-analysis/get_change_log.py

@@ -0,0 +1,181 @@
+import os
+from pathlib import Path
+import requests
+import yaml
+from typing import List, Set, Tuple
+
+RULES_FILE = "rules.yml"
+OWNER = "mozilla-mobile"
+REPO = "firefox-ios"
+
+IGNORED_DIRECTORIES = [
+    ".github/workflows/",
+    "taskcluster/",
+    ".github/workflows",
+]
+
+IGNORED_FILENAMES = [
+    ".swiftlint.yml",
+    "README.md",
+    "bitrise.yml",
+    "version.txt",
+]
+
+IGNORED_EXACT_PATHS = [
+    "firefox-ios/Client.xcodeproj/project.pbxproj",
+    "firefox-ios/Client/Configuration/version.xcconfig",
+    "taskcluster/requirements.txt",
+]
+
+IGNORED_EXTENSIONS = [
+    ".strings",
+    ".stringsdict",
+]
+
+def _github_headers() -> dict:
+    headers = {"Accept": "application/vnd.github+json"}
+    token = os.environ.get("GITHUB_TOKEN")
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    return headers
+
+
+def get_latest_release_tags(owner: str, repo: str, count: int = 2,
+                             prefix: str = "firefox-v") -> List[str]:
+    """Return the `count` most-recent release tag names by version, newest first."""
+    url = f"https://api.github.com/repos/{owner}/{repo}/git/matching-refs/tags/{prefix}"
+    all_tags = []
+
+    while url:
+        response = requests.get(url, headers=_github_headers())
+        response.raise_for_status()
+        all_tags.extend(ref["ref"].removeprefix("refs/tags/") for ref in response.json())
+
+        url = None
+        for part in response.headers.get("Link", "").split(","):
+            if 'rel="next"' in part:
+                url = part[part.index("<") + 1: part.index(">")]
+                break
+
+    def version_key(tag: str) -> tuple:
+        try:
+            return tuple(int(p) for p in tag.removeprefix(prefix).split("."))
+        except ValueError:
+            return (0,)
+
+    all_tags.sort(key=version_key, reverse=True)
+
+    if len(all_tags) < count:
+        raise ValueError(
+            f"Only {len(all_tags)} tag(s) matching '{prefix}' found in {owner}/{repo}; need at least {count}."
+        )
+    return all_tags[:count]
+
+
+def is_ignored_path(path: str) -> bool:
+    p = path.lower()
+
+    # Ignore test directories
+    if "/tests/" in p or p.startswith("tests/"):
+        return True
+    if "/uitests/" in p or p.startswith("uitests/"):
+        return True
+
+    # Ignore specific directories
+    for d in IGNORED_DIRECTORIES:
+        if p.startswith(d) or f"/{d}" in p:
+            return True
+
+    # Ignore exact file paths
+    for exact in IGNORED_EXACT_PATHS:
+        if p == exact.lower():
+            return True
+
+    # Ignore specific filenames
+    if Path(p).name in [f.lower() for f in IGNORED_FILENAMES]:
+        return True
+
+    # Ignore specific extensions
+    if any(p.endswith(ext) for ext in IGNORED_EXTENSIONS):
+        return True
+    return False
+
+
+# -------------------------------
+# Load rules from YAML
+# -------------------------------
+
+def load_rules(filename: str):
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    rules_path = os.path.join(script_dir, filename)
+
+    with open(rules_path, "r") as f:
+        config = yaml.safe_load(f)
+
+    return config.get("rules", [])
+
+
+# -------------------------------
+# Map files to components
+# -------------------------------
+
+def map_files_to_components(files: List[str], rules) -> Tuple[Set[str], List[str]]:
+    impacted_components = set()
+    unmatched_files = []
+
+    for file_path in files:
+        matched = False
+
+        for rule in rules:
+            prefix = rule["prefix"]
+            components = rule["components"]
+
+            if file_path.startswith(prefix):
+                impacted_components.update(components)
+                matched = True
+
+        if not matched:
+            unmatched_files.append(file_path)
+
+    return impacted_components, unmatched_files
+
+
+# -------------------------------
+# Get changed files from GitHub
+# -------------------------------
+
+def get_changed_files(owner, repo, base, head):
+    url = f"https://api.github.com/repos/{owner}/{repo}/compare/{base}...{head}"
+    response = requests.get(url, headers=_github_headers())
+    response.raise_for_status()
+
+    data = response.json()
+
+    all_files = [f["filename"] for f in data.get("files", [])]
+
+    # Apply ignore filtering here
+    filtered_files = [
+        f for f in all_files if not is_ignored_path(f)
+    ]
+
+    print(f"\nTotal changed files (raw): {len(all_files)}")
+    print(f"After filtering ignored paths: {len(filtered_files)}")
+
+    return filtered_files
+
+def get_impacted_components(base_tag: str, head_tag: str,
+                             owner: str = OWNER, repo: str = REPO) -> List[str]:
+    rules = load_rules(RULES_FILE)
+    changed_files = get_changed_files(owner, repo, base_tag, head_tag)
+    components, unmatched = map_files_to_components(changed_files, rules)
+    print(f"Unmatched files: {len(unmatched)}")
+    return sorted(components)
+
+
+if __name__ == "__main__":
+    head_tag, base_tag = get_latest_release_tags(OWNER, REPO)
+    print(f"Comparing {base_tag} → {head_tag}")
+    components = get_impacted_components(base_tag, head_tag)
+    print("\nImpacted Components:")
+    for c in components:
+        print("-", c)

changelog-analysis/get_release_tags.py

@@ -0,0 +1,60 @@
+import os
+import requests
+from typing import List, Tuple
+
+BITRISE_APP_ID = os.environ.get("BITRISE_APP_ID", "6c06d3a40422d10f")
+BITRISE_TOKEN = os.environ["BITRISE_TOKEN"]
+
+VALID_WORKFLOWS = {
+    "release_promotion_push",       # Firefox
+    "release_promotion_push_focus", # Focus
+}
+
+TAG_PREFIX = "firefox-v"
+
+
+def _get_firefox_tags_from_bitrise() -> List[str]:
+    """Return all unique firefox-v* tags from successful Bitrise release builds, newest-version first."""
+    url = f"https://api.bitrise.io/v0.1/apps/{BITRISE_APP_ID}/builds?trigger_event_type=tag&limit=50"
+    headers = {"accept": "application/json", "Authorization": BITRISE_TOKEN}
+
+    response = requests.get(url, headers=headers)
+    response.raise_for_status()
+
+    builds = response.json().get("data", [])
+
+    tags = {
+        b["tag"]
+        for b in builds
+        if (b.get("status_text") or "").lower() == "success"
+        and b.get("tag", "").lower().startswith(TAG_PREFIX)
+        and (b.get("triggered_workflow") or "").lower() in VALID_WORKFLOWS
+    }
+
+    def version_key(tag: str) -> tuple:
+        try:
+            return tuple(int(p) for p in tag.removeprefix(TAG_PREFIX).split("."))
+        except ValueError:
+            return (0,)
+
+    return sorted(tags, key=version_key, reverse=True)
+
+
+def get_tags() -> Tuple[str, str]:
+    """Return (base_tag, head_tag) — the two latest firefox-v* tags from Bitrise."""
+    tags = _get_firefox_tags_from_bitrise()
+
+    if len(tags) < 2:
+        raise RuntimeError(
+            f"Need at least 2 firefox-v* tags from Bitrise, found {len(tags)}: {tags}"
+        )
+
+    head_tag, base_tag = tags[0], tags[1]
+    print(f"Tags from Bitrise: {base_tag} → {head_tag}")
+    return base_tag, head_tag
+
+
+if __name__ == "__main__":
+    base, head = get_tags()
+    print(f"Base: {base}")
+    print(f"Head: {head}")