autogits/integration/tests/test_pr_workflow.py

import pytest
import re
import time
import subprocess
import requests
from pathlib import Path
from tests.lib.common_test_utils import (
    GiteaAPIClient,
    mock_build_result,
)

# =============================================================================
#  TEST CASES
# =============================================================================


def test_pr_workflow_succeeded(gitea_env, mock_build_result):
    """End-to-end test for a successful PR workflow."""
    diff = "diff --git a/test.txt b/test.txt\nnew file mode 100644\nindex 0000000..e69de29\n"
    pr = gitea_env.create_gitea_pr("pool/pkgA", diff, "Test PR - should succeed")
    initial_pr_number = pr["number"]

    compose_dir = Path(__file__).parent.parent

    forwarded_pr_number = None
    print(
        f"Polling pool/pkgA PR #{initial_pr_number} timeline for forwarded PR event..."
    )
    for _ in range(20):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("pool/pkgA", initial_pr_number)
        for event in timeline_events:
            if event.get("type") == "pull_ref":
                if not (ref_issue := event.get("ref_issue")):
                    continue
                url_to_check = ref_issue.get("html_url", "")
                match = re.search(r"products/SLFO/pulls/(\d+)", url_to_check)
                if match:
                    forwarded_pr_number = match.group(1)
                    break
        if forwarded_pr_number:
            break
    assert (
        forwarded_pr_number is not None
    ), "Workflow bot did not create a pull_ref event on the timeline."
    print(f"Found forwarded PR: products/SLFO #{forwarded_pr_number}")

    print(f"Polling products/SLFO PR #{forwarded_pr_number} for reviewer assignment...")
    reviewer_added = False
    for _ in range(15):
        time.sleep(1)
        pr_details = gitea_env.get_pr_details("products/SLFO", forwarded_pr_number)
        if any(
            r.get("login") == "autogits_obs_staging_bot"
            for r in pr_details.get("requested_reviewers", [])
        ):
            reviewer_added = True
            break
    assert reviewer_added, "Staging bot was not added as a reviewer."
    print("Staging bot has been added as a reviewer.")

    mock_build_result(package_name="pkgA", code="succeeded") 

    print("Restarting obs-staging-bot...")
    subprocess.run(
        ["podman-compose", "restart", "obs-staging-bot"],
        cwd=compose_dir,
        check=True,
        capture_output=True,
    )

    print(f"Polling products/SLFO PR #{forwarded_pr_number} for final status...")
    status_comment_found = False
    for _ in range(20):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("products/SLFO", forwarded_pr_number)
        for event in timeline_events:
            print(event.get("body", "not a body"))
            if event.get("body") and "successful" in event["body"]:
                status_comment_found = True
                break
        if status_comment_found:
            break
    assert status_comment_found, "Staging bot did not post a 'successful' comment."


def test_pr_workflow_failed(gitea_env, mock_build_result):
    """End-to-end test for a failed PR workflow."""
    diff = "diff --git a/another_test.txt b/another_test.txt\nnew file mode 100644\nindex 0000000..e69de29\n"
    pr = gitea_env.create_gitea_pr("pool/pkgA", diff, "Test PR - should fail")
    initial_pr_number = pr["number"]

    compose_dir = Path(__file__).parent.parent

    forwarded_pr_number = None
    print(
        f"Polling pool/pkgA PR #{initial_pr_number} timeline for forwarded PR event..."
    )
    for _ in range(20):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("pool/pkgA", initial_pr_number)
        for event in timeline_events:
            if event.get("type") == "pull_ref":
                if not (ref_issue := event.get("ref_issue")):
                    continue
                url_to_check = ref_issue.get("html_url", "")
                match = re.search(r"products/SLFO/pulls/(\d+)", url_to_check)
                if match:
                    forwarded_pr_number = match.group(1)
                    break
        if forwarded_pr_number:
            break
    assert (
        forwarded_pr_number is not None
    ), "Workflow bot did not create a pull_ref event on the timeline."
    print(f"Found forwarded PR: products/SLFO #{forwarded_pr_number}")

    print(f"Polling products/SLFO PR #{forwarded_pr_number} for reviewer assignment...")
    reviewer_added = False
    for _ in range(15):
        time.sleep(1)
        pr_details = gitea_env.get_pr_details("products/SLFO", forwarded_pr_number)
        if any(
            r.get("login") == "autogits_obs_staging_bot"
            for r in pr_details.get("requested_reviewers", [])
        ):
            reviewer_added = True
            break
    assert reviewer_added, "Staging bot was not added as a reviewer."
    print("Staging bot has been added as a reviewer.")

    mock_build_result(package_name="pkgA", code="failed")

    print("Restarting obs-staging-bot...")
    subprocess.run(
        ["podman-compose", "restart", "obs-staging-bot"],
        cwd=compose_dir,
        check=True,
        capture_output=True,
    )

    print(f"Polling products/SLFO PR #{forwarded_pr_number} for final status...")
    status_comment_found = False
    for _ in range(20):
        time.sleep(1)
        timeline_events = gitea_env.get_timeline_events("products/SLFO", forwarded_pr_number)
        for event in timeline_events:
            if event.get("body") and "failed" in event["body"]:
                status_comment_found = True
                break
        if status_comment_found:
            break
    assert status_comment_found, "Staging bot did not post a 'failed' comment."