import sys, os, io, json, uuid import pytest ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.insert(0, os.path.join(ROOT, 'backend')) sys.path.insert(0, ROOT) import server # the Flask app from server import app from session_management.session_manager import SessionManager # ── Shared fixture ──────────────────────────────────────────────────────────── @pytest.fixture(scope='module') def client(): """ Flask test client using an isolated in-memory test session_manager. Module-scoped so state (pid, sid) is shared across the ordered test class. """ server.session_manager = SessionManager(test_mode=True) server.db = server.session_manager.db app.config['TESTING'] = True with app.test_client() as c: yield c # ── UI test class ───────────────────────────────────────────────────────────── class TestFlaskUI: """ Tests for every Flask API endpoint consumed by respira-flask.html. Tests run in definition order; shared state is stored as class attributes. """ # ── Setup: create a unique patient + session before any test runs ────────── @pytest.fixture(autouse=True, scope='class') def setup_patient_and_session(self, client): """Create a unique patient and open a session shared across all tests.""" unique_name = f"UI Test Patient {uuid.uuid4().hex[:8]}" r = client.post( '/api/patients', data=json.dumps({'name': unique_name, 'dob': '1975-06-15', 'gender': 'F'}), content_type='application/json', ) data = r.get_json() assert data['type'] == 'PATIENT_CREATED', f"Setup failed to create patient: {data}" TestFlaskUI.pid = data['pid'] TestFlaskUI.unique_name = unique_name sid_r = client.post( '/api/sessions/start', data=json.dumps({'pid': TestFlaskUI.pid}), content_type='application/json', ).get_json() assert sid_r['type'] == 'SESSION_STARTED', f"Setup failed to start session: {sid_r}" TestFlaskUI.sid = sid_r['sid'] yield client.post( '/api/sessions/end', data=json.dumps({'sid': TestFlaskUI.sid}), content_type='application/json', ) # ── Static / HTML ───────────────────────────────────────────────────────── def test_index_serves_html(self, client): r = client.get('/') assert r.status_code == 200 assert b'' in r.data def test_explicit_html_route(self, client): r = client.get('/frontend/respira-flask.html') assert r.status_code == 200 assert b'Respira' in r.data # ── Auth ────────────────────────────────────────────────────────────────── def test_login_valid_credentials_returns_success(self, client, monkeypatch): """Known valid doctor1 credentials should succeed.""" import bcrypt plain_password = "testpass123" hashed_password = bcrypt.hashpw(plain_password.encode(), bcrypt.gensalt()).decode() monkeypatch.setenv("DOCTOR1_PASSWORD", hashed_password) r = client.post( '/api/auth/login', data=json.dumps({'username': 'doctor1', 'password': plain_password}), content_type='application/json', ) assert r.status_code == 200 data = r.get_json() assert data['success'] is True assert data['name'] == 'Doctor One' def test_login_missing_body_does_not_crash(self, client): r = client.post( '/api/auth/login', data=json.dumps({}), content_type='application/json', ) assert r.status_code == 200 assert 'success' in r.get_json() # ── Patients ────────────────────────────────────────────────────────────── def test_create_patient_returns_pid(self, client): new_name = f"Create Test {uuid.uuid4().hex[:8]}" r = client.post( '/api/patients', data=json.dumps({'name': new_name, 'dob': '1980-03-20', 'gender': 'M'}), content_type='application/json', ) assert r.status_code == 200 data = r.get_json() assert data['type'] == 'PATIENT_CREATED' assert isinstance(data['pid'], int) def test_create_duplicate_patient_returns_error(self, client): r = client.post( '/api/patients', data=json.dumps({'name': TestFlaskUI.unique_name, 'dob': '1975-06-15', 'gender': 'F'}), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] == 'ERROR' def test_list_patients_includes_created_patient(self, client): r = client.get('/api/patients') assert r.status_code == 200 data = r.get_json() assert 'patients' in data pids = [p['pid'] for p in data['patients']] assert TestFlaskUI.pid in pids # ── Sessions ────────────────────────────────────────────────────────────── def test_start_and_end_session(self, client): """Start a second session for the same patient and immediately end it.""" r = client.post( '/api/sessions/start', data=json.dumps({'pid': TestFlaskUI.pid}), content_type='application/json', ) assert r.status_code == 200 data = r.get_json() assert data['type'] == 'SESSION_STARTED' assert isinstance(data['sid'], int) assert 'existing_evidence' in data extra_sid = data['sid'] r2 = client.post( '/api/sessions/end', data=json.dumps({'sid': extra_sid}), content_type='application/json', ) assert r2.status_code == 200 assert r2.get_json()['type'] == 'SESSION_ENDED' # ── Text pipeline ───────────────────────────────────────────────────────── def test_process_text_returns_valid_response_type(self, client): r = client.post( '/api/process_text', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'text': 'Patient has a persistent cough and shortness of breath', }), content_type='application/json', ) assert r.status_code == 200 data = r.get_json() assert 'type' in data assert data['type'] in ( 'RISK_ASSESSMENT_OUTPUT', 'INTENT_CLARIFICATION', 'VAGUE_CLARIFICATION', 'TEXT', 'ERROR', ) def test_process_text_risk_assessment_has_required_keys(self, client): r = client.post( '/api/process_text', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'text': 'Patient smokes 20 cigarettes a day and has haemoptysis and weight loss', }), content_type='application/json', ) data = r.get_json() if data['type'] == 'RISK_ASSESSMENT_OUTPUT': for key in ('risk_level', 'risk_label', 'explanation', 'session_evidence', 'vector_results', 'sources'): assert key in data, f"RISK_ASSESSMENT_OUTPUT missing key: {key}" assert data['risk_label'] in ('Low', 'Medium', 'High', 'Unknown') assert isinstance(data['vector_results'], list) TestFlaskUI.last_vector_output = data.get('vector_search_output') TestFlaskUI.last_info_array = data.get('info_array') TestFlaskUI.last_risk_level = data.get('last_risk_level') def test_process_text_help_request(self, client): r = client.post( '/api/process_text', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'text': 'How do I use this system?', }), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] in ('TEXT', 'INTENT_CLARIFICATION') def test_process_text_source_request_without_prior_output(self, client): """Asking for sources before any evidence should return a graceful message.""" # Start a blank session blank_sid = client.post( '/api/sessions/start', data=json.dumps({'pid': TestFlaskUI.pid}), content_type='application/json', ).get_json()['sid'] r = client.post( '/api/process_text', data=json.dumps({ 'sid': blank_sid, 'pid': TestFlaskUI.pid, 'text': 'Can you show me the sources?', }), content_type='application/json', ) assert r.status_code == 200 data = r.get_json() assert data['type'] in ('TEXT', 'INTENT_CLARIFICATION') client.post( '/api/sessions/end', data=json.dumps({'sid': blank_sid}), content_type='application/json', ) # ── Clarification resolution ────────────────────────────────────────────── def test_resolve_intent_unknown_option_returns_error(self, client): r = client.post( '/api/resolve_intent', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'selected_option': 'Not a real option', 'text_input': 'some query', }), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] == 'ERROR' def test_resolve_intent_add_evidence(self, client): r = client.post( '/api/resolve_intent', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'selected_option': 'Add new evidence', 'text_input': 'Patient has chest pain and wheezing', }), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] in ( 'RISK_ASSESSMENT_OUTPUT', 'VAGUE_CLARIFICATION', 'TEXT', 'ERROR', ) def test_resolve_intent_general_help(self, client): r = client.post( '/api/resolve_intent', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'selected_option': 'General help', 'text_input': 'help me', }), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] == 'TEXT' def test_resolve_vague_symptom_returns_valid_type(self, client): r = client.post( '/api/resolve_vague', data=json.dumps({ 'sid': TestFlaskUI.sid, 'pid': TestFlaskUI.pid, 'vague_symptom': 'cough', 'selected_option': 'persistent cough', 'extracted_evidence': ['cough'], }), content_type='application/json', ) assert r.status_code == 200 assert r.get_json()['type'] in ( 'RISK_ASSESSMENT_OUTPUT', 'VAGUE_CLARIFICATION', 'TEXT', 'ERROR', ) # ── X-ray upload ────────────────────────────────────────────────────────── def test_upload_xray_no_file_returns_error(self, client): r = client.post('/api/upload_xray', data={ 'sid': str(TestFlaskUI.sid), 'pid': str(TestFlaskUI.pid), }) assert r.status_code == 200 data = r.get_json() assert data['type'] == 'ERROR' assert data['message'] == 'No file provided' def test_upload_xray_with_image_file(self, client): """ Upload a minimal 1×1 JPEG. The CNN may find no findings — both RISK_ASSESSMENT_OUTPUT and TEXT are acceptable outcomes. """ minimal_jpeg = ( b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00' b'\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t' b'\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a' b'\x1f\x1e\x1d\x1a\x1c\x1c $.\' ",#\x1c\x1c(7),01444\x1f\'9=82<.342\x1e' b'\xff\xc0\x00\x0b\x08\x00\x01\x00\x01\x01\x01\x11\x00' b'\xff\xc4\x00\x1f\x00\x00\x01\x05\x01\x01\x01\x01\x01\x01\x00\x00' b'\x00\x00\x00\x00\x00\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b' b'\xff\xda\x00\x08\x01\x01\x00\x00?\x00\xfb\xd4P\x00\x00\x00\xff\xd9' ) r = client.post( '/api/upload_xray', data={ 'sid': str(TestFlaskUI.sid), 'pid': str(TestFlaskUI.pid), 'file': (io.BytesIO(minimal_jpeg), 'test.jpg'), }, content_type='multipart/form-data', ) assert r.status_code == 200 assert r.get_json()['type'] in ('RISK_ASSESSMENT_OUTPUT', 'TEXT', 'ERROR') # ── Patient records ─────────────────────────────────────────────────────── def test_patient_records_valid_pid(self, client): r = client.get(f'/api/patients/{TestFlaskUI.pid}/records') assert r.status_code == 200 data = r.get_json() assert 'patient' in data assert 'sessions' in data assert 'evidence' in data assert 'xray_gallery' in data assert data['patient']['pid'] == TestFlaskUI.pid def test_patient_records_unknown_pid_returns_404(self, client): r = client.get('/api/patients/99999999/records') assert r.status_code == 404 # ── Dashboard ───────────────────────────────────────────────────────────── def test_dashboard_counts_present(self, client): r = client.get('/api/dashboard') assert r.status_code == 200 data = r.get_json() assert 'counts' in data for k in ('patients', 'sessions', 'symptoms', 'imaging', 'risk_factors'): assert k in data['counts'], f"Missing dashboard count: {k}" def test_dashboard_chart_data_present(self, client): r = client.get('/api/dashboard') data = r.get_json() for key in ('risk_levels', 'symptoms', 'risk_factors', 'age_distribution', 'gender_split', 'sessions_over_time', 'recent_sessions'): assert key in data, f"Missing dashboard key: {key}"