| import asyncio |
| import urllib.parse |
| import uuid |
|
|
| import httpx |
| import pytest |
|
|
| from app.auth import AuthUnavailable |
| from app.main import create_app |
| from scan_support import OTHER, OWNER, ScanAuth, postgres_server |
|
|
| DEV = '33333333-3333-4333-8333-333333333333' |
|
|
|
|
| class PrompterAuth(ScanAuth): |
| """OWNER and DEV are developers, OTHER is a student; an in-memory math_prompter_texts behind the PostgREST calls.""" |
| developers = frozenset({OWNER + '@test.invalid', DEV + '@test.invalid'}) |
|
|
| def __init__(self): |
| self.rows: list[dict] = [] |
| self.missing = False |
| self.no_outline = False |
|
|
| def user(self, token): |
| return {'id': token, 'email': token + '@test.invalid'} if token in (OWNER, OTHER, DEV) else None |
|
|
| def _admin(self, path, method='GET', body=None, prefer=''): |
| if self.missing: |
| raise AuthUnavailable('supabase answered 404: relation does not exist') |
| if self.no_outline and ('outline' in path or 'outline' in (body or {})): |
| raise AuthUnavailable('supabase answered 400: column math_prompter_texts.outline does not exist') |
| table, _, query = path.partition('?') |
| assert table == '/rest/v1/math_prompter_texts' |
| q = dict(urllib.parse.parse_qsl(query)) |
| match = [r for r in self.rows if all(r[k] == q[k].removeprefix('eq.') for k in ('id', 'user_id') if k in q)] |
| if method == 'POST': |
| row = {**body, 'id': str(uuid.uuid4())} |
| self.rows.append(row) |
| return [row] |
| if method == 'PATCH': |
| for r in match: |
| r.update(body) |
| return match |
| if method == 'DELETE': |
| self.rows = [r for r in self.rows if r not in match] |
| return None |
| return sorted(match, key=lambda r: r['updated_at'], reverse=True) |
|
|
|
|
| def call(app, user, method, url, **kw): |
| async def go(): |
| async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test', |
| headers={'Authorization': f'Bearer {user}'}) as c: |
| return await c.request(method, url, **kw) |
| return asyncio.run(go()) |
|
|
|
|
| @pytest.fixture |
| def app(): |
| a = create_app() |
| a.state.auth = PrompterAuth() |
| return a |
|
|
|
|
| def test_developer_saves_lists_updates_and_deletes(app): |
| r = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': '', 'body': ' ืฉืืื ืืืืื\nืฉืืจื ืฉื ืืื'}) |
| assert r.status_code == 200 |
| first = r.json()['text'] |
| assert first['title'] == 'ืฉืืื ืืืืื' |
| second = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 'ืคืชืืื', 'body': 'ืืงืกื'}).json()['text'] |
| texts = call(app, OWNER, 'GET', '/api/prompter/texts').json()['texts'] |
| assert [t['id'] for t in texts] == [second['id'], first['id']] |
| r = call(app, OWNER, 'PUT', f"/api/prompter/texts/{first['id']}", json={'title': 'ืืืฉ', 'body': 'ืืืฃ ืืืฉ'}) |
| assert r.json()['text']['title'] == 'ืืืฉ' and r.json()['text']['body'] == 'ืืืฃ ืืืฉ' |
| assert call(app, OWNER, 'DELETE', f"/api/prompter/texts/{second['id']}").status_code == 200 |
| assert [t['title'] for t in call(app, OWNER, 'GET', '/api/prompter/texts').json()['texts']] == ['ืืืฉ'] |
|
|
|
|
| def test_each_developer_sees_only_their_own(app): |
| mine = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 'ืฉืื', 'body': 'x'}).json()['text'] |
| assert call(app, DEV, 'GET', '/api/prompter/texts').json()['texts'] == [] |
| assert call(app, DEV, 'PUT', f"/api/prompter/texts/{mine['id']}", json={'title': 'a', 'body': 'b'}).status_code == 404 |
| call(app, DEV, 'DELETE', f"/api/prompter/texts/{mine['id']}") |
| assert len(call(app, OWNER, 'GET', '/api/prompter/texts').json()['texts']) == 1 |
|
|
|
|
| def test_students_and_bad_input_are_refused(app): |
| assert call(app, OTHER, 'GET', '/api/prompter/texts').status_code == 403 |
| assert call(app, OTHER, 'POST', '/api/prompter/texts', json={'title': 'a', 'body': 'b'}).status_code == 403 |
| assert call(app, 'nobody', 'GET', '/api/prompter/texts').status_code == 401 |
| assert call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 'a', 'body': ' '}).status_code == 400 |
| assert call(app, OWNER, 'POST', '/api/prompter/texts', json={'body': 'x' * 100_001}).status_code == 413 |
| assert call(app, OWNER, 'PUT', '/api/prompter/texts/not-an-id', json={'body': 'x'}).status_code == 404 |
|
|
|
|
| def test_long_title_is_cut_and_missing_table_is_503(app): |
| r = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 'ื' * 300, 'body': 'x'}) |
| assert len(r.json()['text']['title']) == 120 |
| app.state.auth.missing = True |
| assert call(app, OWNER, 'GET', '/api/prompter/texts').status_code == 503 |
|
|
|
|
| def test_migration_on_real_postgres(postgres_server): |
| """The table exists, keeps its limits, and only the server's key may use it.""" |
| psycopg = pytest.importorskip("psycopg") |
| from pathlib import Path |
| from scan_support import BOOTSTRAP |
| database = "prm_" + 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) |
| try: |
| with psycopg.connect(uri, autocommit=True) as c: |
| c.execute(BOOTSTRAP) |
| c.execute(Path("supabase/migrations/20261106130000_math_prompter_texts.sql").read_text()) |
| c.execute("insert into auth.users values(%s)", (OWNER,)) |
| c.execute("set role service_role") |
| c.execute("insert into math_prompter_texts(user_id,title,body) values(%s,'a','b')", (OWNER,)) |
| with pytest.raises(psycopg.errors.CheckViolation): |
| c.execute("insert into math_prompter_texts(user_id,title,body) values(%s,'','b')", (OWNER,)) |
| c.execute("reset role") |
| c.execute("set role authenticated") |
| with pytest.raises(psycopg.errors.InsufficientPrivilege): |
| c.execute("select * from math_prompter_texts") |
| c.execute("reset role") |
| c.execute("delete from auth.users") |
| assert c.execute("select count(*) from math_prompter_texts").fetchone()[0] == 0 |
| finally: |
| with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c: |
| c.execute("drop database " + database + " with (force)") |
|
|
|
|
| SECTIONS = [{"part": "Part 1 โ Why", "heading": "Slide 1 โ From Data to Decisions", "time": "Opening ยท 35 seconds", |
| "reminders": ["Good morning", "BI + AI in daily work"]}, |
| {"part": "", "heading": "Slide 2 โ Thank You", "time": "", "reminders": [" Use the data we have ", ""]}, |
| {"part": "", "heading": "", "time": "", "reminders": ["dropped: no heading"]}] |
|
|
|
|
| def test_outline_text_is_a_slide_script(): |
| from app.prompter import outline_text |
| assert outline_text(SECTIONS) == ("Part 1 โ Why\nSlide 1 โ From Data to Decisions\nOpening ยท 35 seconds\n" |
| "โข Good morning\nโข BI + AI in daily work\n\nSlide 2 โ Thank You\nโข Use the data we have") |
|
|
|
|
| def test_claude_writes_the_outline_and_it_is_kept(app, monkeypatch): |
| from app import ai_usage |
| from app.prompter import outline_text |
| seen, recorded = [], [] |
| app.state.prompter_outliner = lambda body: (seen.append(body) or outline_text(SECTIONS), {"model": "m"}) |
| monkeypatch.setattr(ai_usage, "record", lambda auth, user, feature, usage: recorded.append((user, feature))) |
| text = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 't', 'body': 'Slide 1 โ a\nhello'}).json()['text'] |
| r = call(app, OWNER, 'POST', f"/api/prompter/texts/{text['id']}/outline") |
| assert r.status_code == 200 and r.json()['text']['outline'].startswith("Part 1 โ Why\nSlide 1") |
| assert seen == ['Slide 1 โ a\nhello'] and recorded == [(OWNER, "prompter")] |
| assert call(app, DEV, 'POST', f"/api/prompter/texts/{text['id']}/outline").status_code == 404 |
| assert call(app, OTHER, 'POST', f"/api/prompter/texts/{text['id']}/outline").status_code == 403 |
| url = f"/api/prompter/texts/{text['id']}" |
| |
| assert call(app, OWNER, 'PUT', url, json={'title': 'new', 'body': 'Slide 1 โ a\nhello'}).json()['text']['outline'] |
| assert call(app, OWNER, 'PUT', url, json={'title': 'new', 'body': 'Slide 1 โ a\nhello', |
| 'outline': 'Slide 1 โ a\nโข mine'}).json()['text']['outline'] == 'Slide 1 โ a\nโข mine' |
| assert call(app, OWNER, 'PUT', url, json={'title': 'new', 'body': 'Slide 1 โ b\nbye'}).json()['text']['outline'] is None |
| assert call(app, OWNER, 'PUT', url, json={'body': 'x', 'outline': 'y' * 20_001}).status_code == 413 |
| assert call(app, OWNER, 'PUT', url, json={'body': 'x', 'outline': ' '}).json()['text']['outline'] is None |
|
|
|
|
| def test_outline_without_claude_is_503(app, monkeypatch): |
| monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) |
| monkeypatch.delenv("CLAUDE_API_KEY", raising=False) |
| text = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 't', 'body': 'x'}).json()['text'] |
| assert call(app, OWNER, 'POST', f"/api/prompter/texts/{text['id']}/outline").status_code == 503 |
|
|
|
|
| def test_outline_migration_on_real_postgres(postgres_server): |
| psycopg = pytest.importorskip("psycopg") |
| from pathlib import Path |
| from scan_support import BOOTSTRAP |
| database = "pro_" + uuid.uuid4().hex |
| with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c: |
| c.execute("create database " + database) |
| try: |
| with psycopg.connect(postgres_server.get_uri(database=database), autocommit=True) as c: |
| c.execute(BOOTSTRAP) |
| for name in ("20261103220000_math_ai_usage", "20261106130000_math_prompter_texts", |
| "20261107130000_math_prompter_outline"): |
| c.execute(Path(f"supabase/migrations/{name}.sql").read_text()) |
| c.execute("insert into auth.users values(%s)", (OWNER,)) |
| c.execute("set role service_role") |
| c.execute("insert into math_prompter_texts(user_id,title,body) values(%s,'a','b')", (OWNER,)) |
| c.execute("update math_prompter_texts set outline='โข x'") |
| with pytest.raises(psycopg.errors.CheckViolation): |
| c.execute("update math_prompter_texts set outline=''") |
| c.execute("insert into math_ai_usage(feature,model) values('prompter','m')") |
| with pytest.raises(psycopg.errors.CheckViolation): |
| c.execute("insert into math_ai_usage(feature,model) values('other','m')") |
| finally: |
| with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c: |
| c.execute("drop database " + database + " with (force)") |
|
|
|
|
| def test_before_the_outline_migration_texts_still_work(app): |
| app.state.auth.no_outline = True |
| app.state.prompter_outliner = lambda body: ("Slide 1 โ a\nโข x", None) |
| text = call(app, OWNER, 'POST', '/api/prompter/texts', json={'title': 't', 'body': 'x'}).json()['text'] |
| assert 'outline' not in text |
| assert [t['id'] for t in call(app, OWNER, 'GET', '/api/prompter/texts').json()['texts']] == [text['id']] |
| assert call(app, OWNER, 'PUT', f"/api/prompter/texts/{text['id']}", json={'body': 'y'}).json()['text']['body'] == 'y' |
| assert call(app, OWNER, 'POST', f"/api/prompter/texts/{text['id']}/outline").status_code == 503 |
|
|
