Spaces:
Sleeping
Sleeping
File size: 6,703 Bytes
5b34fc5 dea7109 5b34fc5 dea7109 | 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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | 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()
|