mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
1dbde0845e
* test: move guardrails, image, audio, batches, search legacy tests to unit * test: restore the custom guardrail and remaining chunk 1 unit rows * test: move the ocr matrix to e2e * test: drop comments from moved tests and fix vertex ocr credentials * test: move litellm utils and unified google legacy tests to their tiers * test: move logging callback and pass-through unit legacy tests to their tiers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: move store_model_in_db legacy tests and start the admin ui migration Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test: finish the proxy job rows and drop their CircleCI jobs Moves the remaining multi-instance, openai endpoint, otel, pass-through and top-level proxy tests into tests/integration with scripted upstreams, rewrites the weak replacements to assert the outbound wire and exact responses, and removes the build_and_test job group this PR owns from CircleCI * test: tighten the key generation and error log assertions, rename migrated unit files to their mapped names * test: shard the pagerduty unit tests and record the rust harness file on the coverage allowlist * test: drive the SSO sign-in tests through a scripted IdP and isolate the cost center stand-in * test: type the e2e beta header, prompt cache and secret manager values * test: assert the missing model error message on the OpenAI routes * test: drop the deleted otel base class import and restore mcp redaction and exact langfuse usage checks * test: make the langfuse fixture directory a package for the layout check * test: restore collected no-match cases, real langfuse and gcs payload checks, and gate the aws secret lane * test: keep the ocr e2e pdf fixture next to the test * test: put meta below covers on the streaming tool search e2e * test: run the OCR matrix against a scripted upstream and fix two runner-only e2e failures * test: count only OCR calls in the OCR matrix recorder * test: pin OCR parse bodies and billed cost, group new e2e cases into classes * ci: restore the python image anchor dropped in the main sync --------- Co-authored-by: yuneng <yuneng@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
import argparse
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
ALLOWED_COST_CENTERS = {"CC-1001", "CC-1002"}
|
|
|
|
|
|
class CostCenterHandler(BaseHTTPRequestHandler):
|
|
def _respond(self, status: int, body: dict) -> None:
|
|
payload = json.dumps(body).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def do_GET(self):
|
|
if self.path == "/health":
|
|
self._respond(200, {"status": "healthy"})
|
|
return
|
|
self._respond(404, {"error": "not found"})
|
|
|
|
def do_POST(self):
|
|
if self.path != "/validate":
|
|
self._respond(404, {"error": "not found"})
|
|
return
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
body = json.loads(self.rfile.read(length) or b"{}")
|
|
cost_center = (body.get("metadata") or {}).get("cost_center")
|
|
if cost_center is None:
|
|
self._respond(200, {"ok": False, "reason": "cost_center missing per cost center service"})
|
|
elif cost_center not in ALLOWED_COST_CENTERS:
|
|
self._respond(200, {"ok": False, "reason": f"cost center {cost_center} rejected by cost center service"})
|
|
else:
|
|
self._respond(200, {"ok": True})
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--host", default="0.0.0.0")
|
|
parser.add_argument("--port", type=int, default=9414)
|
|
args = parser.parse_args()
|
|
print(f"cost center service listening on {args.host}:{args.port}")
|
|
ThreadingHTTPServer((args.host, args.port), CostCenterHandler).serve_forever()
|