@@ -313,6 +313,7 @@ async def consume() -> None:
313313@pytest .mark .timeout (10 )
314314async def test_owner_context_succeeds_on_all_methods (db_url : str ) -> None :
315315 """The owner's context sees its own events and is accepted everywhere."""
316+ task_id = 'task-owner-ctx'
316317 store_engine = create_async_engine (db_url )
317318 stream_engine = create_async_engine (db_url )
318319 _enforce_sqlite_foreign_keys (store_engine )
@@ -328,31 +329,31 @@ async def test_owner_context_succeeds_on_all_methods(db_url: str) -> None:
328329 await stream .initialize ()
329330
330331 v1 = await store .save (
331- create_task (),
332+ create_task (task_id ),
332333 event = None ,
333334 prev = None ,
334335 prev_version = TaskVersion .MISSING ,
335336 context = TEST_CONTEXT ,
336337 )
337338 v2 = await store .save (
338- create_task (),
339- event = status_event (state = TaskState .TASK_STATE_WORKING ),
339+ create_task (task_id ),
340+ event = status_event (task_id , state = TaskState .TASK_STATE_WORKING ),
340341 prev = None ,
341342 prev_version = v1 ,
342343 context = TEST_CONTEXT ,
343344 )
344345
345- agen = stream .subscribe ('task-abc' , after = v1 , context = TEST_CONTEXT )
346+ agen = stream .subscribe (task_id , after = v1 , context = TEST_CONTEXT )
346347 received = await asyncio .wait_for (agen .__anext__ (), timeout = 5 )
347348 assert received .event .status .state == TaskState .TASK_STATE_WORKING
348349 await agen .aclose ()
349350
350351 await stream .publish (
351- 'task-abc' ,
352- VersionedEvent (create_task (), v2 ),
352+ task_id ,
353+ VersionedEvent (create_task (task_id ), v2 ),
353354 context = TEST_CONTEXT ,
354355 )
355- await stream .destroy ('task-abc' , context = TEST_CONTEXT )
356+ await stream .destroy (task_id , context = TEST_CONTEXT )
356357 finally :
357358 await store_engine .dispose ()
358359 await stream_engine .dispose ()
@@ -362,6 +363,7 @@ async def test_owner_context_succeeds_on_all_methods(db_url: str) -> None:
362363@pytest .mark .timeout (10 )
363364async def test_other_owner_context_does_not_receive_events (db_url : str ) -> None :
364365 """A different owner does not see events belonging to another owner."""
366+ task_id = 'task-other-owner'
365367 store_engine = create_async_engine (db_url )
366368 stream_engine = create_async_engine (db_url )
367369 _enforce_sqlite_foreign_keys (store_engine )
@@ -377,22 +379,22 @@ async def test_other_owner_context_does_not_receive_events(db_url: str) -> None:
377379 await stream .initialize ()
378380
379381 v1 = await store .save (
380- create_task (),
382+ create_task (task_id ),
381383 event = None ,
382384 prev = None ,
383385 prev_version = TaskVersion .MISSING ,
384386 context = TEST_CONTEXT ,
385387 )
386388 await store .save (
387- create_task (),
388- event = status_event (state = TaskState .TASK_STATE_WORKING ),
389+ create_task (task_id ),
390+ event = status_event (task_id , state = TaskState .TASK_STATE_WORKING ),
389391 prev = None ,
390392 prev_version = v1 ,
391393 context = TEST_CONTEXT ,
392394 )
393395 other = ServerCallContext (user = SampleUser ('someone_else' ))
394396
395- agen = stream .subscribe ('task-abc' , after = v1 , context = other )
397+ agen = stream .subscribe (task_id , after = v1 , context = other )
396398 with pytest .raises (asyncio .TimeoutError ):
397399 await asyncio .wait_for (agen .__anext__ (), timeout = 0.15 )
398400 await agen .aclose ()
0 commit comments