Spaces:
Sleeping
Sleeping
File size: 8,666 Bytes
a1b9fa3 | 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 | """tests/test_api.py — pytest suite (no GROQ_API_KEY required to run)"""
import os, math, pytest
from unittest.mock import MagicMock
os.environ.pop("GROQ_API_KEY", None) # ensure clean state
from fastapi.testclient import TestClient
@pytest.fixture(scope="session")
def client():
from app import app, DB, STAGE_MAP
stub = [
{"i":"1","p":"ACARBOSE","s":"ACA_SEQ_01","d":"2025-01-03","sc":"RM",
"it":"Unknown","st":"Finished","sw":501.0,"t":"ASSAY","b":"BATCH001",
"ar":"AR001","in":"INSTR-01","co":"COL-01","d1":25.0,"wl":"210 nm",
"ft":"Isocratic","th":25.0,"ct":35.0,"rp":106.0,"pp":102.0,
"pa":"311000.0","pr":"17.6","pas":"1.19","pp2":"3900","cn":"Acarbose","sw1":"502.41","pn":"Acarbose"},
{"i":"2","p":"METFORMIN","s":"MET_SEQ_01","d":"2025-03-15","sc":"FP",
"it":"CalibrationStandard","st":"Finished","sw":200.0,"t":"DISSOLUTION",
"b":"BATCH002","ar":"AR002","in":"INSTR-02","co":"COL-02","d1":100.0,
"wl":"254 nm","ft":"Gradient","th":37.0,"ct":40.0,"rp":200.0,"pp":198.0,
"pa":"150000.0","pr":"5.3","pas":"1.05","pp2":"5200","cn":"Metformin HCl","sw1":"100.0","pn":"Metformin"},
{"i":"3","p":"ACARBOSE","s":"ACA_SEQ_01","d":"2025-06-20","sc":"RM",
"it":"Blank","st":"Finished","sw":None,"t":"ASSAY","b":"BATCH001",
"ar":"AR001","in":"INSTR-01","co":"COL-01","d1":1.0,"wl":"210 nm",
"ft":"Isocratic","th":25.0,"ct":35.0,"rp":104.0,"pp":101.0,
"pa":None,"pr":None,"pas":None,"pp2":None,"cn":None,"sw1":None,"pn":None},
]
areas=[float(r["pa"]) for r in stub if r.get("pa")]
pv=[float(r["rp"]) for r in stub if r.get("rp")]
cv=[float(r["ct"]) for r in stub if r.get("ct") is not None]
def avg(a): return sum(a)/len(a) if a else 0.0
def std(a):
if len(a)<2: return 0.0
m=avg(a); return math.sqrt(sum((x-m)**2 for x in a)/len(a))
DB.records=stub
DB.agg={"total":3,"finished":3,
"area_mean":avg(areas),"area_std":std(areas),"area_n":len(areas),
"pres_mean":avg(pv),"pres_std":std(pv),"pres_n":len(pv),
"ct_mean":avg(cv),"ct_std":std(cv),"ct_n":len(cv),
"stage_cnt":{"Raw Material":2,"Finished Product":1},
"it_cnt":{"Unknown":1,"CalibrationStandard":1,"Blank":1},
"instr_cnt":{"INSTR-01":2,"INSTR-02":1},"wl_cnt":{"210 nm":2,"254 nm":1},
"prod_cnt":{"ACARBOSE":2,"METFORMIN":1},"ft_cnt":{"Isocratic":2,"Gradient":1},
"thermo_cnt":{"25.0":2,"37.0":1},"ct_cnt":{"35.0":2,"40.0":1},
"daily":[
{"d":"2025-01-03","n":1,"a":311000.0,"p":106.0,"c":35.0,"r":17.6,"as":1.19},
{"d":"2025-03-15","n":1,"a":150000.0,"p":200.0,"c":40.0,"r":5.3,"as":1.05},
{"d":"2025-06-20","n":1,"a":None,"p":104.0,"c":35.0,"r":None,"as":None},
],
"area_sample":areas,"sa_stats":{"Raw Material":{"mean":311000.0,"std":0.0,"n":1}},"it_stats":{}}
DB.meta={"total":3,"sequences":["ACA_SEQ_01","MET_SEQ_01"],"products":["ACARBOSE","METFORMIN"],
"instruments":["INSTR-01","INSTR-02"],"columns":["COL-01","COL-02"],
"inj_types":["Blank","CalibrationStandard","Unknown"],"stage_cats":list(STAGE_MAP.values()),
"tests":["ASSAY","DISSOLUTION"],"wavelengths":["210 nm","254 nm"],"flow_types":["Gradient","Isocratic"]}
DB.loaded=True
return TestClient(app)
class TestHealth:
def test_ok(self, client):
r=client.get("/api/health"); assert r.status_code==200
b=r.json(); assert b["status"]=="ok"; assert b["records"]==3
assert b["port"]==7860; assert "ai_enabled" in b
def test_meta(self, client):
r=client.get("/api/meta"); assert r.status_code==200
assert "ACARBOSE" in r.json()["products"]
def test_agg(self, client):
r=client.get("/api/aggregate"); assert r.status_code==200; assert r.json()["total"]==3
class TestInjections:
def test_list(self, client): assert client.get("/api/injections").json()["total"]==3
def test_page(self, client):
r=client.get("/api/injections?per_page=2"); assert len(r.json()["data"])==2
def test_product(self, client): assert client.get("/api/injections?product=ACARBOSE").json()["total"]==2
def test_inj_type(self, client): assert client.get("/api/injections?inj_type=Blank").json()["total"]==1
def test_stage(self, client): assert client.get("/api/injections?stage=Finished+Product").json()["total"]==1
def test_dates(self, client): assert client.get("/api/injections?date_from=2025-02-01").json()["total"]==2
def test_pressure(self, client): assert client.get("/api/injections?pres_min=150&pres_max=300").json()["total"]==1
def test_instrument(self, client): assert client.get("/api/injections?instrument=INSTR-02").json()["total"]==1
def test_wavelength(self, client): assert client.get("/api/injections?wavelength=254+nm").json()["total"]==1
def test_fulltext(self, client): assert client.get("/api/injections?q=BATCH002").json()["total"]==1
def test_stage_label(self, client):
for rec in client.get("/api/injections").json()["data"]: assert "stage_label" in rec
class TestSingle:
def test_found(self, client):
r=client.get("/api/injections/1"); assert r.status_code==200
assert r.json()["stage_label"]=="Raw Material"
def test_missing(self, client): assert client.get("/api/injections/99999").status_code==404
class TestTrends:
def test_count(self, client): assert client.get("/api/trends/daily?metric=n").json()["values"]==[1,1,1]
def test_area(self, client): assert client.get("/api/trends/daily?metric=a").json()["values"][0]==311000.0
def test_date_filter(self, client):
assert len(client.get("/api/trends/daily?date_from=2025-03-01").json()["labels"])==2
class TestStats:
def test_summary(self, client):
b=client.get("/api/stats/summary").json()
assert b["total"]==3; assert b["finish_rate_pct"]==100.0
assert b["area"]["p10"]<=b["area"]["median"]<=b["area"]["p90"]
class TestExport:
def test_csv(self, client):
r=client.get("/api/injections/export"); assert "text/csv" in r.headers["content-type"]
lines=r.text.strip().split("\n"); assert lines[0].startswith("Product,Batch"); assert len(lines)==4
def test_filtered(self, client):
lines=client.get("/api/injections/export?product=ACARBOSE").text.strip().split("\n"); assert len(lines)==3
class TestCompare:
def test_ok(self, client):
r=client.post("/api/compare",json={"by":"product","groups":["ACARBOSE","METFORMIN"]})
assert r.status_code==200; assert r.json()["groups"]["ACARBOSE"]["count"]==2
def test_too_few(self, client): assert client.post("/api/compare",json={"by":"product","groups":["A"]}).status_code==400
def test_too_many(self, client): assert client.post("/api/compare",json={"by":"product","groups":["A","B","C","D"]}).status_code==400
def test_bad_by(self, client): assert client.post("/api/compare",json={"by":"bad","groups":["A","B"]}).status_code==400
class TestAI:
def test_models_no_key(self, client):
# Works without any key — just lists models
r=client.get("/api/ai/models"); assert r.status_code==200
b=r.json(); assert b["provider"]=="GROQ"
assert any(m["id"]=="llama-3.3-70b-versatile" for m in b["models"])
assert b["enabled"]==False # no key in test env
assert "note" in b
def test_chat_no_key_503(self, client):
# Without key → 503 with clear message
r=client.post("/api/ai/chat",json={"messages":[{"role":"user","content":"hi"}]})
assert r.status_code==503
assert "GROQ_API_KEY" in r.json()["detail"]
assert "console.groq.com" in r.json()["detail"]
def test_chat_with_mock(self, client):
import app as app_mod
mock_resp=MagicMock()
mock_resp.choices=[MagicMock()]
mock_resp.choices[0].message.content="RSD 12% is within acceptable limits."
mock_resp.choices[0].finish_reason="stop"
mock_resp.model="llama-3.3-70b-versatile"
mock_resp.usage.prompt_tokens=50; mock_resp.usage.completion_tokens=30; mock_resp.usage.total_tokens=80
mock_client=MagicMock()
mock_client.chat.completions.create.return_value=mock_resp
original=app_mod._get_groq_client
app_mod._get_groq_client=lambda: mock_client
r=client.post("/api/ai/chat",json={"messages":[{"role":"user","content":"Is RSD ok?"}]})
app_mod._get_groq_client=original
assert r.status_code==200
assert "RSD" in r.json()["reply"]
assert r.json()["model"]=="llama-3.3-70b-versatile"
|