Spaces:
Sleeping
Sleeping
File size: 7,904 Bytes
840101a | 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 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | import streamlit as st
from supabase import create_client, Client
import os
import json
def init_supabase():
"""Initialize Supabase client"""
try:
url = st.secrets.get("SUPABASE_URL")
key = st.secrets.get("SUPABASE_KEY")
except:
url = os.getenv("SUPABASE_URL")
key = os.getenv("SUPABASE_KEY")
if not url or not key:
st.error("Supabase credentials not found.")
return None
return create_client(url, key)
def test_submissions():
"""Test submission functionality"""
st.title("📝 Submission System Test")
supabase = init_supabase()
if not supabase:
st.error("Failed to initialize Supabase client")
return
try:
# Test 1: Check if student_submissions table exists
st.subheader("🔍 Testing Database Connection")
try:
response = supabase.table('student_submissions').select('*').limit(1).execute()
st.success("✅ student_submissions table exists and is accessible")
except Exception as e:
st.error(f"❌ student_submissions table error: {str(e)}")
return
# Test 2: Get all users to find a test user
st.subheader("👥 Available Users")
try:
users_response = supabase.table('users').select('*').execute()
if users_response.data:
st.success(f"Found {len(users_response.data)} users:")
for user in users_response.data:
st.write(f"**{user.get('username')}** - ID: `{user.get('id')}` - Role: {user.get('role')}")
# Use the first student for testing
test_user = None
for user in users_response.data:
if user.get('role') == 'student':
test_user = user
break
# If no student found, use the first user
if not test_user and users_response.data:
test_user = users_response.data[0]
st.warning(f"No student users found, using {test_user.get('username')} instead")
if not test_user:
st.error("No users found for testing")
return
st.info(f"Using {test_user.get('username')} (ID: {test_user.get('id')}) for testing")
else:
st.error("No users found in database")
return
except Exception as e:
st.error(f"❌ Error fetching users: {str(e)}")
return
# Test 3: Try to insert a test submission
st.subheader("📤 Testing Submission Insert")
test_submission_data = {
"questions": [
{
"id": "test_q1",
"type": "textarea",
"question": "Test question",
"student_answer": "Test answer",
"max_length": 500
}
]
}
test_data = {
'username': test_user.get('username'),
'week_number': 1,
'submission_type': 'test_submission',
'submission_data': test_submission_data,
'score': 5,
'max_score': 5
}
st.write("**Attempting to insert test submission:**")
st.json(test_data)
try:
insert_response = supabase.table('student_submissions').insert(test_data).execute()
st.write(f"**Insert Response:** {insert_response}")
st.write(f"**Response Data:** {insert_response.data}")
st.write(f"**Response Count:** {insert_response.count}")
if insert_response.data:
st.success("✅ Test submission inserted successfully!")
else:
st.error("❌ No data returned from insert")
except Exception as e:
st.error(f"❌ Insert failed: {str(e)}")
# Test 4: Try to retrieve the test submission
st.subheader("📥 Testing Submission Retrieval")
try:
retrieve_response = supabase.table('student_submissions').select('*').eq('username', test_user.get('username')).eq('week_number', 1).eq('submission_type', 'test_submission').execute()
st.write(f"**Retrieve Response:** {retrieve_response}")
st.write(f"**Retrieve Data:** {retrieve_response.data}")
st.write(f"**Retrieve Count:** {retrieve_response.count}")
if retrieve_response.data:
st.success("✅ Test submission retrieved successfully!")
st.json(retrieve_response.data[0])
else:
st.warning("⚠️ No submission found (insert may have failed)")
except Exception as e:
st.error(f"❌ Retrieve failed: {str(e)}")
# Test 5: Test upsert functionality
st.subheader("🔄 Testing Upsert Functionality")
updated_submission_data = {
"questions": [
{
"id": "test_q1",
"type": "textarea",
"question": "Updated test question",
"student_answer": "Updated test answer",
"max_length": 500
}
]
}
upsert_data = {
'username': test_user.get('username'),
'week_number': 1,
'submission_type': 'test_submission',
'submission_data': updated_submission_data,
'score': 10,
'max_score': 10
}
st.write("**Attempting to upsert test submission:**")
st.json(upsert_data)
try:
upsert_response = supabase.table('student_submissions').upsert(upsert_data).execute()
st.write(f"**Upsert Response:** {upsert_response}")
st.write(f"**Upsert Data:** {upsert_response.data}")
st.write(f"**Upsert Count:** {upsert_response.count}")
if upsert_response.data:
st.success("✅ Test submission upserted successfully!")
else:
st.error("❌ No data returned from upsert")
except Exception as e:
st.error(f"❌ Upsert failed: {str(e)}")
# Test 6: Check RLS policies
st.subheader("🔒 Testing RLS Policies")
try:
# Try to get all submissions (should be limited by RLS)
all_submissions = supabase.table('student_submissions').select('*').execute()
st.write(f"**All submissions count:** {len(all_submissions.data) if all_submissions.data else 0}")
if all_submissions.data:
st.info("RLS allows viewing submissions (may be too permissive)")
else:
st.info("RLS is working correctly - no submissions visible")
except Exception as e:
st.error(f"❌ RLS test failed: {str(e)}")
# Test 7: Clean up test data
st.subheader("🧹 Cleanup")
if st.button("Clean up test submissions"):
try:
delete_response = supabase.table('student_submissions').delete().eq('submission_type', 'test_submission').execute()
st.success(f"✅ Cleaned up {len(delete_response.data) if delete_response.data else 0} test submissions")
except Exception as e:
st.error(f"❌ Cleanup failed: {str(e)}")
except Exception as e:
st.error(f"❌ Test failed: {str(e)}")
if __name__ == "__main__":
test_submissions() |