Spaces:
Sleeping
Sleeping
| import importlib | |
| import os | |
| import re | |
| from pathlib import Path | |
| from fastapi.testclient import TestClient | |
| def _csrf(html: str) -> str: | |
| match = re.search(r'name="csrf_token" value="([^"]+)"', html) | |
| assert match | |
| return match.group(1) | |
| def test_enrollment_assignment_submission_and_media_authorization(tmp_path: Path) -> None: | |
| os.environ.update( | |
| { | |
| "ANNOTATOR_DB_PATH": str(tmp_path / "app.db"), | |
| "ANNOTATOR_MEDIA_DIR": str(tmp_path / "media"), | |
| "ANNOTATOR_COOKIE_SECURE": "false", | |
| "REQUIRE_ACCESS_CODE": "false", | |
| "STUDY_OPEN": "true", | |
| "STUDY_PHASE": "development", | |
| } | |
| ) | |
| from app import auth as auth_module | |
| from app import db as db_module | |
| from app import main as main_module | |
| importlib.reload(auth_module) | |
| importlib.reload(db_module) | |
| main = importlib.reload(main_module) | |
| with TestClient(main.app, follow_redirects=False) as client: | |
| login_page = client.get("/") | |
| assert login_page.status_code == 200 | |
| response = client.post( | |
| "/login", | |
| data={ | |
| "csrf_token": _csrf(login_page.text), | |
| "participant_id": "P-INTEGRATION", | |
| "access_code": "unused", | |
| "finance_familiarity": "some", | |
| "english_proficiency": "fluent", | |
| "consent": "yes", | |
| }, | |
| ) | |
| assert response.status_code == 303 | |
| assert response.headers["location"] == "/instructions" | |
| practice = client.get("/instructions") | |
| assert practice.status_code == 200 | |
| trained = client.post( | |
| "/instructions", | |
| data={ | |
| "csrf_token": _csrf(practice.text), | |
| "practice_1": "substantive_answer_attempt", | |
| "practice_2": "clarification_repair", | |
| "practice_3": "explicit_disclosure_boundary", | |
| }, | |
| ) | |
| assert trained.status_code == 303 | |
| assert trained.headers["location"] == "/label" | |
| label = client.get("/label") | |
| assert label.status_code == 200 | |
| assert 'value="flag" class="btn btn-flag-confirm" formnovalidate' in label.text | |
| task_id = int(re.search(r'name="task_id" value="(\d+)"', label.text).group(1)) | |
| session = main.db.get_session() | |
| try: | |
| task = session.get(main.db.StudyTask, task_id) | |
| item = session.get(main.db.StudyItem, task.item_id) | |
| media_path = Path(os.environ["ANNOTATOR_MEDIA_DIR"]) / item.audio_filename | |
| media_path.parent.mkdir(parents=True, exist_ok=True) | |
| media_path.write_bytes(b"ID3" + b"\0" * 2000) | |
| if task.condition == "text_audio": | |
| assert client.get(f"/media/{item.id}").status_code == 200 | |
| else: | |
| assert client.get(f"/media/{item.id}").status_code == 404 | |
| finally: | |
| session.close() | |
| submit = client.post( | |
| "/submit", | |
| data={ | |
| "csrf_token": _csrf(label.text), | |
| "task_id": str(task_id), | |
| "action": "label", | |
| "gate": "clarification_repair", | |
| "gate_confidence": "90", | |
| "gate_confidence_touched": "1", | |
| "rationale": "The executive asks the analyst to clarify the intended scope.", | |
| "response_time_ms": "31000", | |
| "audio_played_ms": "12000" if task.condition == "text_audio" else "0", | |
| "audio_completed": "1" if task.condition == "text_audio" else "0", | |
| }, | |
| ) | |
| assert submit.status_code == 303 | |
| session = main.db.get_session() | |
| try: | |
| annotation = session.query(main.db.Annotation).filter_by(task_id=task_id).one() | |
| assert annotation.gate == "clarification_repair" | |
| assert annotation.rasiah is None | |
| assert session.get(main.db.StudyTask, task_id).completed is True | |
| finally: | |
| session.close() | |
| def test_flag_bypasses_normal_annotation_requirements_and_advances(tmp_path: Path) -> None: | |
| os.environ.update( | |
| { | |
| "ANNOTATOR_DB_PATH": str(tmp_path / "app.db"), | |
| "ANNOTATOR_MEDIA_DIR": str(tmp_path / "media"), | |
| "ANNOTATOR_COOKIE_SECURE": "false", | |
| "REQUIRE_ACCESS_CODE": "false", | |
| "STUDY_OPEN": "true", | |
| "STUDY_PHASE": "development", | |
| } | |
| ) | |
| from app import auth as auth_module | |
| from app import db as db_module | |
| from app import main as main_module | |
| importlib.reload(auth_module) | |
| importlib.reload(db_module) | |
| main = importlib.reload(main_module) | |
| with TestClient(main.app, follow_redirects=False) as client: | |
| login_page = client.get("/") | |
| login = client.post( | |
| "/login", | |
| data={ | |
| "csrf_token": _csrf(login_page.text), | |
| "participant_id": "P-FLAG", | |
| "access_code": "unused", | |
| "finance_familiarity": "some", | |
| "english_proficiency": "fluent", | |
| "consent": "yes", | |
| }, | |
| ) | |
| assert login.status_code == 303 | |
| practice = client.get("/instructions") | |
| trained = client.post( | |
| "/instructions", | |
| data={ | |
| "csrf_token": _csrf(practice.text), | |
| "practice_1": "substantive_answer_attempt", | |
| "practice_2": "clarification_repair", | |
| "practice_3": "explicit_disclosure_boundary", | |
| }, | |
| ) | |
| assert trained.status_code == 303 | |
| label = client.get("/label") | |
| task_id = int(re.search(r'name="task_id" value="(\d+)"', label.text).group(1)) | |
| flagged = client.post( | |
| "/submit", | |
| data={ | |
| "csrf_token": _csrf(label.text), | |
| "task_id": str(task_id), | |
| "action": "flag", | |
| "flag_reason": "Wrong question–response pairing", | |
| }, | |
| ) | |
| assert flagged.status_code == 303 | |
| assert flagged.headers["location"] == "/label" | |
| next_label = client.get("/label") | |
| next_task_id = int(re.search(r'name="task_id" value="(\d+)"', next_label.text).group(1)) | |
| assert next_task_id != task_id | |
| session = main.db.get_session() | |
| try: | |
| annotation = session.query(main.db.Annotation).filter_by(task_id=task_id).one() | |
| assert annotation.flagged_broken is True | |
| assert annotation.flag_reason == "Wrong question–response pairing" | |
| assert session.get(main.db.StudyTask, task_id).completed is True | |
| finally: | |
| session.close() | |