# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Integration tests for the cross-entity activity-view API (sc-107283).

US1 — dashboard activity stream: ``GET /api/v1/dashboard/<uuid>/activity/``.
Tests for US2 (chart activity) and US3 (dataset activity) come in later
phases.

Per spec T053 / sc-103156 T062, every test that mutates a fixture entity
wraps the test body in ``try``/``finally`` with
``metadata_db.session.rollback()`` in the ``finally``. The rationale is
documented in the spec — Continuum captures dirty mappers during
autoflush, so leaving an instrumented attribute dirty pollutes
downstream tests via the shadow tables.
"""

from __future__ import annotations

from typing import Any

import pytest

from superset.connectors.sqla.models import SqlaTable
from superset.extensions import db
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from superset.utils import json as _json
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.constants import ADMIN_USERNAME, ALPHA_USERNAME
from tests.integration_tests.fixtures.birth_names_dashboard import (  # noqa: F401
    load_birth_names_dashboard_with_slices,
    load_birth_names_data,
)


def _get_birth_names_dataset() -> SqlaTable:
    return (
        db.session.query(SqlaTable)
        .filter(SqlaTable.table_name == "birth_names")
        .first()
    )


def _persist_fixture_state() -> None:
    """Force the fixture's pending INSERTs to commit so subsequent edits
    produce *new* version rows instead of being batched into the
    creation transaction. Mirrors the same helper in
    ``tests/integration_tests/dashboards/version_history_tests.py``.
    """
    db.session.commit()


def _get_birth_names_dashboard() -> Dashboard:
    return (
        db.session.query(Dashboard)
        .filter(Dashboard.dashboard_title == "USA Births Names")
        .first()
    )


class TestDashboardActivityView(SupersetTestCase):
    """T017–T026 — ``GET /api/v1/dashboard/<uuid>/activity/`` (US1)."""

    @pytest.fixture(autouse=True)
    def _load_data(self, load_birth_names_dashboard_with_slices):  # noqa: PT004, F811
        pass

    def _activity(self, dashboard_uuid: str, **query: Any) -> Any:
        return self.client.get(
            f"/api/v1/dashboard/{dashboard_uuid}/activity/",
            query_string=query,
        )

    # ---- 4xx boundary cases ----

    def test_activity_returns_404_for_unknown_uuid(self) -> None:
        """AV-009: unknown path entity → 404."""
        self.login(ADMIN_USERNAME)
        rv = self._activity("00000000-0000-0000-0000-000000000000")
        assert rv.status_code == 404

    def test_activity_returns_400_for_invalid_uuid(self) -> None:
        """A malformed UUID is rejected by the endpoint, not by Werkzeug."""
        self.login(ADMIN_USERNAME)
        rv = self._activity("not-a-uuid")
        assert rv.status_code == 400

    def test_activity_returns_400_for_invalid_include(self) -> None:
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dashboard.uuid), include="sibling")
        assert rv.status_code == 400

    def test_activity_returns_400_for_invalid_since(self) -> None:
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dashboard.uuid), since="yesterday")
        assert rv.status_code == 400

    def test_activity_denies_non_owner(self) -> None:
        """Mirrors sc-103156 T056 — Alpha doesn't own the admin-fixture
        dashboard, so raise_for_ownership rejects with 403 before the
        activity layer runs."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)

        self.login(ALPHA_USERNAME)
        rv = self._activity(dashboard_uuid)
        assert rv.status_code == 403

    # ---- 200 happy paths ----

    def test_activity_returns_200_with_envelope_shape(self) -> None:
        """Smoke test: the endpoint returns the documented envelope shape
        (``result`` list + ``count`` integer) even when the dashboard has
        no activity yet."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)

        self.login(ADMIN_USERNAME)
        rv = self._activity(dashboard_uuid)
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        assert "result" in body
        assert "count" in body
        assert isinstance(body["result"], list)
        assert isinstance(body["count"], int)

    def test_activity_includes_chart_edit_as_related(self) -> None:
        """T018 / AS-1 of US1: editing a chart on the dashboard surfaces
        the chart-edit record with ``entity_kind=Slice`` and
        ``source=related``."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        chart_on_dashboard = next(iter(dashboard.slices), None)
        assert chart_on_dashboard is not None
        chart_id = chart_on_dashboard.id
        original_name = chart_on_dashboard.slice_name

        try:
            chart_on_dashboard.slice_name = f"{original_name} (edited)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dashboard_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            related = [
                r
                for r in body["result"]
                if r["entity_kind"] == "chart" and r["source"] == "related"
            ]
            assert related, (
                "Expected at least one Slice/related record from the chart "
                "edit; got: "
                f"{[(r['entity_kind'], r['source']) for r in body['result']]}"
            )
            # Spot-check the carry-through of denormalized fields
            sample = related[0]
            assert sample["entity_uuid"] is not None
            assert "transaction_id" in sample
            assert "issued_at" in sample
        finally:
            db.session.rollback()
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = original_name
            db.session.commit()

    def test_activity_include_self_excludes_related(self) -> None:
        """T023 / AV-016: ``?include=self`` filters out related records."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        chart_on_dashboard = next(iter(dashboard.slices), None)
        assert chart_on_dashboard is not None
        chart_id = chart_on_dashboard.id
        original_name = chart_on_dashboard.slice_name

        try:
            chart_on_dashboard.slice_name = f"{original_name} (edited self)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dashboard_uuid, include="self")
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            for record in body["result"]:
                assert record["source"] == "self", (
                    f"include=self leaked a non-self record: {record}"
                )
                assert record["entity_kind"] == "dashboard"
        finally:
            db.session.rollback()
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = original_name
            db.session.commit()

    def test_activity_include_related_excludes_self(self) -> None:
        """T024 / AV-016: ``?include=related`` returns only related records."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        original_title = dashboard.dashboard_title
        dashboard_id = dashboard.id

        try:
            # Edit the dashboard's own field so we have a self record to
            # filter out, and edit a chart on it so we have a related
            # record to keep.
            dashboard.dashboard_title = f"{original_title} (edited dash)"
            db.session.commit()
            chart_on_dashboard = next(iter(dashboard.slices), None)
            assert chart_on_dashboard is not None
            chart_id = chart_on_dashboard.id
            chart_original_name = chart_on_dashboard.slice_name
            chart_on_dashboard.slice_name = f"{chart_original_name} (edited chart)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dashboard_uuid, include="related")
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            for record in body["result"]:
                assert record["source"] == "related", (
                    f"include=related leaked a self record: {record}"
                )
                assert record["entity_kind"] != "dashboard"
        finally:
            db.session.rollback()
            dashboard = (
                db.session.query(Dashboard).filter(Dashboard.id == dashboard_id).one()
            )
            dashboard.dashboard_title = original_title
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = chart_original_name
            db.session.commit()

    def test_activity_pagination_clamps_oversized_page_size(self) -> None:
        """``?page_size=500`` is silently clamped to the contract max
        (200) rather than rejected with 400."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dashboard.uuid), page_size="500")
        assert rv.status_code == 200

    def test_activity_ordering_is_stable_by_issued_at_then_transaction_id(self) -> None:
        """T040 / AV-006: records are ordered ``(issued_at DESC,
        transaction_id DESC)``. When two records share ``issued_at`` the
        tie-break is ``transaction_id`` — never random. We verify this by
        asserting the result list is monotonically non-increasing on the
        composite key, which would only hold under deterministic
        ordering."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dashboard.uuid))
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        records = body["result"]
        # Each pair of adjacent records must satisfy (prev >= cur) on the
        # composite (issued_at, transaction_id) — DESC ordering.
        # ``records[1:]`` is intentionally one element shorter than
        # ``records``; strict=False is the correct semantic for an
        # adjacent-pair iteration.
        for prev, cur in zip(records, records[1:], strict=False):
            assert (prev["issued_at"], prev["transaction_id"]) >= (
                cur["issued_at"],
                cur["transaction_id"],
            ), (
                f"Ordering broke at adjacent pair: "
                f"prev=({prev['issued_at']}, {prev['transaction_id']}) "
                f"cur=({cur['issued_at']}, {cur['transaction_id']})"
            )

    def test_activity_page_size_caps_returned_records_at_200(self) -> None:
        """T041: ``?page_size=500`` must return *at most* 200 records.
        Pairs with the no-400 check above: that test confirms the
        oversized request is accepted, this test confirms the response
        is bounded as the contract guarantees (AV-019 / spec
        ActivityResponseSchema documentation)."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dashboard.uuid), page_size="500")
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        assert len(body["result"]) <= 200, (
            f"page_size=500 returned {len(body['result'])} records; "
            "cap is 200 per the OpenAPI schema"
        )

    def test_activity_marks_hard_deleted_chart_with_tombstone(self) -> None:
        """T042 / D-15: when a chart was on the dashboard and has since
        been hard-deleted, the chart's historical change records still
        surface in the dashboard's activity stream, marked with
        ``entity_deleted: true`` and ``entity_uuid: null``. ``entity_name``
        is preserved from the last shadow row so the UI can show
        "(deleted) Girls" without a live row to query.

        Hard-delete pattern: edit the chart (creates a Slice change
        record), commit, then ``db.session.delete(chart); commit``.
        Continuum end-stamps the M2M row but does not cascade-delete
        the shadow rows, so the history is still reachable. The
        activity-view's tombstone check (``_check_entity_tombstones``)
        detects the missing live row and stamps the record."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        chart_to_delete = (
            db.session.query(Slice).filter(Slice.slice_name == "Girls").first()
        )
        assert chart_to_delete is not None
        original_name = chart_to_delete.slice_name

        try:
            # Step 1: generate a chart-edit change record for "Girls".
            chart_to_delete.slice_name = f"{original_name} (pre-delete edit)"
            db.session.commit()

            # Step 2: hard-delete the chart. The fixture's _cleanup will
            # tolerate this — its `Slice.id.in_(slice_ids)` filter
            # silently skips the missing row.
            db.session.delete(chart_to_delete)
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dashboard_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            tombstoned = [
                r
                for r in body["result"]
                if r["entity_kind"] == "chart" and r["entity_deleted"] is True
            ]
            seen = [
                (r["entity_kind"], r["entity_deleted"]) for r in body["result"][:10]
            ]
            assert tombstoned, (
                "Expected ≥1 tombstoned Slice record after the chart was "
                f"hard-deleted; got entity_deleted values: {seen}"
            )
            sample = tombstoned[0]
            got_uuid = sample["entity_uuid"]
            assert got_uuid is None, (
                f"Hard-deleted entity should have null entity_uuid; got {got_uuid!r}"
            )
            assert sample["entity_name"], (
                "entity_name should be recovered from the last shadow row; "
                f"got empty: {sample!r}"
            )
        finally:
            db.session.rollback()

    def test_activity_excludes_records_after_retention_prune(self) -> None:
        """T051 / AV-010: retention bounds the activity feed. After
        ``_prune_old_versions_impl`` drops shadow / change-record rows
        whose ``version_transaction.issued_at`` is older than the
        retention cutoff, the activity stream stops surfacing them.

        Test pattern: capture the highest ``version_transaction.id``
        before our edits, edit a chart (creating a new transaction),
        backdate that transaction's ``issued_at`` past the retention
        cutoff, run the prune, and assert the chart-edit no longer
        appears in the activity stream."""
        # pylint: disable=import-outside-toplevel
        from datetime import datetime, timedelta

        import sqlalchemy as sa
        from sqlalchemy_continuum import versioning_manager

        from superset.tasks.version_history_retention import (
            _prune_old_versions_impl,
        )

        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        chart = db.session.query(Slice).filter(Slice.slice_name == "Boys").first()
        assert chart is not None
        chart_id = chart.id
        original_name = chart.slice_name

        tx_table = versioning_manager.transaction_cls.__table__

        # Capture pre-edit max tx_id so we can identify the rows produced
        # by THIS test (and not backdate anything else).
        max_tx_before = (
            db.session.connection()
            .execute(sa.select(sa.func.max(tx_table.c.id)))
            .scalar()
            or 0
        )

        try:
            chart.slice_name = f"{original_name} (retention test)"
            db.session.commit()

            # Backdate the new transactions to before the 30-day cutoff.
            old_timestamp = datetime.utcnow() - timedelta(days=60)
            db.session.connection().execute(
                sa.update(tx_table)
                .where(tx_table.c.id > max_tx_before)
                .values(issued_at=old_timestamp)
            )
            db.session.commit()

            # Snapshot the activity-record count BEFORE the prune. With
            # ?page_size=200 + the highest possible page coverage, the
            # count field is the post-visibility filtered total.
            self.login(ADMIN_USERNAME)
            rv_before = self._activity(dashboard_uuid, page_size="200")
            assert rv_before.status_code == 200
            count_before = _json.loads(rv_before.data.decode("utf-8"))["count"]

            # Run the prune. The backdated tx rows are now > 30 days old
            # and should be deleted. AV-010 requires the prune to remove
            # at least the backdated transaction(s) we created.
            stats = _prune_old_versions_impl(retention_days=30)
            assert stats.get("pruned_transactions", 0) >= 1, (
                f"Prune should have removed our backdated tx; stats={stats}"
            )

            # After the prune, the activity endpoint still works and the
            # filtered count has DROPPED — change records joined to the
            # pruned transactions are no longer in the result set (the
            # join in _fetch_change_records drops them).
            rv_after = self._activity(dashboard_uuid, page_size="200")
            assert rv_after.status_code == 200
            count_after = _json.loads(rv_after.data.decode("utf-8"))["count"]
            assert count_after < count_before, (
                f"Activity count should decrease after prune; "
                f"before={count_before} after={count_after}"
            )
        finally:
            db.session.rollback()
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = original_name
            db.session.commit()

    def test_activity_pagination_is_deterministic_and_disjoint(self) -> None:
        """T039 / SC-AV-002 (pragmatic interpretation): two consecutive
        requests for the same page return identical results, and
        consecutive pages do not overlap.

        The spec's stricter "no skip/duplicate under concurrent writes"
        is unprovable with offset pagination — new top-inserted records
        shift every later page by one. Cursor pagination would solve
        this and is deferred per plan §D-10. Under THIS pagination
        scheme, the testable guarantees are: (a) the same request fired
        twice produces the same page (request determinism), and (b)
        page N and page N+1 share no record under the same request
        round. Both come from the stable
        ``(issued_at DESC, transaction_id DESC, sequence DESC)`` sort.
        """
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        self.login(ADMIN_USERNAME)

        rv1a = self._activity(dashboard_uuid, page="0", page_size="25")
        rv1b = self._activity(dashboard_uuid, page="0", page_size="25")
        rv2 = self._activity(dashboard_uuid, page="1", page_size="25")
        assert rv1a.status_code == 200
        assert rv1b.status_code == 200
        assert rv2.status_code == 200

        page0_first = _json.loads(rv1a.data.decode("utf-8"))["result"]
        page0_second = _json.loads(rv1b.data.decode("utf-8"))["result"]
        page1 = _json.loads(rv2.data.decode("utf-8"))["result"]

        # (a) Request determinism: same page twice → same records in same
        #     order. Use (entity_kind, entity_id_internal_proxy, tx, seq)
        #     fingerprint — entity_uuid + transaction_id is sufficient
        #     since entity_id isn't in the API contract.
        fingerprint = lambda r: (  # noqa: E731
            r["entity_kind"],
            r["entity_uuid"],
            r["transaction_id"],
            r["kind"],
            tuple(r["path"]) if r["path"] else (),
        )
        assert [fingerprint(r) for r in page0_first] == [
            fingerprint(r) for r in page0_second
        ], "page=0 fired twice returned different records"

        # (b) Page 0 and page 1 are disjoint under one request round.
        page0_keys = {fingerprint(r) for r in page0_first}
        page1_keys = {fingerprint(r) for r in page1}
        overlap = page0_keys & page1_keys
        assert not overlap, f"page=0 and page=1 returned overlapping records: {overlap}"

    @pytest.mark.xfail(
        reason=(
            "AV-015 requires sc-103156's restore code to emit a synthetic "
            "change record with kind='restore', path=['__meta__', "
            "'restored_from'], and to_value carrying the source version_uuid "
            "+ label. sc-103156's restore_version() currently does not emit "
            "this — it relies on the diff capture for the field changes the "
            "revert produces, which surface as kind='field' records. The "
            "activity-view layer correctly passes through whatever kind "
            "sc-103156 emits; this test will pass once the upstream "
            "emission lands. Tracking via the AV-015 contract in the spec; "
            "no code change required on the sc-107283 side."
        ),
        strict=True,
    )
    def test_activity_surfaces_dashboard_restore_event(self) -> None:
        """T044 / AV-015: restoring a dashboard to a prior version surfaces
        a ``kind='restore'`` record in the dashboard's own activity stream
        (``source='self'``). The restore is emitted by sc-103156's restore
        path and the activity layer passes it through without special-
        casing."""
        _persist_fixture_state()
        dashboard = _get_birth_names_dashboard()
        assert dashboard is not None
        dashboard_uuid = str(dashboard.uuid)
        dashboard_id = dashboard.id
        original_title = dashboard.dashboard_title

        try:
            # Two edits → at least two restorable prior versions.
            dashboard.dashboard_title = f"{original_title} v1"
            db.session.commit()
            dashboard.dashboard_title = f"{original_title} v2"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            # Find a prior version to restore to (version_number 0 is the
            # baseline; we restore to whichever earlier version the list
            # endpoint surfaces).
            versions_rv = self.client.get(
                f"/api/v1/dashboard/{dashboard_uuid}/versions/"
            )
            assert versions_rv.status_code == 200, versions_rv.data
            versions = _json.loads(versions_rv.data.decode("utf-8"))["result"]
            assert len(versions) >= 2, f"expected ≥2 versions, got {versions}"
            target_version_uuid = versions[0]["version_uuid"]  # earliest

            # Restore. The endpoint commits; finally clean up below.
            restore_rv = self.client.post(
                f"/api/v1/dashboard/{dashboard_uuid}"
                f"/versions/{target_version_uuid}/restore"
            )
            assert restore_rv.status_code == 200, restore_rv.data

            # Activity stream should now show a restore record on the
            # dashboard itself.
            rv = self._activity(dashboard_uuid, include="self")
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            restore_records = [
                r
                for r in body["result"]
                if r["kind"] == "restore" and r["entity_kind"] == "dashboard"
            ]
            assert restore_records, (
                "Expected at least one kind='restore' Dashboard record; "
                f"got kinds: {[r['kind'] for r in body['result'][:10]]}"
            )
        finally:
            db.session.rollback()
            dashboard = (
                db.session.query(Dashboard).filter(Dashboard.id == dashboard_id).one()
            )
            dashboard.dashboard_title = original_title
            db.session.commit()


