File size: 5,985 Bytes
f14ef69 afd13cb f14ef69 afd13cb f14ef69 7badcff f14ef69 6f5ba40 0818184 7badcff 6f5ba40 f14ef69 6f5ba40 f14ef69 6f5ba40 7badcff f14ef69 e79873b 6f5ba40 0ba5932 7badcff f14ef69 | 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 | """Real PostgreSQL transactions and a local private-blob stand-in for tests."""
import json
import uuid
from pathlib import Path
import pytest
from app.answer_scans import AnswerScans, ScanError
OWNER = '11111111-1111-4111-8111-111111111111'
OTHER = '22222222-2222-4222-8222-222222222222'
BOOTSTRAP = '''
create schema auth;
create table auth.users(id uuid primary key);
create function auth.uid() returns uuid language sql stable as
$$select nullif(current_setting('request.jwt.claim.sub',true),'')::uuid$$;
grant usage on schema public,auth to anon,authenticated,service_role;
create schema storage;
create table storage.buckets(id text primary key,name text,public boolean,file_size_limit bigint,allowed_mime_types text[]);
create table storage.objects(id uuid primary key default gen_random_uuid(),bucket_id text,name text unique,created_at timestamptz default now());
alter table storage.objects enable row level security;
grant usage on schema storage to anon,authenticated,service_role;
grant all on storage.objects to anon,authenticated,service_role;
create policy preexisting_permissive on storage.objects for all using(true) with check(true);
create table public.classes(id int primary key,name text);
insert into classes values(1,'existing math class');
'''
class ScanAuth:
url = 'https://test.invalid'
key = 'public-test'
secret = 'server-test'
# OWNER is a teacher, so these tests check answers straight with the AI; the students' queue is in test_ai_queue
teachers = frozenset({OWNER + '@test.invalid'})
developers = frozenset()
def _admin(self, path, *args, **kwargs):
from app.auth import AuthUnavailable
raise AuthUnavailable('supabase answered 404: no tables here but the answer RPCs')
def user(self, token):
if token not in (OWNER, OTHER):
return None
return {'id': token, 'email': token + '@test.invalid'}
def classes(self):
return []
def user_email(self, user_id):
return user_id + '@test.invalid'
def access_for(self, _id, **kwargs):
return None
class DatabaseAnswers(AnswerScans):
def __init__(self, uri, files):
super().__init__(ScanAuth(), enabled=True)
self.uri, self.files = uri, files
self.fail_put = self.fail_finalize = False
def connect(self):
import psycopg
return psycopg.connect(self.uri, autocommit=True)
def action(self, user_id, action, **data):
return self.rpc('math_answer_action', user_id, action, data)
def grade(self, user_id, action, **data):
return self.rpc('math_grade_action', user_id, action, data)
def quiz(self, user_id, action, **data):
return self.rpc('math_quiz_action', user_id, action, data)
def review(self, user_id, action, **data):
return self.rpc('math_review_action', user_id, action, data)
def rpc(self, name, user_id, action, data):
import psycopg
if self.fail_finalize and action == 'finalize':
raise ScanError(503, 'answer_storage_unavailable')
with self.connect() as c:
c.execute('set role service_role')
try:
return c.execute(f'select public.{name}(%s,%s,%s::jsonb)',
(user_id, action, json.dumps(data))).fetchone()[0]
except psycopg.errors.NoDataFound as exc:
raise ScanError(404, 'not_found') from exc
except psycopg.errors.RaiseException as exc:
message = str(exc).splitlines()[0]
raise ScanError(429 if message in ('grading_limit', 'review_limit') else 409, message) from exc
def put(self, key, data):
if self.fail_put:
raise ScanError(503, 'answer_storage_unavailable')
p = self.files / key
p.parent.mkdir(parents=True, exist_ok=True)
p.write_bytes(data)
with self.connect() as c:
c.execute("insert into storage.objects(bucket_id,name) values('mathly-answers',%s)", (key,))
def read(self, key):
return (self.files / key).read_bytes()
def delete(self, key):
(self.files / key).unlink(missing_ok=True)
with self.connect() as c:
c.execute('delete from storage.objects where name=%s', (key,))
@pytest.fixture(scope='session')
def postgres_server(tmp_path_factory):
pgserver = pytest.importorskip('pgserver')
psycopg = pytest.importorskip('psycopg')
server = pgserver.get_server(tmp_path_factory.mktemp('postgres') / 'data', cleanup_mode='delete')
with psycopg.connect(server.get_uri(), autocommit=True) as c:
c.execute('create role anon; create role authenticated; create role service_role bypassrls;')
yield server
@pytest.fixture
def answer_store(postgres_server, tmp_path):
import psycopg
database = 'scan_' + uuid.uuid4().hex
with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c:
c.execute('create database ' + database)
uri = postgres_server.get_uri(database=database)
store = DatabaseAnswers(uri, tmp_path / 'images')
with store.connect() as c:
c.execute(BOOTSTRAP)
c.execute(Path('supabase/migrations/20261031133254_math_answer_scans.sql').read_text())
for migration in sorted(Path('supabase/migrations').glob('*_math_answer_*.sql')):
if migration.name.endswith('_math_answer_scans.sql'):
continue
c.execute(migration.read_text())
for name in ('quiz_sittings', 'teacher_reviews', 'quiz_modes'): # in this order: each rewrites a function of the one before
for migration in sorted(Path('supabase/migrations').glob(f'*_math_{name}.sql')):
c.execute(migration.read_text())
c.execute('insert into auth.users values(%s),(%s)', (OWNER, OTHER))
yield store
with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c:
c.execute('drop database ' + database + ' with (force)')
|