diff --git a/src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py b/src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py index 1d03e2d2b26606..0e36292c6bc119 100644 --- a/src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py +++ b/src/sentry/preprod/api/endpoints/snapshots/preprod_artifact_snapshot.py @@ -7,6 +7,7 @@ import orjson from django.conf import settings from django.db import IntegrityError, router, transaction +from django.utils import timezone from rest_framework.request import Request from rest_framework.response import Response @@ -38,6 +39,10 @@ SnapshotDetailsApiResponse, SnapshotImageResponse, ) +from sentry.preprod.api.models.snapshots.snapshot_status import ( + SnapshotStatusInput, + derive_snapshot_status, +) from sentry.preprod.api.schemas import VCS_ERROR_MESSAGES, VCS_SCHEMA_PROPERTIES from sentry.preprod.helpers.deletion import delete_artifacts_and_eap_data from sentry.preprod.models import PreprodArtifact, PreprodComparisonApproval @@ -45,6 +50,7 @@ CategorizedComparison, categorize_comparison_images, ) +from sentry.preprod.snapshots.constants import MISSING_BASE_GRACE_PERIOD_SECONDS from sentry.preprod.snapshots.manifest import ( ComparisonManifest, ImageMetadata, @@ -273,16 +279,15 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) -> comparison_manifest: ComparisonManifest | None = None base_manifest: SnapshotManifest | None = None - comparison = ( - PreprodSnapshotComparison.objects.select_related( - "base_snapshot_metrics", - ) - .filter( - head_snapshot_metrics=snapshot_metrics, - state=PreprodSnapshotComparison.State.SUCCESS, - ) + all_comparisons = list( + PreprodSnapshotComparison.objects.select_related("base_snapshot_metrics") + .filter(head_snapshot_metrics=snapshot_metrics) .order_by("-id") - .first() + ) + latest_comparison = all_comparisons[0] if all_comparisons else None + comparison = next( + (c for c in all_comparisons if c.state == PreprodSnapshotComparison.State.SUCCESS), + None, ) if comparison: comparison_key = (comparison.extras or {}).get("comparison_key") @@ -324,6 +329,26 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) -> ) ) + has_base_sha = bool(commit_comparison and commit_comparison.base_sha) + artifact_age_seconds = (timezone.now() - artifact.date_added).total_seconds() + base_artifact_exists: bool | None = None + if latest_comparison is None and has_base_sha and commit_comparison is not None: + if artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS: + assert commit_comparison.base_sha is not None + base_artifact_exists = ( + find_base_snapshot_artifact( + organization_id=commit_comparison.organization_id, + base_sha=commit_comparison.base_sha, + base_repo_name=commit_comparison.base_repo_name + or commit_comparison.head_repo_name, + project_id=artifact.project_id, + app_id=artifact.app_id, + artifact_type=artifact.artifact_type, + build_configuration=artifact.build_configuration, + ) + is not None + ) + image_list = [ build_snapshot_image_response(key, metadata, manifest.diff_threshold) for key, metadata in sorted(manifest.images.items()) @@ -345,18 +370,18 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) -> if comparison is not None: base_artifact_id = str(comparison.base_snapshot_metrics.preprod_artifact_id) categorized = CategorizedComparison() - pending_or_failed_state = ( - PreprodSnapshotComparison.objects.filter( - head_snapshot_metrics=snapshot_metrics, - state__in=[ + pending_or_failed_state = next( + ( + c.state + for c in all_comparisons + if c.state + in ( PreprodSnapshotComparison.State.PENDING, PreprodSnapshotComparison.State.PROCESSING, PreprodSnapshotComparison.State.FAILED, - ], - ) - .values_list("state", flat=True) - .order_by("-id") - .first() + ) + ), + None, ) if pending_or_failed_state is not None: comparison_state = PreprodSnapshotComparison.State(pending_or_failed_state).name @@ -452,6 +477,17 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) -> approvers=[], ) + sorted_approvals = sorted(all_approvals, key=lambda a: a.id, reverse=True) + derived_status = derive_snapshot_status( + SnapshotStatusInput( + latest_comparison=latest_comparison, + latest_approval=sorted_approvals[0] if sorted_approvals else None, + has_base_sha=has_base_sha, + artifact_age_seconds=artifact_age_seconds, + base_artifact_exists=base_artifact_exists, + ) + ) + response_data = SnapshotDetailsApiResponse( head_artifact_id=str(artifact.id), base_artifact_id=base_artifact_id, @@ -479,6 +515,10 @@ def get(self, request: Request, organization: Organization, snapshot_id: str) -> comparison_run_info=run_info, approval_info=approval_info, diff_threshold=manifest.diff_threshold, + comparison_state=derived_status.comparison_state, + approval_status=derived_status.approval_status, + comparison_error_message=derived_status.comparison_error_message, + approvers=approver_list if approved else [], ).dict() if compact: diff --git a/src/sentry/preprod/api/models/project_preprod_build_details_models.py b/src/sentry/preprod/api/models/project_preprod_build_details_models.py index 1d9712cbab33e9..308e26d0a66672 100644 --- a/src/sentry/preprod/api/models/project_preprod_build_details_models.py +++ b/src/sentry/preprod/api/models/project_preprod_build_details_models.py @@ -6,6 +6,12 @@ from django.utils import timezone from pydantic import BaseModel, Field +from sentry.preprod.api.models.snapshots.snapshot_status import ( + ApprovalStatusLiteral, + ComparisonStateLiteral, + SnapshotStatusInput, + derive_snapshot_status, +) from sentry.preprod.build_distribution_utils import ( get_download_count_for_artifact, is_installable_artifact, @@ -94,16 +100,13 @@ class PostedStatusChecks(BaseModel): class SnapshotComparisonInfo(BaseModel): image_count: int - comparison_state: ( - Literal["pending", "processing", "success", "failed", "waiting_for_base", "no_base_build"] - | None - ) = None + comparison_state: ComparisonStateLiteral | None = None comparison_error_message: str | None = None images_added: int = 0 images_removed: int = 0 images_changed: int = 0 images_unchanged: int = 0 - approval_status: Literal["approved", "auto_approved", "requires_approval"] | None = None + approval_status: ApprovalStatusLiteral | None = None class SizeInfoSizeMetric(BaseModel): @@ -292,29 +295,23 @@ def to_snapshot_comparison_info(head_artifact: PreprodArtifact) -> SnapshotCompa except PreprodSnapshotMetrics.DoesNotExist: return None - comparison_state = None - comparison_error_message = None - images_added = 0 - images_removed = 0 - images_changed = 0 - images_unchanged = 0 - comparisons = sorted( snapshot_metrics.snapshot_comparisons_head_metrics.all(), key=lambda c: c.id, reverse=True, ) comparison = comparisons[0] if comparisons else None - if not comparison: - cc = head_artifact.commit_comparison - if cc and cc.base_sha: - grace_period_expired = ( - timezone.now() - head_artifact.date_added - ).total_seconds() > MISSING_BASE_GRACE_PERIOD_SECONDS - - if ( - grace_period_expired - and find_base_snapshot_artifact( + + cc = head_artifact.commit_comparison + has_base_sha = bool(cc and cc.base_sha) + artifact_age_seconds = (timezone.now() - head_artifact.date_added).total_seconds() + + base_artifact_exists: bool | None = None + if comparison is None and has_base_sha and cc is not None: + if artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS: + assert cc.base_sha is not None + base_artifact_exists = ( + find_base_snapshot_artifact( organization_id=cc.organization_id, base_sha=cc.base_sha, base_repo_name=cc.base_repo_name or cc.head_repo_name, @@ -323,47 +320,45 @@ def to_snapshot_comparison_info(head_artifact: PreprodArtifact) -> SnapshotCompa artifact_type=head_artifact.artifact_type, build_configuration=head_artifact.build_configuration, ) - is None - ): - comparison_state = "no_base_build" - else: - comparison_state = "waiting_for_base" - elif comparison: - comparison_state = PreprodSnapshotComparison.State(comparison.state).name.lower() - if comparison.state == PreprodSnapshotComparison.State.SUCCESS: - images_added = comparison.images_added - images_removed = comparison.images_removed - images_changed = comparison.images_changed - images_unchanged = comparison.images_unchanged - elif comparison.state == PreprodSnapshotComparison.State.FAILED: - comparison_error_message = comparison.error_message - - approval_status = None - # REJECTED is no longer used; all non-APPROVED statuses are treated as requires_approval + is not None + ) + approvals = [ a for a in head_artifact.preprodcomparisonapproval_set.all() if a.preprod_feature_type == PreprodComparisonApproval.FeatureType.SNAPSHOTS ] approvals.sort(key=lambda a: a.id, reverse=True) - if approvals: - if approvals[0].approval_status == PreprodComparisonApproval.ApprovalStatus.APPROVED: - if (approvals[0].extras or {}).get("auto_approval") is True: - approval_status = "auto_approved" - else: - approval_status = "approved" - else: - approval_status = "requires_approval" + + derived = derive_snapshot_status( + SnapshotStatusInput( + latest_comparison=comparison, + latest_approval=approvals[0] if approvals else None, + has_base_sha=has_base_sha, + artifact_age_seconds=artifact_age_seconds, + base_artifact_exists=base_artifact_exists, + ) + ) + + images_added = 0 + images_removed = 0 + images_changed = 0 + images_unchanged = 0 + if comparison is not None and comparison.state == PreprodSnapshotComparison.State.SUCCESS: + images_added = comparison.images_added + images_removed = comparison.images_removed + images_changed = comparison.images_changed + images_unchanged = comparison.images_unchanged return SnapshotComparisonInfo( image_count=snapshot_metrics.image_count, - comparison_state=comparison_state, - comparison_error_message=comparison_error_message, + comparison_state=derived.comparison_state, + comparison_error_message=derived.comparison_error_message, images_added=images_added, images_removed=images_removed, images_changed=images_changed, images_unchanged=images_unchanged, - approval_status=approval_status, + approval_status=derived.approval_status, ) diff --git a/src/sentry/preprod/api/models/snapshots/project_preprod_snapshot_models.py b/src/sentry/preprod/api/models/snapshots/project_preprod_snapshot_models.py index b67e4430283b34..4eaf457b2a107d 100644 --- a/src/sentry/preprod/api/models/snapshots/project_preprod_snapshot_models.py +++ b/src/sentry/preprod/api/models/snapshots/project_preprod_snapshot_models.py @@ -6,6 +6,10 @@ from pydantic import BaseModel from sentry.preprod.api.models.project_preprod_build_details_models import BuildDetailsVcsInfo +from sentry.preprod.api.models.snapshots.snapshot_status import ( + ApprovalStatusLiteral, + ComparisonStateLiteral, +) from sentry.preprod.models import PreprodArtifact @@ -132,5 +136,10 @@ class SnapshotDetailsApiResponse(BaseModel): diff_threshold: float | None = None + comparison_state: ComparisonStateLiteral | None = None + approval_status: ApprovalStatusLiteral | None = None + comparison_error_message: str | None = None + approvers: list[SnapshotApprover] = [] + # TODO: POST request in the future when we migrate away from current schemas diff --git a/src/sentry/preprod/api/models/snapshots/snapshot_status.py b/src/sentry/preprod/api/models/snapshots/snapshot_status.py new file mode 100644 index 00000000000000..05d62ef191bc17 --- /dev/null +++ b/src/sentry/preprod/api/models/snapshots/snapshot_status.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from pydantic import BaseModel + +from sentry.preprod.models import PreprodComparisonApproval +from sentry.preprod.snapshots.constants import MISSING_BASE_GRACE_PERIOD_SECONDS +from sentry.preprod.snapshots.models import PreprodSnapshotComparison + +ComparisonStateLiteral = Literal[ + "pending", "processing", "success", "failed", "waiting_for_base", "no_base_build" +] +ApprovalStatusLiteral = Literal["approved", "auto_approved", "requires_approval"] + + +@dataclass(frozen=True) +class SnapshotStatusInput: + latest_comparison: PreprodSnapshotComparison | None + latest_approval: PreprodComparisonApproval | None + has_base_sha: bool + artifact_age_seconds: float + base_artifact_exists: bool | None + + +class SnapshotDerivedStatus(BaseModel): + comparison_state: ComparisonStateLiteral | None = None + approval_status: ApprovalStatusLiteral | None = None + comparison_error_message: str | None = None + + +def derive_snapshot_status(status_input: SnapshotStatusInput) -> SnapshotDerivedStatus: + comparison_state = None + comparison_error_message = None + + if status_input.latest_comparison is not None: + comparison_state = PreprodSnapshotComparison.State( + status_input.latest_comparison.state + ).name.lower() + if status_input.latest_comparison.state == PreprodSnapshotComparison.State.FAILED: + comparison_error_message = status_input.latest_comparison.error_message + elif status_input.has_base_sha: + grace_period_expired = status_input.artifact_age_seconds > MISSING_BASE_GRACE_PERIOD_SECONDS + if grace_period_expired and status_input.base_artifact_exists is False: + comparison_state = "no_base_build" + else: + comparison_state = "waiting_for_base" + + approval_status = None + if status_input.latest_approval is not None: + if ( + status_input.latest_approval.approval_status + == PreprodComparisonApproval.ApprovalStatus.APPROVED + ): + if (status_input.latest_approval.extras or {}).get("auto_approval") is True: + approval_status = "auto_approved" + else: + approval_status = "approved" + else: + approval_status = "requires_approval" + + return SnapshotDerivedStatus( + comparison_state=comparison_state, + comparison_error_message=comparison_error_message, + approval_status=approval_status, + ) diff --git a/tests/sentry/preprod/api/endpoints/test_preprod_artifact_snapshot.py b/tests/sentry/preprod/api/endpoints/test_preprod_artifact_snapshot.py index 4e4e7caf8acc09..0de90d6e6f790b 100644 --- a/tests/sentry/preprod/api/endpoints/test_preprod_artifact_snapshot.py +++ b/tests/sentry/preprod/api/endpoints/test_preprod_artifact_snapshot.py @@ -4,7 +4,7 @@ from django.urls import reverse from sentry.models.commitcomparison import CommitComparison -from sentry.preprod.models import PreprodArtifact +from sentry.preprod.models import PreprodArtifact, PreprodComparisonApproval from sentry.preprod.snapshots.models import PreprodSnapshotComparison, PreprodSnapshotMetrics from sentry.testutils.cases import APITestCase @@ -597,6 +597,118 @@ def test_get_snapshot_returns_404_for_member_without_project_access(self) -> Non assert response.status_code == 404 + @patch("sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot.get_preprod_session") + def test_get_snapshot_flat_fields_solo_no_approval(self, mock_get_session): + artifact, _, _, manifest_json, _ = self._create_artifact_with_manifest() + mock_get_session.return_value = self._create_mock_session(manifest_json) + + url = self._get_detail_url(artifact.id) + with self.feature("organizations:preprod-snapshots"): + response = self.client.get(url) + + assert response.status_code == 200 + assert response.data["comparison_state"] is None + assert response.data["approval_status"] is None + assert response.data["comparison_error_message"] is None + assert response.data["approvers"] == [] + assert response.data["comparison_type"] == "solo" + + @patch("sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot.get_preprod_session") + def test_get_snapshot_flat_fields_pending_comparison(self, mock_get_session): + artifact, snapshot_metrics, _, manifest_json, _ = self._create_artifact_with_manifest( + commit_comparison=CommitComparison.objects.create( + organization_id=self.org.id, + head_sha="a" * 40, + base_sha="b" * 40, + provider="github", + head_repo_name="org/repo", + head_ref="feature", + ), + ) + base_artifact = PreprodArtifact.objects.create( + project=self.project, + state=PreprodArtifact.ArtifactState.UPLOADED, + app_id="com.example.app", + ) + base_metrics = PreprodSnapshotMetrics.objects.create( + preprod_artifact=base_artifact, + image_count=1, + extras={"manifest_key": "base-key"}, + ) + self.create_preprod_snapshot_comparison( + head_snapshot_metrics=snapshot_metrics, + base_snapshot_metrics=base_metrics, + state=PreprodSnapshotComparison.State.PENDING, + ) + mock_get_session.return_value = self._create_mock_session(manifest_json) + + url = self._get_detail_url(artifact.id) + with self.feature("organizations:preprod-snapshots"): + response = self.client.get(url) + + assert response.status_code == 200 + assert response.data["comparison_state"] == "pending" + + @patch("sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot.get_preprod_session") + def test_get_snapshot_flat_fields_with_approval(self, mock_get_session): + artifact, _, _, manifest_json, _ = self._create_artifact_with_manifest() + self.create_preprod_comparison_approval( + preprod_artifact=artifact, + approval_status=PreprodComparisonApproval.ApprovalStatus.APPROVED, + approved_by_id=self.user.id, + ) + mock_get_session.return_value = self._create_mock_session(manifest_json) + + url = self._get_detail_url(artifact.id) + with self.feature("organizations:preprod-snapshots"): + response = self.client.get(url) + + assert response.status_code == 200 + assert response.data["approval_status"] == "approved" + assert len(response.data["approvers"]) == 1 + assert response.data["approvers"][0]["source"] == "sentry" + assert response.data["approval_info"] is not None + assert response.data["approval_info"]["status"] == "approved" + + @patch("sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot.get_preprod_session") + def test_get_snapshot_flat_fields_auto_approved(self, mock_get_session): + artifact, _, _, manifest_json, _ = self._create_artifact_with_manifest() + self.create_preprod_comparison_approval( + preprod_artifact=artifact, + approval_status=PreprodComparisonApproval.ApprovalStatus.APPROVED, + extras={"auto_approval": True}, + ) + mock_get_session.return_value = self._create_mock_session(manifest_json) + + url = self._get_detail_url(artifact.id) + with self.feature("organizations:preprod-snapshots"): + response = self.client.get(url) + + assert response.status_code == 200 + assert response.data["approval_status"] == "auto_approved" + + @patch("sentry.preprod.api.endpoints.snapshots.preprod_artifact_snapshot.get_preprod_session") + def test_get_snapshot_flat_fields_waiting_for_base(self, mock_get_session): + artifact, _, _, manifest_json, _ = self._create_artifact_with_manifest( + commit_comparison=CommitComparison.objects.create( + organization_id=self.org.id, + head_sha="a" * 40, + base_sha="b" * 40, + provider="github", + head_repo_name="org/repo", + head_ref="feature", + ), + ) + mock_get_session.return_value = self._create_mock_session(manifest_json) + + url = self._get_detail_url(artifact.id) + with self.feature("organizations:preprod-snapshots"): + response = self.client.get(url) + + assert response.status_code == 200 + assert response.data["comparison_state"] == "waiting_for_base" + assert response.data["comparison_type"] == "waiting_for_base" + class ProjectPreprodSnapshotDeleteTest(APITestCase): def setUp(self) -> None: diff --git a/tests/sentry/preprod/api/models/test_snapshot_status.py b/tests/sentry/preprod/api/models/test_snapshot_status.py new file mode 100644 index 00000000000000..838df36ebd86c8 --- /dev/null +++ b/tests/sentry/preprod/api/models/test_snapshot_status.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +from sentry.preprod.api.models.snapshots.snapshot_status import ( + SnapshotStatusInput, + derive_snapshot_status, +) +from sentry.preprod.models import PreprodComparisonApproval +from sentry.preprod.snapshots.constants import MISSING_BASE_GRACE_PERIOD_SECONDS +from sentry.preprod.snapshots.models import PreprodSnapshotComparison +from sentry.testutils.cases import TestCase +from sentry.testutils.silo import cell_silo_test + + +def _make_input( + latest_comparison: PreprodSnapshotComparison | None = None, + latest_approval: PreprodComparisonApproval | None = None, + has_base_sha: bool = False, + artifact_age_seconds: float = 0.0, + base_artifact_exists: bool | None = None, +) -> SnapshotStatusInput: + return SnapshotStatusInput( + latest_comparison=latest_comparison, + latest_approval=latest_approval, + has_base_sha=has_base_sha, + artifact_age_seconds=artifact_age_seconds, + base_artifact_exists=base_artifact_exists, + ) + + +@cell_silo_test +class TestDeriveSnapshotStatusComparisonState(TestCase): + def _create_comparison( + self, state: int = PreprodSnapshotComparison.State.SUCCESS + ) -> PreprodSnapshotComparison: + artifact = self.create_preprod_artifact(project=self.project) + head_metrics = self.create_preprod_snapshot_metrics( + preprod_artifact=artifact, image_count=1 + ) + base_artifact = self.create_preprod_artifact(project=self.project) + base_metrics = self.create_preprod_snapshot_metrics( + preprod_artifact=base_artifact, image_count=1 + ) + return self.create_preprod_snapshot_comparison( + head_snapshot_metrics=head_metrics, + base_snapshot_metrics=base_metrics, + state=state, + ) + + def test_comparison_state_mapping(self) -> None: + for state, expected in [ + (PreprodSnapshotComparison.State.PENDING, "pending"), + (PreprodSnapshotComparison.State.PROCESSING, "processing"), + (PreprodSnapshotComparison.State.SUCCESS, "success"), + (PreprodSnapshotComparison.State.FAILED, "failed"), + ]: + comparison = self._create_comparison(state=state) + result = derive_snapshot_status(_make_input(latest_comparison=comparison)) + assert result.comparison_state == expected + + def test_failed_state_with_error_message(self) -> None: + comparison = self._create_comparison(state=PreprodSnapshotComparison.State.FAILED) + comparison.error_message = "Something went wrong" + comparison.save(update_fields=["error_message"]) + result = derive_snapshot_status(_make_input(latest_comparison=comparison)) + assert result.comparison_state == "failed" + assert result.comparison_error_message == "Something went wrong" + + def test_no_comparison_no_base_sha(self) -> None: + result = derive_snapshot_status(_make_input()) + assert result.comparison_state is None + + def test_no_comparison_has_base_sha_within_grace_period(self) -> None: + result = derive_snapshot_status(_make_input(has_base_sha=True, artifact_age_seconds=0.0)) + assert result.comparison_state == "waiting_for_base" + + def test_no_comparison_has_base_sha_past_grace_period_no_base_artifact(self) -> None: + result = derive_snapshot_status( + _make_input( + has_base_sha=True, + artifact_age_seconds=MISSING_BASE_GRACE_PERIOD_SECONDS + 1, + base_artifact_exists=False, + ) + ) + assert result.comparison_state == "no_base_build" + + def test_no_comparison_has_base_sha_past_grace_period_base_artifact_exists(self) -> None: + result = derive_snapshot_status( + _make_input( + has_base_sha=True, + artifact_age_seconds=MISSING_BASE_GRACE_PERIOD_SECONDS + 1, + base_artifact_exists=True, + ) + ) + assert result.comparison_state == "waiting_for_base" + + +@cell_silo_test +class TestDeriveSnapshotStatusApprovalStatus(TestCase): + def _create_approval( + self, + approval_status: int = PreprodComparisonApproval.ApprovalStatus.APPROVED, + extras: dict | None = None, + ) -> PreprodComparisonApproval: + artifact = self.create_preprod_artifact(project=self.project) + return self.create_preprod_comparison_approval( + preprod_artifact=artifact, + approval_status=approval_status, + extras=extras, + ) + + def test_no_approval_returns_none(self) -> None: + result = derive_snapshot_status(_make_input()) + assert result.approval_status is None + + def test_manual_approval(self) -> None: + approval = self._create_approval() + result = derive_snapshot_status(_make_input(latest_approval=approval)) + assert result.approval_status == "approved" + + def test_auto_approval(self) -> None: + approval = self._create_approval(extras={"auto_approval": True}) + result = derive_snapshot_status(_make_input(latest_approval=approval)) + assert result.approval_status == "auto_approved" + + def test_extras_none_treated_as_manual(self) -> None: + approval = self._create_approval(extras=None) + result = derive_snapshot_status(_make_input(latest_approval=approval)) + assert result.approval_status == "approved" + + def test_needs_approval(self) -> None: + approval = self._create_approval( + approval_status=PreprodComparisonApproval.ApprovalStatus.NEEDS_APPROVAL, + ) + result = derive_snapshot_status(_make_input(latest_approval=approval)) + assert result.approval_status == "requires_approval"