Re-fetch info on access elevator cache miss

Author: Piccirello

src/main.py

@@ -768,6 +768,34 @@ def classify_auto_approved_permission_sets(  # noqa: PLR0913
     return auto_approved
 
 
+def _get_cached_user_info(view_key: str, user_id: str, client: "WebClient") -> tuple[set[str], str | None]:
+    """Get user group IDs and email from cache, re-fetching from Identity Center on miss."""
+    user_group_ids = user_view_map.get(f"{view_key}:group_ids")
+    user_email = user_view_map.get(f"{view_key}:user_email")
+    if user_group_ids is not None:
+        return user_group_ids, user_email
+
+    logger.info("User info cache miss (container recycled), re-fetching from Identity Center")
+    identity_store_id = sso.get_identity_store_id(cfg, sso_client)
+    user_email = slack_helpers.get_user(client, id=user_id).email
+    user_principal_id, _ = sso.get_user_principal_id_by_email(
+        identity_store_client=identity_store_client,
+        identity_store_id=identity_store_id,
+        email=user_email,
+        cfg=cfg,
+    )
+    user_group_ids = sso.get_user_group_ids(
+        identity_store_client=identity_store_client,
+        identity_store_id=identity_store_id,
+        user_principal_id=user_principal_id,
+    )
+    # Re-populate cache for subsequent calls in this container
+    user_view_map[f"{view_key}:group_ids"] = user_group_ids
+    user_view_map[f"{view_key}:user_principal_id"] = user_principal_id
+    user_view_map[f"{view_key}:user_email"] = user_email
+    return user_group_ids, user_email
+
+
 @app.action(slack_helpers.RequestForAccessView.ACCOUNT_ACTION_ID)
 def handle_account_selection(ack: Ack, body: dict, client: WebClient) -> SlackResponse:
     ack()
@@ -791,19 +819,12 @@ def handle_account_selection(ack: Ack, body: dict, client: WebClient) -> SlackRe
         updated_view = slack_helpers.RequestForAccessView.build_no_permission_sets_view(view_blocks=body["view"]["blocks"])
         return client.views_update(view_id=view_id, view=updated_view)
 
-    # Get cached user_group_ids
+    # Get cached user info, re-fetching from Identity Center on cache miss
+    # (e.g. when Lambda container was recycled between form load and account selection)
     user_id = body.get("user", {}).get("id")
     callback_id = slack_helpers.RequestForAccessView.CALLBACK_ID
     view_key = f"{user_id}:{callback_id}"
-    group_ids_key = f"{view_key}:group_ids"
-    user_group_ids = user_view_map.get(group_ids_key)
-    if user_group_ids is None:
-        logger.warning(
-            f"User group IDs not found in cache for key: {group_ids_key}. "
-            "This may happen if Lambda container was recycled between form load and account selection. "
-            "Defaulting to empty set, which will restrict visibility to statements without required_group_membership."
-        )
-        user_group_ids = set()
+    user_group_ids, user_email = _get_cached_user_info(view_key, user_id, client)
 
     # Compute intersection of permission sets across all selected accounts
     valid_ps_names = statement.get_permission_sets_for_accounts_and_user(cfg.statements, account_ids, user_group_ids)
@@ -825,7 +846,6 @@ def handle_account_selection(ack: Ack, body: dict, client: WebClient) -> SlackRe
         return client.views_update(view_id=view_id, view=updated_view)
 
     # Classify permission sets as auto-approved vs requires-approval
-    user_email = user_view_map.get(f"{view_key}:user_email")
     auto_approved_arns: set[str] | None = None
     if user_email:
         resolver_cache: dict[frozenset[str], set[str]] = {}

src/tests/test_main.py

@@ -627,3 +627,112 @@ def side_effect(*_args, **kwargs):
             )
         assert ps_auto.arn in result
         assert ps_manual.arn not in result
