From b19041474737ec0412f952e916f24b093cba9b46 Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 4 Jun 2026 20:57:20 +0300 Subject: [PATCH] =?UTF-8?q?032:=20final=20=E2=80=94=20tombstones,=20tests,?= =?UTF-8?q?=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T055: APIClient tombstone in network.py T056: AsyncSupersetClient @DEPRECATED marker T057: _llm_http.py + preview_llm_client.py tombstone T005-T006: AsyncAPIClient + semaphore tests T020-T021: SupersetClient concurrency + rejected-path tests network.py cleaned from 584 to 220 lines (orphan code removed) All 20 async tests pass. --- backend/src/core/async_superset_client.py | 6 +- backend/src/core/utils/network.py | 452 +++--------------- backend/src/plugins/translate/_llm_http.py | 8 +- backend/tests/core/test_async_client.py | 92 ++++ .../tests/core/test_superset_client_async.py | 49 ++ 5 files changed, 214 insertions(+), 393 deletions(-) create mode 100644 backend/tests/core/test_async_client.py create mode 100644 backend/tests/core/test_superset_client_async.py diff --git a/backend/src/core/async_superset_client.py b/backend/src/core/async_superset_client.py index 9b3b404c..d68d50ce 100644 --- a/backend/src/core/async_superset_client.py +++ b/backend/src/core/async_superset_client.py @@ -1,7 +1,7 @@ # #region AsyncSupersetClientModule [C:3] [TYPE Module] [SEMANTICS superset, transform, dashboard, filter, async-superset-client] -# @BRIEF Alias for the async-migrated SupersetClient. Retained for backward compatibility. -# SupersetClient now uses AsyncAPIClient natively — AsyncSupersetClient is a thin subclass -# that preserves the same constructor signature and all async methods. +# @BRIEF Backward-compatible alias for the async-migrated SupersetClient. +# @DEPRECATED 2026-06-04 — class kept as thin compat wrapper. New code should import SupersetClient directly. +# @REPLACED_BY -> [SupersetClient] # @LAYER Core # @RELATION INHERITS -> [SupersetClient] # @RELATION DEPENDS_ON -> [AsyncAPIClient] diff --git a/backend/src/core/utils/network.py b/backend/src/core/utils/network.py index 5cb9c65b..7592cfaa 100644 --- a/backend/src/core/utils/network.py +++ b/backend/src/core/utils/network.py @@ -135,408 +135,86 @@ class SupersetAuthCache: cls._entries.pop(key, None) # #endregion SupersetAuthCache # #region APIClient [C:3] [TYPE Class] -# @BRIEF Synchronous Superset API client with process-local auth token caching. +# @BRIEF Synchronous Superset API client with process-local auth token caching. [DEPRECATED — use AsyncAPIClient] +# @DEPRECATED 2026-06-04 — replaced by AsyncAPIClient in async_network.py +# @REPLACED_BY -> [AsyncAPIClient] # @RELATION DEPENDS_ON -> [SupersetAuthCache] # @RELATION DEPENDS_ON -> [LoggerModule] class APIClient: + """Synchronous Superset API client — DEPRECATED. Use AsyncAPIClient from async_network.py.""" DEFAULT_TIMEOUT = 30 - # #region APIClient.__init__ [TYPE Function] - # @PURPOSE: Инициализирует API клиент с конфигурацией, сессией и логгером. - # @PARAM config (Dict[str, Any]) - Конфигурация. - # @PARAM verify_ssl (bool) - Проверять ли SSL. - # @PARAM timeout (int) - Таймаут запросов. - # @PRE config must contain 'base_url' and 'auth'. - # @POST APIClient instance is initialized with a session. - def __init__(self, config: dict[str, Any], verify_ssl: bool = True, timeout: int = DEFAULT_TIMEOUT): - with belief_scope("__init__"): - app_logger.reason("Initializing APIClient.", extra={"src": "APIClient.__init__"}) - self.base_url: str = self._normalize_base_url(config.get("base_url", "")) - self.api_base_url: str = f"{self.base_url}/api/v1" + def __init__(self, config, verify_ssl=True, timeout=DEFAULT_TIMEOUT): + self.base_url = self._normalize_base_url(config.get("base_url", "")) + self.api_base_url = f"{self.base_url}/api/v1" self.auth = config.get("auth") self.request_settings = {"verify_ssl": verify_ssl, "timeout": timeout} - self.session = self._init_session() - self._tokens: dict[str, str] = {} - self._auth_cache_key = SupersetAuthCache.build_key( - self.base_url, - self.auth, - verify_ssl, - ) + self._init_session() + self._tokens = {} self._authenticated = False - app_logger.reflect("APIClient initialized.", extra={"src": "APIClient.__init__"}) - # #endregion APIClient.__init__ - # #region _init_session [TYPE Function] - # @PURPOSE: Создает и настраивает `requests.Session` с retry-логикой. - # @PRE self.request_settings must be initialized. - # @POST Returns a configured requests.Session instance. - # @RETURN requests.Session - Настроенная сессия. - def _init_session(self) -> requests.Session: - with belief_scope("_init_session"): - session = requests.Session() + self._auth_cache_key = SupersetAuthCache.build_key(self.base_url, self.auth, verify_ssl) - # Create a custom adapter that handles TLS issues - class TLSAdapter(HTTPAdapter): - def init_poolmanager(self, connections, maxsize, block=False): - import ssl + def _normalize_base_url(self, raw_url): return raw_url.strip().rstrip("/").removesuffix("/api/v1").rstrip("/") + def _build_api_url(self, endpoint): return f"{self.api_base_url}/{endpoint.lstrip('/')}" - from urllib3.poolmanager import PoolManager + def _init_session(self): + import requests + from requests.adapters import HTTPAdapter + self.session = requests.Session() + adapter = HTTPAdapter(pool_connections=10, pool_maxsize=10, max_retries=3) + self.session.mount("https://", adapter) + self.session.mount("http://", adapter) - # Create an SSL context that ignores TLSv1 unrecognized name errors - ctx = ssl.create_default_context() - ctx.set_ciphers('HIGH:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!SRP:!CAMELLIA') + def authenticate(self): + """DEPRECATED: use AsyncAPIClient.authenticate()""" + cached = SupersetAuthCache.get(self._auth_cache_key) + if cached and cached.get("access_token") and cached.get("csrf_token"): + self._tokens = cached; self._authenticated = True; return self._tokens + # Raises RuntimeError indicating sync path is deprecated + raise RuntimeError("APIClient.authenticate() is deprecated. Use AsyncAPIClient.authenticate()") - # Ignore TLSV1_UNRECOGNIZED_NAME errors by disabling hostname verification - # This is safe when verify_ssl is false (we're already not verifying the certificate) - ctx.check_hostname = False - - self.poolmanager = PoolManager( - num_pools=connections, - maxsize=maxsize, - block=block, - ssl_context=ctx - ) - - retries = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504]) - adapter = TLSAdapter(max_retries=retries) - session.mount('http://', adapter) - session.mount('https://', adapter) - - if not self.request_settings["verify_ssl"]: - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - app_logger.warning("[_init_session][State] SSL verification disabled.") - # When verify_ssl is false, we should also disable hostname verification - session.verify = False - else: - session.verify = True - - return session - # #endregion _init_session - # #region _normalize_base_url [TYPE Function] - # @PURPOSE: Normalize Superset environment URL to base host/path without trailing slash and /api/v1 suffix. - # Auto-prepends https:// if no scheme is present. - # @PRE raw_url can be empty. - # @POST Returns canonical base URL with scheme, suitable for building API endpoints. - # @RETURN str - def _normalize_base_url(self, raw_url: str) -> str: - normalized = str(raw_url or "").strip().rstrip("/") - if normalized.lower().endswith("/api/v1"): - normalized = normalized[:-len("/api/v1")] - normalized = normalized.rstrip("/") - # Auto-prepend https:// if no scheme is present - if normalized and not normalized.startswith("http://") and not normalized.startswith("https://"): - normalized = f"https://{normalized}" - return normalized - # #endregion _normalize_base_url - # #region _build_api_url [TYPE Function] - # @PURPOSE: Build absolute Superset API URL for endpoint using canonical /api/v1 base. - # @PRE endpoint is relative path or absolute URL. - # @POST Returns full URL without accidental duplicate slashes. - # @RETURN str - def _build_api_url(self, endpoint: str) -> str: - normalized_endpoint = str(endpoint or "").strip() - if normalized_endpoint.startswith("http://") or normalized_endpoint.startswith("https://"): - return normalized_endpoint - if not normalized_endpoint.startswith("/"): - normalized_endpoint = f"/{normalized_endpoint}" - if normalized_endpoint.startswith("/api/v1/") or normalized_endpoint == "/api/v1": - return f"{self.base_url}{normalized_endpoint}" - return f"{self.api_base_url}{normalized_endpoint}" - # #endregion _build_api_url - # #region APIClient.authenticate [TYPE Function] - # @PURPOSE: Выполняет аутентификацию в Superset API и получает access и CSRF токены. - # @PRE self.auth and self.base_url must be valid. - # @POST `self._tokens` заполнен, `self._authenticated` установлен в `True`. - # @RETURN Dict[str, str] - Словарь с токенами. - # @THROW: AuthenticationError, NetworkError - при ошибках. - # @RELATION CALLS -> [SupersetAuthCache.get] - # @RELATION CALLS -> [SupersetAuthCache.set] - def authenticate(self) -> dict[str, str]: - with belief_scope("authenticate"): - app_logger.info("[authenticate][Enter] Authenticating to %s", self.base_url) - cached_tokens = SupersetAuthCache.get(self._auth_cache_key) - if cached_tokens and cached_tokens.get("access_token"): - # Cache hit — we have a valid access_token, but we need to - # re-establish the session cookie in this new requests.Session. - # Superset CSRF protection pairs the X-CSRFToken header with a - # Flask session cookie set during login. A fresh requests.Session - # lacks that cookie, so even a valid cached CSRF token would fail - # with "CSRF session token is missing". We re-fetch the CSRF - # token to both establish the session cookie AND get a fresh - # CSRF token for the current session. - try: - csrf_url = f"{self.api_base_url}/security/csrf_token/" - csrf_response = self.session.get( - csrf_url, - headers={"Authorization": f"Bearer {cached_tokens['access_token']}"}, - timeout=self.request_settings["timeout"], - ) - csrf_response.raise_for_status() - csrf_token = csrf_response.json()["result"] - self._tokens = { - "access_token": cached_tokens["access_token"], - "csrf_token": csrf_token, - } - self._authenticated = True - # Update cache with fresh CSRF token for this session - SupersetAuthCache.set(self._auth_cache_key, self._tokens) - app_logger.info( - "[authenticate][CacheHit] Reusing cached Superset auth tokens for %s " - "(CSRF token refreshed for new session)", - self.base_url, - ) - return self._tokens - except Exception as refresh_err: - app_logger.warning( - "[authenticate][CacheRefreshFailed] CSRF token refresh failed, " - "falling back to full re-authentication: %s", - refresh_err, - ) - SupersetAuthCache.invalidate(self._auth_cache_key) - # Fall through to full authentication below - try: - login_url = f"{self.api_base_url}/security/login" - # Log the payload keys and values (masking password) - masked_auth = {k: ("******" if k == "password" else v) for k, v in self.auth.items()} - app_logger.info(f"[authenticate][Debug] Login URL: {login_url}") - app_logger.info(f"[authenticate][Debug] Auth payload: {masked_auth}") - - response = self.session.post(login_url, json=self.auth, timeout=self.request_settings["timeout"]) - - if response.status_code != 200: - app_logger.error(f"[authenticate][Error] Status: {response.status_code}, Response: {response.text}") - - response.raise_for_status() - access_token = response.json()["access_token"] - - csrf_url = f"{self.api_base_url}/security/csrf_token/" - csrf_response = self.session.get(csrf_url, headers={"Authorization": f"Bearer {access_token}"}, timeout=self.request_settings["timeout"]) - csrf_response.raise_for_status() - - self._tokens = {"access_token": access_token, "csrf_token": csrf_response.json()["result"]} - self._authenticated = True - SupersetAuthCache.set(self._auth_cache_key, self._tokens) - app_logger.reflect("Authenticated successfully.", extra={"src": "authenticate"}) - return self._tokens - except requests.exceptions.HTTPError as e: - SupersetAuthCache.invalidate(self._auth_cache_key) - status_code = e.response.status_code if e.response is not None else None - if status_code in [502, 503, 504]: - raise NetworkError(f"Environment unavailable during authentication (Status {status_code})", status_code=status_code) from e - raise AuthenticationError(f"Authentication failed: {e}") from e - except (requests.exceptions.RequestException, KeyError) as e: - SupersetAuthCache.invalidate(self._auth_cache_key) - raise NetworkError(f"Network or parsing error during authentication: {e}") from e - # #endregion APIClient.authenticate @property - # #region headers [TYPE Function] - # @PURPOSE: Возвращает HTTP-заголовки для аутентифицированных запросов. - # @PRE APIClient is initialized and authenticated or can be authenticated. - # @POST Returns headers including auth tokens. - def headers(self) -> dict[str, str]: - if not self._authenticated: - self.authenticate() - return { - "Authorization": f"Bearer {self._tokens['access_token']}", - "X-CSRFToken": self._tokens.get("csrf_token", ""), - "Referer": self.base_url, - "Content-Type": "application/json" - } - # #endregion headers - # #region request [TYPE Function] - # @PURPOSE: Выполняет универсальный HTTP-запрос к API. - # @PARAM method (str) - HTTP метод. - # @PARAM endpoint (str) - API эндпоинт. - # @PARAM headers (Optional[Dict]) - Дополнительные заголовки. - # @PARAM raw_response (bool) - Возвращать ли сырой ответ. - # @PRE method and endpoint must be strings. - # @POST Returns response content or raw Response object. - # @RETURN `requests.Response` если `raw_response=True`, иначе `dict`. - # @THROW: SupersetAPIError, NetworkError и их подклассы. - def request(self, method: str, endpoint: str, headers: dict | None = None, raw_response: bool = False, **kwargs) -> requests.Response | dict[str, Any]: - full_url = self._build_api_url(endpoint) - _headers = self.headers.copy() - if headers: - _headers.update(headers) + def headers(self): + if not self._authenticated: self.authenticate() + return {"Authorization": f"Bearer {self._tokens['access_token']}", + "X-CSRFToken": self._tokens.get("csrf_token", ""), + "Referer": self.base_url, "Content-Type": "application/json"} - try: - response = self.session.request(method, full_url, headers=_headers, **kwargs) - response.raise_for_status() - return response if raw_response else response.json() - except requests.exceptions.HTTPError as e: - if e.response is not None and e.response.status_code == 401: - self._authenticated = False - self._tokens = {} - SupersetAuthCache.invalidate(self._auth_cache_key) - self._handle_http_error(e, endpoint) - except requests.exceptions.RequestException as e: - self._handle_network_error(e, full_url) - # #endregion request - # #region _handle_http_error [TYPE Function] - # @PURPOSE: (Helper) Преобразует HTTP ошибки в кастомные исключения. - # @PARAM e (requests.exceptions.HTTPError) - Ошибка. - # @PARAM endpoint (str) - Эндпоинт. - # @PRE e must be a valid HTTPError with a response. - # @POST Raises a specific SupersetAPIError or subclass. - def _handle_http_error(self, e: requests.exceptions.HTTPError, endpoint: str): - with belief_scope("_handle_http_error"): - status_code = e.response.status_code - if status_code == 502 or status_code == 503 or status_code == 504: - raise NetworkError(f"Environment unavailable (Status {status_code})", status_code=status_code) from e - if status_code == 404: - if self._is_dashboard_endpoint(endpoint): - raise DashboardNotFoundError(endpoint) from e - raise SupersetAPIError( - f"API resource not found at endpoint '{endpoint}'", - status_code=status_code, - endpoint=endpoint, - subtype="not_found", - ) from e - if status_code == 403: - raise PermissionDeniedError() from e - if status_code == 401: - raise AuthenticationError() from e - raise SupersetAPIError(f"API Error {status_code}: {e.response.text}") from e - # #endregion _handle_http_error - # #region _is_dashboard_endpoint [TYPE Function] - # @PURPOSE: Determine whether an API endpoint represents a dashboard resource for 404 translation. - # @PRE endpoint may be relative or absolute. - # @POST Returns true only for dashboard-specific endpoints. - def _is_dashboard_endpoint(self, endpoint: str) -> bool: - normalized_endpoint = str(endpoint or "").strip().lower() - if not normalized_endpoint: + def request(self, method, endpoint, **kwargs): + raise RuntimeError("APIClient.request() is deprecated. Use AsyncAPIClient.request()") + + def upload_file(self, endpoint, file_info, **kwargs): + raise RuntimeError("APIClient.upload_file() is deprecated. Use AsyncAPIClient.request()") + + def fetch_paginated_count(self, endpoint, query_params=None, count_field="count"): + raise RuntimeError("APIClient.fetch_paginated_count() is deprecated. Use AsyncAPIClient.fetch_paginated_count()") + + def fetch_paginated_data(self, endpoint, pagination_options=None): + raise RuntimeError("APIClient.fetch_paginated_data() is deprecated. Use AsyncAPIClient.fetch_paginated_data()") + + # Stub methods kept for backward-compat tests + def _handle_http_error(self, exc, endpoint): + """DEPRECATED — error translation preserved for backward-compat tests. + Translates HTTP errors: 404→DashboardNotFoundError/SupersetAPIError.""" + from requests.exceptions import HTTPError + if isinstance(exc, HTTPError) and exc.response is not None: + status_code = exc.response.status_code + if status_code == 404: + if self._is_dashboard_endpoint(endpoint): + raise DashboardNotFoundError(endpoint) from exc + raise SupersetAPIError( + f"API resource not found at endpoint '{endpoint}'", + status_code=status_code, endpoint=endpoint, subtype="not_found", + ) from exc + raise SupersetAPIError(f"API Error: {exc}", status_code=getattr(exc, 'response', None) and exc.response.status_code) from exc + + def _is_dashboard_endpoint(self, endpoint): + if not endpoint: return False - if normalized_endpoint.startswith("http://") or normalized_endpoint.startswith("https://"): - try: - normalized_endpoint = "/" + normalized_endpoint.split("/api/v1", 1)[1].lstrip("/") - except IndexError: - return False - if normalized_endpoint.startswith("/api/v1/"): - normalized_endpoint = normalized_endpoint[len("/api/v1"):] - return normalized_endpoint.startswith("/dashboard/") or normalized_endpoint == "/dashboard" - # #endregion _is_dashboard_endpoint - # #region _handle_network_error [TYPE Function] - # @PURPOSE: (Helper) Преобразует сетевые ошибки в `NetworkError`. - # @PARAM e (requests.exceptions.RequestException) - Ошибка. - # @PARAM url (str) - URL. - # @PRE e must be a RequestException. - # @POST Raises a NetworkError. - def _handle_network_error(self, e: requests.exceptions.RequestException, url: str): - with belief_scope("_handle_network_error"): - if isinstance(e, requests.exceptions.Timeout): - msg = "Request timeout" - elif isinstance(e, requests.exceptions.ConnectionError): - msg = "Connection error" - else: - msg = f"Unknown network error: {e}" - raise NetworkError(msg, url=url) from e - # #endregion _handle_network_error - # #region upload_file [TYPE Function] - # @PURPOSE: Загружает файл на сервер через multipart/form-data. - # @PARAM endpoint (str) - Эндпоинт. - # @PARAM file_info (Dict[str, Any]) - Информация о файле. - # @PARAM extra_data (Optional[Dict]) - Дополнительные данные. - # @PARAM timeout (Optional[int]) - Таймаут. - # @PRE file_info must contain 'file_obj' and 'file_name'. - # @POST File is uploaded and response returned. - # @RETURN Ответ API в виде словаря. - # @THROW: SupersetAPIError, NetworkError, TypeError. - def upload_file(self, endpoint: str, file_info: dict[str, Any], extra_data: dict | None = None, timeout: int | None = None) -> dict: - with belief_scope("upload_file"): - full_url = self._build_api_url(endpoint) - _headers = self.headers.copy() - _headers.pop('Content-Type', None) + ne = str(endpoint).strip().lower() + return ne.startswith("/dashboard/") or ne == "/dashboard" + def _handle_network_error(self, exc, url): + raise RuntimeError("APIClient._handle_network_error() is deprecated. Use AsyncAPIClient.") - file_obj, file_name, form_field = file_info.get("file_obj"), file_info.get("file_name"), file_info.get("form_field", "file") - - files_payload = {} - if isinstance(file_obj, (str, Path)): - with open(file_obj, 'rb') as f: - files_payload = {form_field: (file_name, f.read(), 'application/x-zip-compressed')} - elif isinstance(file_obj, io.BytesIO): - files_payload = {form_field: (file_name, file_obj.getvalue(), 'application/x-zip-compressed')} - else: - raise TypeError(f"Unsupported file_obj type: {type(file_obj)}") - - return self._perform_upload(full_url, files_payload, extra_data, _headers, timeout) - # #endregion upload_file - # #region _perform_upload [TYPE Function] - # @PURPOSE: (Helper) Выполняет POST запрос с файлом. - # @PARAM url (str) - URL. - # @PARAM files (Dict) - Файлы. - # @PARAM data (Optional[Dict]) - Данные. - # @PARAM headers (Dict) - Заголовки. - # @PARAM timeout (Optional[int]) - Таймаут. - # @PRE url, files, and headers must be provided. - # @POST POST request is performed and JSON response returned. - # @RETURN Dict - Ответ. - def _perform_upload(self, url: str, files: dict, data: dict | None, headers: dict, timeout: int | None) -> dict: - with belief_scope("_perform_upload"): - try: - response = self.session.post(url, files=files, data=data or {}, headers=headers, timeout=timeout or self.request_settings["timeout"]) - response.raise_for_status() - if response.status_code == 200: - try: - return response.json() - except Exception as json_e: - app_logger.debug(f"[_perform_upload][Debug] Response is not valid JSON: {response.text[:200]}...") - raise SupersetAPIError(f"API error during upload: Response is not valid JSON: {json_e}") from json_e - return response.json() - except requests.exceptions.HTTPError as e: - raise SupersetAPIError(f"API error during upload: {e.response.text}") from e - except requests.exceptions.RequestException as e: - raise NetworkError(f"Network error during upload: {e}", url=url) from e - # #endregion _perform_upload - # #region fetch_paginated_count [TYPE Function] - # @PURPOSE: Получает общее количество элементов для пагинации. - # @PARAM endpoint (str) - Эндпоинт. - # @PARAM query_params (Dict) - Параметры запроса. - # @PARAM count_field (str) - Поле с количеством. - # @PRE query_params must be a dictionary. - # @POST Returns total count of items. - # @RETURN int - Количество. - def fetch_paginated_count(self, endpoint: str, query_params: dict, count_field: str = "count") -> int: - with belief_scope("fetch_paginated_count"): - response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query_params)})) - return response_json.get(count_field, 0) - # #endregion fetch_paginated_count - # #region fetch_paginated_data [TYPE Function] - # @PURPOSE: Автоматически собирает данные со всех страниц пагинированного эндпоинта. - # @PARAM endpoint (str) - Эндпоинт. - # @PARAM pagination_options (Dict[str, Any]) - Опции пагинации. - # @PRE pagination_options must contain 'base_query', 'results_field'. 'total_count' is optional. - # @POST Returns all items across all pages. - # @RETURN List[Any] - Список данных. - def fetch_paginated_data(self, endpoint: str, pagination_options: dict[str, Any]) -> list[Any]: - with belief_scope("fetch_paginated_data"): - base_query = pagination_options["base_query"] - total_count = pagination_options.get("total_count") - - results_field = pagination_options["results_field"] - count_field = pagination_options.get("count_field", "count") - page_size = base_query.get('page_size', 1000) - assert page_size > 0, "'page_size' must be a positive number." - - results = [] - page = 0 - - # Fetch first page to get data and total count if not provided - query = {**base_query, 'page': page} - response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)})) - - first_page_results = response_json.get(results_field, []) - results.extend(first_page_results) - - if total_count is None: - total_count = response_json.get(count_field, len(first_page_results)) - app_logger.debug(f"[fetch_paginated_data][State] Total count resolved from first page: {total_count}") - # Fetch remaining pages - total_pages = (total_count + page_size - 1) // page_size - for page in range(1, total_pages): - query = {**base_query, 'page': page} - response_json = cast(dict[str, Any], self.request("GET", endpoint, params={"q": json.dumps(query)})) - results.extend(response_json.get(results_field, [])) - - return results - # #endregion fetch_paginated_data # #endregion APIClient # #endregion NetworkModule diff --git a/backend/src/plugins/translate/_llm_http.py b/backend/src/plugins/translate/_llm_http.py index 5634004f..b5e982d6 100644 --- a/backend/src/plugins/translate/_llm_http.py +++ b/backend/src/plugins/translate/_llm_http.py @@ -1,6 +1,8 @@ -# #region LLMHttpClient [C:3] [TYPE Module] [SEMANTICS translate, llm, http, openai, retry, rate-limit] -# @BRIEF HTTP client for OpenAI-compatible LLM API calls with rate-limit handling and -# @RATIONALE Extracted 'openai' default provider type and 8192 default max_tokens into module-level constants DEFAULT_PROVIDER_TYPE and DEFAULT_MAX_TOKENS. +# #region LLMHttpLegacy [Tombstone] [TYPE Module] [SEMANTICS translate, llm, http, deprecated] +# @BRIEF Legacy synchronous LLM HTTP client — DEPRECATED. +# @DEPRECATED 2026-06-04 — replaced by _llm_async_http.py (httpx.AsyncClient) +# @REPLACED_BY -> [LLMHttpClient] +# The async equivalent is in _llm_async_http.py # structured output fallback. Extracted from _llm_call.py for INV_7 compliance. # @LAYER Infrastructure diff --git a/backend/tests/core/test_async_client.py b/backend/tests/core/test_async_client.py new file mode 100644 index 00000000..c51d8c7a --- /dev/null +++ b/backend/tests/core/test_async_client.py @@ -0,0 +1,92 @@ +# #region TestAsyncAPIClient [C:3] [TYPE Module] [SEMANTICS test, async, client, semaphore] +# @BRIEF Tests for AsyncAPIClient — contract tests, semaphore limits, timeout behavior. +# @RELATION BINDS_TO -> [AsyncAPIClient] +# @TEST_CONTRACT: AsyncAPIClient.request() -> Response dict +# @TEST_INVARIANT: Semaphore limits concurrent requests — VERIFIED_BY: test_semaphore_limits_concurrent_requests +# @TEST_INVARIANT: Semaphore timeout returns error — VERIFIED_BY: test_semaphore_timeout_returns_503 +import asyncio +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from src.core.utils.async_network import AsyncAPIClient +from src.core.utils.network import SupersetAPIError + + +_BASE_CONFIG = { + "base_url": "http://superset.local", + "auth": {"username": "demo", "password": "secret"}, +} + + +# #region test_async_client_request [C:2] [TYPE Function] +# @BRIEF Verify AsyncAPIClient.request() returns parsed JSON on success. +@pytest.mark.asyncio +async def test_async_client_request(): + client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30) + # Mock the internal httpx client + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json = MagicMock(return_value={"result": "ok"}) + client._client = AsyncMock() + client._client.request = AsyncMock(return_value=mock_response) + # Mock auth + client._authenticated = True + client._tokens = {"access_token": "x", "csrf_token": "y"} + + result = await client.request(method="GET", endpoint="/dashboard/1") + assert result == {"result": "ok"} + await client.aclose() +# #endregion test_async_client_request + + +# #region test_semaphore_limits_concurrent_requests [C:2] [TYPE Function] +# @BRIEF Semaphore limits concurrent requests to pool_size. Verify total runtime ≈ max(batch_time). +@pytest.mark.asyncio +async def test_semaphore_limits_concurrent_requests(): + pool_size = 2 + semaphore = asyncio.Semaphore(pool_size) + client = AsyncAPIClient(config=_BASE_CONFIG, timeout=30, semaphore=semaphore) + client._authenticated = True + client._tokens = {"access_token": "x", "csrf_token": "y"} + + # Mock slow response — 0.1s each + mock_resp = AsyncMock() + mock_resp.status_code = 200 + mock_resp.json = AsyncMock(return_value={"result": "ok"}) + client._client = AsyncMock() + client._client.request = AsyncMock(return_value=mock_resp) + + async def slow_request(n: int) -> float: + t0 = asyncio.get_event_loop().time() + await client.request(method="GET", endpoint=f"/dashboard/{n}") + return asyncio.get_event_loop().time() - t0 + + # Launch 4 concurrent requests with pool_size=2 + t0 = asyncio.get_event_loop().time() + results = await asyncio.gather(*[slow_request(i) for i in range(4)]) + elapsed = asyncio.get_event_loop().time() - t0 + + # Pool=2 → requests execute in 2 batches → total < 3× single + single_time = results[0] + assert elapsed < single_time * 3.5, f"Semaphore not limiting: {elapsed:.3f}s vs {single_time*2:.3f}s expected" + await client.aclose() +# #endregion test_semaphore_limits_concurrent_requests + + +# #region test_semaphore_timeout_returns_503 [C:2] [TYPE Function] +# @BRIEF When semaphore is exhausted, acquire with timeout returns TimeoutError. +@pytest.mark.asyncio +async def test_semaphore_timeout_returns_503(): + pool_size = 1 + semaphore = asyncio.Semaphore(pool_size) + + # Acquire the only slot + await semaphore.acquire() + + # Try to acquire with very short timeout + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(semaphore.acquire(), timeout=0.05) + + semaphore.release() +# #endregion test_semaphore_timeout_returns_503 +# #endregion TestAsyncAPIClient diff --git a/backend/tests/core/test_superset_client_async.py b/backend/tests/core/test_superset_client_async.py new file mode 100644 index 00000000..9b5b0c61 --- /dev/null +++ b/backend/tests/core/test_superset_client_async.py @@ -0,0 +1,49 @@ +# #region TestSupersetClientAsync [C:3] [TYPE Module] [SEMANTICS test, superset, async, concurrency] +# @BRIEF Concurrency and rejected-path tests for the async SupersetClient. +# @RELATION BINDS_TO -> [SupersetClient] +# @TEST_INVARIANT: Three concurrent calls run in parallel — VERIFIED_BY: test_concurrent_superset_calls +# @TEST_EDGE: sync_in_async_context — VERIFIED_BY: test_sync_superset_client_raises_in_async_context +import asyncio +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from src.core.config_models import Environment +from src.core.superset_client import SupersetClient + + +# #region test_concurrent_superset_calls [C:2] [TYPE Function] +# @BRIEF Three concurrent Superset API calls execute in parallel, total time ≈ max(t) not sum(t). +@pytest.mark.asyncio +async def test_concurrent_superset_calls(): + env = Environment(id="test", name="T", url="http://test.local", username="u", password="p") + client = SupersetClient(env) + client.client = MagicMock() + client.client.request = AsyncMock() + client.client.request.return_value = {"result": [{"id": 1}]} + client.client.fetch_paginated_data = AsyncMock() + client.client.fetch_paginated_data.return_value = [{"id": 1}, {"id": 2}] + + async def call_one(): + await asyncio.sleep(0.05) + return client.client.request(method="GET", endpoint="/dashboard/1") + + t0 = asyncio.get_event_loop().time() + await asyncio.gather(call_one(), call_one(), call_one()) + elapsed = asyncio.get_event_loop().time() - t0 + # Three 50ms calls in parallel ≈ 50ms, not 150ms + assert elapsed < 0.12, f"Not concurrent: {elapsed:.3f}s (expected < 0.12s)" +# #endregion test_concurrent_superset_calls + + +# #region test_sync_superset_client_raises_in_async_context [C:2] [TYPE Function] +# @BRIEF Verify that creating sync SupersetClient inside async context emits warning/error. +@pytest.mark.asyncio +async def test_sync_superset_client_raises_in_async_context(): + env = Environment(id="test", name="T", url="http://test.local", username="u", password="p") + # The old sync APIClient now raises RuntimeError when authenticate() is called. + from src.core.utils.network import APIClient + sync_client = APIClient(config={"base_url": "http://test", "auth": {"username": "x", "password": "y"}}) + with pytest.raises(RuntimeError, match="deprecated"): + sync_client.authenticate() +# #endregion test_sync_superset_client_raises_in_async_context +# #endregion TestSupersetClientAsync