Tests listing keys in the artifact service.
(service_type, artifact_service_factory)
| 319 | ], |
| 320 | ) |
| 321 | async def test_list_keys(service_type, artifact_service_factory): |
| 322 | """Tests listing keys in the artifact service.""" |
| 323 | artifact_service = artifact_service_factory(service_type) |
| 324 | artifact = types.Part.from_bytes(data=b"test_data", mime_type="text/plain") |
| 325 | app_name = "app0" |
| 326 | user_id = "user0" |
| 327 | session_id = "123" |
| 328 | filename = "filename" |
| 329 | filenames = [filename + str(i) for i in range(5)] |
| 330 | |
| 331 | for f in filenames: |
| 332 | await artifact_service.save_artifact( |
| 333 | app_name=app_name, |
| 334 | user_id=user_id, |
| 335 | session_id=session_id, |
| 336 | filename=f, |
| 337 | artifact=artifact, |
| 338 | ) |
| 339 | |
| 340 | assert ( |
| 341 | await artifact_service.list_artifact_keys( |
| 342 | app_name=app_name, user_id=user_id, session_id=session_id |
| 343 | ) |
| 344 | == filenames |
| 345 | ) |
| 346 | |
| 347 | |
| 348 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected