test(reports): restore WORKING-row invariant check, use UUID in log filter

Address copilot review feedback on #41966: assert_log dropped its check that
WORKING execution-log rows carry no value/value_row_json, and the single-log
regression test filtered ReportExecutionLog.uuid with a raw string instead of
a UUID (production casts task_id to UUID before writing).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

superset/commands/report/execute.py

@@ -223,22 +223,47 @@ class BaseReportState:
     def create_log(self, error_message: Optional[str] = None) -> None:
         """
         Creates a Report execution log, uses the current computed last_value for Alerts
+
+        A single execution transitions through the WORKING state before reaching a
+        terminal state (SUCCESS/ERROR/NOOP/GRACE). To avoid duplicated rows in the
+        execution-log view (see issue #29857), the terminal result is written onto
+        the WORKING "trigger" row created earlier in the same execution rather than
+        inserting a second row. The WORKING row is only used as a placeholder while
+        the execution is in flight (``find_last_entered_working_log`` relies on it
+        for working-timeout detection), so promoting it in place keeps one row per
+        execution ``uuid`` without losing that behavior. The intentional
+        error-notification marker row is a terminal-to-terminal transition, so it is
+        still recorded as a distinct row.
         """
         from sqlalchemy.orm.exc import StaleDataError
 
         try:
-            log = ReportExecutionLog(
-                scheduled_dttm=self._scheduled_dttm,
-                start_dttm=self._start_dttm,
-                end_dttm=datetime.utcnow(),
-                value=self._report_schedule.last_value,
-                value_row_json=self._report_schedule.last_value_row_json,
-                state=self._report_schedule.last_state,
-                error_message=error_message,
-                report_schedule=self._report_schedule,
-                uuid=self._execution_id,
+            # Reuse the in-flight WORKING trigger row for this execution, if any,
+            # so a single execution surfaces as a single log entry.
+            log = (
+                db.session.query(ReportExecutionLog)
+                .filter(
+                    ReportExecutionLog.uuid == self._execution_id,
+                    ReportExecutionLog.state == ReportState.WORKING,
+                    ReportExecutionLog.error_message.is_(None),
+                )
+                .first()
+                if self._report_schedule.last_state != ReportState.WORKING
+                else None
             )
-            db.session.add(log)
+            if log is None:
+                log = ReportExecutionLog(
+                    scheduled_dttm=self._scheduled_dttm,
+                    start_dttm=self._start_dttm,
+                    report_schedule=self._report_schedule,
+                    uuid=self._execution_id,
+                )
+                db.session.add(log)
+            log.end_dttm = datetime.utcnow()
+            log.value = self._report_schedule.last_value
+            log.value_row_json = self._report_schedule.last_value_row_json
+            log.state = self._report_schedule.last_state
+            log.error_message = error_message
             db.session.commit()  # pylint: disable=consider-using-transaction
         except StaleDataError as ex:
             # Report schedule was modified or deleted by another process

tests/integration_tests/reports/commands_tests.py

@@ -18,7 +18,7 @@ from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 from unittest.mock import call, Mock, patch
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 from flask.ctx import AppContext
@@ -159,15 +159,26 @@ def assert_log(state: str, error_message: Optional[str] = None):
     db.session.commit()
     logs = db.session.query(ReportExecutionLog).all()
 
-    if state == ReportState.ERROR:
-        # On error we send an email
-        assert len(logs) == 3
-    else:
+    if state == ReportState.WORKING:
+        # A report that is already in the WORKING state logs an extra WORKING row
+        # for the refused re-computation, on top of the row seeded by the fixture.
         assert len(logs) == 2
+    elif state == ReportState.ERROR:
+        # On error we also send a notification, which is recorded as a separate
+        # error-notification marker row.
+        assert len(logs) == 2
+    else:
+        # A single execution yields a single log row: the terminal result replaces
+        # the WORKING "trigger" row rather than adding a second row (issue #29857).
+        assert len(logs) == 1
     log_states = [log.state for log in logs]
-    assert ReportState.WORKING in log_states
     assert state in log_states
-    assert error_message in [log.error_message for log in logs]
+    # Previously a standalone WORKING "trigger" row always contributed a ``None``
+    # error_message, so the default ``None`` match was trivially satisfied and never
+    # verified the terminal row. With the result now written onto that single row,
+    # only assert an explicitly expected message.
+    if error_message is not None:
+        assert error_message in [log.error_message for log in logs]
 
     for log in logs:
         if log.state == ReportState.WORKING:
@@ -772,6 +783,45 @@ def test_email_chart_report_schedule(
         assert_log(ReportState.SUCCESS)
 
 
+@pytest.mark.usefixtures(
+    "load_birth_names_dashboard_with_slices", "create_report_email_chart"
+)
+@patch("superset.reports.notifications.email.send_email_smtp")
+@patch("superset.utils.screenshots.ChartScreenshot.get_screenshot")
+def test_email_chart_report_schedule_single_log_per_execution(
+    screenshot_mock,
+    email_mock,
+    create_report_email_chart,
+):
+    """
+    ExecuteReport Command: a single execution should produce a single log row.
+
+    Regression for #29857: the Alerts & Reports execution log shows duplicated
+    entries for a single execution. Each execution transitions through the
+    WORKING state and then a terminal state (SUCCESS/ERROR), and every
+    transition writes a ReportExecutionLog row sharing the same execution
+    ``uuid``. As a result one execution surfaces as two rows in the log view
+    (the "trigger" row and the "result" row). This test asserts that one
+    execution -- identified by its execution uuid -- yields exactly one log row.
+    """
+    screenshot_mock.return_value = SCREENSHOT_FILE
+
+    with freeze_time("2020-01-01T00:00:00Z"):
+        AsyncExecuteReportScheduleCommand(
+            TEST_ID, create_report_email_chart.id, datetime.utcnow()
+        ).run()
+
+        db.session.commit()
+        logs = (
+            db.session.query(ReportExecutionLog)
+            .filter(ReportExecutionLog.uuid == UUID(TEST_ID))
+            .all()
+        )
+        # A single execution (one uuid) must map to a single log entry, not one
+        # row per intermediate state transition.
+        assert len(logs) == 1
+
+
 @pytest.mark.usefixtures(
     "load_birth_names_dashboard_with_slices", "create_report_email_chart_alpha_owner"
 )

tests/unit_tests/commands/report/execute_test.py

@@ -2186,6 +2186,9 @@ def test_create_log_success_commits(mocker: MockerFixture) -> None:
     state._report_schedule = schedule
 
     mock_db = mocker.patch("superset.commands.report.execute.db")
+    # No in-flight WORKING "trigger" row exists for this execution, so create_log
+    # inserts a fresh row rather than promoting an existing one.
+    mock_db.session.query.return_value.filter.return_value.first.return_value = None
     mock_log_cls = mocker.patch(
         "superset.commands.report.execute.ReportExecutionLog",
         return_value=mocker.Mock(),