File size: 3,619 Bytes
d5adeb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | import json
from pathlib import Path
from backend.app.core import interview_flow
def _write_state(tmp_path: Path, session_id: str, state: dict) -> None:
session_dir = tmp_path / session_id
session_dir.mkdir(parents=True, exist_ok=True)
(session_dir / "interview_state.json").write_text(json.dumps(state), encoding="utf-8")
def test_next_write_returns_existing_active_question_for_concurrent_followup(tmp_path):
session_id = "session-a"
active_followup = {
"id": "followup_python_4",
"question": "Could you give me a concrete example of working with Python?",
"skill_target": "Python",
}
state = {
"session_id": session_id,
"completed": False,
"cursor": {"stage": "technical", "last_question_id": "technical_1"},
"questions_asked": [active_followup],
"answers": {},
"turns": [
{
"role": "interviewer",
"id": active_followup["id"],
"text": active_followup["question"],
"skill_target": "Python",
}
],
}
_write_state(tmp_path, session_id, state)
stale_decision = {
"action": "followup",
"original_q": {"id": "technical_1", "question": "Explain Python.", "skill_target": "Python"},
"original_q_id": "technical_1",
"depth": 1,
}
result = interview_flow._decide_next_write(
tmp_path,
session_id,
{"questions": []},
stale_decision,
active_followup["question"],
)
assert result["id"] == active_followup["id"]
assert result["question"] == active_followup["question"]
def test_duplicate_followup_text_uses_alternate_not_completed(tmp_path):
session_id = "session-b"
duplicate_text = "Could you give me a concrete example of working with Python?"
state = {
"session_id": session_id,
"completed": False,
"followup_count": 1,
"cursor": {"stage": "technical", "last_question_id": "technical_1"},
"questions_asked": [
{"id": "technical_1", "question": "Explain Python.", "skill_target": "Python"},
{"id": "followup_python_2", "question": duplicate_text, "skill_target": "Python"},
],
"answers": {
"technical_1": {"answer": "Python answer", "score": 5},
"followup_python_2": {"answer": "Follow-up answer", "score": 5},
},
"turns": [
{"role": "interviewer", "id": "technical_1", "text": "Explain Python."},
{"role": "candidate", "id": "technical_1", "text": "Python answer", "score": 5},
{"role": "interviewer", "id": "followup_python_2", "text": duplicate_text},
{"role": "candidate", "id": "followup_python_2", "text": "Follow-up answer", "score": 5},
],
}
_write_state(tmp_path, session_id, state)
decision = {
"action": "followup",
"original_q": {"id": "technical_1", "question": "Explain Python.", "skill_target": "Python"},
"original_q_id": "technical_1",
"depth": 2,
}
plan = {
"questions": [
{"id": "technical_1", "type": "technical", "question": "Explain Python."},
{"id": "technical_2", "type": "technical", "question": "Explain APIs."},
]
}
result = interview_flow._decide_next_write(
tmp_path,
session_id,
plan,
decision,
duplicate_text,
)
assert result["id"].startswith("followup_python_")
assert result["question"] != duplicate_text
assert "completed" not in result
|