import csv import io import json from pathlib import Path import hiring_funnel as hf import settings class FakeSession: def __init__(self, code='S1'): self.code = code self.config = {'use_dynamic_condition_assignment': True} self.vars = {} self._participants = [] def get_participants(self): return list(self._participants) class FakeParticipant: def __init__(self, code='P1', id_in_session=1, label='label', session=None, vars=None): self.code = code self.id_in_session = id_in_session self.label = label self.session = session or FakeSession() self.vars = vars or {} self.history_json = '' self.main_data_json = '' self.selected_round = '' self.selected_candidate = '' self.selected_true_quality = '' self.selected_reward_string = '' self.dataset_unique_id = '' if self not in self.session._participants: self.session._participants.append(self) def get_players(self): return list(getattr(self, '_players', [])) class LegacyParticipant: def __init__(self, session=None): self.session = session or FakeSession('LEGACY') self.vars = {} self.code = 'LEGACY_P' def __getattr__(self, name): raise KeyError(name) class FakePlayer: def __init__(self, participant, round_number, fields=None): self.participant = participant self.session = participant.session self.round_number = round_number self.id_in_group = 1 try: self.id_in_subsession = getattr(participant, 'id_in_session', 1) except Exception: self.id_in_subsession = 1 self._meta = type('FakeMeta', (), {'app_label': hf.C.NAME_IN_URL})() self._fields = fields or {} self.has_hiring_experience = self._fields.get('has_hiring_experience') self.hiring_experience_type = self._fields.get('hiring_experience_type', '') self.hiring_experience_other = self._fields.get('hiring_experience_other', '') try: if not hasattr(participant, '_players'): participant._players = [] participant._players.append(self) except Exception: pass def field_maybe_none(self, field_name): value = self._fields.get(field_name, None) if isinstance(value, Exception): raise value return value def assert_equal(actual, expected, message): if actual != expected: raise AssertionError(f'{message}: expected {expected!r}, got {actual!r}') def assert_true(value, message): if not value: raise AssertionError(message) def make_player(condition, round_number, fields=None): session = FakeSession('TEST') participant = FakeParticipant( session=session, vars={'condition': condition, 'assignment_finalized': True}, ) return FakePlayer(participant, round_number, fields=fields) def test_round_structure_and_active_pages(): assert_equal(hf.C.BLIND_REPLAY_SOURCES_PER_PARTICIPANT, 1, 'Blind replay uses one Standard source') assert_equal(hf.C.BLIND_MAIN_ROUNDS, 7, 'Blind main task has 7 rounds') assert_equal(hf.C.NUM_ROUNDS, 7, 'C.NUM_ROUNDS stays structural at 7') players = [make_player(hf.CONDITION_STANDARD, round_number) for round_number in range(1, hf.C.NUM_ROUNDS + 1)] assert_equal(len(players), 7, 'session creation shape has 7 Player rows per participant') assert_true(hf.MainGate1.is_displayed(make_player(hf.CONDITION_STANDARD, 7)), 'Standard round 7 is active') assert_true(not hf.MainGate1.is_displayed(make_player(hf.CONDITION_STANDARD, 8)), 'Standard round 8 is not active') assert_true(hf.WeightsJustification.is_displayed(make_player(hf.CONDITION_STANDARD, 7)), 'Standard final survey appears at structural round 7') assert_true(hf.MainStage3Decision.is_displayed(make_player(hf.CONDITION_BLIND, 7)), 'Blind round 7 is active') assert_true(not hf.MainStage3Decision.is_displayed(make_player(hf.CONDITION_BLIND, 8)), 'Blind round 8 is not active') def test_post_task_validation(): assert_equal( hf.WeightsJustification.form_fields, ['open_ended_explanation'], 'Decision Strategy submits only the open-ended explanation', ) assert_equal( hf.WeightsJustification.error_message(None, { 'open_ended_explanation': 'I compared scores carefully.', }), None, 'valid explanation passes without score weights', ) assert_true( hf.WeightsJustification.error_message(None, { 'open_ended_explanation': ' ', }), 'empty explanation fails', ) base = {'age_range': '26-35', 'gender': 'female', 'device_type': 'computer', 'survey_comments': ''} assert_equal( hf.Demographics.form_fields, ['age_range', 'gender', 'device_type', 'survey_comments'], 'Demographics submits only retained demographic fields', ) assert_equal( hf.Demographics.error_message(None, dict(base)), None, 'retained demographics pass with empty optional comment', ) assert_true( hf.Demographics.error_message(None, dict(base, age_range='')), 'age remains required', ) assert_true( hf.Demographics.error_message(None, dict(base, gender='')), 'gender remains required', ) assert_true( hf.Demographics.error_message(None, dict(base, device_type='')), 'device type remains required', ) original_standard = hf._mark_standard_completed_if_ready original_blind = hf._mark_blind_completed_if_ready try: hf._mark_standard_completed_if_ready = lambda player: None hf._mark_blind_completed_if_ready = lambda player: None player = make_player(hf.CONDITION_STANDARD, hf.C.NUM_ROUNDS) hf.Demographics.before_next_page(player, False) finally: hf._mark_standard_completed_if_ready = original_standard hf._mark_blind_completed_if_ready = original_blind def test_click_to_reveal_instruction_copy(): sentence = 'You do NOT need to click open every score before making a selection.' main_instructions = Path('hiring_funnel/InstructionsPart1.html').read_text(encoding='utf-8') noise_instructions = Path('hiring_funnel/Instructions.html').read_text(encoding='utf-8') review_instructions = Path('hiring_funnel/Comprehension.html').read_text(encoding='utf-8') recap_template = Path('_templates/global/Page.html').read_text(encoding='utf-8') assert_equal(main_instructions.count('Click-to-Reveal'), 0, 'main Instruction Part 1 has no Click-to-Reveal section') assert_equal(main_instructions.count(sentence), 0, 'main Instruction Part 1 has no click reminder') assert_equal(noise_instructions.count(sentence), 1, 'score-noise instructions have one click reminder') assert_equal(recap_template.count('Click-to-Reveal'), 1, 'instruction recap has one Click-to-Reveal section') assert_equal(recap_template.count(sentence), 2, 'instruction recap has the reminder once per condition branch') assert_true( 'When you move to the next stage, all score cells will be hidden again.' not in review_instructions, 'review instructions remove the old hidden-again sentence from Your Decision Process', ) assert_equal( review_instructions.count('You may re-reveal any score cell you want in the new stage.'), 2, 'review instructions include the re-reveal sentence once per condition branch', ) assert_true( 'In this condition, you make the final hiring decision from the finalist set shown on the page.' in recap_template and 'There is no additional screening stage after this page.' in recap_template, 'Blind instruction copy keeps final-decision reveal wording', ) assert_true( 'instruction-recap-open' in recap_template and 'instruction-recap-backdrop' in recap_template, 'instruction recap trigger and panel remain present', ) def test_removed_noise_comprehension_question(): player = make_player(hf.CONDITION_STANDARD, 1) assert_true( 'Q_score_interpretation' in hf.Instructions.form_fields, 'score interpretation question remains on Instructions quiz', ) assert_true( 'Q_noise_level_difference' not in hf.Instructions.form_fields, 'score-noise question is no longer submitted', ) assert_equal( hf.Instructions.error_message(player, {'Q_score_interpretation': 1}), None, 'Instructions quiz proceeds without Q_noise_level_difference', ) assert_equal( hf.Instructions.error_message(player, {}), 'Please answer the score interpretation question before continuing.', 'remaining Instructions quiz question is still required', ) def test_prolific_id_dev_override(): original_allow_any = hf.DEV_ALLOW_ANY_PROLIFIC_ID try: demo_config = next( config for config in settings.SESSION_CONFIGS if config.get('name') == 'hiring_funnel_demo' ) assert_equal( settings.SESSION_CONFIG_DEFAULTS['require_prolific_pid'], True, 'production default keeps Prolific ID required', ) assert_equal( demo_config['require_prolific_pid'], False, 'demo/test session config disables Prolific ID requirement', ) strict_player = make_player(hf.CONDITION_STANDARD, 1) strict_player.session.config['require_prolific_pid'] = True hf.DEV_ALLOW_ANY_PROLIFIC_ID = False assert_equal( hf.ProlificIntro.error_message(strict_player, { 'prolific_pid': '', 'prolific_pid_from_url': '', }), 'Please provide your Prolific ID to continue.', 'strict mode blocks missing Prolific ID', ) assert_true( hf.ProlificIntro.error_message(strict_player, { 'prolific_pid': 'LOCAL_TEST_PID', 'prolific_pid_from_url': '', }), 'strict mode blocks fake Prolific ID shape', ) assert_equal( hf.ProlificIntro.error_message(strict_player, { 'prolific_pid': 'ABCDEFGHIJKLMNOPQRSTUVWX', 'prolific_pid_from_url': 'ZZZZZZZZZZZZZZZZZZZZZZZZ', }), 'The Prolific ID must match the PROLIFIC_PID in your study link.', 'strict mode blocks mismatched linked Prolific ID', ) optional_player = make_player(hf.CONDITION_STANDARD, 1) optional_player.session.config['require_prolific_pid'] = False optional_player.participant.label = 'LABEL_PID' hf.DEV_ALLOW_ANY_PROLIFIC_ID = False for test_pid, url_pid, message in ( ('', '', 'optional mode allows blank Prolific ID'), ('test', '', 'optional mode allows short test ID'), ('test-id!', '', 'optional mode allows non-alphanumeric test ID'), ('typed_id', 'URL_ID', 'optional mode allows mismatch with URL ID'), ('different_from_label', '', 'optional mode allows mismatch with participant label'), ): assert_equal( hf.ProlificIntro.error_message(optional_player, { 'prolific_pid': test_pid, 'prolific_pid_from_url': url_pid, }), None, message, ) stored_optional = make_player(hf.CONDITION_STANDARD, 1, { 'prolific_pid': 'test-id!', 'prolific_pid_from_url': 'URL_ID', 'prolific_study_id': 'STUDY', 'prolific_session_id': 'SESSION', }) stored_optional.session.config['require_prolific_pid'] = False hf.ProlificIntro.before_next_page(stored_optional, False) assert_equal( stored_optional.participant.vars['prolific_pid'], 'test-id!', 'optional mode still stores typed Prolific ID', ) assert_true( stored_optional.participant.vars['prolific_pid_mismatch'], 'optional mode still records mismatch diagnostics', ) local_player = make_player(hf.CONDITION_STANDARD, 1) local_player.session.config['require_prolific_pid'] = True local_player.participant.label = '' hf.DEV_ALLOW_ANY_PROLIFIC_ID = True assert_equal( hf.ProlificIntro.error_message(local_player, { 'prolific_pid': '', 'prolific_pid_from_url': '', }), None, 'local override allows missing Prolific ID and URL parameters', ) hf.ProlificIntro.before_next_page(local_player, False) assert_equal( local_player.participant.vars['prolific_pid'], f'LOCAL_TEST_{local_player.participant.code}', 'local override records a safe local placeholder when all ID sources are missing', ) local_player_with_numeric_url = make_player(hf.CONDITION_STANDARD, 1, { 'prolific_pid': '9', 'prolific_pid_from_url': '9', 'prolific_study_id': '', 'prolific_session_id': '', }) local_player_with_numeric_url.session.config['require_prolific_pid'] = True assert_equal( hf.ProlificIntro.error_message(local_player_with_numeric_url, { 'prolific_pid': '9', 'prolific_pid_from_url': '9', }), None, 'local override allows locked numeric URL Prolific ID', ) hf.ProlificIntro.before_next_page(local_player_with_numeric_url, False) assert_equal( local_player_with_numeric_url.participant.vars['prolific_pid'], '9', 'local override records locked numeric URL Prolific ID', ) local_player_with_fake = make_player(hf.CONDITION_STANDARD, 1, { 'prolific_pid': 'LOCAL_TEST_PID', 'prolific_pid_from_url': 'DIFFERENT_URL_PID', 'prolific_study_id': '', 'prolific_session_id': '', }) local_player_with_fake.participant.label = 'LABEL_PID' assert_equal( hf.ProlificIntro.error_message(local_player_with_fake, { 'prolific_pid': 'LOCAL_TEST_PID', 'prolific_pid_from_url': 'DIFFERENT_URL_PID', }), None, 'local override allows fake typed IDs and mismatch with URL/label', ) hf.ProlificIntro.before_next_page(local_player_with_fake, False) assert_equal( local_player_with_fake.participant.vars['prolific_pid'], 'LOCAL_TEST_PID', 'local override still records the typed Prolific ID', ) assert_equal( local_player_with_fake.participant.vars['prolific_pid_from_url'], 'DIFFERENT_URL_PID', 'local override still records URL Prolific metadata when supplied', ) finally: hf.DEV_ALLOW_ANY_PROLIFIC_ID = original_allow_any def test_dev_condition_chooser_paths(): original_chooser = hf.DEV_CONDITION_CHOOSER try: hf.DEV_CONDITION_CHOOSER = True assert_equal( hf.page_sequence.count(hf.DevConditionChoice), 1, 'DevConditionChoice appears once in page_sequence', ) assert_equal( hf.page_sequence.index(hf.DevConditionChoice), hf.page_sequence.index(hf.Consent) + 1, 'DevConditionChoice remains immediately after Consent', ) assert_true( hf.page_sequence.index(hf.DevConditionChoice) < hf.page_sequence.index(hf.AssignmentGate), 'DevConditionChoice remains before AssignmentGate', ) standard_session = FakeSession('DEV_STANDARD') standard_participant = FakeParticipant( code='DEV_STD_P', id_in_session=1, label='', session=standard_session, vars={'consent_completed': True}, ) standard_player = FakePlayer(standard_participant, 1) assert_true( hf.DevConditionChoice.is_displayed(standard_player), 'chooser displays for a fresh round-1 participant', ) standard_player.dev_condition_choice = hf.CONDITION_STANDARD hf.DevConditionChoice.before_next_page(standard_player, False) assert_equal(standard_player.dev_condition_choice, hf.CONDITION_STANDARD, 'player stores Standard chooser value') assert_equal(standard_participant.vars['dev_condition_choice'], hf.CONDITION_STANDARD, 'participant vars store Standard chooser value') assert_equal(standard_participant.vars['condition'], hf.CONDITION_STANDARD, 'participant vars store Standard condition') assert_equal(standard_participant.condition, hf.CONDITION_STANDARD, 'participant field stores Standard condition') assert_true(standard_participant.main_data_json, 'Standard chooser path initializes main data') assert_true(not hf.DevConditionChoice.is_displayed(standard_player), 'chooser does not display again after assignment') assert_true(not hf.AssignmentGate.is_displayed(standard_player), 'AssignmentGate does not intervene in chooser mode') assert_true(not hf.BlindReplayWait.is_displayed(standard_player), 'BlindReplayWait does not intervene for Standard chooser path') assert_true(hf.PracticeGate1.is_displayed(standard_player), 'Standard practice funnel is active') assert_true(hf.MainGate1.is_displayed(standard_player), 'Standard main funnel is active') blind_session = FakeSession('DEV_BLIND') blind_participant = FakeParticipant( code='DEV_BLIND_P', id_in_session=1, label='', session=blind_session, vars={'consent_completed': True}, ) blind_player = FakePlayer(blind_participant, 1) assert_true(hf.DevConditionChoice.is_displayed(blind_player), 'chooser displays for fresh Blind test participant') blind_player.dev_condition_choice = hf.CONDITION_BLIND hf.DevConditionChoice.before_next_page(blind_player, False) assert_equal(blind_player.dev_condition_choice, hf.CONDITION_BLIND, 'player stores Blind chooser value') assert_equal(blind_participant.vars['dev_condition_choice'], hf.CONDITION_BLIND, 'participant vars store Blind chooser value') assert_equal(blind_participant.vars['condition'], hf.CONDITION_BLIND, 'participant vars store Blind condition') assert_equal(blind_participant.condition, hf.CONDITION_BLIND, 'participant field stores Blind condition') assert_true(blind_participant.main_data_json, 'Blind chooser path initializes fallback main data') assert_true(not blind_participant.vars.get('matched_standard_code'), 'Blind chooser path does not require a matched Standard participant') assert_true(not hf.AssignmentGate.is_displayed(blind_player), 'AssignmentGate does not intervene in Blind chooser mode') assert_true(not hf.BlindReplayWait.is_displayed(blind_player), 'BlindReplayWait does not intervene after fallback data initialization') assert_true(hf.PracticeStage3Decision.is_displayed(blind_player), 'Blind practice finalist task is active') assert_true(hf.MainStage3Decision.is_displayed(blind_player), 'Blind main finalist task is active') hf.DEV_CONDITION_CHOOSER = False production_like_participant = FakeParticipant( code='PROD_FLOW_P', id_in_session=1, label='', session=FakeSession('PROD_FLOW'), vars={'consent_completed': True}, ) production_like_player = FakePlayer(production_like_participant, 1) assert_true( not hf.DevConditionChoice.is_displayed(production_like_player), 'setting chooser flag back to False hides the development chooser', ) finally: hf.DEV_CONDITION_CHOOSER = original_chooser def test_safe_exports(): special_text = 'Strategy with commas, line breaks\nUnicode: café, and "quotes".' session = FakeSession('EXPORT1') participant = FakeParticipant( code='P1', id_in_session=1, label='PROLIFIC_R1', session=session, vars={'condition': hf.CONDITION_STANDARD, 'assignment_finalized': True}, ) participant.main_data_json = '{"rounds": []}' participant.history_json = '[{"round": 1}]' participant.selected_round = 7 participant.selected_candidate = 'C03' participant.selected_true_quality = 88 participant.selected_reward_string = '$8.40' round1 = FakePlayer(participant, 1, { 'prolific_pid': 'PROLIFIC_R1', 'reveal_log': 'not json', 'gate1_reveal_log': ValueError('bad cell'), }) final_round = FakePlayer(participant, hf.C.NUM_ROUNDS, { 'resume_weight': 20, 'zoom_weight': 30, 'inperson_weight': 50, 'open_ended_explanation': special_text, 'has_hiring_experience': 1, 'hiring_experience_type': 'other', 'hiring_experience_other': 'Interview panels', 'age_range': '26-35', 'gender': 'female', 'device_type': 'computer', 'survey_comments': '', 'reveal_log': '{malformed', }) incomplete = FakePlayer( FakeParticipant(code='P2', id_in_session=2, label='', session=session, vars={'condition': hf.CONDITION_STANDARD}), 1, {}, ) legacy = FakePlayer(LegacyParticipant(FakeSession('LEGACY')), 1, {'open_ended_explanation': KeyError('missing')}) recovery_rows = list(hf.custom_export_recovery([round1, final_round, incomplete, legacy])) post_rows = list(hf.custom_export_post_survey([round1, final_round, incomplete, legacy])) general_rows = list(hf.custom_export([round1, final_round, incomplete, legacy])) assert_true(len(recovery_rows) == 5, 'recovery export includes header plus all supplied Player rows') assert_true(len(post_rows) == 4, 'post-survey export includes header plus all supplied participants') assert_true(len(general_rows) == 5, 'general export handles malformed reveal logs') recovery_header = recovery_rows[0] assert_true('participant_vars_json' in recovery_header, 'recovery export includes participant vars column') assert_true(recovery_rows[1][recovery_header.index('participant_vars_json')], 'participant JSON appears on round 1') assert_equal(recovery_rows[2][recovery_header.index('participant_vars_json')], '', 'participant JSON is blank after round 1') post_header = post_rows[0] p1_post = next(row for row in post_rows[1:] if row[post_header.index('participant_code')] == 'P1') assert_equal(p1_post[post_header.index('prolific_pid')], 'PROLIFIC_R1', 'post export combines round-1 Prolific ID') assert_equal(p1_post[post_header.index('open_ended_explanation')], special_text, 'post export combines final-round explanation') assert_equal(p1_post[post_header.index('survey_comments')], '', 'empty optional survey comment exports as blank') assert_equal(p1_post[post_header.index('resume_weight')], '20', 'resume weight exports') assert_equal(p1_post[post_header.index('zoom_weight')], '30', 'zoom weight exports') assert_equal(p1_post[post_header.index('inperson_weight')], '50', 'in-person weight exports') assert_equal(p1_post[post_header.index('participant_selected_round')], '7', 'participant selected round exports') buffer = io.StringIO() csv.writer(buffer).writerows(post_rows) assert_true('"Strategy with commas, line breaks' in buffer.getvalue(), 'CSV writer handles special characters') def test_reveal_log_normalization_and_filtering(): payload = { 'revealed_cells': ['C99:resume'], 'revealed_stages': ['resume'], 'reveal_log': [ { 'event_index': 1, 'round_number': 3, 'display_round': 3, 'condition': 'standard', 'gate': 'gate2', 'candidate': 'C01', 'stage': 'resume', 'cell_key': 'C01:resume', 'event_type': 'reveal', 'client_timestamp_ms': 123, 'elapsed_ms_on_page': 10, 'was_already_revealed': False, 'is_first_reveal_for_cell': True, }, { 'candidate': 'C01', 'stage': 'resume', 'cell_key': 'C01:resume', 't_ms': 12, 'order': 2, }, { 'candidate': 'C02', 'stage': 'zoom', 'cell_key': 'C02:zoom', 't_ms': 30, 'order': 3, }, { 'candidate': 'C03', 'stage': 'inperson', 'cell_key': 'C03:inperson', 't_ms': 40, 'order': 4, }, ], } filtered = hf.filter_reveal_state(payload, ['C01', 'C02'], ['resume', 'zoom']) assert_equal( [event['cell_key'] for event in filtered['reveal_log']], ['C01:resume', 'C02:zoom'], 'filter preserves valid event order and removes invalid/duplicate first reveals', ) assert_equal(filtered['revealed_cells'], ['C01:resume', 'C02:zoom'], 'revealed cells derive from filtered events') assert_equal(filtered['revealed_stages'], ['resume', 'zoom'], 'revealed stages derive from filtered cells') assert_equal(filtered['reveal_log'][0]['elapsed_ms_on_page'], 10, 'new elapsed field preserved') assert_equal(filtered['reveal_log'][1]['elapsed_ms_on_page'], 30, 'legacy t_ms maps to elapsed_ms_on_page') legacy_list = hf.normalize_reveal_log_data([ {'candidate': 'C01', 'stage': 'resume', 'cell_key': 'C01:resume', 'order': 1, 't_ms': 5} ]) assert_equal(legacy_list['reveal_log'][0]['event_index'], 1, 'legacy list gets event_index') def test_standard_gate2_before_next_page_smoke(): labels = [f'C{i:02d}' for i in range(1, 41)] data = { label: { 'resume': 40 + index, 'zoom': 45 + index, 'inperson': 50 + index, 'true_q': 55 + index, } for index, label in enumerate(labels, start=1) } gate1_choices = labels[:20] gate2_choices = labels[:10] reveal_state = { 'reveal_log': [ { 'event_index': 1, 'round_number': 1, 'display_round': 1, 'condition': hf.CONDITION_STANDARD, 'gate': 'gate2', 'candidate': 'C01', 'stage': 'resume', 'cell_key': 'C01:resume', 'event_type': 'reveal', 'client_timestamp_ms': 1000, 'elapsed_ms_on_page': 25, 'was_already_revealed': False, 'is_first_reveal_for_cell': True, }, { 'event_index': 2, 'round_number': 1, 'display_round': 1, 'condition': hf.CONDITION_STANDARD, 'gate': 'gate2', 'candidate': 'C02', 'stage': 'zoom', 'cell_key': 'C02:zoom', 'event_type': 'reveal', 'client_timestamp_ms': 1010, 'elapsed_ms_on_page': 35, 'was_already_revealed': False, 'is_first_reveal_for_cell': True, }, ], 'revealed_cells': ['C01:resume', 'C02:zoom'], 'revealed_stages': ['resume', 'zoom'], } player = make_player(hf.CONDITION_STANDARD, 1, { 'gate1_choices': json.dumps(gate1_choices), 'gate2_choices': json.dumps(gate2_choices), 'gate2_reveal_log': json.dumps(reveal_state), }) player.gate2_choices = json.dumps(gate2_choices) original_get_round_data = hf.get_round_data original_record_stage_duration = hf.record_stage_duration try: hf.get_round_data = lambda participant, round_number=None: data hf.record_stage_duration = lambda player, stage_name: None hf.MainGate2.before_next_page(player, False) context = json.loads(player.gate2_decision_context) assert_equal(context['selected_candidates'], gate2_choices, 'Gate 2 selected candidates are sanitized against Gate 1 choices') assert_equal( [candidate['label'] for candidate in context['visible_candidates']], gate1_choices, 'Gate 2 visible candidates are the candidates advanced from Gate 1', ) assert_equal(context['reveal_log'][0]['cell_key'], 'C01:resume', 'Gate 2 reveal log remains valid JSON and ordered') assert_true(json.loads(player.field_maybe_none('gate2_reveal_log')), 'raw Gate 2 reveal log remains valid JSON') gate3_context = hf.MainGate3.vars_for_template(player) assert_equal(gate3_context['available_candidates'], gate2_choices, 'Standard smoke path reaches Gate 3 candidates') finally: hf.get_round_data = original_get_round_data hf.record_stage_duration = original_record_stage_duration def make_standard_replay_participant(session, code, id_in_session): participant = FakeParticipant( code=code, id_in_session=id_in_session, label='', session=session, vars={'condition': hf.CONDITION_STANDARD, 'assignment_finalized': True}, ) labels = [f'C{i:02d}' for i in range(1, 41)] round_data = { label: { 'resume': 40 + index, 'zoom': 45 + index, 'inperson': 50 + index, 'true_q': 55 + index, } for index, label in enumerate(labels, start=1) } participant.main_data_json = json.dumps([ round_data for _ in range(hf.C.STANDARD_MAIN_ROUNDS) ]) gate2_choices = json.dumps(labels[:hf.C.GATE2_SELECT]) for round_number in range(1, hf.C.STANDARD_MAIN_ROUNDS + 1): fields = {'gate2_choices': gate2_choices} if round_number == hf.C.STANDARD_MAIN_ROUNDS: fields['hired_candidate'] = labels[0] player = FakePlayer(participant, round_number, fields) player.gate2_choices = gate2_choices if round_number == hf.C.STANDARD_MAIN_ROUNDS: player.hired_candidate = labels[0] return participant def test_admin_report_is_read_only_and_failure_resistant(): session = FakeSession('ADMIN') session.vars['replay_inventory_summary'] = { 'total_slots_created': 2, 'total_slots_unused': 1, 'total_slots_assigned': 0, 'total_slots_completed': 0, 'standard_assigned_count': 0, 'blind_assigned_count': 0, } session.vars['replay_inventory_controller'] = { 'slots': [ { 'slot_id': 'BAD_USAGE', 'standard_code': 'P1', 'status': 'available', 'usage_records': '{not-json', 'max_uses': 2, }, 'not-a-slot-dict', ], 'active_phase': 'adaptive_randomization', 'total_standard_replay_ready': 1, } malformed = FakeParticipant( code='MALFORMED', id_in_session=1, label='', session=session, vars={'condition': hf.CONDITION_STANDARD, 'assignment_finalized': True}, ) malformed.main_data_json = '{not-json' before_session_vars = json.dumps(session.vars, sort_keys=True, default=str) before_participant_vars = json.dumps(malformed.vars, sort_keys=True, default=str) original_sync = hf._sync_replay_inventory_from_participants try: hf._sync_replay_inventory_from_participants = lambda session: (_ for _ in ()).throw(AssertionError('admin report must not sync inventory')) context = hf.vars_for_admin_report(type('FakeSubsession', (), {'session': session})()) finally: hf._sync_replay_inventory_from_participants = original_sync assert_true(isinstance(context['replay_inventory_summary'], dict), 'admin report supplies safe summary') assert_true(len(context['participant_rows']) == 0, 'admin report does not scan participant rows') assert_equal(context['slot_rows'], [], 'admin report hides slot rows outside debug mode') assert_true(context['admin_error'], 'admin report surfaces warnings instead of raising') assert_equal(json.dumps(session.vars, sort_keys=True, default=str), before_session_vars, 'admin report does not mutate session vars') assert_equal(json.dumps(malformed.vars, sort_keys=True, default=str), before_participant_vars, 'admin report does not mutate participant vars') empty_session = FakeSession('EMPTY_ADMIN') empty_context = hf.vars_for_admin_report(type('FakeSubsession', (), {'session': empty_session})()) assert_true('Cached replay inventory summary is not available yet.' in empty_context['admin_error'], 'admin report warns when cached summary is missing') assert_true(isinstance(empty_context['replay_inventory_summary'], dict), 'admin report renders with missing cached summary') def test_lightweight_inventory_monitor_payload_is_read_only(): session = FakeSession('MONITOR') session.config['enable_lightweight_inventory_monitor'] = True session.vars['replay_inventory_summary'] = { 'active_phase': 'adaptive_randomization', 'total_standard_replay_ready': 2, 'blind_bundles_available_now': 1, 'additional_blind_capacity_after_waiters': 1, 'pending_blind_participants': 0, 'total_slots_created': 2, 'total_slots_unused': 1, 'total_slots_assigned': 1, 'total_slots_completed': 0, 'available_replay_slots': 1, 'assigned_replay_uses': 1, 'completed_replay_uses': 0, 'standard_assigned_count': 3, 'blind_assigned_count': 1, 'active_phase': 'adaptive_randomization', 'last_inventory_reconciliation_at': 12345, 'manual_assignment_override_active': False, 'adaptive_forced_condition': '', } session.vars['replay_inventory_controller'] = { 'slots': [ { 'slot_id': 'S1', 'standard_code': 'STD1', 'status': 'available', 'usage_records': '{malformed', 'max_uses': 2, 'use_count': 1, 'remaining_uses': 1, } ], 'active_phase': 'adaptive_randomization', } before = json.dumps(session.vars, sort_keys=True, default=str) original_sync = hf._sync_replay_inventory_from_participants original_lock = hf._assignment_state_lock original_save_replay = hf._save_replay_inventory_controller original_refresh_batch = hf._refresh_batch_controller original_sync_batch = hf._sync_batch_controller_from_participants try: hf._sync_replay_inventory_from_participants = lambda session: (_ for _ in ()).throw(AssertionError('monitor must not sync')) hf._assignment_state_lock = lambda: (_ for _ in ()).throw(AssertionError('monitor must not lock')) hf._save_replay_inventory_controller = lambda session, controller: (_ for _ in ()).throw(AssertionError('monitor must not save replay controller')) hf._refresh_batch_controller = lambda session: (_ for _ in ()).throw(AssertionError('monitor must not refresh batch controller')) hf._sync_batch_controller_from_participants = lambda session: (_ for _ in ()).throw(AssertionError('monitor must not sync batch controller')) first = hf.build_cached_inventory_monitor_payload(session) second = hf.build_cached_inventory_monitor_payload(session) assert_equal(first['monitor_mode'], 'cached_read_only', 'monitor payload advertises cached read-only mode') assert_equal(first['total_standard_replay_ready'], 2, 'monitor payload reads cached Standard ready count') assert_equal(first['last_updated'], second['last_updated'], 'repeated monitor payload does not change cached last_updated') assert_equal(json.dumps(session.vars, sort_keys=True, default=str), before, 'monitor payload does not mutate session vars') assert_equal(first['slot_rows'], [], 'monitor payload hides slot rows outside debug mode') session.config['debug_batch_matching'] = True debug_payload = hf.build_cached_inventory_monitor_payload(session) assert_true(debug_payload['slot_rows'][0]['row_error'], 'monitor payload reports malformed cached slot records in debug mode') assert_true('slot_id' not in debug_payload['slot_rows'][0], 'monitor debug slot rows omit slot IDs') assert_true('standard_code' not in debug_payload['slot_rows'][0], 'monitor debug slot rows omit participant/source codes') player = make_player(hf.CONDITION_STANDARD, 1) player.session = session player.participant.session = session response = hf.live_inventory_monitor(player, {'type': 'poll_cached_inventory_monitor'}) assert_equal(response[player.id_in_group]['monitor_mode'], 'cached_read_only', 'monitor live poll returns cached payload') finally: hf._sync_replay_inventory_from_participants = original_sync hf._assignment_state_lock = original_lock hf._save_replay_inventory_controller = original_save_replay hf._refresh_batch_controller = original_refresh_batch hf._sync_batch_controller_from_participants = original_sync_batch missing_session = FakeSession('MONITOR_MISSING') missing_payload = hf.build_cached_inventory_monitor_payload(missing_session) assert_equal(missing_payload['total_slots_created'], 0, 'monitor payload uses zero defaults when cache missing') assert_true(missing_payload['warning'], 'monitor payload warns when cached summary is missing') monitor_player = make_player(hf.CONDITION_STANDARD, 1) monitor_player.participant.label = 'INVENTORY_MONITOR' monitor_player.session.config['enable_lightweight_inventory_monitor'] = True assert_true(hf.LightweightInventoryMonitor.is_displayed(monitor_player), 'monitor page is visible only when enabled for monitor label') normal_player = make_player(hf.CONDITION_STANDARD, 1) normal_player.session.config['enable_lightweight_inventory_monitor'] = True assert_true(not hf.LightweightInventoryMonitor.is_displayed(normal_player), 'monitor page is hidden from ordinary participants') def test_assignment_and_blind_wait_fail_safe(): player = make_player(hf.CONDITION_STANDARD, 1) player.participant.vars = {'consent_completed': True} original_run_assignment_entry = hf._run_assignment_entry original_chooser = hf.DEV_CONDITION_CHOOSER try: hf.DEV_CONDITION_CHOOSER = False hf._run_assignment_entry = lambda player, event='': (_ for _ in ()).throw(RuntimeError('temporary assignment failure')) displayed = hf.AssignmentGate.is_displayed(player) assert_true(displayed, 'assignment failure shows safe hold page') assert_equal(player.participant.vars['assignment_last_error']['exception_type'], 'RuntimeError', 'assignment failure records error type') assert_true(player.participant.vars['assignment_hold'], 'assignment failure places participant on hold') finally: hf._run_assignment_entry = original_run_assignment_entry hf.DEV_CONDITION_CHOOSER = original_chooser blind_player = make_player(hf.CONDITION_BLIND, 1) blind_player.participant.vars = { 'condition': hf.CONDITION_BLIND, 'assignment_finalized': True, 'blind_replay_status': 'pending', } original_safe_status = hf._safe_blind_wait_allocation_status try: hf._safe_blind_wait_allocation_status = lambda player, event: (_ for _ in ()).throw(RuntimeError('poll failure')) response = hf.BlindReplayWait.live_method(blind_player, {'type': 'poll_replay_inventory'}) payload = response[blind_player.id_in_group] assert_equal(payload['status'], 'waiting', 'blind wait polling failure returns waiting') assert_true(payload['transient_error'], 'blind wait polling failure marks transient error') finally: hf._safe_blind_wait_allocation_status = original_safe_status def test_replay_inventory_standard_register_and_blind_allocate(): session = FakeSession('REPLAY') session.config.update({ 'use_adaptive_replay_inventory': True, 'use_batch_matching': False, }) standard_one = make_standard_replay_participant(session, 'STD1', 1) standard_two = make_standard_replay_participant(session, 'STD2', 2) assert_true(hf._register_adaptive_standard_replay_slot_if_ready(standard_one), 'first Standard completion registers replay slot') assert_true(hf._register_adaptive_standard_replay_slot_if_ready(standard_two), 'second Standard completion registers replay slot') controller = session.vars['replay_inventory_controller'] assert_equal(len(controller['slots']), 2, 'two Standard completions create two replay slots') assert_equal(session.vars['replay_inventory_summary']['total_slots_created'], 2, 'Standard completion updates cached slot count') blind = FakeParticipant( code='BLIND1', id_in_session=3, label='', session=session, vars={ 'condition': hf.CONDITION_BLIND, 'assignment_finalized': True, 'blind_replay_status': 'pending', 'assignment_randomized': True, 'assignment_probability_blind': 1.0, 'assignment_random_draw': 0.01, }, ) blind_player = FakePlayer(blind, 1) allocated = hf._ensure_pending_blind_replay_allocation(blind_player) assert_true(allocated, 'pending Blind participant consumes available replay bundle') assert_true(blind.main_data_json, 'Blind allocation writes combined replay main data') assert_equal(len(json.loads(blind.main_data_json)), 7, 'Blind allocation writes 7 replay rounds') assert_equal(json.loads(blind.vars['matched_standard_codes']), ['STD1'], 'Blind allocation uses exactly one Standard source') controller = session.vars['replay_inventory_controller'] assert_equal(sum(hf._slot_use_count(slot) for slot in controller['slots']), 1, 'Blind allocation records one source-slot use') assert_equal(session.vars['replay_inventory_summary']['total_slots_assigned'], 1, 'Blind allocation updates cached assigned uses') final_player = FakePlayer(blind, hf.C.NUM_ROUNDS, {'hired_candidate': 'C01'}) final_player.hired_candidate = 'C01' hf._mark_blind_completed_if_ready(final_player) assert_equal(session.vars['replay_inventory_summary']['total_slots_completed'], 1, 'Blind completion updates cached completed uses') def make_adaptive_assignment_session(code, forced_condition, source_count=1): session = FakeSession(code) session.config.update({ 'use_adaptive_replay_inventory': True, 'use_batch_matching': False, 'adaptive_forced_condition': forced_condition, 'blind_replay_sources_per_participant': source_count, 'blind_replay_max_uses_per_standard': 1, 'initial_completed_standard_threshold': 1, }) return session def make_unassigned_player(session, code, id_in_session): participant = FakeParticipant( code=code, id_in_session=id_in_session, label='', session=session, vars={}, ) return FakePlayer(participant, 1) def add_replay_inventory(session, count): for index in range(1, count + 1): participant = make_standard_replay_participant( session, f'STD_FORCE_{index}', index, ) assert_true( hf._register_adaptive_standard_replay_slot_if_ready(participant), f'Standard replay source {index} registers', ) def test_forced_condition_normalization_aliases(): assert_equal(hf._coerce_forced_condition('Standard_Condition'), hf.CONDITION_STANDARD, 'Standard alias normalizes') assert_equal(hf._coerce_forced_condition('Blind_Condition'), hf.CONDITION_BLIND, 'Blind alias normalizes') assert_equal(hf._coerce_forced_condition('Random_Condition'), hf.CONDITION_RANDOM, 'Random alias normalizes') assert_equal(hf._coerce_forced_condition('1'), hf.CONDITION_STANDARD, 'numeric Standard alias normalizes') assert_equal(hf._coerce_forced_condition('2'), hf.CONDITION_BLIND, 'numeric Blind alias normalizes') assert_equal(hf._coerce_forced_condition('0'), hf.CONDITION_RANDOM, 'numeric Random alias normalizes') assert_equal(hf._coerce_forced_condition('normal'), None, 'normal keeps default adaptive behavior') def test_forced_condition_standard_and_blind_modes(): standard_session = make_adaptive_assignment_session('FORCED_STANDARD', hf.CONDITION_STANDARD) standard_player = make_unassigned_player(standard_session, 'P_FORCED_STANDARD', 1) standard_result = hf._ensure_adaptive_replay_inventory_assignment(standard_player) assert_true(standard_result['assigned'], 'forced Standard assigns immediately') assert_equal(standard_player.participant.vars['condition'], hf.CONDITION_STANDARD, 'forced Standard selects Standard') assert_equal( standard_session.vars['replay_inventory_summary']['total_slots_assigned'], 0, 'forced Standard consumes no Blind inventory', ) assert_equal( standard_player.participant.vars['adaptive_forced_condition'], hf.CONDITION_STANDARD, 'forced Standard metadata records override', ) blind_session = make_adaptive_assignment_session('FORCED_BLIND', hf.CONDITION_BLIND, source_count=1) add_replay_inventory(blind_session, 1) blind_player = make_unassigned_player(blind_session, 'P_FORCED_BLIND', 2) blind_result = hf._ensure_adaptive_replay_inventory_assignment(blind_player) assert_true(blind_result['assigned'], 'forced Blind with inventory assigns') assert_equal(blind_player.participant.vars['condition'], hf.CONDITION_BLIND, 'forced Blind selects Blind') assert_true(blind_player.participant.main_data_json, 'forced Blind receives replay data') assert_true(blind_player.participant.vars.get('matched_replay_slot_id'), 'forced Blind saves matched slot metadata') assert_equal( blind_session.vars['replay_inventory_summary']['total_slots_assigned'], 1, 'forced Blind consumes one replay source when configured for one source', ) no_inventory_session = make_adaptive_assignment_session('FORCED_BLIND_EMPTY', hf.CONDITION_BLIND, source_count=1) no_inventory_player = make_unassigned_player(no_inventory_session, 'P_FORCED_BLIND_EMPTY', 3) no_inventory_result = hf._ensure_adaptive_replay_inventory_assignment(no_inventory_player) assert_true(no_inventory_result['hold'], 'forced Blind without inventory uses safe hold behavior') assert_equal(no_inventory_player.participant.vars['blind_replay_status'], 'pending', 'forced Blind without inventory is pending, not crashed') assert_true(not no_inventory_player.participant.main_data_json, 'forced Blind without inventory does not create empty replay data') def test_forced_random_inventory_gated_assignment(): original_random = hf.random.random original_save_replay_inventory_controller = hf._save_replay_inventory_controller try: blind_session = make_adaptive_assignment_session('FORCED_RANDOM_BLIND', hf.CONDITION_RANDOM, source_count=1) add_replay_inventory(blind_session, 1) hf.random.random = lambda: 0.0 blind_player = make_unassigned_player(blind_session, 'P_FORCED_RANDOM_BLIND', 2) blind_result = hf._ensure_adaptive_replay_inventory_assignment(blind_player) assert_true(blind_result['assigned'], 'forced Random Blind draw assigns') assert_equal(blind_player.participant.vars['condition'], hf.CONDITION_BLIND, 'forced Random draw below probability selects Blind') assert_true(blind_player.participant.vars['forced_randomization_attempted'], 'forced Random with inventory records attempted randomization') assert_equal(blind_player.participant.vars['forced_condition_random_draw'], 0.0, 'forced Random records draw when inventory exists') assert_equal(blind_session.vars['replay_inventory_summary']['total_slots_assigned'], 1, 'forced Random Blind draw consumes inventory') assert_equal(hf._get_blind_replay_sources_per_participant(blind_session), 1, 'one-source Blind replay design is used') two_source_config_session = make_adaptive_assignment_session( 'FORCED_RANDOM_TWO_SOURCE_CONFIG_IGNORED', hf.CONDITION_RANDOM, source_count=2, ) assert_equal( hf._get_blind_replay_sources_per_participant(two_source_config_session), 1, 'session config cannot request a second Blind replay source', ) standard_session = make_adaptive_assignment_session('FORCED_RANDOM_STANDARD', hf.CONDITION_RANDOM, source_count=1) add_replay_inventory(standard_session, 1) hf.random.random = lambda: 0.99 standard_player = make_unassigned_player(standard_session, 'P_FORCED_RANDOM_STANDARD', 2) standard_result = hf._ensure_adaptive_replay_inventory_assignment(standard_player) assert_true(standard_result['assigned'], 'forced Random Standard draw assigns') assert_equal(standard_player.participant.vars['condition'], hf.CONDITION_STANDARD, 'forced Random draw above probability selects Standard') assert_equal(standard_player.participant.vars['forced_condition_random_draw'], 0.99, 'forced Random Standard draw records draw') assert_equal(standard_session.vars['replay_inventory_summary']['total_slots_assigned'], 0, 'forced Random Standard draw does not consume inventory') empty_session = make_adaptive_assignment_session('FORCED_RANDOM_EMPTY', hf.CONDITION_RANDOM, source_count=1) def fail_if_called(): raise AssertionError('random draw should not occur without Blind inventory') hf.random.random = fail_if_called empty_player = make_unassigned_player(empty_session, 'P_FORCED_RANDOM_EMPTY', 1) hf._save_replay_inventory_controller = lambda *args, **kwargs: (_ for _ in ()).throw( AssertionError('forced Random without inventory should not save replay controller') ) try: empty_result = hf._ensure_adaptive_replay_inventory_assignment(empty_player) assert_true(empty_result['assigned'], 'forced Random without inventory assigns Standard directly') assert_equal(empty_player.participant.vars['condition'], hf.CONDITION_STANDARD, 'forced Random without inventory selects Standard') assert_true(not empty_player.participant.vars['forced_randomization_attempted'], 'forced Random without inventory does not attempt randomization') assert_true('forced_condition_random_draw' not in empty_player.participant.vars, 'forced Random without inventory records no random draw') assert_equal( empty_player.participant.vars['forced_condition_fallback_reason'], 'cached_inventory_unavailable_assign_standard', 'forced Random without cached inventory records fallback reason', ) assert_equal(empty_session.vars.get('replay_inventory_summary', {}).get('total_slots_assigned', 0), 0, 'forced Random without inventory consumes no replay slots') finally: hf._save_replay_inventory_controller = original_save_replay_inventory_controller finally: hf.random.random = original_random hf._save_replay_inventory_controller = original_save_replay_inventory_controller def main(): test_round_structure_and_active_pages() test_post_task_validation() test_click_to_reveal_instruction_copy() test_removed_noise_comprehension_question() test_prolific_id_dev_override() test_dev_condition_chooser_paths() test_safe_exports() test_reveal_log_normalization_and_filtering() test_standard_gate2_before_next_page_smoke() test_admin_report_is_read_only_and_failure_resistant() test_lightweight_inventory_monitor_payload_is_read_only() test_assignment_and_blind_wait_fail_safe() test_replay_inventory_standard_register_and_blind_allocate() test_forced_condition_normalization_aliases() test_forced_condition_standard_and_blind_modes() test_forced_random_inventory_gated_assignment() print('regression checks passed') if __name__ == '__main__': main()