28 lines
1.0 KiB
Python
28 lines
1.0 KiB
Python
from sqlalchemy import select
|
|
|
|
import services.routing_service.app as routing_module
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import RoutingCounter
|
|
|
|
|
|
def test_postgres_counter_path_uses_atomic_upsert(monkeypatch):
|
|
init_sql_schema()
|
|
session = get_session()
|
|
monkeypatch.setattr(routing_module, "_db_backend_name", lambda: "postgresql")
|
|
try:
|
|
indexes = [
|
|
routing_module._next_agent_index(session, channel="voice_pg_atomic", agent_count=3),
|
|
routing_module._next_agent_index(session, channel="voice_pg_atomic", agent_count=3),
|
|
routing_module._next_agent_index(session, channel="voice_pg_atomic", agent_count=3),
|
|
]
|
|
session.commit()
|
|
|
|
stored = session.execute(
|
|
select(RoutingCounter).where(RoutingCounter.channel == "voice_pg_atomic")
|
|
).scalar_one()
|
|
assert indexes == [0, 1, 2]
|
|
assert stored.counter == 3
|
|
finally:
|
|
session.close()
|