Extend fleet and instance permission tests

src/tests/_internal/server/routers/test_fleets.py

@@ -192,6 +192,51 @@ async def test_returns_40x_if_not_authenticated(
         response = await client.post("/api/project/main/fleets/get")
         assert response.status_code in [401, 403]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    @pytest.mark.parametrize(
+        "by_id", [pytest.param(False, id="by-name"), pytest.param(True, id="by-id")]
+    )
+    async def test_returns_403_on_nonexistent_fleet_in_foreign_project(
+        self, test_db, session: AsyncSession, client: AsyncClient, by_id: bool
+    ):
+        await create_project(session, name="test-project")
+        user = await create_user(session, global_role=GlobalRole.USER)  # not a project member
+        if by_id:
+            body = {"id": str(uuid4())}
+        else:
+            body = {"name": "nonexistent"}
+        response = await client.post(
+            "/api/project/test-project/fleets/get",
+            headers=get_auth_headers(user.token),
+            json=body,
+        )
+        assert response.status_code == 403
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    @pytest.mark.parametrize(
+        "by_id", [pytest.param(False, id="by-name"), pytest.param(True, id="by-id")]
+    )
+    async def test_returns_403_on_deleted_fleet_in_foreign_project(
+        self, test_db, session: AsyncSession, client: AsyncClient, by_id: bool
+    ):
+        project = await create_project(session, name="test-project")
+        user = await create_user(session, global_role=GlobalRole.USER)  # not a project member
+        fleet = await create_fleet(
+            session=session, project=project, deleted=True, name="deleted-fleet"
+        )
+        if by_id:
+            body = {"id": str(fleet.id)}
+        else:
+            body = {"name": "deleted-fleet"}
+        response = await client.post(
+            "/api/project/test-project/fleets/get",
+            headers=get_auth_headers(user.token),
+            json=body,
+        )
+        assert response.status_code == 403
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
     @pytest.mark.parametrize("deleted", [False, True])
@@ -303,6 +348,29 @@ async def test_not_returns_by_name_if_fleet_does_not_exist(
         )
         assert response.status_code == 400
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    @pytest.mark.parametrize(
+        "by_id", [pytest.param(False, id="by-name"), pytest.param(True, id="by-id")]
+    )
+    async def test_returns_foreign_fleet_to_global_admin(
+        self, test_db, session: AsyncSession, client: AsyncClient, by_id: bool
+    ):
+        admin = await create_user(session, global_role=GlobalRole.ADMIN)
+        project = await create_project(session, name="test-project")
+        fleet = await create_fleet(session=session, project=project, name="test-fleet")
+        if by_id:
+            body = {"id": str(fleet.id)}
+        else:
+            body = {"name": "test-fleet"}
+        response = await client.post(
+            "/api/project/test-project/fleets/get",
+            headers=get_auth_headers(admin.token),
+            json=body,
+        )
+        assert response.status_code == 200
+        assert response.json()["name"] == "test-fleet"
+
 
 class TestApplyFleetPlan:
     @pytest.mark.asyncio

src/tests/_internal/server/routers/test_instances.py

@@ -422,6 +422,23 @@ async def test_returns_instance_by_id(
         assert resp_data["project_name"] == project.name
         assert resp_data["fleet_name"] == fleet.name
 
+    async def test_returns_instance_to_global_admin(
+        self, session: AsyncSession, client: AsyncClient
+    ) -> None:
+        admin = await create_user(session, global_role=GlobalRole.ADMIN, name="global-admin")
+        project = await create_project(session)
+        fleet = await create_fleet(session, project)
+        instance = await create_instance(session=session, project=project, fleet=fleet)
+
+        resp = await client.post(
+            f"/api/project/{project.name}/instances/get",
+            headers=get_auth_headers(admin.token),
+            json={"id": str(instance.id)},
+        )
+        assert resp.status_code == 200
+        resp_data = resp.json()
+        assert resp_data["id"] == str(instance.id)
+
     async def test_returns_400_if_instance_not_found(
         self, session: AsyncSession, client: AsyncClient
     ) -> None:
@@ -479,3 +496,16 @@ async def test_returns_403_if_not_project_member(
             json={"id": str(instance.id)},
         )
         assert resp.status_code == 403
+
+    async def test_returns_403_if_not_project_member_and_instance_not_exists(
+        self, session: AsyncSession, client: AsyncClient
+    ) -> None:
+        user = await create_user(session, name="non_member", global_role=GlobalRole.USER)
+        project = await create_project(session)
+
+        resp = await client.post(
+            f"/api/project/{project.name}/instances/get",
+            headers=get_auth_headers(user.token),
+            json={"id": str(uuid.uuid4())},
+        )
+        assert resp.status_code == 403