class TestChartActivityView(SupersetTestCase):
    """T028–T032 — ``GET /api/v1/chart/<uuid>/activity/`` (US2).

    Chart activity = chart's own edits + datasets the chart pointed at
    during association. **No** dashboard records — even when the chart
    is on a dashboard, sibling-traversal is excluded per the spec's
    Relationship Traversal section (T032).
    """

    @pytest.fixture(autouse=True)
    def _load_data(self, load_birth_names_dashboard_with_slices):  # noqa: PT004, F811
        pass

    def _activity(self, chart_uuid: str, **query: Any) -> Any:
        return self.client.get(
            f"/api/v1/chart/{chart_uuid}/activity/",
            query_string=query,
        )

    def _get_birth_names_chart(self) -> Slice:
        return db.session.query(Slice).filter(Slice.slice_name == "Girls").first()

    # ---- 4xx boundary cases ----

    def test_chart_activity_returns_404_for_unknown_uuid(self) -> None:
        self.login(ADMIN_USERNAME)
        rv = self._activity("00000000-0000-0000-0000-000000000000")
        assert rv.status_code == 404

    def test_chart_activity_returns_400_for_invalid_uuid(self) -> None:
        self.login(ADMIN_USERNAME)
        rv = self._activity("not-a-uuid")
        assert rv.status_code == 400

    def test_chart_activity_returns_400_for_invalid_include(self) -> None:
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        assert chart is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(chart.uuid), include="upstream")
        assert rv.status_code == 400

    def test_chart_activity_denies_non_owner(self) -> None:
        """Same shape as the dashboard endpoint: Alpha lacks ownership
        on the admin-fixture chart so raise_for_ownership returns 403."""
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        assert chart is not None
        self.login(ALPHA_USERNAME)
        rv = self._activity(str(chart.uuid))
        assert rv.status_code == 403

    # ---- 200 happy paths ----

    def test_chart_activity_returns_200_with_envelope_shape(self) -> None:
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        assert chart is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(chart.uuid))
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        assert isinstance(body["result"], list)
        assert isinstance(body["count"], int)

    def test_chart_activity_self_edit_appears_as_self_record(self) -> None:
        """Editing the chart itself surfaces a ``source=self``,
        ``entity_kind=Slice`` record."""
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        assert chart is not None
        chart_id = chart.id
        chart_uuid = str(chart.uuid)
        original_name = chart.slice_name

        try:
            chart.slice_name = f"{original_name} (edited self)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(chart_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            self_records = [
                r
                for r in body["result"]
                if r["entity_kind"] == "chart" and r["source"] == "self"
            ]
            got = [(r["entity_kind"], r["source"]) for r in body["result"]]
            assert self_records, (
                f"Expected ≥1 Slice/self record from the chart edit; got: {got}"
            )
        finally:
            db.session.rollback()
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = original_name
            db.session.commit()

    def test_chart_activity_includes_dataset_edit_as_related(self) -> None:
        """T030 / AS-1 of US2: editing the chart's dataset surfaces a
        ``source=related``, ``entity_kind=SqlaTable`` record."""
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        dataset = _get_birth_names_dataset()
        assert chart is not None
        assert dataset is not None
        chart_uuid = str(chart.uuid)
        dataset_id = dataset.id
        original_description = dataset.description

        try:
            dataset.description = "edited for activity-view test"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(chart_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            related = [
                r
                for r in body["result"]
                if r["entity_kind"] == "dataset" and r["source"] == "related"
            ]
            assert related, (
                "Expected at least one SqlaTable/related record from the "
                "dataset edit; got: "
                f"{[(r['entity_kind'], r['source']) for r in body['result']]}"
            )
        finally:
            db.session.rollback()
            dataset = (
                db.session.query(SqlaTable).filter(SqlaTable.id == dataset_id).one()
            )
            dataset.description = original_description
            db.session.commit()

    def test_chart_activity_excludes_sibling_dashboards(self) -> None:
        """T032: Even when the chart is on a dashboard, dashboard edits
        do NOT appear in the chart's activity. Per the spec's Relationship
        Traversal section: charts don't see "sideways" to the dashboards
        they happen to be on."""
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        dashboard = _get_birth_names_dashboard()
        assert chart is not None
        assert dashboard is not None
        chart_uuid = str(chart.uuid)
        dashboard_id = dashboard.id
        original_title = dashboard.dashboard_title

        try:
            # Mutate the dashboard the chart is on — that edit MUST NOT
            # appear in the chart's activity stream.
            dashboard.dashboard_title = f"{original_title} (edited sibling)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(chart_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            for record in body["result"]:
                assert record["entity_kind"] != "dashboard", (
                    f"Dashboard edit leaked into chart's activity stream: {record}"
                )
        finally:
            db.session.rollback()
            dashboard = (
                db.session.query(Dashboard).filter(Dashboard.id == dashboard_id).one()
            )
            dashboard.dashboard_title = original_title
            db.session.commit()

    def test_chart_activity_include_self_excludes_related(self) -> None:
        """``?include=self`` filters out the dataset records."""
        _persist_fixture_state()
        chart = self._get_birth_names_chart()
        dataset = _get_birth_names_dataset()
        assert chart is not None
        assert dataset is not None
        chart_uuid = str(chart.uuid)
        dataset_id = dataset.id
        original_description = dataset.description

        try:
            dataset.description = "edited (self filter test)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(chart_uuid, include="self")
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            for record in body["result"]:
                assert record["source"] == "self"
                assert record["entity_kind"] == "chart"
        finally:
            db.session.rollback()
            dataset = (
                db.session.query(SqlaTable).filter(SqlaTable.id == dataset_id).one()
            )
            dataset.description = original_description
            db.session.commit()


class TestDatasetActivityView(SupersetTestCase):
    """T033–T036 — ``GET /api/v1/dataset/<uuid>/activity/`` (US3).

    Dataset activity = dataset's own edits only. **No** transitive layer
    in V2 (AV-004) — even when charts use the dataset, those chart edits
    do NOT appear here. ``?include=related`` and ``?include=all``
    collapse to the same self-only stream as ``?include=self``.
    """

    @pytest.fixture(autouse=True)
    def _load_data(self, load_birth_names_dashboard_with_slices):  # noqa: PT004, F811
        pass

    def _activity(self, dataset_uuid: str, **query: Any) -> Any:
        return self.client.get(
            f"/api/v1/dataset/{dataset_uuid}/activity/",
            query_string=query,
        )

    # ---- 4xx boundary cases ----

    def test_dataset_activity_returns_404_for_unknown_uuid(self) -> None:
        self.login(ADMIN_USERNAME)
        rv = self._activity("00000000-0000-0000-0000-000000000000")
        assert rv.status_code == 404

    def test_dataset_activity_returns_400_for_invalid_uuid(self) -> None:
        self.login(ADMIN_USERNAME)
        rv = self._activity("not-a-uuid")
        assert rv.status_code == 400

    def test_dataset_activity_returns_400_for_invalid_include(self) -> None:
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        assert dataset is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dataset.uuid), include="upstream")
        assert rv.status_code == 400

    def test_dataset_activity_denies_non_owner(self) -> None:
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        assert dataset is not None
        self.login(ALPHA_USERNAME)
        rv = self._activity(str(dataset.uuid))
        assert rv.status_code == 403

    # ---- 200 happy paths ----

    def test_dataset_activity_returns_200_with_envelope_shape(self) -> None:
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        assert dataset is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dataset.uuid))
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        assert isinstance(body["result"], list)
        assert isinstance(body["count"], int)

    def test_dataset_activity_includes_dataset_self_edits(self) -> None:
        """T036: the dataset's own scalar edits appear as ``source=self``,
        ``entity_kind=SqlaTable``."""
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        assert dataset is not None
        dataset_id = dataset.id
        dataset_uuid = str(dataset.uuid)
        original_description = dataset.description

        try:
            dataset.description = "edited self for dataset activity"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dataset_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            self_records = [
                r
                for r in body["result"]
                if r["entity_kind"] == "dataset" and r["source"] == "self"
            ]
            got = [(r["entity_kind"], r["source"]) for r in body["result"]]
            assert self_records, (
                f"Expected ≥1 SqlaTable/self record from the dataset edit; got: {got}"
            )
        finally:
            db.session.rollback()
            dataset = (
                db.session.query(SqlaTable).filter(SqlaTable.id == dataset_id).one()
            )
            dataset.description = original_description
            db.session.commit()

    def test_dataset_activity_excludes_chart_edits(self) -> None:
        """T035 / AS-1 / AV-004: When a chart that uses the dataset is
        edited, that edit does NOT appear in the dataset's activity stream.
        Datasets are read-only upstream in V2."""
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        chart = db.session.query(Slice).filter(Slice.slice_name == "Girls").first()
        assert dataset is not None
        assert chart is not None
        dataset_uuid = str(dataset.uuid)
        chart_id = chart.id
        chart_original_name = chart.slice_name

        try:
            # Edit the chart — generates a Slice change record. The
            # dataset's activity MUST NOT surface it.
            chart.slice_name = f"{chart_original_name} (edited from dataset test)"
            db.session.commit()

            self.login(ADMIN_USERNAME)
            rv = self._activity(dataset_uuid)
            assert rv.status_code == 200
            body = _json.loads(rv.data.decode("utf-8"))
            for record in body["result"]:
                assert record["entity_kind"] == "dataset", (
                    "Non-dataset record leaked into dataset's activity "
                    f"stream: {record}"
                )
                assert record["source"] == "self", (
                    f"Dataset activity contains a related record: {record}"
                )
        finally:
            db.session.rollback()
            chart = db.session.query(Slice).filter(Slice.id == chart_id).one()
            chart.slice_name = chart_original_name
            db.session.commit()

    def test_dataset_activity_related_only_returns_empty(self) -> None:
        """AV-004: datasets have no transitive layer. ``?include=related``
        returns an empty result list because there are no related entities
        to draw from."""
        _persist_fixture_state()
        dataset = _get_birth_names_dataset()
        assert dataset is not None
        self.login(ADMIN_USERNAME)
        rv = self._activity(str(dataset.uuid), include="related")
        assert rv.status_code == 200
        body = _json.loads(rv.data.decode("utf-8"))
        assert body["result"] == []
        assert body["count"] == 0


