Spaces:
Running
Running
| """Anchor-vector (bipolar) runs end to end (spec 0006). | |
| Contract under test: | |
| * an anchored job (target construct + opposite_construct_id) completes and the | |
| JobOut echoes the anchored fields, | |
| * the summary is bipolar (per-pole item loadings, target/opposite names, most | |
| target = top_docs / most opposite = bottom_docs), | |
| * export columns are target_sim_item_N / opposite_sim_item_N / target_ccr_score | |
| / opposite_ccr_score / anchor_score under schema version 1.2, and for the dot | |
| metric anchor_score == target_ccr_score - opposite_ccr_score row by row, | |
| * metadata carries the scoring block, the anchor block, and BOTH snapshots, | |
| * the reproduction script embeds both item sets and stays valid Python, | |
| * validation rejects self-opposition, anchored+multi, bad metric, missing pole. | |
| """ | |
| import ast | |
| import csv | |
| 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") | |
| def client(): | |
| with TestClient(app) as c: # lifespan seeds the construct library | |
| yield c | |
| def setup(client): | |
| project = client.post("/api/projects", json={"name": "Anchor"}).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") | |
| return {"project": project, "swls": swls, "care": care} | |
| def _fresh_corpus(client, project_id): | |
| return client.post( | |
| f"/api/projects/{project_id}/corpora", | |
| files={"file": ("c.csv", io.BytesIO(CSV.encode()), "application/octet-stream")}, | |
| ).json() | |
| def _anchored_job(client, setup, metric): | |
| corpus = _fresh_corpus(client, setup["project"]["id"]) | |
| resp = client.post( | |
| "/api/jobs", | |
| json={ | |
| "project_id": setup["project"]["id"], | |
| "corpus_id": corpus["id"], | |
| "construct_id": setup["swls"]["id"], | |
| "opposite_construct_id": setup["care"]["id"], | |
| "similarity_metric": metric, | |
| "text_column": "text", | |
| "model_name": "fake-deterministic", | |
| }, | |
| ) | |
| assert resp.status_code == 201, resp.text | |
| return wait_for_job(client, resp.json()["id"]) | |
| def cosine_job(client, setup): | |
| return _anchored_job(client, setup, "cosine") | |
| def dot_job(client, setup): | |
| return _anchored_job(client, setup, "dot") | |
| def test_anchored_job_completes_and_echoes(cosine_job, setup): | |
| assert cosine_job["status"] == "completed", cosine_job["error"] | |
| assert cosine_job["anchored"] is True | |
| assert cosine_job["opposite_construct_id"] == setup["care"]["id"] | |
| assert cosine_job["opposite_construct_name"] == "Moral Foundations - Care" | |
| assert cosine_job["similarity_metric"] == "cosine" | |
| def test_anchored_summary_is_bipolar(client, cosine_job, setup): | |
| summary = client.get(f"/api/jobs/{cosine_job['id']}/results").json()["summary"] | |
| assert summary["anchored"] is True | |
| assert summary["target_name"] == "Satisfaction with Life" | |
| assert summary["opposite_name"] == "Moral Foundations - Care" | |
| assert summary["metric"] == "cosine" | |
| assert len(summary["target_item_means"]) == len(setup["swls"]["items"]) | |
| assert len(summary["opposite_item_means"]) == len(setup["care"]["items"]) | |
| assert summary["top_docs"] and summary["bottom_docs"] | |
| assert "histogram" in summary and "score_mean" not in summary.get("constructs", {}) | |
| def test_anchored_export_columns_and_dot_selfcheck(client, dot_job, setup): | |
| text = client.get(f"/api/jobs/{dot_job['id']}/export").text | |
| header = text.splitlines()[0].split(",") | |
| assert header[:2] == ["id", "text"] | |
| assert "anchor_score" in header | |
| assert "target_ccr_score" in header and "opposite_ccr_score" in header | |
| assert sum(1 for h in header if h.startswith("target_sim_item_")) == len(setup["swls"]["items"]) | |
| assert sum(1 for h in header if h.startswith("opposite_sim_item_")) == len(setup["care"]["items"]) | |
| assert "ccr_score" not in header # no bare (ambiguous) column | |
| # dot anchor score is exactly the per-pole difference, row by row. | |
| for r in csv.DictReader(io.StringIO(text)): | |
| diff = float(r["target_ccr_score"]) - float(r["opposite_ccr_score"]) | |
| assert abs(float(r["anchor_score"]) - diff) < 1e-4 | |
| def test_anchored_metadata(client, cosine_job, setup): | |
| md = client.get(f"/api/jobs/{cosine_job['id']}/results").json()["metadata"] | |
| assert md["output_schema_version"] == "1.2" | |
| assert md["scoring"]["method"] == "anchored_vector" | |
| assert md["scoring"]["similarity"] == "cosine" | |
| assert md["anchor"]["target_construct_id"] == setup["swls"]["id"] | |
| assert md["anchor"]["opposite_construct_id"] == setup["care"]["id"] | |
| assert md["target_construct"]["snapshot"]["items"] | |
| assert md["opposite_construct"]["snapshot"]["items"] | |
| assert md["target_items_sha256_16"] and md["opposite_items_sha256_16"] | |
| assert "anchor_score" in md["output_schema"] | |
| def test_anchored_script_reproduces(client, cosine_job): | |
| script = client.get(f"/api/jobs/{cosine_job['id']}/script").text | |
| ast.parse(script) # valid offline Python | |
| assert "TARGET_ITEMS" in script and "OPPOSITE_ITEMS" in script | |
| assert "AV = target_emb.mean" in script | |
| assert script.count("doc_emb = model.encode") == 1 # corpus embedded once | |
| def test_anchor_validation_errors(client, setup): | |
| corpus = _fresh_corpus(client, setup["project"]["id"]) | |
| base = { | |
| "project_id": setup["project"]["id"], | |
| "corpus_id": corpus["id"], | |
| "text_column": "text", | |
| "model_name": "fake-deterministic", | |
| } | |
| swls, care = setup["swls"]["id"], setup["care"]["id"] | |
| r = client.post("/api/jobs", json={**base, "construct_id": swls, "opposite_construct_id": swls}) | |
| assert r.status_code == 400 and "differ" in r.json()["detail"] | |
| r = client.post( | |
| "/api/jobs", | |
| json={**base, "construct_ids": [swls, care], "opposite_construct_id": care}, | |
| ) | |
| assert r.status_code == 400 # anchored is single-target vs one opposite | |
| r = client.post( | |
| "/api/jobs", | |
| json={**base, "construct_id": swls, "opposite_construct_id": care, "similarity_metric": "euclid"}, | |
| ) | |
| assert r.status_code == 400 and "cosine" in r.json()["detail"] | |
| r = client.post( | |
| "/api/jobs", | |
| json={**base, "construct_id": swls, "opposite_construct_id": "nonexistent"}, | |
| ) | |
| assert r.status_code == 404 | |