(
*,
name: str,
url: str,
method: str = "GET",
headers: Optional[Dict[str, str]] = None,
json_body: Optional[Dict[str, Any]] = None,
timeout_seconds: int = 12,
expected_codes: Optional[List[int]] = None,
proxy_url: Optional[str] = None,
allow_direct_fallback: bool = True,
critical: bool = False,
)
| 139 | |
| 140 | |
| 141 | def _probe_endpoint( |
| 142 | *, |
| 143 | name: str, |
| 144 | url: str, |
| 145 | method: str = "GET", |
| 146 | headers: Optional[Dict[str, str]] = None, |
| 147 | json_body: Optional[Dict[str, Any]] = None, |
| 148 | timeout_seconds: int = 12, |
| 149 | expected_codes: Optional[List[int]] = None, |
| 150 | proxy_url: Optional[str] = None, |
| 151 | allow_direct_fallback: bool = True, |
| 152 | critical: bool = False, |
| 153 | ) -> Dict[str, Any]: |
| 154 | expected = set(expected_codes or [200]) |
| 155 | endpoint_result: Dict[str, Any] = { |
| 156 | "name": name, |
| 157 | "url": url, |
| 158 | "method": method.upper(), |
| 159 | "critical": bool(critical), |
| 160 | "expected_codes": sorted(expected), |
| 161 | "ok": False, |
| 162 | "via": None, |
| 163 | "http_status": None, |
| 164 | "error": "", |
| 165 | "proxy": {"used": bool(proxy_url), "status": None, "error": ""}, |
| 166 | "direct": {"status": None, "error": ""}, |
| 167 | } |
| 168 | |
| 169 | def _request_once(use_proxy: bool) -> Dict[str, Any]: |
| 170 | session = _build_http_session(proxy_url if use_proxy else None) |
| 171 | started = time.perf_counter() |
| 172 | req_kwargs: Dict[str, Any] = { |
| 173 | "url": url, |
| 174 | "headers": headers or {}, |
| 175 | "timeout": timeout_seconds, |
| 176 | } |
| 177 | if json_body is not None: |
| 178 | req_kwargs["json"] = json_body |
| 179 | if method.upper() == "POST": |
| 180 | resp = session.post(**req_kwargs) |
| 181 | else: |
| 182 | resp = session.get(**req_kwargs) |
| 183 | cost = int((time.perf_counter() - started) * 1000) |
| 184 | return {"status": int(resp.status_code), "elapsed_ms": cost} |
| 185 | |
| 186 | if proxy_url: |
| 187 | try: |
| 188 | proxy_out = _request_once(True) |
| 189 | endpoint_result["proxy"]["status"] = proxy_out["status"] |
| 190 | endpoint_result["proxy"]["elapsed_ms"] = proxy_out["elapsed_ms"] |
| 191 | endpoint_result["via"] = "proxy" |
| 192 | endpoint_result["http_status"] = proxy_out["status"] |
| 193 | endpoint_result["ok"] = proxy_out["status"] in expected |
| 194 | if endpoint_result["ok"]: |
| 195 | return endpoint_result |
| 196 | if not allow_direct_fallback: |
| 197 | endpoint_result["error"] = f"unexpected_status:{proxy_out['status']}" |
| 198 | return endpoint_result |
no test coverage detected