Spaces:
Running
Running
File size: 2,710 Bytes
2978bba | 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 |
import requests
import json
import time
import os
BASE_URL = "http://127.0.0.1:5001"
API_KEY = "dummy_key_if_needed" # Assuming developer mode might bypass or we need a real key.
# In dev environment usually require_api_key might be relaxed or we can use a known key.
# For this test, we will assume we can hit the endpoint. If auth fails, we'll see 401.
# In `src/middleware/api_auth.py`, checking logic... often header 'X-API-Key'.
def check_status():
try:
r = requests.get(f"{BASE_URL}/api/demo/retrain_status")
return r.json()
except Exception as e:
print(f"Status check failed: {e}")
return None
def test_feedback_loop():
print("1. Checking initial status...")
status = check_status()
print(f" Initial Status: {status}")
# Simulate an image path (in a real scenario, this comes from /api/detect)
# We will just use a dummy path. The backend doesn't validate file existence strictly for logging.
test_image_path = "/tmp/test_feedback_image.jpg"
# We need to send enough feedback to trigger the threshold (default 10)
print("\n2. Sending feedback batch to trigger threshold...")
for i in range(11):
payload = {
"image_path": test_image_path,
"label": "real",
"prediction": "morph"
}
# Assuming we might need headers if auth is active.
# If running locally without strict auth config, this might pass.
# If strict, we might need to skip auth or valid key.
# Let's try without key first, if 401, we abort or fix.
headers = {}
# Try adding typical dev key if configured, or just empty.
try:
r = requests.post(f"{BASE_URL}/api/feedback", json=payload, headers=headers)
if r.status_code == 401:
print(" Auth required. Skipping strictly automated test or need API Key.")
return 401
elif r.status_code == 200:
print(f" Feedback {i+1}: OK")
else:
print(f" Feedback {i+1}: Failed {r.status_code} {r.text}")
except Exception as e:
print(f" Request failed: {e}")
break
# Wait for async thread to potentially pick it up
print("\n3. Waiting for pipeline trigger...")
time.sleep(2)
status = check_status()
print(f" Post-Feedback Status: {status}")
if status['detector']['status'] in ['running', 'complete']:
print("\nSUCCESS: Pipeline triggered correctly!")
else:
print("\nWARNING: Pipeline did not trigger. Check threshold or threading.")
if __name__ == "__main__":
test_feedback_loop()
|