fix(runtime): restore lifecycle and Telegram verification
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 3m17s
CD Pipeline / build-and-deploy (push) Successful in 13m9s
CD Pipeline / post-deploy-checks (push) Successful in 1m59s
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 3m17s
CD Pipeline / build-and-deploy (push) Successful in 13m9s
CD Pipeline / post-deploy-checks (push) Successful in 1m59s
This commit is contained in:
@@ -539,6 +539,40 @@ def test_signal_worker_gracefully_cancels_agent99_reconciler() -> None:
|
||||
assert source.index(await_task) < source.index("await worker.stop()")
|
||||
|
||||
|
||||
def test_signal_worker_is_single_incident_lifecycle_reconciler_owner() -> None:
|
||||
import inspect
|
||||
|
||||
from src import main as api_main
|
||||
from src.workers import signal_worker
|
||||
|
||||
api_source = inspect.getsource(api_main.lifespan)
|
||||
worker_source = inspect.getsource(signal_worker._main)
|
||||
|
||||
assert "run_incident_lifecycle_reconciler_loop" not in api_source
|
||||
assert "incident_lifecycle_reconciler_owner_delegated" in api_source
|
||||
assert worker_source.count("run_incident_lifecycle_reconciler_loop()") == 1
|
||||
assert worker_source.index("if settings.DATABASE_BOOTSTRAP_DDL_ENABLED") < (
|
||||
worker_source.index("run_incident_lifecycle_reconciler_loop()")
|
||||
)
|
||||
assert "signal_worker_incident_lifecycle_reconciler_started" in worker_source
|
||||
assert 'owner="signal_worker"' in worker_source
|
||||
|
||||
|
||||
def test_signal_worker_gracefully_cancels_incident_lifecycle_reconciler() -> None:
|
||||
import inspect
|
||||
|
||||
from src.workers import signal_worker
|
||||
|
||||
source = inspect.getsource(signal_worker._main)
|
||||
cancel = "incident_lifecycle_reconciler_task.cancel()"
|
||||
await_task = "await incident_lifecycle_reconciler_task"
|
||||
|
||||
assert cancel in source
|
||||
assert await_task in source
|
||||
assert source.index(cancel) < source.index(await_task)
|
||||
assert source.index(await_task) < source.index("await worker.stop()")
|
||||
|
||||
|
||||
def test_production_signal_worker_is_single_agent99_reconciler_owner() -> None:
|
||||
import inspect
|
||||
|
||||
|
||||
Reference in New Issue
Block a user