feat(ai-onboarding): enhance conversation flow, prompt rules & update Postman environment port to 8001

2026-09-01 15:22:40 +05:30
parent 1d794ca425
commit c23d979338
31 changed files with 4686 additions and 749 deletions
@@ -1,98 +1,92 @@
-import urllib.request
-import urllib.parse
+import os
+import sys
 import json
-import time
+import pytest
+from fastapi.testclient import TestClient

-BASE_URL = 'http://127.0.0.1:8000'
-VALID_KEY = 'skyheal_secure_api_key_2026'
+# Add project root to sys.path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))

-def http_post(url, body_dict=None, token=None, headers=None):
-    if headers is None:
-        headers = {'Content-Type': 'application/json'}
-    else:
-        headers = dict(headers)
-        if 'Content-Type' not in headers:
-            headers['Content-Type'] = 'application/json'
+from app.main import app

-    if token:
-        headers['Authorization'] = f"Bearer {token}"
+client = TestClient(app)
+VALID_KEY = "skyheal_secure_api_key_2026"

-    data_bytes = json.dumps(body_dict).encode('utf-8') if body_dict is not None else b''
-    req = urllib.request.Request(url, data=data_bytes, headers=headers, method='POST')
-    try:
-        with urllib.request.urlopen(req) as res:
-            return res.status, json.loads(res.read().decode('utf-8')), res.headers
-    except urllib.error.HTTPError as e:
-        body = e.read().decode('utf-8', errors='ignore')
-        try:
-            parsed = json.loads(body)
-        except:
-            parsed = body
-        return e.code, parsed, e.headers

-def http_get(url, token=None, headers=None):
-    if headers is None:
-        headers = {}
-    else:
-        headers = dict(headers)
-    if token:
-        headers['Authorization'] = f"Bearer {token}"
-    req = urllib.request.Request(url, headers=headers, method='GET')
-    try:
-        with urllib.request.urlopen(req) as res:
-            ct = res.headers.get('Content-Type', '')
-            body = res.read().decode('utf-8', errors='ignore')
-            if 'json' in ct:
-                return res.status, json.loads(body), res.headers
-            return res.status, body, res.headers
-    except urllib.error.HTTPError as e:
-        body = e.read().decode('utf-8', errors='ignore')
-        try:
-            parsed = json.loads(body)
-        except:
-            parsed = body
-        return e.code, parsed, e.headers
+def test_reject_unauthenticated_request():
+    """1. Reject Unauthenticated Request (401)"""
+    res = client.post("/v1/onboarding/start", json={"patient": {"id": "P1"}})
+    assert res.status_code == 401
+
+
+def test_authenticated_session_creation():
+    """2. Authenticated Session Creation (200)"""
+    res = client.post(
+        "/v1/onboarding/start",
+        json={
+            "patient": {"id": "PAT-PROD-01", "name": "Prod Patient", "age": 32, "gender": "female"},
+            "category": {"type": "general", "code": "general_medicine", "name": "General Medicine"}
+        },
+        headers={"Authorization": "Bearer test_token_PAT-PROD-01"}
+    )
+    assert res.status_code == 200
+    data = res.json()
+    assert data.get("session_id") is not None
+
+
+def test_idor_prevention_cross_session_access():
+    """3. IDOR Prevention Cross-Session Access (403)"""
+    # Create session as PAT-PROD-01
+    start_res = client.post(
+        "/v1/onboarding/start",
+        json={
+            "patient": {"id": "PAT-PROD-01", "name": "Prod Patient", "age": 32, "gender": "female"},
+            "category": {"type": "general", "code": "general_medicine", "name": "General Medicine"}
+        },
+        headers={"Authorization": "Bearer test_token_PAT-PROD-01"}
+    )
+    s_id = start_res.json().get("session_id")
+    assert s_id is not None
+
+    # Attempt access as PAT-PROD-02
+    msg_res = client.post(
+        f"/v1/onboarding/{s_id}/message",
+        json={"message": "Unauthorized message"},
+        headers={"Authorization": "Bearer test_token_PAT-PROD-02"}
+    )
+    assert msg_res.status_code == 403
+
+
+def test_security_headers_and_request_id():
+    """4. Security Headers & X-Request-ID Correlation"""
+    res = client.get("/health")
+    assert res.status_code == 200
+    assert "X-Request-ID" in res.headers
+    assert res.headers.get("X-Content-Type-Options") == "nosniff"
+    assert res.headers.get("X-Frame-Options") == "DENY"
+
+
+def test_schema_validation_negative_age_rejection():
+    """5. Schema Validation Negative Age Rejection (422)"""
+    res = client.post(
+        "/v1/onboarding/start",
+        json={
+            "patient": {"id": "PAT-NEG", "name": "Invalid Age", "age": -10, "gender": "female"},
+            "category": {"type": "general", "code": "general_medicine", "name": "General Medicine"}
+        },
+        headers={"Authorization": f"Bearer {VALID_KEY}"}
+    )
+    assert res.status_code == 422
+

 def run_automated_production_tests():
