# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license

import subprocess
from unittest.mock import MagicMock, call, patch

import pytest

from actions import first_interaction, review_pr
from actions.first_interaction import get_event_content, get_first_interaction_response, get_pr_first_comment_template


def test_pr_first_comment_preserves_contributor_guidance():
    """Test PR acknowledgments retain the complete contributor checklist and links."""
    comment = get_pr_first_comment_template("owner/repo", "contributor")

    assert comment.count("- ✅") == 7
    assert "[Continuous Integration (CI)](https://docs.ultralytics.com/help/CI)" in comment
    assert "[Contributor License Agreement](https://docs.ultralytics.com/help/CLA)" in comment
    assert "[Contributing Guide](https://docs.ultralytics.com/help/contributing)" in comment
    assert "Bruce Lee" in comment


def test_get_event_content_issue():
    """Test extracting content from issue event."""
    # Create mock event
    mock_event = MagicMock()
    mock_event.event_name = "issues"
    mock_event.event_data = {
        "action": "opened",
        "issue": {
            "number": 123,
            "node_id": "node123",
            "title": "Test Issue",
            "body": "Issue description",
            "user": {"login": "testuser"},
        },
    }

    number, node_id, title, body, username, issue_type, action = get_event_content(mock_event)

    assert number == 123
    assert node_id == "node123"
    assert title == "Test Issue"
    assert body == "Issue description"
    assert username == "testuser"
    assert issue_type == "issue"
    assert action == "opened"


def test_get_event_content_pr():
    """Test extracting content from PR event."""
    # Create mock event
    mock_event = MagicMock()
    mock_event.event_name = "pull_request"
    mock_event.event_data = {
        "action": "opened",
        "pull_request": {
            "number": 456,
            "node_id": "node456",
            "title": "Test PR",
            "body": "PR description",
            "user": {"login": "testuser"},
        },
    }

    number, node_id, title, body, username, issue_type, action = get_event_content(mock_event)

    assert number == 456
    assert node_id == "node456"
    assert title == "Test PR"
    assert body == "PR description"
    assert username == "testuser"
    assert issue_type == "pull request"
    assert action == "opened"
    mock_event.get_repo_data.assert_not_called()


@patch("actions.first_interaction.review_pr.run_review")
@patch("actions.first_interaction.get_pr_open_response")
@patch("actions.first_interaction.get_event_content")
@patch("actions.first_interaction.Action")
@pytest.mark.parametrize(
    ("body", "expected"),
    [
        (
            "PR description\n\nDeleted: superseded registry entries.",
            "PR description\n\nDeleted: superseded registry entries.",
        ),
        ("a" * 30, "a" * 30),
        ("a" * 29, f"{'a' * 29}\n\nGenerated summary"),
        ("a " * 19 + "a", f"{'a ' * 19}a\n\nGenerated summary"),
        ("", "Generated summary"),
    ],
)
def test_open_pr_review_description(mock_action, mock_content, mock_response, mock_review, body, expected):
    """Test automatic reviews use the best available author and generated description context."""
    mock_content.return_value = (456, "node456", "Test PR", body, "testuser", "pull request", "opened")
    mock_response.return_value = {"summary": "Generated summary", "labels": []}
    event = mock_action.return_value
    event.should_skip_llm.return_value = False
    event.should_skip_pr_author.return_value = False
    event.paginate.return_value = []
    event.pr = {"body": body}
    event.get_pr_diff.return_value = ("diff", [])

    first_interaction.main()

    mock_review.assert_called_once_with(event, "Test PR", expected)


@patch("actions.first_interaction.check_links_in_string", return_value="Thank you for your issue")
@patch("actions.first_interaction.get_response")
def test_get_first_interaction_response(mock_get_response, mock_check_links):
    """Test labels and first response share one structured model call."""
    mock_get_response.return_value = {
        "labels": ["BUG", "not-a-label"],
        "first_comment": "Thank you for your issue",
    }

    mock_event = MagicMock()
    mock_event.repository = "test/repo"

    response = get_first_interaction_response(
        event=mock_event,
        issue_type="issue",
        title="Test Issue",
        body="Issue description",
        username="testuser",
        available_labels={"bug": "Something is broken"},
        current_labels=[],
        repository_context="A test repository; primary language: Rust",
    )

    assert response == {"labels": ["bug"], "first_comment": "Thank you for your issue"}
    mock_get_response.assert_called_once()
    mock_check_links.assert_called_once_with("Thank you for your issue", replace=True)
    assert mock_get_response.call_args.kwargs["text_format"]["format"]["type"] == "json_schema"
    prompt = mock_get_response.call_args.args[0][1]["content"]
    assert "primary language: Rust" in prompt
    assert "never assume Python, pip, PyPI" in prompt