class TestActivityOpenApiSpec(SupersetTestCase):
    """T049 — confirm the three ``/activity/`` endpoints are surfaced by
    FAB-generated OpenAPI at ``/api/v1/_openapi``.

    ``base_api_tests.py::TestOpenApiSpec::test_open_api_spec`` already
    validates the full spec's YAML correctness on every CI run. This
    class adds activity-specific assertions: the paths exist, are
    documented with the expected query parameters, and reference an
    ``ActivityResponse``-shaped 200 response.
    """

    def _spec(self) -> dict[str, Any]:
        self.login(ADMIN_USERNAME)
        rv = self.client.get("/api/v1/_openapi")
        assert rv.status_code == 200, rv.status_code
        return _json.loads(rv.data.decode("utf-8"))

    def test_three_activity_paths_appear_in_openapi(self) -> None:
        """One path per endpoint family. Paths are keyed by the URL
        template, not the method name, so the FAB-generated keys are
        the ``/<uuid_str>/activity/`` route templates."""
        spec = self._spec()
        paths = spec.get("paths", {})
        # FAB templates the path-arg as ``{uuid_str}`` in the OpenAPI dict.
        expected = {
            "/api/v1/dashboard/{uuid_str}/activity/",
            "/api/v1/chart/{uuid_str}/activity/",
            "/api/v1/dataset/{uuid_str}/activity/",
        }
        missing = expected - paths.keys()
        assert not missing, f"missing activity paths in OpenAPI: {missing}"

    def test_activity_endpoints_document_query_params(self) -> None:
        """Each endpoint declares since / until / include / page /
        page_size as query parameters. Spot-check on the dashboard
        endpoint — the YAML docstring is the same shape across all
        three so this assertion is sufficient."""
        spec = self._spec()
        op = spec["paths"]["/api/v1/dashboard/{uuid_str}/activity/"]["get"]
        params = {p["name"]: p for p in op.get("parameters", [])}
        for expected in ("since", "until", "include", "page", "page_size"):
            assert expected in params, (
                f"query param {expected!r} missing from dashboard /activity/"
            )
        # include enum is the published contract — verify it's correct.
        include_param = params["include"]
        assert include_param["in"] == "query"
        assert set(include_param["schema"]["enum"]) == {"self", "related", "all"}

    def test_activity_endpoints_declare_200_response(self) -> None:
        """Each endpoint declares a 200 response. The exact schema
        reference depends on how FAB resolves ``schema: ActivityResponseSchema``
        in the YAML docstring; here we just confirm the 200 + the 4xx
        error responses are all present."""
        spec = self._spec()
        op = spec["paths"]["/api/v1/dashboard/{uuid_str}/activity/"]["get"]
        responses = op.get("responses", {})
        for code in ("200", "400", "401", "403", "404"):
            assert code in responses, (
                f"response code {code} missing on dashboard /activity/"
            )