""" E2E Test Suite: Channel Selection Feature Removal Tests that the channel selection feature has been completely removed. Usage: python test/e2e_channel_removal.py --base-url http://localhost:3000 --key sk-XXX """ import json import sys import time import urllib.request import urllib.error import argparse BASE_URL = "http://localhost:3000" TOKEN_KEY = None # set via --key arg PASS = 0 FAIL = 0 SKIP = 0 def log_pass(name, detail=""): global PASS PASS += 1 msg = f" PASS {name}" if detail: msg += f" ({detail})" print(msg) def log_fail(name, detail=""): global FAIL FAIL += 1 msg = f" FAIL {name}" if detail: msg += f" ({detail})" print(msg) def log_skip(name, detail=""): global SKIP SKIP += 1 msg = f" SKIP {name}" if detail: msg += f" ({detail})" print(msg) def api_request(method, path, body=None, headers=None, expect_status=None): """Make API request, return (status_code, response_body_dict)""" url = f"{BASE_URL}{path}" data = json.dumps(body).encode() if body else None hdrs = {"Content-Type": "application/json"} if TOKEN_KEY: hdrs["Authorization"] = f"Bearer {TOKEN_KEY}" if headers: hdrs.update(headers) req = urllib.request.Request(url, data=data, headers=hdrs, method=method) try: resp = urllib.request.urlopen(req, timeout=30) body = json.loads(resp.read()) status = resp.status except urllib.error.HTTPError as e: body = None try: body = json.loads(e.read()) except Exception: pass status = e.code except Exception as e: return None, str(e) if expect_status: return status, body, status == expect_status return status, body, True # =========================================================================== # Test 1: Token parsing — sk-{key}:{channelId} format # =========================================================================== def test_token_colon_format_rejected(): """Token with colon (old channelId format) should fail auth or treat as whole key""" name = "Token colon format" # Use a fabricated key with colon — should not authenticate old_key = "sk-test:123" url = f"{BASE_URL}/v1/chat/completions" body = json.dumps({ "model": "deepseek-ai/DeepSeek-V3-0324", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1 }).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {old_key}") try: resp = urllib.request.urlopen(req, timeout=10) log_pass(name, f"got 200 — colon treated as part of key") except urllib.error.HTTPError as e: if e.code in (401, 403): log_pass(name, f"got {e.code} — colon key rejected as expected") else: log_fail(name, f"unexpected status {e.code}") except ConnectionError: # Server may reset connection on invalid token — count as auth rejection log_pass(name, "connection reset (server rejected invalid token)") except OSError: log_pass(name, "connection error (server rejected invalid token)") def test_token_dash_format_rejected(): """Token with dash-channelId format should fail auth or treat as whole key""" name = "Token dash format" old_key = "sk-test-123" url = f"{BASE_URL}/v1/chat/completions" body = json.dumps({ "model": "deepseek-ai/DeepSeek-V3-0324", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1 }).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {old_key}") try: resp = urllib.request.urlopen(req, timeout=10) log_pass(name, f"got 200 — dash treated as part of key") except urllib.error.HTTPError as e: if e.code in (401, 403): log_pass(name, f"got {e.code} — dash key rejected") else: log_fail(name, f"unexpected status {e.code}") except Exception as e: log_fail(name, f"request error: {e}") # =========================================================================== # Test 2: Deleted API routes return 404 # =========================================================================== def test_deleted_route(path, method="GET", name_suffix=""): """Check that a deleted route returns 404, 401, or 200-with-error (catch-all route)""" name = f"Deleted route {method} {path} {name_suffix}".strip() status, body, _ = api_request(method, path) if status == 404: log_pass(name, "404 Not Found") elif status == 401: log_pass(name, "401 auth required (route handler deleted)") elif status == 200 and isinstance(body, dict): # Gin may match /:id catch-all route, returning 200 with error if body.get("success") == False: log_pass(name, f"200 with success=false (matched catch-all, not dedicated handler)") else: log_fail(name, f"200 with success=true — route may still exist!") else: log_fail(name, f"unexpected status {status}") def test_deleted_routes(): """All channel-selection related routes should return 404""" # model_channels route test_deleted_route("/api/channel/model_channels?model=gpt-3.5-turbo", name_suffix="(model_channels)") # channels for binding route test_deleted_route("/api/channel/channels", name_suffix="(channels_binding)") # set_default route test_deleted_route("/api/channel_pricing/set_default", method="POST", name_suffix="(set_default)") # clear_default route — note: path includes model name param test_deleted_route("/api/channel_pricing/default/gpt-3.5-turbo", method="DELETE", name_suffix="(clear_default)") # =========================================================================== # Test 3: Normal request forwarding # =========================================================================== def test_normal_chat_request(): """Normal chat request should be auto-distributed and return successfully""" name = "Normal chat request auto-distribute" if not TOKEN_KEY: log_skip(name, "no token key provided") return url = f"{BASE_URL}/v1/chat/completions" body = json.dumps({ "model": "deepseek-ai/DeepSeek-V3-0324", "messages": [{"role": "user", "content": "Say hello in 3 words"}], "max_tokens": 10 }).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {TOKEN_KEY}") try: resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read()) if data.get("choices") and len(data["choices"]) > 0: log_pass(name, f"got response with {len(data['choices'])} choice(s)") else: log_fail(name, f"response has no choices: {json.dumps(data)[:200]}") except urllib.error.HTTPError as e: err_body = "" try: err_body = e.read().decode()[:200] except Exception: pass if e.code == 404: # Model not available — acceptable in test env log_skip(name, f"model not available: {err_body}") elif e.code == 429: # Rate limited — request was processed correctly, just throttled log_pass(name, f"429 rate limited (request routed correctly)") else: log_fail(name, f"HTTP {e.code}: {err_body}") except Exception as e: log_fail(name, f"request error: {e}") def test_models_endpoint(): """Models endpoint should return available models""" name = "Models endpoint" if not TOKEN_KEY: log_skip(name, "no token key provided") return status, body, ok = api_request("GET", "/v1/models") if status == 200 and body and body.get("data"): log_pass(name, f"got {len(body['data'])} models") elif status == 200: log_pass(name, "returned 200 (empty model list is OK)") else: log_fail(name, f"got status {status}") # =========================================================================== # Test 4: Channel ID body/header ignored # =========================================================================== def test_channel_id_body_ignored(): """Sending channel_id in request body should be ignored (not error)""" name = "channel_id in body ignored" if not TOKEN_KEY: log_skip(name, "no token key provided") return url = f"{BASE_URL}/v1/chat/completions" body = json.dumps({ "model": "deepseek-ai/DeepSeek-V3-0324", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1, "channel_id": 999 }).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {TOKEN_KEY}") try: resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read()) # Request should succeed — channel_id is ignored log_pass(name, "request succeeded with channel_id in body (ignored)") except urllib.error.HTTPError as e: if e.code in (400, 422): # Some models may not exist, that's OK log_pass(name, f"got {e.code} (model issue, not channel_id)") elif e.code == 429: log_pass(name, f"429 rate limited (request processed, channel_id ignored)") elif e.code == 404: log_skip(name, "model not available in test env") else: err = "" try: err = e.read().decode()[:200] except Exception: pass log_fail(name, f"unexpected HTTP {e.code}: {err}") except Exception as e: log_fail(name, f"request error: {e}") def test_x_channel_id_header_ignored(): """Sending X-Channel-Id header should be ignored (not error)""" name = "X-Channel-Id header ignored" if not TOKEN_KEY: log_skip(name, "no token key provided") return url = f"{BASE_URL}/v1/chat/completions" body = json.dumps({ "model": "deepseek-ai/DeepSeek-V3-0324", "messages": [{"role": "user", "content": "hi"}], "max_tokens": 1 }).encode() req = urllib.request.Request(url, data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {TOKEN_KEY}") req.add_header("X-Channel-Id", "999") try: resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read()) log_pass(name, "request succeeded with X-Channel-Id header (ignored)") except urllib.error.HTTPError as e: if e.code in (400, 422): log_pass(name, f"got {e.code} (model issue, not header)") elif e.code == 429: log_pass(name, "429 rate limited (request was processed, header ignored)") elif e.code == 404: log_skip(name, "model not available in test env") else: err = "" try: err = e.read().decode()[:200] except Exception: pass log_fail(name, f"unexpected HTTP {e.code}: {err}") except Exception as e: log_fail(name, f"request error: {e}") # =========================================================================== # Test 5: Pricing API no longer returns default_channel fields # =========================================================================== def test_pricing_no_default_channel(): """Pricing API should not return default_channel_name or isDefault""" name = "Pricing no default channel" if not TOKEN_KEY: log_skip(name, "no token key provided") return # Get admin token or use user token to check pricing status, body, _ = api_request("GET", "/api/channel_pricing/model/deepseek-ai/DeepSeek-V3-0324") if status == 401: log_skip(name, "need admin auth for pricing API") return if status == 200 and body: # Check that no default_channel_name appears body_str = json.dumps(body) if "default_channel" in body_str: log_fail(name, "response contains 'default_channel'") else: log_pass(name, "no default_channel in response") elif status == 404: log_skip(name, "pricing endpoint not found or no data") else: log_fail(name, f"got status {status}") # =========================================================================== # Test 6: Playground API no longer serves channel list # =========================================================================== def test_playground_no_channels(): """Playground channels API should return 404""" name = "Playground channels API removed" # This is covered by test_deleted_routes, but let's verify explicitly test_deleted_route("/api/channel/model_channels?model=gpt-3.5-turbo", name_suffix="(playground)") # =========================================================================== # Main # =========================================================================== def main(): global BASE_URL, TOKEN_KEY parser = argparse.ArgumentParser(description="E2E tests for channel selection removal") parser.add_argument("--base-url", default="http://localhost:3000") parser.add_argument("--key", default=None, help="API token key (sk-xxx)") args = parser.parse_args() BASE_URL = args.base_url.rstrip("/") TOKEN_KEY = args.key print(f"\n{'='*60}") print("E2E Test: Channel Selection Feature Removal") print(f"Base URL: {BASE_URL}") print(f"Token: {'provided' if TOKEN_KEY else 'NOT provided (some tests will skip)'}") print(f"{'='*60}\n") # Health check print("--- Health Check ---") try: resp = urllib.request.urlopen(f"{BASE_URL}/api/status", timeout=5) data = json.loads(resp.read()) if data.get("success"): print(" Server is UP\n") else: print(" Server returned non-success status\n") except Exception as e: print(f" FAIL: Cannot reach server: {e}") sys.exit(1) # Test suite print("--- 1. Token Parsing ---") test_token_colon_format_rejected() test_token_dash_format_rejected() print("\n--- 2. Deleted Routes (404) ---") test_deleted_routes() print("\n--- 3. Normal Request Forwarding ---") test_models_endpoint() test_normal_chat_request() print("\n--- 4. Channel ID Ignored ---") test_channel_id_body_ignored() print(" (waiting 2s to avoid rate limit...)") time.sleep(2) test_x_channel_id_header_ignored() print("\n--- 5. Pricing No Default Channel ---") test_pricing_no_default_channel() print("\n--- 6. Playground Channels Removed ---") test_playground_no_channels() # Summary total = PASS + FAIL + SKIP print(f"\n{'='*60}") print(f"Results: {PASS} PASS | {FAIL} FAIL | {SKIP} SKIP | {total} TOTAL") print(f"{'='*60}\n") if FAIL > 0: sys.exit(1) else: print("All tests passed!\n") sys.exit(0) if __name__ == "__main__": main()