@patch("actions.review_pr._read_head_file", return_value=None)
@patch("actions.review_pr._verified_local_checkout", return_value=True)
@patch("actions.review_pr.get_agent_response")
def test_generate_pr_review_uses_synchronous_response(
    mock_get_agent_response, mock_checkout, mock_read, tmp_path, monkeypatch
):
    """Test PR reviews avoid background polling for code diffs."""
    monkeypatch.chdir(tmp_path)
    mock_get_agent_response.return_value = {"comments": [], "summary": "LGTM"}
    mock_event = MagicMock()
    mock_event.repository = "ultralytics/actions"
    mock_event.pr = {"number": 1, "head": {"sha": "headsha"}}
    api_response = MagicMock(status_code=200, text="")
    api_response.json.return_value = {"head": {"sha": "headsha"}}
    mock_event.get.return_value = api_response
    diff = """diff --git a/test.py b/test.py
--- a/test.py
+++ b/test.py
@@ -1 +1 @@
-old = True
+old = False
"""

    review = review_pr.generate_pr_review("ultralytics/actions", (diff, []), "Test PR", "", event=mock_event)

    assert review["summary"] == "LGTM"
    mock_get_agent_response.assert_called_once()
    kwargs = mock_get_agent_response.call_args.kwargs
    assert "background" not in kwargs
    assert kwargs["tools"][0]["type"] == "web_search"
    assert "filters" not in kwargs["tools"][0]
    assert {tool.get("name") for tool in kwargs["tools"] if tool.get("type") == "function"} == {
        "list_changed_files",
        "list_files",
        "read_diff",
        "read_file",
        "search_repo",
    }
    assert kwargs["max_turns"] == review_pr.MAX_AGENT_TURNS
    assert kwargs["max_cost"] == review_pr.REVIEW_COST_SOFT_LIMIT
    assert kwargs["reasoning_effort"] == "high"
    assert "FULL FILE CONTENTS" not in mock_get_agent_response.call_args.args[0][1]["content"]
    assert (
        "Do not flag package or version availability based on web search"
        in mock_get_agent_response.call_args.args[0][0]["content"]
    )
    assert "Never approve any added file larger than 1 MB" in mock_get_agent_response.call_args.args[0][0]["content"]


