Spaces:

fatimaxa
/

Lung-Cancer-Risk-Diagnosis-Assistant

Running

App Files Files Community

Lung-Cancer-Risk-Diagnosis-Assistant / testing /test_flask_ui.py

fatimaxa

Upload 112 files

00bd0c6 verified 2 months ago

raw

history blame contribute delete

15.9 kB


	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'<!DOCTYPE html>' 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}"