Bug 1066272 - Add the possibility to display unscheduled tasks (#9080)

* Bug 1066272 - part 1: Add Pulse binding for task-defined exchange

* Bug 1066272 - part 2: Map task-defined exchange to unscheduled state

* Bug 1066272 - part 3: Implement handle_task_defined for unscheduled tasks

* Bug 1066272 - part 4: Route task-defined messages to handler

* Bug 1066272 - part 5: Allow unscheduled to pending/running/completed transitions

* Bug 1066272 - part 6: Add unscheduled to job counter

* Bug 1066272 - part 7: Add CSS class for unscheduled jobs

Use turquoise as more contrastful color for unscheduled tasks

Co-authored-by: Sebastian Hengst <aryx.github@gmx-topmail.de>

* Bug 1066272 - part 8: Add unscheduled filter with separate toggle

* Bug 1066272 - part 9: Add integration tests for unscheduled task lifecycle

* Bug 1066272 - part 10: Hide unscheduled tasks by default

* Bug 1066272 - part 11: Add shortcut key 's' to toggle display of unscheduled tasks

* Bug 1066272 - part 12: Add unscheduled task to user guide

---------

Co-authored-by: Sebastian Hengst <aryx.github@gmx-topmail.de>

Author: JohanLorenzo

tests/etl/taskcluster_pulse/__init__.py

@@ -0,0 +1,87 @@
+import pytest
+
+from treeherder.etl.taskcluster_pulse.handler import handle_message, handle_task_defined
+
+
+@pytest.mark.asyncio
+async def test_handle_message_routes_task_defined():
+    task = {
+        "metadata": {
+            "name": "test-task",
+            "description": "Test task",
+            "owner": "test@example.com",
+        },
+        "created": "2025-01-01T00:00:00.000Z",
+        "workerType": "test-worker",
+        "tags": {},
+        "routes": ["tc-treeherder.v2.autoland.abc123"],
+        "extra": {
+            "treeherder": {
+                "symbol": "T",
+                "tier": 1,
+            }
+        },
+    }
+
+    message = {
+        "exchange": "exchange/taskcluster-queue/v1/task-defined",
+        "root_url": "https://firefox-ci-tc.services.mozilla.com",
+        "payload": {
+            "runId": 0,
+            "status": {
+                "taskId": "AJBb7wqZT6K9kz4niYAatg",
+                "state": "unscheduled",
+                "runs": [],
+            },
+        },
+    }
+
+    result = await handle_message(message, task)
+
+    assert len(result) == 1
+    assert result[0]["state"] == "unscheduled"
+    assert result[0]["result"] == "unknown"
+
+
+def test_handle_task_defined():
+    push_info = {
+        "project": "autoland",
+        "revision": "abc123",
+        "origin": "hg.mozilla.org",
+        "id": "12345",
+    }
+
+    task = {
+        "metadata": {
+            "name": "test-task",
+            "description": "Test task",
+            "owner": "test@example.com",
+        },
+        "created": "2025-01-01T00:00:00.000Z",
+        "workerType": "test-worker",
+        "tags": {},
+        "extra": {
+            "treeherder": {
+                "symbol": "T",
+                "tier": 1,
+            }
+        },
+    }
+
+    message = {
+        "exchange": "exchange/taskcluster-queue/v1/task-defined",
+        "payload": {
+            "status": {
+                "taskId": "AJBb7wqZT6K9kz4niYAatg",
+                "state": "unscheduled",
+                "runs": [],
+            },
+        },
+    }
+
+    result = handle_task_defined(push_info, task, message)
+
+    assert result is not None
+    assert isinstance(result, dict)
+    assert result["buildMachine"]["name"] == "unknown"
+    assert result["origin"]["project"] == "autoland"

tests/etl/taskcluster_pulse/test_handler.py

@@ -307,3 +307,132 @@ def test_ingest_job_with_updated_job_group(
 
     assert second_job.job_group.name == second_job_datum["job"]["group_name"]
     assert first_job.job_group.name == first_job_datum["job"]["group_name"]
+
+
+def test_unscheduled_task_full_lifecycle(
+    test_repository, failure_classifications, sample_data, sample_push, mock_log_parser
+):
+    """Test complete lifecycle: unscheduled -> pending -> running -> completed"""
+    job_datum = copy.deepcopy(sample_data.job_data[0])
+    job_guid = job_datum["job"]["job_guid"]
+    original_logs = job_datum["job"].get("logs", [])
+
+    # 1. Ingest as unscheduled
+    job_datum["job"]["state"] = "unscheduled"
+    job_datum["job"]["result"] = "unknown"
+    if "logs" in job_datum["job"]:
+        del job_datum["job"]["logs"]
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    assert Job.objects.count() == 1
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "unscheduled"
+    assert job.result == "unknown"
+
+    # 2. Transition to pending
+    job_datum["job"]["state"] = "pending"
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "pending"
+    assert job.result == "unknown"
+
+    # 3. Transition to running
+    job_datum["job"]["state"] = "running"
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "running"
+    assert job.result == "unknown"
+
+    # 4. Transition to completed
+    job_datum["job"]["state"] = "completed"
+    job_datum["job"]["result"] = "success"
+    if original_logs:
+        job_datum["job"]["logs"] = original_logs
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "completed"
+    assert job.result == "success"
+
+
+def test_unscheduled_task_out_of_order(
+    test_repository, failure_classifications, sample_data, sample_push, mock_log_parser
+):
+    """Test receiving pending before task-defined"""
+    job_datum = copy.deepcopy(sample_data.job_data[0])
+    job_guid = job_datum["job"]["job_guid"]
+
+    # 1. Ingest as pending first
+    job_datum["job"]["state"] = "pending"
+    job_datum["job"]["result"] = "unknown"
+    if "logs" in job_datum["job"]:
+        del job_datum["job"]["logs"]
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    assert Job.objects.count() == 1
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "pending"
+
+    # 2. Try to transition back to unscheduled (should be rejected)
+    job_datum["job"]["state"] = "unscheduled"
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "pending"  # Should stay pending
+
+
+def test_unscheduled_to_completed_direct(
+    test_repository, failure_classifications, sample_data, sample_push, mock_log_parser
+):
+    """Test task that completes without pending/running states"""
+    job_datum = copy.deepcopy(sample_data.job_data[0])
+    job_guid = job_datum["job"]["job_guid"]
+
+    # 1. Ingest as unscheduled
+    job_datum["job"]["state"] = "unscheduled"
+    job_datum["job"]["result"] = "unknown"
+    if "logs" in job_datum["job"]:
+        del job_datum["job"]["logs"]
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    assert Job.objects.count() == 1
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "unscheduled"
+
+    # 2. Jump directly to completed (valid for tasks that fail immediately)
+    job_datum["job"]["state"] = "completed"
+    job_datum["job"]["result"] = "exception"
+    job_datum["job"]["logs"] = []
+    test_utils.do_job_ingestion(test_repository, [job_datum], sample_push)
+
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "completed"
+    assert job.result == "exception"
+
+
+def test_unscheduled_ingestion_idempotent(
+    test_repository, failure_classifications, sample_data, sample_push, mock_log_parser
+):
+    """Test that ingesting the same unscheduled job twice doesn't create duplicates"""
+    job_data = copy.deepcopy(sample_data.job_data[:1])
+    job_guid = job_data[0]["job"]["job_guid"]
+
+    # Create unscheduled job
+    job_data[0]["job"]["state"] = "unscheduled"
+    job_data[0]["job"]["result"] = "unknown"
+    if "logs" in job_data[0]["job"]:
+        del job_data[0]["job"]["logs"]
+    test_utils.do_job_ingestion(test_repository, job_data, sample_push)
+
+    assert Job.objects.count() == 1
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "unscheduled"
+
+    # Ingest same unscheduled job again (should not create duplicate)
+    test_utils.do_job_ingestion(test_repository, job_data, sample_push)
+
+    assert Job.objects.count() == 1
+    job = Job.objects.get(guid=job_guid)
+    assert job.state == "unscheduled"

tests/etl/test_job_ingestion.py

@@ -360,23 +360,45 @@ def test_transition_pending_retry_fail_stays_retry(
     change_state_result(first_job, jl, "completed", "fail", "completed", "retry")
 
 
-def test_skip_unscheduled(first_job, failure_classifications, mock_log_parser):
+def test_transition_unscheduled_pending_running_complete(
+    first_job, failure_classifications, mock_log_parser
+):
+    jl = JobLoader()
+
+    change_state_result(first_job, jl, "unscheduled", "unknown", "unscheduled", "unknown")
+    change_state_result(first_job, jl, "pending", "unknown", "pending", "unknown")
+    change_state_result(first_job, jl, "running", "unknown", "running", "unknown")
+    change_state_result(first_job, jl, "completed", "success", "completed", "success")
+
+
+def test_transition_pending_unscheduled_stays_pending(
+    first_job, failure_classifications, mock_log_parser
+):
     jl = JobLoader()
-    first_job["state"] = "unscheduled"
-    jl.process_job(first_job, "https://firefox-ci-tc.services.mozilla.com")
 
-    assert not Job.objects.count()
+    change_state_result(first_job, jl, "pending", "unknown", "pending", "unknown")
+    change_state_result(first_job, jl, "unscheduled", "unknown", "pending", "unknown")
+
+
+def test_transition_running_unscheduled_stays_running(
+    first_job, failure_classifications, mock_log_parser
+):
+    jl = JobLoader()
+
+    change_state_result(first_job, jl, "running", "unknown", "running", "unknown")
+    change_state_result(first_job, jl, "unscheduled", "unknown", "running", "unknown")
 
 
 def change_state_result(test_job, job_loader, new_state, new_result, exp_state, exp_result):
     # make a copy so we can modify it and not affect other tests
     job = copy.deepcopy(test_job)
     job["state"] = new_state
     job["result"] = new_result
-    if new_state == "pending":
-        # pending jobs wouldn't have logs and our store_job_data doesn't
+    if new_state in ("pending", "unscheduled"):
+        # pending/unscheduled jobs wouldn't have logs and our store_job_data doesn't
         # support it.
-        del job["logs"]
+        if "logs" in job:
+            del job["logs"]
         errorsummary_indices = [
             i
             for i, item in enumerate(job["jobInfo"].get("links", []))

tests/etl/test_job_loader.py

@@ -0,0 +1,12 @@
+import { thAllResultStatuses } from '../../../ui/helpers/constants';
+import { thDefaultFilterResultStatuses } from '../../../ui/helpers/filter';
+
+describe('Filter constants', () => {
+  it('thAllResultStatuses includes unscheduled', () => {
+    expect(thAllResultStatuses).toContain('unscheduled');
+  });
+
+  it('thDefaultFilterResultStatuses does not include unscheduled', () => {
+    expect(thDefaultFilterResultStatuses).not.toContain('unscheduled');
+  });
+});