ci: add pytests for the gantt chart scripts

Add some initial pytests to test new options and to prevent future
regressions.

Part-of: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/32637>

bin/ci/ci_gantt_chart.py

@@ -11,7 +11,6 @@ import argparse
 from datetime import datetime, timedelta, timezone
 from typing import Dict, List
 
-import gitlab
 import plotly.express as px
 import plotly.graph_objs as go
 from gitlab import Gitlab, base
@@ -159,7 +158,7 @@ def main(
     ci_timeout: float = 60,
 ):
     token = read_token(token)
-    gl = gitlab.Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True)
+    gl = Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True)
 
     pipeline, _ = get_gitlab_pipeline_from_url(gl, pipeline_url)
     fig: go.Figure = generate_gantt_chart(pipeline, ci_timeout)

bin/ci/ci_post_gantt.py

@@ -23,6 +23,11 @@ from gitlab.base import RESTObject
 from gitlab.v4.objects import Project, ProjectPipeline
 from gitlab_common import GITLAB_URL, get_gitlab_pipeline_from_url, read_token
 
+
+class MockGanttExit(Exception):
+    pass
+
+
 LAST_MARGE_EVENT_FILE = os.path.expanduser("~/.config/last_marge_event")
 
 
@@ -110,10 +115,8 @@ def main(
     ci_timeout: float = 60,
 ):
     log.basicConfig(level=log.INFO)
-
     token = read_token(token)
-
-    gl = gitlab.Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True)
+    gl = Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True)
 
     user = gl.users.get(marge_user_id)
     last_event_at = since if since else read_last_event_date_from_file()
