Spaces:
Running
Running
| 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() | |