+
+
+class TestGetCachedUserInfo:
+    """Tests for _get_cached_user_info re-fetching user info on cache miss."""
+
+    def test_cache_hit_returns_cached_values(self, import_main):
+        """When group_ids are cached, returns cached values without API calls."""
+        main = import_main
+        main.user_view_map.clear()
+
+        view_key = "U_CACHED:request_for__account_access_submitted"
+        main.user_view_map[f"{view_key}:group_ids"] = {"group-1"}
+        main.user_view_map[f"{view_key}:user_email"] = "cached@test.com"
+
+        mock_client = MagicMock()
+
+        with (
+            patch.object(main.sso, "get_identity_store_id") as mock_get_id,
+            patch.object(main.sso, "get_user_principal_id_by_email") as mock_get_principal,
+            patch.object(main.sso, "get_user_group_ids") as mock_get_groups,
+        ):
+            group_ids, email = main._get_cached_user_info(view_key, "U_CACHED", mock_client)
+
+            mock_get_id.assert_not_called()
+            mock_get_principal.assert_not_called()
+            mock_get_groups.assert_not_called()
+
+        assert group_ids == {"group-1"}
+        assert email == "cached@test.com"
+
+    def test_cache_miss_refetches_from_identity_center(self, import_main):
+        """When group_ids are not cached, re-fetches from Identity Center."""
+        main = import_main
+        main.user_view_map.clear()
+
+        view_key = "U_COLD:request_for__account_access_submitted"
+        refetched_groups = {"group-dev", "group-admin"}
+
+        mock_client = MagicMock()
+
+        with (
+            patch.object(main.sso, "get_identity_store_id", return_value="d-123456"),
+            patch.object(
+                main.slack_helpers,
+                "get_user",
+                return_value=entities.slack.User(id="U_COLD", email="cold@test.com", real_name="Cold User"),
+            ),
+            patch.object(main.sso, "get_user_principal_id_by_email", return_value=("principal-cold", None)),
+            patch.object(main.sso, "get_user_group_ids", return_value=refetched_groups) as mock_get_groups,
+        ):
+            group_ids, email = main._get_cached_user_info(view_key, "U_COLD", mock_client)
+
+            mock_get_groups.assert_called_once()
+
+        assert group_ids == refetched_groups
+        assert email == "cold@test.com"
+
+    def test_cache_miss_repopulates_cache(self, import_main):
+        """After re-fetching, user info is stored back in the cache."""
+        main = import_main
+        main.user_view_map.clear()
+
+        view_key = "U_REPOP:request_for__account_access_submitted"
+        mock_client = MagicMock()
+
+        with (
+            patch.object(main.sso, "get_identity_store_id", return_value="d-123456"),
+            patch.object(
+                main.slack_helpers,
+                "get_user",
+                return_value=entities.slack.User(id="U_REPOP", email="repop@test.com", real_name="Repop User"),
+            ),
+            patch.object(main.sso, "get_user_principal_id_by_email", return_value=("principal-repop", None)),
+            patch.object(main.sso, "get_user_group_ids", return_value={"group-x"}),
+        ):
+            main._get_cached_user_info(view_key, "U_REPOP", mock_client)
+
+        assert main.user_view_map[f"{view_key}:group_ids"] == {"group-x"}
+        assert main.user_view_map[f"{view_key}:user_principal_id"] == "principal-repop"
+        assert main.user_view_map[f"{view_key}:user_email"] == "repop@test.com"
+
+    def test_second_call_after_refetch_uses_cache(self, import_main):
+        """After a cache miss repopulates, the next call uses the cache."""
+        main = import_main
+        main.user_view_map.clear()
+
+        view_key = "U_TWICE:request_for__account_access_submitted"
+        mock_client = MagicMock()
+
+        with (
+            patch.object(main.sso, "get_identity_store_id", return_value="d-123456"),
+            patch.object(
+                main.slack_helpers,
+                "get_user",
+                return_value=entities.slack.User(id="U_TWICE", email="twice@test.com", real_name="Twice User"),
+            ),
+            patch.object(main.sso, "get_user_principal_id_by_email", return_value=("principal-twice", None)),
+            patch.object(main.sso, "get_user_group_ids", return_value={"group-y"}) as mock_get_groups,
+        ):
+            # First call: cache miss, re-fetches
+            main._get_cached_user_info(view_key, "U_TWICE", mock_client)
+            assert mock_get_groups.call_count == 1
+
+            # Second call: cache hit, no re-fetch
+            group_ids, email = main._get_cached_user_info(view_key, "U_TWICE", mock_client)
+            assert mock_get_groups.call_count == 1  # still 1, not 2
+
+        assert group_ids == {"group-y"}
+        assert email == "twice@test.com"