@@ -160,6 +163,8 @@ def main(
                 log.info("Posting reply ...")
                 message = compose_message(file_name, file_url)
                 gitlab_post_reply_to_note(gl, event, message)
+            except MockGanttExit:
+                pass  # Allow tests to exit early without printing a traceback
             except Exception as e:
                 log.info(f"Failed to generate gantt chart, not posting reply.{e}")
                 traceback.print_exc()

bin/ci/test/test_gantt_chart.py

@@ -0,0 +1,201 @@
+from contextlib import suppress
+from datetime import datetime, timedelta
+from unittest import mock
+from unittest.mock import MagicMock, patch
+
+import ci_post_gantt
+import pytest
+from ci_gantt_chart import generate_gantt_chart
+from ci_post_gantt import Gitlab, MockGanttExit
+
+
+def create_mock_job(
+    name, id, status, created_at, queued_duration, started_at, finished_at=None
+):
+    mock_job = MagicMock()
+    mock_job.name = name
+    mock_job.status = status
+    mock_job.id = id
+    mock_job.created_at = created_at
+    mock_job.queued_duration = queued_duration
+    mock_job.started_at = started_at
+    mock_job.finished_at = finished_at
+    return mock_job
+
+
+@pytest.fixture
+def fake_pipeline():
+    current_time = datetime.fromisoformat("2024-12-17 23:54:13.940091+00:00")
+    created_at = current_time - timedelta(minutes=10)
+
+    job1 = create_mock_job(
+        name="job1",
+        id="1",
+        status="success",
+        created_at=created_at.isoformat(),
+        queued_duration=1,  # seconds
+        started_at=(created_at + timedelta(seconds=2)).isoformat(),
+        finished_at=(created_at + timedelta(minutes=1)).isoformat(),
+    )
+
+    mock_pipeline = MagicMock()
+    mock_pipeline.web_url = "https://gitlab.freedesktop.org/mesa/mesa/-/pipelines/9999"
+    mock_pipeline.duration = 600  # Total pipeline duration in seconds
+    mock_pipeline.created_at = created_at.isoformat()
+    mock_pipeline.yaml_errors = False
+    mock_pipeline.jobs.list.return_value = [job1]
+    return mock_pipeline
+
+
+def test_generate_gantt_chart(fake_pipeline):
+    fig = generate_gantt_chart(fake_pipeline)
+
+    fig_dict = fig.to_dict()
+    assert "data" in fig_dict
+
+    # Extract all job names from the "y" axis in the Gantt chart data
+    all_job_names = set()
+    for trace in fig_dict["data"]:
+        if "y" in trace:
+            all_job_names.update(trace["y"])
+
+    assert any(
+        "job1" in job for job in all_job_names
+    ), "job1 should be present in the Gantt chart"
+
+
+def test_ci_timeout(fake_pipeline):
+    fig = generate_gantt_chart(fake_pipeline, ci_timeout=1)
+
+    fig_dict = fig.to_dict()
+
+    timeout_line = None
+    for shape in fig_dict.get("layout", {}).get("shapes", []):
+        if shape.get("line", {}).get("dash") == "dash":
+            timeout_line = shape
+            break
+
+    assert timeout_line is not None, "Timeout line should exist in the Gantt chart"
+    timeout_x = timeout_line["x0"]
+
+    # Check that the timeout line is 1 minute after the pipeline creation time
+    pipeline_created_at = datetime.fromisoformat(fake_pipeline.created_at)
+    expected_timeout = pipeline_created_at + timedelta(minutes=1)
+    assert (
+        timeout_x == expected_timeout
+    ), f"Timeout should be at {expected_timeout}, got {timeout_x}"
+
+
+def test_marge_bot_user_id():
+    with patch("ci_post_gantt.Gitlab") as MockGitlab:
+        mock_gitlab_instance = MagicMock(spec=Gitlab)
+        mock_gitlab_instance.users = MagicMock()
+        MockGitlab.return_value = mock_gitlab_instance
+
+        marge_bot_user_id = 12345
+        ci_post_gantt.main("fake_token", None, marge_bot_user_id)
+        mock_gitlab_instance.users.get.assert_called_once_with(marge_bot_user_id)
+
+
+def test_project_ids():
+    current_time = datetime.now()
+    project_id_1 = 176
+    event_1 = MagicMock()
+    event_1.project_id = project_id_1
+    event_1.created_at = (current_time - timedelta(days=1)).isoformat()
+    event_1.note = {"body": f"Event for project {project_id_1}"}
+
+    project_id_2 = 166
+    event_2 = MagicMock()
+    event_2.project_id = project_id_2
+    event_2.created_at = (current_time - timedelta(days=2)).isoformat()
+    event_2.note = {"body": f"Event for project {project_id_2}"}
+
+    with patch("ci_post_gantt.Gitlab") as MockGitlab:
+        mock_user = MagicMock()
+        mock_user.events = MagicMock()
+        mock_user.events.list.return_value = [event_1, event_2]
+
+        mock_gitlab_instance = MagicMock(spec=Gitlab)
+        mock_gitlab_instance.users = MagicMock()
+        mock_gitlab_instance.users.get.return_value = mock_user
+        MockGitlab.return_value = mock_gitlab_instance
+
+        last_event_date = (current_time - timedelta(days=3)).isoformat()
+
+        # Test a single project id
+        ci_post_gantt.main("fake_token", last_event_date)
+        marge_bot_single_project_scope = [
+            event.note["body"]
+            for event in mock_user.events.list.return_value
+            if event.project_id == project_id_1
+        ]
+        assert f"Event for project {project_id_1}" in marge_bot_single_project_scope
+        assert f"Event for project {project_id_2}" not in marge_bot_single_project_scope
+
+        # Test multiple project ids
+        ci_post_gantt.main(
+            "fake_token", last_event_date, 9716, [project_id_1, project_id_2]
+        )
+
+        marge_bot_multiple_project_scope = [
+            event.note["body"] for event in mock_user.events.list.return_value
+        ]
+        assert f"Event for project {project_id_1}" in marge_bot_multiple_project_scope
+        assert f"Event for project {project_id_2}" in marge_bot_multiple_project_scope
+
+
+def test_add_gantt_after_pipeline_message():
+    current_time = datetime.now()
+
+    plain_url = "https://gitlab.freedesktop.org/mesa/mesa/-/pipelines/12345"
+    plain_message = (
+        f"I couldn't merge this branch: CI failed! See pipeline {plain_url}."
+    )
+    event_plain = MagicMock()
+    event_plain.project_id = 176
+    event_plain.created_at = (current_time - timedelta(days=1)).isoformat()
+    event_plain.note = {"body": plain_message}
+
+    summary_url = "https://gitlab.freedesktop.org/mesa/mesa/-/pipelines/99999"
+    summary_message = (
+        "I couldn't merge this branch: "
+        f"CI failed! See pipeline {summary_url}.<br>There were problems with job:"
+        "[lavapipe](https://gitlab.freedesktop.org/mesa/mesa/-/jobs/68141218)<details><summary> "
+        "3 crashed tests</summary>dEQP-VK.ray_query.builtin.instancecustomindex.frag.aabbs,Crash<br>dEQP"
+        "-VK.ray_query.builtin.objecttoworld.frag.aabbs,Crash<br>dEQP-VK.sparse_resources.shader_intrinsics."
+        "2d_array_sparse_fetch.g16_b16r16_2plane_444_unorm.11_37_3_nontemporal,Crash<br></details>"
+    )
+    event_with_summary = MagicMock()
+    event_with_summary.project_id = 176
+    event_with_summary.created_at = (current_time - timedelta(days=1)).isoformat()
+    event_with_summary.note = {"body": summary_message}
+
+    with patch("ci_post_gantt.Gitlab") as MockGitlab, patch(
+        "ci_post_gantt.get_gitlab_pipeline_from_url", return_value=None
+    ) as mock_get_gitlab_pipeline_from_url:
+
+        def safe_mock(*args, **kwargs):
+            with suppress(TypeError):
+                raise MockGanttExit("Exiting for test purposes")
+
+        mock_get_gitlab_pipeline_from_url.side_effect = safe_mock
+
+        mock_user = MagicMock()
+        mock_user.events = MagicMock()
+        mock_user.events.list.return_value = [event_plain, event_with_summary]
+
+        mock_gitlab_instance = MagicMock(spec=Gitlab)
+        mock_gitlab_instance.users = MagicMock()
+        mock_gitlab_instance.users.get.return_value = mock_user
+        MockGitlab.return_value = mock_gitlab_instance
+
+        last_event_date = (current_time - timedelta(days=3)).isoformat()
+        ci_post_gantt.main("fake_token", last_event_date, 12345)
+        mock_get_gitlab_pipeline_from_url.assert_has_calls(
+            [
+                mock.call(mock_gitlab_instance, plain_url),
+                mock.call(mock_gitlab_instance, summary_url),
+            ],
+            any_order=True,
+        )