pcb-afm-simulation-system/web/backend/test_integration.py

"""Integration test for dual-system workflow (System 2 <-> System 1).

Tests the full closed loop:
1. Create project
2. Create simulation plan
3. Upload scan results CSV (simulating System 2 execution)
4. Import results to experience library
5. Query experience library (list/get/update/delete)
6. Query analytics (trend/pareto/sensitivity/overview/stats/similar)

Uses FastAPI TestClient (no real server needed).
All source is ASCII.
"""

import io
import json
import sys
import os

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

# Sample CSV content simulating scan_results.csv from System 2
SAMPLE_CSV = """run_index,status,seconds,Airgap,RMSCurrent,MagnetThickness,tavg_nm,ripple_pct,efficiency_pct,total_losses_w,copper_loss_w,iron_loss_w
0,OK,120,0.8,20,5,2.800,3.50,86.0,50.0,30.0,15.0
1,OK,118,1.0,20,5,2.500,3.20,85.2,48.0,28.0,14.0
2,OK,121,1.2,20,5,2.300,2.90,84.8,45.0,26.0,13.0
3,OK,125,1.0,25,5,3.000,4.10,86.5,55.0,35.0,15.0
4,FAILED,0,1.5,20,5,0,0,0,0,0,0
"""

passed = 0
failed = 0


def test(name, condition, detail=""):
    global passed, failed
    if condition:
        passed += 1
        print(f"  [PASS] {name}")
    else:
        failed += 1
        print(f"  [FAIL] {name} {detail}")


print("=" * 60)
print("Dual-System Integration Test (System 2 <-> System 1)")
print("=" * 60)

# 1. Health check
print("\n[1] Health Check")
r = client.get("/api/health")
test("health endpoint returns 200", r.status_code == 200, f"status={r.status_code}")
test("health status is ok", r.json().get("status") == "ok")

# 2. Create project
print("\n[2] Create Project")
r = client.post("/api/projects", json={
    "name": "Integration Test Project",
    "topology": "SSSR",
    "description": "Created by integration test",
    "boundary_conditions": {"outer_radius_mm": 100, "speed_rpm": 3000}
})
test("create project returns 201", r.status_code == 201, f"status={r.status_code}")
project_id = r.json().get("id")
test("project has id", project_id is not None)
test("project name correct", r.json().get("name") == "Integration Test Project")

# 3. Create plan
print("\n[3] Create Simulation Plan")
r = client.post("/api/plans", json={
    "project_id": project_id,
    "name": "Test Airgap Scan",
    "plan_id": "INT-TEST-001",
    "topology": "SSSR",
    "model_path": "models/test.mot",
    "plan_data": {
        "plan_id": "INT-TEST-001",
        "model_path": "models/test.mot",
        "variables": [
            {"name": "Airgap", "display_name": "Airgap", "unit": "mm", "values": [0.8, 1.0, 1.2]},
            {"name": "RMSCurrent", "display_name": "RMS Current", "unit": "A", "values": [20, 25]}
        ]
    },
    "estimated_points": 6
})
test("create plan returns 201", r.status_code == 201, f"status={r.status_code}, body={r.text}")
plan_id = r.json().get("id")
test("plan has id", plan_id is not None)

# 4. Download plan (System 2 fetches plan)
print("\n[4] Download Plan (System 2 -> System 1)")
r = client.get(f"/api/plans/{plan_id}/download")
test("download plan returns 200", r.status_code == 200, f"status={r.status_code}")
plan_data = r.json()
test("download has plan_data", "plan_data" in plan_data)
test("plan_data has variables", "variables" in plan_data.get("plan_data", {}))

# 5. Upload results CSV (System 2 uploads results)
print("\n[5] Upload Results CSV (System 2 -> System 1)")
csv_file = ("scan_results.csv", SAMPLE_CSV.encode("utf-8"), "text/csv")
r = client.post(
    f"/api/plans/{plan_id}/upload-results",
    files={"file": csv_file}
)
test("upload results returns 201", r.status_code == 201, f"status={r.status_code}, body={r.text}")
upload_result = r.json()
test("upload reports count", upload_result.get("count", 0) > 0, f"result={upload_result}")
test("upload reports 5 results", upload_result.get("count") == 5, f"count={upload_result.get('count')}")

# 6. Query plan results
print("\n[6] Query Plan Results")
r = client.get(f"/api/plans/{plan_id}/results")
test("get results returns 200", r.status_code == 200)
results = r.json().get("items", [])
test("results has 5 items", len(results) == 5, f"count={len(results)}")
ok_results = [r for r in results if r.get("status") == "OK"]
test("4 OK results", len(ok_results) == 4, f"ok_count={len(ok_results)}")

