@@ -87,6 +87,7 @@ def mocked_function(request) -> dict[str, Any]:
8787async def test_function_workflow (
8888 client : TestClient ,
8989 logged_user : UserInfoDict ,
90+ other_logged_user : UserInfoDict ,
9091 mocked_function : dict [str , Any ],
9192 expected_register : HTTPStatus ,
9293 expected_get : HTTPStatus ,
@@ -118,6 +119,43 @@ async def test_function_workflow(
118119 retrieved_function = TypeAdapter (RegisteredFunctionGet ).validate_python (data )
119120 assert retrieved_function .uid == returned_function .uid
120121
122+ # Set group permissions for other user
123+ new_group_id = other_logged_user ["primary_gid" ]
124+ new_group_access_rights = {"read" : True , "write" : True , "execute" : False }
125+
126+ url = client .app .router ["update_function_group" ].url_for (
127+ function_id = f"{ returned_function_uid } " , group_id = f"{ new_group_id } "
128+ )
129+
130+ response = await client .put (url , json = new_group_access_rights )
131+ data , error = await assert_status (response , expected_update )
132+ if not error :
133+ assert data == new_group_access_rights
134+
135+ # Remove group permissions for original user
136+ url = client .app .router ["delete_function_group" ].url_for (
137+ function_id = f"{ returned_function_uid } " , group_id = f"{ logged_user ['primary_gid' ]} "
138+ )
139+
140+ response = await client .delete (url )
141+ data , error = await assert_status (response , expected_delete )
142+ if not error :
143+ assert data is None
144+
145+ # Check that original user no longer has access
146+ url = client .app .router ["get_function" ].url_for (
147+ function_id = f"{ returned_function_uid } "
148+ )
149+ response = await client .get (url )
150+ data , error = await assert_status (response , expected_get )
151+ if not error :
152+ retrieved_function = (
153+ TypeAdapter (RegisteredFunctionGet ).validate_python (data ).model_dump ()
154+ )
155+ assert retrieved_function ["access_rights" ] == {
156+ new_group_id : new_group_access_rights
157+ }
158+
121159 # List existing functions
122160 url = client .app .router ["list_functions" ].url_for ()
123161 response = await client .get (url )
@@ -140,9 +178,11 @@ async def test_function_workflow(
140178 )
141179 data , error = await assert_status (response , expected_update )
142180 if not error :
143- updated_function = TypeAdapter (RegisteredFunctionGet ).validate_python (data )
144- assert updated_function .title == new_title
145- assert updated_function .description == new_description
181+ updated_group_access_rights = TypeAdapter (
182+ RegisteredFunctionGet
183+ ).validate_python (data )
184+ assert updated_group_access_rights .title == new_title
185+ assert updated_group_access_rights .description == new_description
146186
147187 # Delete existing function
148188 url = client .app .router ["delete_function" ].url_for (
0 commit comments