Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 8,263 Bytes
a685524 | 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 | """Multi-construct runs: one job scores N constructs on a single corpus pass
and reports how they interrelate (per-text score correlations).
Contract under test:
* summary gains constructs[] + correlations{}; single-construct summaries
keep the original flat shape (including when sent as a one-element list),
* export columns are prefixed per construct ({slug}_sim_item_N,
{slug}_ccr_score) with no ambiguous bare ccr_score,
* metadata carries every construct snapshot under schema version 1.1,
* the reproduction script reproduces ALL constructs and stays valid Python.
"""
import ast
import io
import time
import pytest
from fastapi.testclient import TestClient
from app.main import app
CSV = (
"id,text\n"
"1,I am deeply satisfied with my life and grateful every day.\n"
"2,The bus was late again this morning.\n"
"3,My life is close to my ideal in most ways.\n"
"4,We fixed the printer in the lab office.\n"
"5,Caring for others and protecting the vulnerable matters to me.\n"
)
def wait_for_job(client, job_id, timeout=10.0):
deadline = time.time() + timeout
while time.time() < deadline:
job = client.get(f"/api/jobs/{job_id}").json()
if job["status"] in ("completed", "failed"):
return job
time.sleep(0.05)
raise TimeoutError(f"Job {job_id} did not finish within {timeout}s")
@pytest.fixture(scope="module")
def client():
with TestClient(app) as c: # lifespan seeds the construct library
yield c
@pytest.fixture(scope="module")
def flow(client):
project = client.post(
"/api/projects", json={"name": "Multi", "description": "multi-construct e2e"}
).json()
corpus = client.post(
f"/api/projects/{project['id']}/corpora",
files={"file": ("corpus.csv", io.BytesIO(CSV.encode()), "application/octet-stream")},
).json()
constructs = client.get("/api/constructs").json()
swls = next(c for c in constructs if c["name"] == "Satisfaction with Life")
care = next(c for c in constructs if c["name"] == "Moral Foundations - Care")
job = client.post(
"/api/jobs",
json={
"project_id": project["id"],
"corpus_id": corpus["id"],
"construct_ids": [swls["id"], care["id"]],
"text_column": "text",
"model_name": "fake-deterministic",
},
).json()
job = wait_for_job(client, job["id"])
return {"project": project, "corpus": corpus, "constructs": [swls, care], "job": job}
def test_multi_job_completes_and_lists_constructs(flow):
job = flow["job"]
assert job["status"] == "completed", job["error"]
assert job["construct_ids"] == [c["id"] for c in flow["constructs"]]
assert job["construct_names"] == ["Satisfaction with Life", "Moral Foundations - Care"]
assert job["construct_name"] == "Satisfaction with Life" # legacy field = first
def test_multi_summary_has_per_construct_stats_and_correlations(client, flow):
body = client.get(f"/api/jobs/{flow['job']['id']}/results").json()
summary = body["summary"]
assert summary["n_docs"] == 5
assert "score_mean" not in summary # per-construct now, not flat
blocks = summary["constructs"]
assert [b["construct_name"] for b in blocks] == [
"Satisfaction with Life", "Moral Foundations - Care",
]
for b in blocks:
assert {"score_mean", "score_sd", "histogram", "item_means",
"top_docs", "bottom_docs", "column_prefix"} <= set(b)
assert len(b["item_means"]) == b["n_items"]
corr = summary["correlations"]
assert corr["method"] == "pearson"
assert corr["n_texts"] == 5
assert corr["constructs"] == [b["construct_name"] for b in blocks]
m = corr["matrix"]
assert len(m) == 2 and len(m[0]) == 2
assert m[0][0] == 1.0 and m[1][1] == 1.0
assert m[0][1] == m[1][0]
assert m[0][1] is None or -1.0 <= m[0][1] <= 1.0
def test_multi_export_columns_are_prefixed(client, flow):
body = client.get(f"/api/jobs/{flow['job']['id']}/results").json()
prefixes = [b["column_prefix"] for b in body["summary"]["constructs"]]
assert len(set(prefixes)) == 2
header = client.get(f"/api/jobs/{flow['job']['id']}/export").text.splitlines()[0].split(",")
assert header[:2] == ["id", "text"]
assert "ccr_score" not in header # would be ambiguous with 2 constructs
for prefix, construct in zip(prefixes, flow["constructs"]):
assert f"{prefix}_ccr_score" in header
n_sim = sum(1 for h in header if h.startswith(f"{prefix}_sim_item_"))
assert n_sim == len(construct["items"])
def test_multi_metadata_versions_and_snapshots(client, flow):
metadata = client.get(f"/api/jobs/{flow['job']['id']}/results").json()["metadata"]
assert metadata["output_schema_version"] == "1.1"
assert metadata["construct"] == "Satisfaction with Life + Moral Foundations - Care"
assert "items_sha256_16" not in metadata # ambiguous at top level; lives per construct
blocks = metadata["constructs"]
assert len(blocks) == 2
for b in blocks:
assert b["items_sha256_16"]
assert b["snapshot"]["item_hash"] and b["snapshot"]["items"]
assert f"{b['column_prefix']}_ccr_score" in metadata["output_schema"]
assert metadata["correlations"]["matrix"][0][0] == 1.0
assert metadata["n_items"] == sum(b["n_items"] for b in blocks)
def test_multi_script_reproduces_all_constructs(client, flow):
job_id = flow["job"]["id"]
script = client.get(f"/api/jobs/{job_id}/script").text
ast.parse(script) # must be valid offline Python
assert "CONSTRUCTS" in script
prefixes = [
b["column_prefix"]
for b in client.get(f"/api/jobs/{job_id}/results").json()["summary"]["constructs"]
]
for prefix in prefixes:
assert prefix in script
# embeds the corpus once, not once per construct
assert script.count("doc_emb = model.encode") == 1
def test_single_construct_via_list_keeps_flat_shape(client, flow):
"""The UI now always sends construct_ids; a one-element list must behave
exactly like the original single-construct API (shape and schema 1.0)."""
# Fresh upload: the flow corpus's file was deleted after its run
# (anonymous retention), so it can't be re-run.
corpus = client.post(
f"/api/projects/{flow['project']['id']}/corpora",
files={"file": ("again.csv", io.BytesIO(CSV.encode()), "application/octet-stream")},
).json()
resp = client.post(
"/api/jobs",
json={
"project_id": flow["project"]["id"],
"corpus_id": corpus["id"],
"construct_ids": [flow["constructs"][0]["id"]],
"text_column": "text",
"model_name": "fake-deterministic",
},
)
assert resp.status_code == 201, resp.text
job = wait_for_job(client, resp.json()["id"])
assert job["status"] == "completed", job["error"]
body = client.get(f"/api/jobs/{job['id']}/results").json()
assert "constructs" not in body["summary"]
assert "score_mean" in body["summary"]
assert body["metadata"]["output_schema_version"] == "1.0"
header = client.get(f"/api/jobs/{job['id']}/export").text.splitlines()[0].split(",")
assert "ccr_score" in header and "sim_item_1" in header
def test_multi_validation_errors(client, flow):
base = {
"project_id": flow["project"]["id"],
"corpus_id": flow["corpus"]["id"],
"text_column": "text",
"model_name": "fake-deterministic",
}
swls_id = flow["constructs"][0]["id"]
resp = client.post("/api/jobs", json={**base, "construct_ids": []})
assert resp.status_code == 400 # no construct at all
resp = client.post("/api/jobs", json={**base, "construct_ids": [swls_id, swls_id]})
assert resp.status_code == 400
assert "Duplicate" in resp.json()["detail"]
all_ids = [c["id"] for c in client.get("/api/constructs").json()][:11]
resp = client.post("/api/jobs", json={**base, "construct_ids": all_ids})
assert resp.status_code == 400
assert "At most 10" in resp.json()["detail"]
resp = client.post("/api/jobs", json={**base, "construct_ids": [swls_id, "nonexistent"]})
assert resp.status_code == 404
|