# 7. Import results to experience library
print("\n[7] Import Results to Experience Library")
r = client.post(f"/api/experience/from-plan/{plan_id}", json={
    "tags": ["integration-test", "auto-imported"],
    "rating": 3,
    "auto_conclusion": True
})
test("import returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
import_result = r.json()
test("imported 4 cases", import_result.get("imported") == 4, f"imported={import_result.get('imported')}")

# 8. Query experience library
print("\n[8] Query Experience Library")
r = client.get("/api/experience?limit=50")
test("list experience returns 200", r.status_code == 200)
cases = r.json().get("items", [])
test("at least 4 cases", len(cases) >= 4, f"count={len(cases)}")

# Find our imported cases
test_cases = [c for c in cases if "integration-test" in c.get("tags", [])]
test("found integration-test cases", len(test_cases) >= 4, f"count={len(test_cases)}")

if test_cases:
    case_id = test_cases[0]["id"]
    # Get single case
    r = client.get(f"/api/experience/{case_id}")
    test("get single case returns 200", r.status_code == 200)
    test("case has params", bool(r.json().get("params")))
    test("case has metrics", bool(r.json().get("metrics")))
    test("case has auto-generated conclusion", bool(r.json().get("conclusion")))

    # Update case
    r = client.put(f"/api/experience/{case_id}", json={
        "conclusion": "Updated by integration test",
        "tags": ["integration-test", "updated"],
        "rating": 5
    })
    test("update case returns 200", r.status_code == 200)
    test("updated conclusion", r.json().get("conclusion") == "Updated by integration test")
    test("updated rating", r.json().get("rating") == 5)

# 9. Analytics: experience stats
print("\n[9] Analytics: Experience Stats")
r = client.get("/api/analytics/experience/stats")
test("experience stats returns 200", r.status_code == 200)
stats = r.json()
test("stats has total", "total" in stats)
test("stats has topology_distribution", "topology_distribution" in stats)
test("stats has metric_ranges", "metric_ranges" in stats)

# 10. Analytics: similar case search
print("\n[10] Analytics: Similar Case Search")
r = client.post("/api/analytics/experience/similar?top_k=3&tolerance=0.5", json={
    "params": {"Airgap": 1.0, "RMSCurrent": 20}
})
test("similar search returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
similar = r.json().get("items", [])
test("found similar cases", len(similar) >= 1, f"count={len(similar)}")
if similar:
    test("similar has similarity_score", "similarity_score" in similar[0])

# 11. Analytics: plan trend
print("\n[11] Analytics: Plan Trend")
r = client.get(f"/api/analytics/plans/{plan_id}/trend?param_key=Airgap&metric_key=tavg_nm")
test("trend returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
trend = r.json()
test("trend has points", "points" in trend and len(trend["points"]) >= 3)
test("trend points sorted by x", all(trend["points"][i][0] <= trend["points"][i+1][0] for i in range(len(trend["points"])-1)))

# 12. Analytics: Pareto frontier
print("\n[12] Analytics: Pareto Frontier")
r = client.get(f"/api/analytics/plans/{plan_id}/pareto?x_metric=total_losses_w&y_metric=efficiency_pct")
test("pareto returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
pareto = r.json()
test("pareto has all_points", "all_points" in pareto)
test("pareto has pareto_points", "pareto_points" in pareto)
test("pareto total count = 4", pareto.get("total_count") == 4, f"count={pareto.get('total_count')}")

# 13. Analytics: sensitivity
print("\n[13] Analytics: Parameter Sensitivity")
r = client.get(f"/api/analytics/plans/{plan_id}/sensitivity?metric_key=tavg_nm")
test("sensitivity returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
sens = r.json()
test("sensitivity has items", "items" in sens)
if sens.get("items"):
    test("sensitivity sorted by abs_correlation desc",
         all(sens["items"][i]["abs_correlation"] >= sens["items"][i+1]["abs_correlation"]
             for i in range(len(sens["items"])-1)))

# 14. Analytics: project overview
print("\n[14] Analytics: Project Overview")
r = client.get(f"/api/analytics/projects/{project_id}/overview")
test("project overview returns 200", r.status_code == 200, f"status={r.status_code}, body={r.text}")
overview = r.json()
test("overview has total_plans", overview.get("total_plans") >= 1)
test("overview has total_results", overview.get("total_results") >= 5)
test("overview has ok_results", overview.get("ok_results") >= 4)
test("overview has best_efficiency_pct", "best_efficiency_pct" in overview)
test("overview has best_torque_nm", "best_torque_nm" in overview)

# 15. Metric definitions
print("\n[15] Metric Definitions")
r = client.get("/api/analytics/metrics")
test("metrics returns 200", r.status_code == 200)
metrics = r.json().get("metrics", [])
test("at least 10 metrics", len(metrics) >= 10, f"count={len(metrics)}")
test("metric has key/label/unit", all("key" in m and "label" in m and "unit" in m for m in metrics))

# 16. Delete test cases (cleanup)
print("\n[16] Cleanup: Delete Test Cases")
r = client.get("/api/experience?limit=50")
test_cases = [c for c in r.json().get("items", []) if "integration-test" in c.get("tags", [])]
deleted = 0
for c in test_cases:
    r = client.delete(f"/api/experience/{c['id']}")
    if r.status_code == 204:
        deleted += 1
test(f"deleted {len(test_cases)} test cases", deleted == len(test_cases), f"deleted={deleted}")

# Summary
print("\n" + "=" * 60)
print(f"TEST SUMMARY: {passed} passed, {failed} failed")
print("=" * 60)

if failed > 0:
    sys.exit(1)
else:
    print("\nAll integration tests passed!")
    sys.exit(0)