-    test_results = []
+    test_reject_unauthenticated_request()
+    test_authenticated_session_creation()
+    test_idor_prevention_cross_session_access()
+    test_security_headers_and_request_id()
+    test_schema_validation_negative_age_rejection()
+    print("ALL PRODUCTION SECURITY TESTS PASSED SUCCESSFULLY!")

-    # 1. Test Authentication Security
-    c_no_auth, _, _ = http_post(f'{BASE_URL}/v1/onboarding/start', {"patient": {"id": "P1"}})
-    test_results.append({"test": "1. Reject Unauthenticated Request (401)", "passed": c_no_auth == 401, "status": c_no_auth})

-    c_auth, r_auth, _ = http_post(f'{BASE_URL}/v1/onboarding/start', {
-        "patient": {"id": "PAT-PROD-01", "name": "Prod Patient", "age": 32, "gender": "female"},
-        "category": {"type": "general", "code": "general_medicine", "name": "General Medicine"}
-    }, token="test_token_PAT-PROD-01")
-    s_id = r_auth.get("session_id") if isinstance(r_auth, dict) else None
-    test_results.append({"test": "2. Authenticated Session Creation (200)", "passed": c_auth == 200, "status": c_auth})
-
-    # 2. Test IDOR Session Isolation
-    if s_id:
-        c_idor, r_idor, _ = http_post(f'{BASE_URL}/v1/onboarding/{s_id}/message', {"message": "Unauthorized message"}, token="test_token_PAT-PROD-02")
-        test_results.append({"test": "3. IDOR Prevention Cross-Session Access (403)", "passed": c_idor == 403, "status": c_idor})
-
-    # 3. Test Response Security Headers & X-Request-ID
-    c_hdr, _, hdrs = http_get(f'{BASE_URL}/health')
-    req_id = hdrs.get("X-Request-ID")
-    nosniff = hdrs.get("X-Content-Type-Options")
-    frame = hdrs.get("X-Frame-Options")
-    hdr_passed = req_id is not None and nosniff == "nosniff" and frame == "DENY"
-    test_results.append({"test": "4. Security Headers & X-Request-ID Correlation", "passed": hdr_passed, "status": c_hdr})
-
-    # 4. Test Age Validation Schema Bounds
-    c_age_neg, _, _ = http_post(f'{BASE_URL}/v1/onboarding/start', {
-        "patient": {"id": "PAT-NEG", "name": "Invalid Age", "age": -10, "gender": "female"},
-        "category": {"type": "general", "code": "general_medicine", "name": "General Medicine"}
-    }, token=VALID_KEY)
-    test_results.append({"test": "5. Schema Validation Negative Age Rejection (422)", "passed": c_age_neg == 422, "status": c_age_neg})
-
-    total = len(test_results)
-    passed = sum(1 for t in test_results if t['passed'])
-    print(f"AUTOMATED PRODUCTION TESTS COMPLETED: {passed}/{total} Passed.")
-
-    with open('tests_production_results.json', 'w', encoding='utf-8') as f:
-        json.dump({"total": total, "passed": passed, "results": test_results}, f, indent=2)
-
-if __name__ == '__main__':
+if __name__ == "__main__":
    run_automated_production_tests()