File size: 9,181 Bytes
afd13cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0ba5932
 
 
afd13cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0ba5932
afd13cb
 
 
 
 
 
 
 
 
 
8fb90d0
afd13cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0ba5932
afd13cb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0ba5932
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""In the browser: a student's quiz answers wait in the teacher's AI queue; the teacher sends them; the grade shows."""
import socket
import threading
import time

import pytest
import uvicorn

from app.answer_grading import validate_assessment
from app.main import create_app
from scan_support import OWNER, OTHER, answer_store, postgres_server
from test_ai_queue import ClassAuth, MemoryQueue
from test_answer_grading import assessment
from test_unit_quiz import photograph

playwright = pytest.importorskip("playwright.sync_api")


@pytest.fixture(autouse=True)
def grading_on(monkeypatch):
    monkeypatch.setenv("MATHLY_OPEN_ENDED_LOCKED", "0")
    monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key")


@pytest.fixture
def queue_server(answer_store):
    app = create_app(auth=ClassAuth()); app.state.answers = answer_store
    app.state.ai_queue = MemoryQueue()
    calls = []

    class Grader:
        def grade(self, image, snapshot):
            if "fail" in calls:
                from app.math_photo import PhotoUnavailable
                raise PhotoUnavailable("x", "api_key_invalid", "AuthenticationError 401: invalid x-api-key")
            calls.append(snapshot["name"])
            return validate_assessment(assessment(score=100))
    app.state.answer_grader = Grader()
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        server = uvicorn.Server(uvicorn.Config(app, log_level="error"))
        thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
        thread.start()
        try:
            while not server.started:
                time.sleep(.01)
            yield f"http://127.0.0.1:{sock.getsockname()[1]}", calls
        finally:
            server.should_exit = True; thread.join(timeout=5)


def page_of(browser, url, owner, errors):
    from test_answer_browser import session
    page = session(browser, owner=owner).new_page()
    page.on("pageerror", lambda e: errors.append(str(e)))
    page.goto(url)
    page.locator('.mx-nav [data-view="home"]').wait_for()
    tab = page.locator('.mx-nav [data-view="path"]')   # a student's trail; a class teacher has only home
    (tab if tab.count() else page.locator('.mx-nav [data-view="home"]')).click()
    return page


def test_the_teacher_sends_a_students_quiz_answer_to_the_ai(queue_server, tmp_path):
    url, calls = queue_server
    with playwright.sync_playwright() as p:
        browser = p.chromium.launch()
        errors = []
        try:
            student = page_of(browser, url, OWNER, errors)
            playwright.expect(student.locator('[data-act="ai-queue"]')).to_have_count(0)   # a student has no queue
            student.locator('.mx-uquiz[data-uquiz="t"]').click()
            student.locator("[data-act=uq-start]").click()
            photograph(student, 0, tmp_path)
            paper = student.locator(".mx-uq-paper")
            paper.locator("[data-act=uq-submit]").click()
            paper.locator("[data-act=uq-submit]").click()   # past the warning about unanswered questions
            sheet = student.locator(".mx-uq-sheet")
            playwright.expect(sheet.locator(".mx-uq-verdict").first).to_have_text("诪诪转讬谉 诇讗讬砖讜专 讛诪讜专讛 诇讘讚讬拽讛", timeout=15000)
            assert not calls

            teacher = page_of(browser, url, OTHER, errors)
            button = teacher.locator('.mx-acts [data-act="ai-queue"]').first
            playwright.expect(button).to_contain_text("1")
            button.click()
            card = teacher.locator(".mx-aq-card.pending")
            playwright.expect(card).to_have_count(1)
            playwright.expect(card.locator(".mx-aq-img")).to_be_visible()
            card.locator("[data-aq-send]").click()
            teacher.locator('[data-aq-tab="done"]').click()
            playwright.expect(teacher.locator(".mx-aq-card.done")).to_contain_text("谞讻讜谞讛 路 100/100")
            assert len(calls) == 1

            sheet.locator('[data-act="close"]').last.click()
            student.locator('.mx-uquiz[data-uquiz="t"]').click()
            student.locator("[data-uq-result]").first.click()
            playwright.expect(student.locator(".mx-uq-verdict").first).to_have_text("谞讻讜谉 讜诪诇讗", timeout=10000)
            assert len(calls) == 1 and not errors, errors
        finally:
            browser.close()


def test_a_failed_send_has_an_error_details_button(queue_server, tmp_path):
    from app.math_photo import PhotoUnavailable
    url, calls = queue_server
    with playwright.sync_playwright() as p:
        browser = p.chromium.launch()
        errors = []
        try:
            student = page_of(browser, url, OWNER, errors)
            student.locator('.mx-uquiz[data-uquiz="t"]').click()
            student.locator("[data-act=uq-start]").click()
            photograph(student, 0, tmp_path)
            paper = student.locator(".mx-uq-paper")
            paper.locator("[data-act=uq-submit]").click()
            paper.locator("[data-act=uq-submit]").click()
            playwright.expect(student.locator(".mx-uq-verdict").first).to_have_text("诪诪转讬谉 诇讗讬砖讜专 讛诪讜专讛 诇讘讚讬拽讛", timeout=15000)
            calls.append("fail")   # the grader below fails while this is in the list

            teacher = page_of(browser, url, OTHER, errors)
            teacher.locator('.mx-acts [data-act="ai-queue"]').first.click()
            teacher.locator(".mx-aq-card.pending [data-aq-send]").click()
            card = teacher.locator(".mx-aq-card.failed")
            playwright.expect(card.locator(".mx-aq-error")).to_have_text("讛讘讚讬拽讛 谞讻砖诇讛. 讗驻砖专 诇谞住讜转 砖讜讘.")
            details = card.locator(".mx-error-details")
            playwright.expect(details.locator("code")).to_be_hidden()
            details.locator("summary").click()
            playwright.expect(details).to_contain_text("诪驻转讞 讛-API 诇讗 转拽讬谉")
            playwright.expect(details.locator("code")).to_have_text("AuthenticationError 401: invalid x-api-key")
            assert not errors, errors
        finally:
            browser.close()


def test_the_teacher_sends_a_whole_practice_quiz_in_one_batch_from_the_class_budget(answer_store, tmp_path):
    from test_ai_batches import BudgetAuth, MemoryBudget
    auth = BudgetAuth()
    app = create_app(auth=auth)
    app.state.answers, app.state.ai_queue, app.state.class_budget = answer_store, MemoryQueue(), MemoryBudget()
    calls = []

    class Grader:
        def grade(self, image, snapshot):
            calls.append(snapshot["name"])
            return validate_assessment(assessment(score=100))
    app.state.answer_grader = Grader()
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        server = uvicorn.Server(uvicorn.Config(app, log_level="error"))
        thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
        thread.start()
        while not server.started:
            time.sleep(.01)
        url = f"http://127.0.0.1:{sock.getsockname()[1]}"
        try:
            with playwright.sync_playwright() as p:
                browser = p.chromium.launch()
                errors = []
                try:
                    student = page_of(browser, url, OWNER, errors)
                    student.locator('.mx-uquiz[data-uquiz="t"]').click()
                    student.locator("[data-act=uq-start]").click()
                    photograph(student, 0, tmp_path)
                    photograph(student, 1, tmp_path)
                    paper = student.locator(".mx-uq-paper")
                    paper.locator("[data-act=uq-submit]").click()
                    paper.locator("[data-act=uq-submit]").click()
                    playwright.expect(student.locator(".mx-uq-verdict").first).to_have_text("诪诪转讬谉 诇讗讬砖讜专 讛诪讜专讛 诇讘讚讬拽讛", timeout=15000)

                    teacher = page_of(browser, url, OTHER, errors)
                    teacher.on("dialog", lambda d: d.accept())
                    teacher.locator('.mx-acts [data-act="ai-queue"]').first.click()
                    panel = teacher.locator(".mx-aq-budget")
                    playwright.expect(panel).to_contain_text("谞讜爪诇讜 0 诪转讜讱 150 讛讬讜诐")
                    playwright.expect(panel.locator(".mx-aq-scopes li")).to_contain_text("讘讞谞讬 讞讬讝讜拽: 讟讻谞讬拽讛 讗诇讙讘专讬转 1: 2 诪诪转讬谞讜转")
                    panel.locator("[data-batch]").click()
                    playwright.expect(panel.locator("[role=status]").last).to_contain_text("谞砖诇讞讜 2 讘讚讬拽讜转")
                    teacher.locator('[data-aq-tab="done"]').click()
                    playwright.expect(teacher.locator(".mx-aq-card.done")).to_have_count(2, timeout=10000)
                    assert len(calls) == 2 and app.state.class_budget.used[1] == 2
                    assert app.state.teacher_ai.left({"email": OTHER + "@test.invalid", "teacher": True}) == 3   # the teacher's own 3 untouched
                    assert not errors, errors
                finally:
                    browser.close()
        finally:
            server.should_exit = True
            thread.join(timeout=5)