47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
from sqlalchemy import select
|
|
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import VoiceTranscriptSegmentRow
|
|
from services.shared.voice_transcripts import add_transcript_segment
|
|
|
|
|
|
def test_add_transcript_segment_retries_on_sequence_conflict():
|
|
init_sql_schema()
|
|
session = get_session()
|
|
try:
|
|
first = add_transcript_segment(
|
|
session,
|
|
session_id="avs_sequence_retry",
|
|
call_id="call_sequence_retry",
|
|
interaction_id="int_sequence_retry",
|
|
speaker="caller",
|
|
source_type="asr",
|
|
text="first",
|
|
sequence_no=1,
|
|
)
|
|
session.commit()
|
|
assert first.sequence_no == 1
|
|
|
|
second = add_transcript_segment(
|
|
session,
|
|
session_id="avs_sequence_retry",
|
|
call_id="call_sequence_retry",
|
|
interaction_id="int_sequence_retry",
|
|
speaker="assistant",
|
|
source_type="tts",
|
|
text="second",
|
|
sequence_no=1,
|
|
)
|
|
session.commit()
|
|
|
|
rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == "avs_sequence_retry")
|
|
.order_by(VoiceTranscriptSegmentRow.sequence_no.asc())
|
|
).scalars().all()
|
|
assert [row.sequence_no for row in rows] == [1, 2]
|
|
assert second.sequence_no == 2
|
|
finally:
|
|
session.close()
|