def test_review_agent_search_scans_local_checkout_only(tmp_path, monkeypatch):
    """Test repo search scans the local checkout without escaping it and is dropped without a checkout."""
    monkeypatch.chdir(tmp_path)
    (tmp_path / "secret.py").write_text("secret = True\n", encoding="utf-8")
    (tmp_path / "patterns.py").write_text("literal = '(a+)+$'\n", encoding="utf-8")
    (tmp_path / "large.json").write_text("x" * 500_001, encoding="utf-8")
    workflow = tmp_path / ".github" / "workflows" / "format.yml"
    workflow.parent.mkdir(parents=True)
    workflow.write_text("workflow = True\n", encoding="utf-8")
    outside = tmp_path.parent / "outside_secret.py"
    outside.write_text("outside_secret = True\n", encoding="utf-8")
    (tmp_path / "linked_secret.py").symlink_to(outside)

    monkeypatch.setenv("GIT_CONFIG_GLOBAL", "/dev/null")
    monkeypatch.setenv("GIT_CONFIG_SYSTEM", "/dev/null")
    subprocess.run(["git", "init", "-q"], check=True)
    subprocess.run(["git", "add", "."], check=True)
    subprocess.run(
        ["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "Fixture"], check=True
    )
    head_sha = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
    (tmp_path / "secret.py").write_text("uncommitted = True\n", encoding="utf-8")
    assert review_pr._verified_local_checkout(head_sha)
    tools, handlers = review_pr.build_review_agent_tools(head_sha=head_sha, local_checkout=True)

    assert "search_repo" in {t.get("name") for t in tools}
    assert "1: secret = True" in handlers["read_file"](path="secret.py", start_line=None, end_line=None)
    assert "is not a readable file at the PR head" in handlers["read_file"](
        path="linked_secret.py", start_line=None, end_line=None
    )
    assert "secret.py" in handlers["list_files"](path_glob=None).splitlines()
    assert ".github/workflows/format.yml" in handlers["list_files"](path_glob=".github/**").splitlines()
    assert "secret.py:1:secret = True" in handlers["search_repo"](query="secret = True", path_glob=None)
    assert "No matches found." == handlers["search_repo"](query="secret.*True", path_glob=None)
    assert "patterns.py:1:literal = '(a+)+$'" in handlers["search_repo"](query="(a+)+$", path_glob=None)
    assert "too large" in handlers["read_file"](path="large.json", start_line=None, end_line=None)
    assert ".github/workflows/format.yml:1:workflow = True" in handlers["search_repo"](
        query="workflow = True", path_glob=".github/**"
    )
    assert "No matches found." == handlers["search_repo"](query="outside_secret = True", path_glob=None)

    tools, handlers = review_pr.build_review_agent_tools(local_checkout=False)
    assert "search_repo" not in handlers
    assert "search_repo" not in {t.get("name") for t in tools}


def test_pr_head_sha_prefers_live_value():
    """Test head SHA resolution requires the live PR value."""
    event = MagicMock()
    event.repository = "org/repo"
    event.pr = {"number": 5, "head": {"sha": "old"}}
    live = MagicMock(status_code=200)
    live.json.return_value = {"head": {"sha": "new"}}
    event.get.return_value = live
    assert review_pr.Action.get_pr_head_sha(event) == "new"
    event.get.return_value = MagicMock(status_code=500)
    assert review_pr.Action.get_pr_head_sha(event) is None


def test_pr_diff_is_filtered_at_retrieval():
    """Test every PR diff consumer receives the same filtered diff and skipped-file list."""
    event = MagicMock(
        repository="org/repo",
        pr={"number": 1, "base": {"sha": "base"}, "head": {"sha": "head"}},
        headers_diff={},
        _pr_diff_cache=None,
    )
    event.get.return_value = MagicMock(
        status_code=200,
        text="diff --git a/code.py b/code.py\n+kept\ndiff --git a/generated/client.py b/generated/client.py\n+omitted",
    )

    event.get.return_value.json.return_value = {"number": 1, "base": {"sha": "base"}, "head": {"sha": "formatted"}}
    diff, skipped_files = review_pr.Action.get_pr_diff(event)
    assert review_pr.Action.get_pr_diff(event) == (diff, skipped_files)
    assert event.get.call_args_list == [
        call("https://api.github.com/repos/org/repo/pulls/1", hard=True),
        call("https://api.github.com/repos/org/repo/compare/base...formatted", headers={}),
    ]

    assert "+kept" in diff
    assert "generated/client.py" not in diff
    assert skipped_files == ["generated/client.py"]


def test_post_review_summary_numbers_reviews_and_logs_findings():
    """Test review publication is required and re-reviews are numbered with a findings log that outlives them."""
    event = MagicMock()
    event.repository = "org/repo"
    event.pr = {"number": 7}
    event.get_pr_head_sha.return_value = "abc"

    review_pr.post_review_summary(event, {"head_sha": "abc", "summary": "LGTM", "comments": []})

    assert event.post.call_args.kwargs["hard"] is True
    assert event.post.call_args.kwargs["json"]["body"].startswith(f"{review_pr.REVIEW_MARKER}\n")  # first: no number

    comment = {"file": "a.py", "line": 3, "side": "RIGHT", "severity": "HIGH", "message": "retry loop never exits"}
    review_pr.post_review_summary(event, {"head_sha": "abc", "summary": "Fix", "comments": [comment]}, 3)

    body = event.post.call_args.kwargs["json"]["body"]
    assert body.startswith(f"{review_pr.REVIEW_MARKER} 3")
    assert "**HIGH** `a.py:3` retry loop never exits" in body


def test_clear_previous_review_captures_history_and_deletes_inline_comments():
    """Test replacement reviews capture prior context, invalidate bot decisions, and remove only bot inline comments."""
    event = MagicMock()
    event.repository = "org/repo"
    event.pr = {"number": 7}
    event.get_username.return_value = "review-bot"
    event.paginate.side_effect = [
        [
            {
                "id": 1,
                "state": "APPROVED",
                "commit_id": "a" * 40,
                "body": f"{review_pr.REVIEW_MARKER} 1\n\nRetry loop needs a bound",
                "user": {"login": "review-bot"},
            },
            {
                "id": 2,
                "state": "COMMENTED",
                "commit_id": "b" * 40,
                "body": f"{review_pr.REVIEW_MARKER} 2\n\nBound is correct now",
                "user": {"login": "review-bot"},
            },
            {"id": 3, "state": "APPROVED", "body": "Human review", "user": {"login": "human"}},
            {"id": 6, "state": "COMMENTED", "body": "Other automation", "user": {"login": "review-bot"}},
        ],
        [
            {
                "id": 4,
                "pull_request_review_id": 1,
                "user": {"login": "review-bot"},
                "path": "a.py",
                "line": 3,
                "body": "⚠️ **HIGH**: retry loop never exits",
            },
            {"id": 5, "pull_request_review_id": 6, "user": {"login": "review-bot"}},
            {
                "id": 8,
                "pull_request_review_id": 3,
                "in_reply_to_id": 4,
                "user": {"login": "human"},
                "path": "a.py",
                "line": 3,
                "body": "intentional, the caller times out",
                "author_association": "MEMBER",
            },
            {
                "id": 9,
                "pull_request_review_id": 3,
                "user": {"login": "human"},
                "path": "b.py",
                "line": 9,
                "body": "unrelated comment on someone else's review",
            },
        ],
    ]

    history = review_pr.clear_previous_review(event)

    assert [r["number"] for r in history["reviews"]] == [1, 2]
    assert history["reviews"][1]["commit"] == "b" * 7
    assert history["reviews"][1]["body"] == "Bound is correct now"
    assert history["replies"] == [
        '- a.py:3 @human (MEMBER) on "⚠️ **HIGH**: retry loop never exits": intentional, the caller times out'
    ]
    assert history["others"] == ["- b.py:9 @human (NONE): unrelated comment on someone else's review"]
    event.put.assert_called_once_with(
        "https://api.github.com/repos/org/repo/pulls/7/reviews/1/dismissals",
        json={"message": "Superseded by new review"},
        hard=True,
    )
    event.delete.assert_called_once_with(
        "https://api.github.com/repos/org/repo/pulls/comments/4",
        expected_status=[200, 204, 404],
        hard=True,
    )


def test_incomplete_review_evidence_cannot_approve():
    """Test unavailable or truncated diffs produce comments rather than approvals."""
    event = MagicMock()
    event.repository = "org/repo"
    event.pr = {"number": 7}
    event.get_pr_head_sha.return_value = "abc"

    error = review_pr.generate_pr_review("org/repo", ("ERROR: UNABLE TO RETRIEVE DIFF.", []), "PR", "", event, "abc")
    review_pr.post_review_summary(event, error)
    assert event.post.call_args.kwargs["json"]["event"] == "COMMENT"

    review_pr.post_review_summary(event, {"head_sha": "abc", "summary": "LGTM", "comments": [], "diff_truncated": True})
    assert event.post.call_args.kwargs["json"]["event"] == "COMMENT"

    binary_review = review_pr.generate_pr_review("org/repo", ("", ["image.png"]), "PR", "", event, "abc")
    review_pr.post_review_summary(event, binary_review)
    assert event.post.call_args.kwargs["json"]["event"] == "COMMENT"

    with pytest.raises(KeyError):
        review_pr.post_review_summary(event, {"summary": "stale result", "comments": []})

    review_pr.post_review_summary(
        event, {"head_sha": "abc", "summary": "All changed files were skipped", "comments": []}
    )
    assert event.post.call_args.kwargs["json"]["event"] == "COMMENT"

    # A push during review generation posts against the reviewed commit instead of approving or failing the job
    event.get_pr_head_sha.return_value = "def"
    review_pr.post_review_summary(
        event, {"head_sha": "abc", "summary": "LGTM", "comments": [], "diff_files": {"a": {}}}
    )
    assert event.post.call_args.kwargs["json"]["event"] == "COMMENT"
    assert event.post.call_args.kwargs["json"]["commit_id"] == "abc"


def test_review_agent_tools_read_pr_head_via_api(tmp_path, monkeypatch):
    """Test review tools serve PR-head content via the GitHub API regardless of local checkout."""
    monkeypatch.chdir(tmp_path)
    event = MagicMock()
    event.repository = "org/repo"
    event.headers = {"Authorization": "Bearer x"}
    file_response = MagicMock(status_code=200, text="import torch\ntorch.zeros(1)\n")
    missing_response = MagicMock(status_code=404)
    tree_response = MagicMock(status_code=200)
    tree_response.json.return_value = {
        "tree": [{"path": "tests/test_python.py", "type": "blob"}, {"path": "docs", "type": "tree"}]
    }

    def fake_get(url, **kwargs):
        if "/git/trees/" in url:
            return tree_response
        if "missing.py" in url:
            return missing_response
        return MagicMock(status_code=500) if "flaky.py" in url else file_response

    event.get.side_effect = fake_get

    _, handlers = review_pr.build_review_agent_tools({}, "", event, "a" * 40, local_checkout=False)

    excerpt = handlers["read_file"](path="tests/test_python.py", start_line=1, end_line=1)
    assert "1: import torch" in excerpt
    assert handlers["read_file"](path="missing.py", start_line=None, end_line=None).endswith(
        "does not exist at the PR head."
    )
    with pytest.raises(RuntimeError):  # transient API errors must not read as "file missing"
        handlers["read_file"](path="flaky.py", start_line=None, end_line=None)

    tree_response.status_code = 500
    with pytest.raises(RuntimeError, match="list_files failed: HTTP 500"):
        handlers["list_files"](path_glob=None)
    tree_response.status_code = 200
    assert handlers["list_files"](path_glob=None).splitlines() == ["tests/test_python.py"]


@patch("actions.review_pr._fetch_head_file")
def test_get_repo_guidelines_fetches_pr_head(mock_fetch):
    """Test guidelines are fetched from the PR head via the GitHub API."""
    mock_fetch.side_effect = lambda event, sha, path: "Be nice" if path == "AGENTS.md" else None
    section = review_pr.get_repo_guidelines("gpt-5.6-sol", event=MagicMock(), head_sha="abc")
    assert "AGENTS.md" in section and "Be nice" in section
    assert "CONTRIBUTING.md" not in section
    assert review_pr.get_repo_guidelines("gpt-5.6-sol", event=None, head_sha="abc") == ""


def test_review_agent_tools_can_list_and_read_changed_file_diffs():
    """Test PR diff tools let the agent inspect every changed file on demand."""
    diff = "diff --git a/early.py b/early.py\n--- a/early.py\n+++ b/early.py\n@@ -1 +1 @@\n-old = True\n+old = False\ndiff --git a/nested/late.py b/nested/late.py\n--- a/nested/late.py\n+++ b/nested/late.py\n@@ -10 +10 @@\n-value = 1\n+value = 2"
    diff_files, augmented_diff = review_pr.parse_diff_files(diff)
    _, handlers = review_pr.build_review_agent_tools(diff_files, augmented_diff)

    changed_files = handlers["list_changed_files"](path_glob=None)
    assert "early.py (+1/-1)" in changed_files
    assert "nested/late.py (+1/-1)" in changed_files
    assert handlers["list_changed_files"](path_glob="nested/**") == "nested/late.py (+1/-1)"
    late_diff = handlers["read_diff"](path="nested/late.py", start_line=None, end_line=None)
    assert "R   10 +value = 2" in late_diff
    assert "L   10 -value = 1" in late_diff