Add tests for filtering by access level

posthog/api/dashboards/dashboard.py

@@ -515,8 +515,8 @@ def dangerously_get_queryset(self):
                 ),
             )
 
-            # Add access level filtering for list actions
-            queryset = self._filter_queryset_by_access_level(queryset)
+        # Add access level filtering for list actions
+        queryset = self._filter_queryset_by_access_level(queryset)
 
         return queryset
 

posthog/api/test/dashboards/test_dashboard.py

@@ -24,6 +24,7 @@
     QueryMatchingTest,
     snapshot_postgres_queries,
 )
+from ee.models.rbac.access_control import AccessControl
 
 valid_template: dict = {
     "template_name": "Sign up conversion template with variables",
@@ -1444,3 +1445,37 @@ def test_dashboard_variables(self):
             assert value["code_name"] == variable.code_name
             assert value["variableId"] == str(variable.id)
             assert value["value"] == "some override value"
+
+    def test_dashboard_access_control_filtering(self) -> None:
+        """Test that dashboards are properly filtered based on access control."""
+
+        user2 = User.objects.create_and_join(self.organization, "[email protected]", None)
+
+        visible_dashboard = Dashboard.objects.create(
+            team=self.team,
+            name="Public Dashboard",
+            created_by=self.user,
+        )
+        hidden_dashboard = Dashboard.objects.create(
+            team=self.team,
+            name="Hidden Dashboard",
+            created_by=self.user,
+        )
+        AccessControl.objects.create(
+            resource="dashboard", resource_id=hidden_dashboard.id, team=self.team, access_level="none"
+        )
+
+        # Verify we can access visible dashboards
+        self.client.force_login(user2)
+        response = self.client.get(f"/api/projects/{self.team.pk}/dashboards/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        dashboard_ids = [dashboard["id"] for dashboard in response.json()["results"]]
+        self.assertIn(visible_dashboard.id, dashboard_ids)
+        self.assertNotIn(hidden_dashboard.id, dashboard_ids)
+
+        # Verify we can access all dashboards as creator
+        self.client.force_login(self.user)
+        response = self.client.get(f"/api/projects/{self.team.pk}/dashboards/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn(visible_dashboard.id, [dashboard["id"] for dashboard in response.json()["results"]])
+        self.assertIn(hidden_dashboard.id, [dashboard["id"] for dashboard in response.json()["results"]])

posthog/api/test/test_insight.py

@@ -32,6 +32,7 @@
     Text,
     User,
 )
+from ee.models.rbac.access_control import AccessControl
 from posthog.models.insight_caching_state import InsightCachingState
 from posthog.models.insight_variable import InsightVariable
 from posthog.models.project import Project
@@ -1258,7 +1259,7 @@ def test_insight_refreshing_legacy_conversion(self) -> None:
             self.assertEqual(response["last_refresh"], None)
             self.assertEqual(response["last_modified_at"], "2012-01-15T04:01:34Z")  # did not change
 
-        #  Test property filter
+        #  Test property filter
 
         dashboard = Dashboard.objects.get(pk=dashboard_id)
         dashboard.filters = {
@@ -1421,7 +1422,7 @@ def test_insight_refreshing_query(self, properties_filter, spy_execute_hogql_que
             self.assertEqual(response["last_modified_at"], "2012-01-15T04:01:34Z")  # did not change
             self.assertFalse(response["is_cached"])
 
-        #  Test property filter
+        #  Test property filter
 
         Dashboard.objects.update(
             id=dashboard_id,
@@ -2692,7 +2693,7 @@ def test_soft_delete_cannot_be_reversed_for_another_team(self) -> None:
 
     def test_cancel_running_query(self) -> None:
         # There is no good way of writing a test that tests this without it being very slow
-        #  Just verify it doesn't throw an error
+        #  Just verify it doesn't throw an error
         response = self.client.post(
             f"/api/projects/{self.team.id}/insights/cancel",
             {"client_query_id": f"testid"},
@@ -3520,3 +3521,39 @@ def test_insight_variables_overrides(self):
             assert value["code_name"] == variable.code_name
             assert value["variableId"] == str(variable.id)
             assert value["value"] == "override value!"
+
+    def test_insight_access_control_filtering(self) -> None:
+        """Test that insights are properly filtered based on access control."""
+
+        user2 = self._create_user("[email protected]")
+
+        visible_insight = Insight.objects.create(
+            team=self.team,
+            name="Public Insight",
+            created_by=self.user,
+            filters={"events": [{"id": "$pageview"}]},
+        )
+        hidden_insight = Insight.objects.create(
+            team=self.team,
+            name="Hidden Insight",
+            created_by=self.user,
+            filters={"events": [{"id": "$pageview"}]},
+        )
+        AccessControl.objects.create(
+            resource="insight", resource_id=hidden_insight.id, team=self.team, access_level="none"
+        )
+
+        # Verify we can access visible insights
+        self.client.force_login(user2)
+        response = self.client.get(f"/api/projects/{self.team.pk}/insights/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        insight_ids = [insight["id"] for insight in response.json()["results"]]
+        self.assertIn(visible_insight.id, insight_ids)
+        self.assertNotIn(hidden_insight.id, insight_ids)
+
+        # Verify we can access all insights as creator
+        self.client.force_login(self.user)
+        response = self.client.get(f"/api/projects/{self.team.pk}/insights/")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn(visible_insight.id, [insight["id"] for insight in response.json()["results"]])
+        self.assertIn(hidden_insight.id, [insight["id"] for insight in response.json()["results"]])