Fix LLM validation and dashboard health hot paths
This commit is contained in:
@@ -0,0 +1,30 @@
|
||||
# [DEF:backend.src.plugins.llm_analysis.__tests__.test_client_headers:Module]
|
||||
# @TIER: STANDARD
|
||||
# @SEMANTICS: tests, llm-client, openrouter, headers
|
||||
# @PURPOSE: Verify OpenRouter client initialization includes provider-specific headers.
|
||||
|
||||
from src.plugins.llm_analysis.models import LLMProviderType
|
||||
from src.plugins.llm_analysis.service import LLMClient
|
||||
|
||||
|
||||
# [DEF:test_openrouter_client_includes_referer_and_title_headers:Function]
|
||||
# @PURPOSE: OpenRouter requests should carry site/app attribution headers for compatibility.
|
||||
# @PRE: Client is initialized for OPENROUTER provider.
|
||||
# @POST: Async client headers include Authorization, HTTP-Referer, and X-Title.
|
||||
def test_openrouter_client_includes_referer_and_title_headers(monkeypatch):
|
||||
monkeypatch.setenv("OPENROUTER_SITE_URL", "http://localhost:8000")
|
||||
monkeypatch.setenv("OPENROUTER_APP_NAME", "ss-tools-test")
|
||||
|
||||
client = LLMClient(
|
||||
provider_type=LLMProviderType.OPENROUTER,
|
||||
api_key="sk-test-provider-key-123456",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
default_model="nvidia/nemotron-nano-12b-v2-vl:free",
|
||||
)
|
||||
|
||||
headers = dict(client.client.default_headers)
|
||||
assert headers["Authorization"] == "Bearer sk-test-provider-key-123456"
|
||||
assert headers["HTTP-Referer"] == "http://localhost:8000"
|
||||
assert headers["X-Title"] == "ss-tools-test"
|
||||
# [/DEF:test_openrouter_client_includes_referer_and_title_headers:Function]
|
||||
# [/DEF:backend.src.plugins.llm_analysis.__tests__.test_client_headers:Module]
|
||||
@@ -0,0 +1,344 @@
|
||||
# [DEF:backend.src.plugins.llm_analysis.__tests__.test_screenshot_service:Module]
|
||||
# @TIER: STANDARD
|
||||
# @SEMANTICS: tests, screenshot-service, navigation, timeout-regression
|
||||
# @PURPOSE: Protect dashboard screenshot navigation from brittle networkidle waits.
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugins.llm_analysis.service import ScreenshotService
|
||||
|
||||
|
||||
# [DEF:test_iter_login_roots_includes_child_frames:Function]
|
||||
# @PURPOSE: Login discovery must search embedded auth frames, not only the main page.
|
||||
# @PRE: Page exposes child frames list.
|
||||
# @POST: Returned roots include page plus child frames in order.
|
||||
def test_iter_login_roots_includes_child_frames():
|
||||
frame_a = object()
|
||||
frame_b = object()
|
||||
fake_page = type("FakePage", (), {"frames": [frame_a, frame_b]})()
|
||||
service = ScreenshotService(env=type("Env", (), {})())
|
||||
|
||||
roots = service._iter_login_roots(fake_page)
|
||||
|
||||
assert roots == [fake_page, frame_a, frame_b]
|
||||
# [/DEF:test_iter_login_roots_includes_child_frames:Function]
|
||||
|
||||
|
||||
# [DEF:test_response_looks_like_login_page_detects_login_markup:Function]
|
||||
# @PURPOSE: Direct login fallback must reject responses that render the login screen again.
|
||||
# @PRE: Response body contains stable login-page markers.
|
||||
# @POST: Helper returns True so caller treats fallback as failed authentication.
|
||||
def test_response_looks_like_login_page_detects_login_markup():
|
||||
service = ScreenshotService(env=type("Env", (), {})())
|
||||
|
||||
result = service._response_looks_like_login_page(
|
||||
"""
|
||||
<html>
|
||||
<body>
|
||||
<p>Enter your login and password below</p>
|
||||
<label>Username:</label>
|
||||
<label>Password:</label>
|
||||
<button>Sign in</button>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
)
|
||||
|
||||
assert result is True
|
||||
# [/DEF:test_response_looks_like_login_page_detects_login_markup:Function]
|
||||
|
||||
|
||||
# [DEF:test_find_first_visible_locator_skips_hidden_first_match:Function]
|
||||
# @PURPOSE: Locator helper must not reject a selector collection just because its first element is hidden.
|
||||
# @PRE: First matched element is hidden and second matched element is visible.
|
||||
# @POST: Helper returns the second visible candidate.
|
||||
@pytest.mark.anyio
|
||||
async def test_find_first_visible_locator_skips_hidden_first_match():
|
||||
class _FakeElement:
|
||||
def __init__(self, visible, label):
|
||||
self._visible = visible
|
||||
self.label = label
|
||||
|
||||
async def is_visible(self):
|
||||
return self._visible
|
||||
|
||||
class _FakeLocator:
|
||||
def __init__(self, elements):
|
||||
self._elements = elements
|
||||
|
||||
async def count(self):
|
||||
return len(self._elements)
|
||||
|
||||
def nth(self, index):
|
||||
return self._elements[index]
|
||||
|
||||
service = ScreenshotService(env=type("Env", (), {})())
|
||||
hidden_then_visible = _FakeLocator([
|
||||
_FakeElement(False, "hidden"),
|
||||
_FakeElement(True, "visible"),
|
||||
])
|
||||
|
||||
result = await service._find_first_visible_locator([hidden_then_visible])
|
||||
|
||||
assert result.label == "visible"
|
||||
# [/DEF:test_find_first_visible_locator_skips_hidden_first_match:Function]
|
||||
|
||||
|
||||
# [DEF:test_submit_login_via_form_post_uses_browser_context_request:Function]
|
||||
# @PURPOSE: Fallback login must submit hidden fields and credentials through the context request cookie jar.
|
||||
# @PRE: Login DOM exposes csrf hidden field and request context returns authenticated HTML.
|
||||
# @POST: Helper returns True and request payload contains csrf_token plus credentials plus request options.
|
||||
@pytest.mark.anyio
|
||||
async def test_submit_login_via_form_post_uses_browser_context_request():
|
||||
class _FakeInput:
|
||||
def __init__(self, name, value):
|
||||
self._name = name
|
||||
self._value = value
|
||||
|
||||
async def get_attribute(self, attr_name):
|
||||
return self._name if attr_name == "name" else None
|
||||
|
||||
async def input_value(self):
|
||||
return self._value
|
||||
|
||||
class _FakeLocator:
|
||||
def __init__(self, items):
|
||||
self._items = items
|
||||
|
||||
async def count(self):
|
||||
return len(self._items)
|
||||
|
||||
def nth(self, index):
|
||||
return self._items[index]
|
||||
|
||||
class _FakeResponse:
|
||||
status = 200
|
||||
url = "https://example.test/welcome/"
|
||||
|
||||
async def text(self):
|
||||
return "<html><body>Welcome</body></html>"
|
||||
|
||||
class _FakeRequest:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
async def post(self, url, form=None, headers=None, timeout=None, fail_on_status_code=None, max_redirects=None):
|
||||
self.calls.append({
|
||||
"url": url,
|
||||
"form": dict(form or {}),
|
||||
"headers": dict(headers or {}),
|
||||
"timeout": timeout,
|
||||
"fail_on_status_code": fail_on_status_code,
|
||||
"max_redirects": max_redirects,
|
||||
})
|
||||
return _FakeResponse()
|
||||
|
||||
class _FakeContext:
|
||||
def __init__(self):
|
||||
self.request = _FakeRequest()
|
||||
|
||||
class _FakePage:
|
||||
def __init__(self):
|
||||
self.frames = []
|
||||
self.context = _FakeContext()
|
||||
|
||||
def locator(self, selector):
|
||||
if selector == "input[type='hidden'][name]":
|
||||
return _FakeLocator([
|
||||
_FakeInput("csrf_token", "csrf-123"),
|
||||
_FakeInput("next", "/superset/welcome/"),
|
||||
])
|
||||
return _FakeLocator([])
|
||||
|
||||
env = type("Env", (), {"username": "admin", "password": "secret"})()
|
||||
service = ScreenshotService(env=env)
|
||||
page = _FakePage()
|
||||
|
||||
result = await service._submit_login_via_form_post(page, "https://example.test/login/")
|
||||
|
||||
assert result is True
|
||||
assert page.context.request.calls == [{
|
||||
"url": "https://example.test/login/",
|
||||
"form": {
|
||||
"csrf_token": "csrf-123",
|
||||
"next": "/superset/welcome/",
|
||||
"username": "admin",
|
||||
"password": "secret",
|
||||
},
|
||||
"headers": {
|
||||
"Origin": "https://example.test",
|
||||
"Referer": "https://example.test/login/",
|
||||
},
|
||||
"timeout": 10000,
|
||||
"fail_on_status_code": False,
|
||||
"max_redirects": 0,
|
||||
}]
|
||||
# [/DEF:test_submit_login_via_form_post_uses_browser_context_request:Function]
|
||||
|
||||
|
||||
# [DEF:test_submit_login_via_form_post_accepts_authenticated_redirect:Function]
|
||||
# @PURPOSE: Fallback login must treat non-login 302 redirect as success without waiting for redirect target.
|
||||
# @PRE: Request response is 302 with Location outside login path.
|
||||
# @POST: Helper returns True.
|
||||
@pytest.mark.anyio
|
||||
async def test_submit_login_via_form_post_accepts_authenticated_redirect():
|
||||
class _FakeInput:
|
||||
def __init__(self, name, value):
|
||||
self._name = name
|
||||
self._value = value
|
||||
|
||||
async def get_attribute(self, attr_name):
|
||||
return self._name if attr_name == "name" else None
|
||||
|
||||
async def input_value(self):
|
||||
return self._value
|
||||
|
||||
class _FakeLocator:
|
||||
def __init__(self, items):
|
||||
self._items = items
|
||||
|
||||
async def count(self):
|
||||
return len(self._items)
|
||||
|
||||
def nth(self, index):
|
||||
return self._items[index]
|
||||
|
||||
class _FakeResponse:
|
||||
status = 302
|
||||
url = "https://example.test/login/"
|
||||
headers = {"location": "/superset/welcome/"}
|
||||
|
||||
async def text(self):
|
||||
return ""
|
||||
|
||||
class _FakeRequest:
|
||||
async def post(self, url, form=None, headers=None, timeout=None, fail_on_status_code=None, max_redirects=None):
|
||||
return _FakeResponse()
|
||||
|
||||
class _FakeContext:
|
||||
def __init__(self):
|
||||
self.request = _FakeRequest()
|
||||
|
||||
class _FakePage:
|
||||
def __init__(self):
|
||||
self.frames = []
|
||||
self.context = _FakeContext()
|
||||
|
||||
def locator(self, selector):
|
||||
if selector == "input[type='hidden'][name]":
|
||||
return _FakeLocator([_FakeInput("csrf_token", "csrf-123")])
|
||||
return _FakeLocator([])
|
||||
|
||||
env = type("Env", (), {"username": "admin", "password": "secret"})()
|
||||
service = ScreenshotService(env=env)
|
||||
|
||||
result = await service._submit_login_via_form_post(_FakePage(), "https://example.test/login/")
|
||||
|
||||
assert result is True
|
||||
# [/DEF:test_submit_login_via_form_post_accepts_authenticated_redirect:Function]
|
||||
|
||||
|
||||
# [DEF:test_submit_login_via_form_post_rejects_login_markup_response:Function]
|
||||
# @PURPOSE: Fallback login must fail when POST response still contains login form content.
|
||||
# @PRE: Login DOM exposes csrf hidden field and request response renders login markup.
|
||||
# @POST: Helper returns False.
|
||||
@pytest.mark.anyio
|
||||
async def test_submit_login_via_form_post_rejects_login_markup_response():
|
||||
class _FakeInput:
|
||||
def __init__(self, name, value):
|
||||
self._name = name
|
||||
self._value = value
|
||||
|
||||
async def get_attribute(self, attr_name):
|
||||
return self._name if attr_name == "name" else None
|
||||
|
||||
async def input_value(self):
|
||||
return self._value
|
||||
|
||||
class _FakeLocator:
|
||||
def __init__(self, items):
|
||||
self._items = items
|
||||
|
||||
async def count(self):
|
||||
return len(self._items)
|
||||
|
||||
def nth(self, index):
|
||||
return self._items[index]
|
||||
|
||||
class _FakeResponse:
|
||||
status = 200
|
||||
url = "https://example.test/login/"
|
||||
|
||||
async def text(self):
|
||||
return """
|
||||
<html>
|
||||
<body>
|
||||
Enter your login and password below
|
||||
Username:
|
||||
Password:
|
||||
Sign in
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
class _FakeRequest:
|
||||
async def post(self, url, form=None, headers=None, timeout=None, fail_on_status_code=None, max_redirects=None):
|
||||
return _FakeResponse()
|
||||
|
||||
class _FakeContext:
|
||||
def __init__(self):
|
||||
self.request = _FakeRequest()
|
||||
|
||||
class _FakePage:
|
||||
def __init__(self):
|
||||
self.frames = []
|
||||
self.context = _FakeContext()
|
||||
|
||||
def locator(self, selector):
|
||||
if selector == "input[type='hidden'][name]":
|
||||
return _FakeLocator([_FakeInput("csrf_token", "csrf-123")])
|
||||
return _FakeLocator([])
|
||||
|
||||
env = type("Env", (), {"username": "admin", "password": "secret"})()
|
||||
service = ScreenshotService(env=env)
|
||||
|
||||
result = await service._submit_login_via_form_post(_FakePage(), "https://example.test/login/")
|
||||
|
||||
assert result is False
|
||||
# [/DEF:test_submit_login_via_form_post_rejects_login_markup_response:Function]
|
||||
|
||||
|
||||
# [DEF:test_goto_resilient_falls_back_from_domcontentloaded_to_load:Function]
|
||||
# @PURPOSE: Pages with unstable primary wait must retry with fallback wait strategy.
|
||||
# @PRE: First page.goto call raises; second succeeds.
|
||||
# @POST: Helper returns second response and attempts both wait modes in order.
|
||||
@pytest.mark.anyio
|
||||
async def test_goto_resilient_falls_back_from_domcontentloaded_to_load():
|
||||
class _FakePage:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
async def goto(self, url, wait_until, timeout):
|
||||
self.calls.append((url, wait_until, timeout))
|
||||
if wait_until == "domcontentloaded":
|
||||
raise RuntimeError("primary wait failed")
|
||||
return {"ok": True, "url": url, "wait_until": wait_until}
|
||||
|
||||
page = _FakePage()
|
||||
service = ScreenshotService(env=type("Env", (), {})())
|
||||
|
||||
response = await service._goto_resilient(
|
||||
page,
|
||||
"https://example.test/dashboard",
|
||||
primary_wait_until="domcontentloaded",
|
||||
fallback_wait_until="load",
|
||||
timeout=1234,
|
||||
)
|
||||
|
||||
assert response["ok"] is True
|
||||
assert page.calls == [
|
||||
("https://example.test/dashboard", "domcontentloaded", 1234),
|
||||
("https://example.test/dashboard", "load", 1234),
|
||||
]
|
||||
# [/DEF:test_goto_resilient_falls_back_from_domcontentloaded_to_load:Function]
|
||||
# [/DEF:backend.src.plugins.llm_analysis.__tests__.test_screenshot_service:Module]
|
||||
67
backend/src/plugins/llm_analysis/__tests__/test_service.py
Normal file
67
backend/src/plugins/llm_analysis/__tests__/test_service.py
Normal file
@@ -0,0 +1,67 @@
|
||||
# [DEF:backend.src.plugins.llm_analysis.__tests__.test_service:Module]
|
||||
# @TIER: STANDARD
|
||||
# @SEMANTICS: tests, llm-analysis, fallback, provider-error, unknown-status
|
||||
# @PURPOSE: Verify LLM analysis transport/provider failures do not masquerade as dashboard FAIL results.
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugins.llm_analysis.models import LLMProviderType
|
||||
from src.plugins.llm_analysis.service import LLMClient
|
||||
|
||||
|
||||
# [DEF:test_test_runtime_connection_uses_json_completion_transport:Function]
|
||||
# @PURPOSE: Provider self-test must exercise the same chat completion transport as runtime analysis.
|
||||
# @PRE: get_json_completion is available on initialized client.
|
||||
# @POST: Self-test forwards a lightweight user message into get_json_completion and returns its payload.
|
||||
@pytest.mark.anyio
|
||||
async def test_test_runtime_connection_uses_json_completion_transport(monkeypatch):
|
||||
client = LLMClient(
|
||||
provider_type=LLMProviderType.OPENROUTER,
|
||||
api_key="sk-test-provider-key-123456",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
default_model="nvidia/nemotron-nano-12b-v2-vl:free",
|
||||
)
|
||||
recorded = {}
|
||||
|
||||
async def _fake_get_json_completion(messages):
|
||||
recorded["messages"] = messages
|
||||
return {"ok": True}
|
||||
|
||||
monkeypatch.setattr(client, "get_json_completion", _fake_get_json_completion)
|
||||
|
||||
result = await client.test_runtime_connection()
|
||||
|
||||
assert result == {"ok": True}
|
||||
assert recorded["messages"][0]["role"] == "user"
|
||||
assert "Return exactly this JSON object" in recorded["messages"][0]["content"]
|
||||
# [/DEF:test_test_runtime_connection_uses_json_completion_transport:Function]
|
||||
|
||||
|
||||
# [DEF:test_analyze_dashboard_provider_error_maps_to_unknown:Function]
|
||||
# @PURPOSE: Infrastructure/provider failures must produce UNKNOWN analysis status rather than FAIL.
|
||||
# @PRE: LLMClient.get_json_completion raises provider/auth exception.
|
||||
# @POST: Returned payload uses status=UNKNOWN and issue severity UNKNOWN.
|
||||
@pytest.mark.anyio
|
||||
async def test_analyze_dashboard_provider_error_maps_to_unknown(monkeypatch, tmp_path):
|
||||
screenshot_path = tmp_path / "shot.jpg"
|
||||
screenshot_path.write_bytes(b"fake-image")
|
||||
|
||||
client = LLMClient(
|
||||
provider_type=LLMProviderType.OPENROUTER,
|
||||
api_key="sk-test-provider-key-123456",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
default_model="nvidia/nemotron-nano-12b-v2-vl:free",
|
||||
)
|
||||
|
||||
async def _raise_provider_error(_messages):
|
||||
raise RuntimeError("Error code: 401 - {'error': {'message': 'User not found.', 'code': 401}}")
|
||||
|
||||
monkeypatch.setattr(client, "get_json_completion", _raise_provider_error)
|
||||
|
||||
result = await client.analyze_dashboard(str(screenshot_path), logs=["line-1"])
|
||||
|
||||
assert result["status"] == "UNKNOWN"
|
||||
assert "Failed to get response from LLM" in result["summary"]
|
||||
assert result["issues"][0]["severity"] == "UNKNOWN"
|
||||
# [/DEF:test_analyze_dashboard_provider_error_maps_to_unknown:Function]
|
||||
# [/DEF:backend.src.plugins.llm_analysis.__tests__.test_service:Module]
|
||||
@@ -35,6 +35,7 @@ class ValidationStatus(str, Enum):
|
||||
PASS = "PASS"
|
||||
WARN = "WARN"
|
||||
FAIL = "FAIL"
|
||||
UNKNOWN = "UNKNOWN"
|
||||
# [/DEF:ValidationStatus:Class]
|
||||
|
||||
# [DEF:DetectedIssue:Class]
|
||||
|
||||
@@ -307,7 +307,7 @@ class DashboardValidationPlugin(PluginBase):
|
||||
await notification_service.dispatch_report(
|
||||
record=db_record,
|
||||
policy=policy,
|
||||
background_tasks=context.background_tasks if context else None
|
||||
background_tasks=getattr(context, "background_tasks", None) if context else None
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(f"Failed to dispatch notifications: {e}")
|
||||
|
||||
@@ -12,6 +12,8 @@ import asyncio
|
||||
import base64
|
||||
import json
|
||||
import io
|
||||
import os
|
||||
from urllib.parse import urlsplit
|
||||
from typing import List, Dict, Any
|
||||
import httpx
|
||||
from PIL import Image
|
||||
@@ -33,6 +35,242 @@ class ScreenshotService:
|
||||
self.env = env
|
||||
# [/DEF:ScreenshotService.__init__:Function]
|
||||
|
||||
# [DEF:ScreenshotService._find_first_visible_locator:Function]
|
||||
# @PURPOSE: Resolve the first visible locator from multiple Playwright locator strategies.
|
||||
# @PRE: candidates is a non-empty list of locator-like objects.
|
||||
# @POST: Returns a locator ready for interaction or None when nothing matches.
|
||||
async def _find_first_visible_locator(self, candidates) -> Any:
|
||||
for locator in candidates:
|
||||
try:
|
||||
match_count = await locator.count()
|
||||
for index in range(match_count):
|
||||
candidate = locator.nth(index)
|
||||
if await candidate.is_visible():
|
||||
return candidate
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
# [/DEF:ScreenshotService._find_first_visible_locator:Function]
|
||||
|
||||
# [DEF:ScreenshotService._iter_login_roots:Function]
|
||||
# @PURPOSE: Enumerate page and child frames where login controls may be rendered.
|
||||
# @PRE: page is a Playwright page-like object.
|
||||
# @POST: Returns ordered roots starting with main page followed by frames.
|
||||
def _iter_login_roots(self, page) -> List[Any]:
|
||||
roots = [page]
|
||||
page_frames = getattr(page, "frames", [])
|
||||
try:
|
||||
for frame in page_frames:
|
||||
if frame not in roots:
|
||||
roots.append(frame)
|
||||
except Exception:
|
||||
pass
|
||||
return roots
|
||||
# [/DEF:ScreenshotService._iter_login_roots:Function]
|
||||
|
||||
# [DEF:ScreenshotService._extract_hidden_login_fields:Function]
|
||||
# @PURPOSE: Collect hidden form fields required for direct login POST fallback.
|
||||
# @PRE: Login page is loaded.
|
||||
# @POST: Returns hidden input name/value mapping aggregated from page and child frames.
|
||||
async def _extract_hidden_login_fields(self, page) -> Dict[str, str]:
|
||||
hidden_fields: Dict[str, str] = {}
|
||||
for root in self._iter_login_roots(page):
|
||||
try:
|
||||
locator = root.locator("input[type='hidden'][name]")
|
||||
count = await locator.count()
|
||||
for index in range(count):
|
||||
candidate = locator.nth(index)
|
||||
field_name = str(await candidate.get_attribute("name") or "").strip()
|
||||
if not field_name or field_name in hidden_fields:
|
||||
continue
|
||||
hidden_fields[field_name] = str(await candidate.input_value()).strip()
|
||||
except Exception:
|
||||
continue
|
||||
return hidden_fields
|
||||
# [/DEF:ScreenshotService._extract_hidden_login_fields:Function]
|
||||
|
||||
# [DEF:ScreenshotService._extract_csrf_token:Function]
|
||||
# @PURPOSE: Resolve CSRF token value from main page or embedded login frame.
|
||||
# @PRE: Login page is loaded.
|
||||
# @POST: Returns first non-empty csrf token or empty string.
|
||||
async def _extract_csrf_token(self, page) -> str:
|
||||
hidden_fields = await self._extract_hidden_login_fields(page)
|
||||
return str(hidden_fields.get("csrf_token") or "").strip()
|
||||
# [/DEF:ScreenshotService._extract_csrf_token:Function]
|
||||
|
||||
# [DEF:ScreenshotService._response_looks_like_login_page:Function]
|
||||
# @PURPOSE: Detect when fallback login POST returned the login form again instead of an authenticated page.
|
||||
# @PRE: response_text is normalized HTML or text from login POST response.
|
||||
# @POST: Returns True when login-page markers dominate the response body.
|
||||
def _response_looks_like_login_page(self, response_text: str) -> bool:
|
||||
normalized = str(response_text or "").strip().lower()
|
||||
if not normalized:
|
||||
return False
|
||||
|
||||
markers = [
|
||||
"enter your login and password below",
|
||||
"username:",
|
||||
"password:",
|
||||
"sign in",
|
||||
'name="csrf_token"',
|
||||
]
|
||||
return sum(marker in normalized for marker in markers) >= 3
|
||||
# [/DEF:ScreenshotService._response_looks_like_login_page:Function]
|
||||
|
||||
# [DEF:ScreenshotService._redirect_looks_authenticated:Function]
|
||||
# @PURPOSE: Treat non-login redirects after form POST as successful authentication without waiting for redirect target.
|
||||
# @PRE: redirect_location may be empty or relative.
|
||||
# @POST: Returns True when redirect target does not point back to login flow.
|
||||
def _redirect_looks_authenticated(self, redirect_location: str) -> bool:
|
||||
normalized = str(redirect_location or "").strip().lower()
|
||||
if not normalized:
|
||||
return True
|
||||
return "/login" not in normalized
|
||||
# [/DEF:ScreenshotService._redirect_looks_authenticated:Function]
|
||||
|
||||
# [DEF:ScreenshotService._submit_login_via_form_post:Function]
|
||||
# @PURPOSE: Fallback login path that submits credentials directly with csrf token.
|
||||
# @PRE: login_url is same-origin and csrf token can be read from DOM.
|
||||
# @POST: Browser context receives authenticated cookies when login succeeds.
|
||||
async def _submit_login_via_form_post(self, page, login_url: str) -> bool:
|
||||
hidden_fields = await self._extract_hidden_login_fields(page)
|
||||
csrf_token = str(hidden_fields.get("csrf_token") or "").strip()
|
||||
if not csrf_token:
|
||||
logger.warning("[DEBUG] Direct form login fallback skipped: csrf_token not found")
|
||||
return False
|
||||
|
||||
try:
|
||||
request_context = page.context.request
|
||||
except Exception as context_error:
|
||||
logger.warning(f"[DEBUG] Direct form login fallback skipped: request context unavailable: {context_error}")
|
||||
return False
|
||||
|
||||
parsed_url = urlsplit(login_url)
|
||||
origin = f"{parsed_url.scheme}://{parsed_url.netloc}" if parsed_url.scheme and parsed_url.netloc else login_url
|
||||
payload = dict(hidden_fields)
|
||||
payload["username"] = self.env.username
|
||||
payload["password"] = self.env.password
|
||||
logger.info(
|
||||
f"[DEBUG] Attempting direct form login fallback via browser context request with hidden fields: {sorted(hidden_fields.keys())}"
|
||||
)
|
||||
|
||||
response = await request_context.post(
|
||||
login_url,
|
||||
form=payload,
|
||||
headers={
|
||||
"Origin": origin,
|
||||
"Referer": login_url,
|
||||
},
|
||||
timeout=10000,
|
||||
fail_on_status_code=False,
|
||||
max_redirects=0,
|
||||
)
|
||||
response_url = str(getattr(response, "url", "") or "")
|
||||
response_status = int(getattr(response, "status", 0) or 0)
|
||||
response_headers = dict(getattr(response, "headers", {}) or {})
|
||||
redirect_location = str(
|
||||
response_headers.get("location")
|
||||
or response_headers.get("Location")
|
||||
or ""
|
||||
).strip()
|
||||
redirect_statuses = {301, 302, 303, 307, 308}
|
||||
if response_status in redirect_statuses:
|
||||
redirect_authenticated = self._redirect_looks_authenticated(redirect_location)
|
||||
logger.info(
|
||||
f"[DEBUG] Direct form login fallback redirect response: status={response_status} url={response_url} location={redirect_location!r} authenticated={redirect_authenticated}"
|
||||
)
|
||||
return redirect_authenticated
|
||||
|
||||
response_text = await response.text()
|
||||
text_snippet = " ".join(response_text.split())[:200]
|
||||
looks_like_login_page = self._response_looks_like_login_page(response_text)
|
||||
logger.info(
|
||||
f"[DEBUG] Direct form login fallback response: status={response_status} url={response_url} login_markup={looks_like_login_page} snippet={text_snippet!r}"
|
||||
)
|
||||
return not looks_like_login_page
|
||||
# [/DEF:ScreenshotService._submit_login_via_form_post:Function]
|
||||
|
||||
# [DEF:ScreenshotService._find_login_field_locator:Function]
|
||||
# @PURPOSE: Resolve login form input using semantic label text plus generic visible-input fallbacks.
|
||||
# @PRE: field_name is `username` or `password`.
|
||||
# @POST: Returns a locator for the corresponding input or None.
|
||||
async def _find_login_field_locator(self, page, field_name: str) -> Any:
|
||||
normalized = str(field_name or "").strip().lower()
|
||||
for root in self._iter_login_roots(page):
|
||||
if normalized == "username":
|
||||
input_candidates = [
|
||||
root.get_by_label("Username", exact=False),
|
||||
root.get_by_label("Login", exact=False),
|
||||
root.locator("label:text-matches('Username|Login', 'i')").locator("xpath=following::input[1]"),
|
||||
root.locator("text=/Username|Login/i").locator("xpath=following::input[1]"),
|
||||
root.locator("input[name='username']"),
|
||||
root.locator("input#username"),
|
||||
root.locator("input[placeholder*='Username']"),
|
||||
root.locator("input[type='text']"),
|
||||
root.locator("input:not([type='password'])"),
|
||||
]
|
||||
locator = await self._find_first_visible_locator(input_candidates)
|
||||
if locator:
|
||||
return locator
|
||||
|
||||
if normalized == "password":
|
||||
input_candidates = [
|
||||
root.get_by_label("Password", exact=False),
|
||||
root.locator("label:text-matches('Password', 'i')").locator("xpath=following::input[1]"),
|
||||
root.locator("text=/Password/i").locator("xpath=following::input[1]"),
|
||||
root.locator("input[name='password']"),
|
||||
root.locator("input#password"),
|
||||
root.locator("input[placeholder*='Password']"),
|
||||
root.locator("input[type='password']"),
|
||||
]
|
||||
locator = await self._find_first_visible_locator(input_candidates)
|
||||
if locator:
|
||||
return locator
|
||||
|
||||
return None
|
||||
# [/DEF:ScreenshotService._find_login_field_locator:Function]
|
||||
|
||||
# [DEF:ScreenshotService._find_submit_locator:Function]
|
||||
# @PURPOSE: Resolve login submit button from main page or embedded auth frame.
|
||||
# @PRE: page is ready for login interaction.
|
||||
# @POST: Returns visible submit locator or None.
|
||||
async def _find_submit_locator(self, page) -> Any:
|
||||
selectors = [
|
||||
lambda root: root.get_by_role("button", name="Sign in", exact=False),
|
||||
lambda root: root.get_by_role("button", name="Login", exact=False),
|
||||
lambda root: root.locator("button[type='submit']"),
|
||||
lambda root: root.locator("button#submit"),
|
||||
lambda root: root.locator(".btn-primary"),
|
||||
lambda root: root.locator("input[type='submit']"),
|
||||
]
|
||||
for root in self._iter_login_roots(page):
|
||||
locator = await self._find_first_visible_locator([factory(root) for factory in selectors])
|
||||
if locator:
|
||||
return locator
|
||||
return None
|
||||
# [/DEF:ScreenshotService._find_submit_locator:Function]
|
||||
|
||||
# [DEF:ScreenshotService._goto_resilient:Function]
|
||||
# @PURPOSE: Navigate without relying on networkidle for pages with long-polling or persistent requests.
|
||||
# @PRE: page is a valid Playwright page and url is non-empty.
|
||||
# @POST: Returns last navigation response or raises when both primary and fallback waits fail.
|
||||
async def _goto_resilient(
|
||||
self,
|
||||
page,
|
||||
url: str,
|
||||
primary_wait_until: str = "domcontentloaded",
|
||||
fallback_wait_until: str = "load",
|
||||
timeout: int = 60000,
|
||||
):
|
||||
try:
|
||||
return await page.goto(url, wait_until=primary_wait_until, timeout=timeout)
|
||||
except Exception as primary_error:
|
||||
logger.warning(
|
||||
f"[ScreenshotService._goto_resilient] Primary navigation wait '{primary_wait_until}' failed for {url}: {primary_error}"
|
||||
)
|
||||
return await page.goto(url, wait_until=fallback_wait_until, timeout=timeout)
|
||||
# [/DEF:ScreenshotService._goto_resilient:Function]
|
||||
|
||||
# [DEF:ScreenshotService.capture_dashboard:Function]
|
||||
# @PURPOSE: Captures a full-page screenshot of a dashboard using Playwright and CDP.
|
||||
# @PRE: dashboard_id is a valid string, output_path is a writable path.
|
||||
@@ -85,7 +323,13 @@ class ScreenshotService:
|
||||
login_url = f"{base_ui_url.rstrip('/')}/login/"
|
||||
logger.info(f"[DEBUG] Navigating to login page: {login_url}")
|
||||
|
||||
response = await page.goto(login_url, wait_until="networkidle", timeout=60000)
|
||||
response = await self._goto_resilient(
|
||||
page,
|
||||
login_url,
|
||||
primary_wait_until="domcontentloaded",
|
||||
fallback_wait_until="load",
|
||||
timeout=60000,
|
||||
)
|
||||
if response:
|
||||
logger.info(f"[DEBUG] Login page response status: {response.status}")
|
||||
|
||||
@@ -101,57 +345,59 @@ class ScreenshotService:
|
||||
logger.info("[DEBUG] Attempting to find login form elements...")
|
||||
|
||||
try:
|
||||
used_direct_form_login = False
|
||||
# Find and fill username
|
||||
u_selector = None
|
||||
for s in selectors["username"]:
|
||||
count = await page.locator(s).count()
|
||||
logger.info(f"[DEBUG] Selector '{s}': {count} elements found")
|
||||
if count > 0:
|
||||
u_selector = s
|
||||
break
|
||||
username_locator = await self._find_login_field_locator(page, "username")
|
||||
|
||||
if not username_locator:
|
||||
roots = self._iter_login_roots(page)
|
||||
logger.info(f"[DEBUG] Found {len(roots)} login roots including child frames")
|
||||
for root_index, root in enumerate(roots[:5]):
|
||||
all_inputs = await root.locator('input').all()
|
||||
logger.info(f"[DEBUG] Root {root_index}: found {len(all_inputs)} input fields")
|
||||
for i, inp in enumerate(all_inputs[:5]): # Log first 5
|
||||
inp_type = await inp.get_attribute('type')
|
||||
inp_name = await inp.get_attribute('name')
|
||||
inp_id = await inp.get_attribute('id')
|
||||
logger.info(f"[DEBUG] Root {root_index} input {i}: type={inp_type}, name={inp_name}, id={inp_id}")
|
||||
used_direct_form_login = await self._submit_login_via_form_post(page, login_url)
|
||||
if not used_direct_form_login:
|
||||
raise RuntimeError("Could not find username input field on login page")
|
||||
username_locator = None
|
||||
|
||||
if not u_selector:
|
||||
# Log all input fields on the page for debugging
|
||||
all_inputs = await page.locator('input').all()
|
||||
logger.info(f"[DEBUG] Found {len(all_inputs)} input fields on page")
|
||||
for i, inp in enumerate(all_inputs[:5]): # Log first 5
|
||||
inp_type = await inp.get_attribute('type')
|
||||
inp_name = await inp.get_attribute('name')
|
||||
inp_id = await inp.get_attribute('id')
|
||||
logger.info(f"[DEBUG] Input {i}: type={inp_type}, name={inp_name}, id={inp_id}")
|
||||
raise RuntimeError("Could not find username input field on login page")
|
||||
|
||||
logger.info(f"[DEBUG] Filling username field with selector: {u_selector}")
|
||||
await page.fill(u_selector, self.env.username)
|
||||
if username_locator is not None:
|
||||
logger.info("[DEBUG] Filling username field")
|
||||
await username_locator.fill(self.env.username)
|
||||
|
||||
# Find and fill password
|
||||
p_selector = None
|
||||
for s in selectors["password"]:
|
||||
if await page.locator(s).count() > 0:
|
||||
p_selector = s
|
||||
break
|
||||
|
||||
if not p_selector:
|
||||
password_locator = await self._find_login_field_locator(page, "password") if username_locator is not None else None
|
||||
|
||||
if username_locator is not None and not password_locator:
|
||||
raise RuntimeError("Could not find password input field on login page")
|
||||
|
||||
logger.info(f"[DEBUG] Filling password field with selector: {p_selector}")
|
||||
await page.fill(p_selector, self.env.password)
|
||||
if password_locator is not None:
|
||||
logger.info("[DEBUG] Filling password field")
|
||||
await password_locator.fill(self.env.password)
|
||||
|
||||
# Click submit
|
||||
s_selector = selectors["submit"][0]
|
||||
for s in selectors["submit"]:
|
||||
if await page.locator(s).count() > 0:
|
||||
s_selector = s
|
||||
break
|
||||
|
||||
logger.info(f"[DEBUG] Clicking submit button with selector: {s_selector}")
|
||||
await page.click(s_selector)
|
||||
submit_locator = await self._find_submit_locator(page) if username_locator is not None else None
|
||||
|
||||
if username_locator is not None and not submit_locator:
|
||||
raise RuntimeError("Could not find submit button on login page")
|
||||
|
||||
if submit_locator is not None:
|
||||
logger.info("[DEBUG] Clicking submit button")
|
||||
await submit_locator.click()
|
||||
|
||||
# Wait for navigation after login
|
||||
await page.wait_for_load_state("networkidle", timeout=30000)
|
||||
if not used_direct_form_login:
|
||||
try:
|
||||
await page.wait_for_load_state("load", timeout=30000)
|
||||
except Exception as load_wait_error:
|
||||
logger.warning(f"[DEBUG] Login post-submit load wait timed out: {load_wait_error}")
|
||||
|
||||
# Check if login was successful
|
||||
if "/login" in page.url:
|
||||
if not used_direct_form_login and "/login" in page.url:
|
||||
# Check for error messages on page
|
||||
error_msg = await page.locator(".alert-danger, .error-message").text_content() if await page.locator(".alert-danger, .error-message").count() > 0 else "Unknown error"
|
||||
logger.error(f"[DEBUG] Login failed. Still on login page. Error: {error_msg}")
|
||||
@@ -183,11 +429,24 @@ class ScreenshotService:
|
||||
|
||||
logger.info(f"[DEBUG] Navigating to dashboard: {dashboard_url}")
|
||||
|
||||
# Use networkidle to ensure all initial assets are loaded
|
||||
response = await page.goto(dashboard_url, wait_until="networkidle", timeout=60000)
|
||||
# Dashboard pages can keep polling/network activity open indefinitely.
|
||||
response = await self._goto_resilient(
|
||||
page,
|
||||
dashboard_url,
|
||||
primary_wait_until="domcontentloaded",
|
||||
fallback_wait_until="load",
|
||||
timeout=60000,
|
||||
)
|
||||
|
||||
if response:
|
||||
logger.info(f"[DEBUG] Dashboard navigation response status: {response.status}, URL: {response.url}")
|
||||
|
||||
if "/login" in page.url:
|
||||
debug_path = output_path.replace(".png", "_debug_failed_dashboard_auth.png")
|
||||
await page.screenshot(path=debug_path)
|
||||
raise RuntimeError(
|
||||
f"Dashboard navigation redirected to login page after authentication. Debug screenshot saved to {debug_path}"
|
||||
)
|
||||
|
||||
try:
|
||||
# Wait for the dashboard grid to be present
|
||||
@@ -440,6 +699,13 @@ class LLMClient:
|
||||
|
||||
# Some OpenAI-compatible gateways are strict about auth header naming.
|
||||
default_headers = {"Authorization": f"Bearer {self.api_key}"}
|
||||
if self.provider_type == LLMProviderType.OPENROUTER:
|
||||
default_headers["HTTP-Referer"] = (
|
||||
os.getenv("OPENROUTER_SITE_URL", "").strip()
|
||||
or os.getenv("APP_BASE_URL", "").strip()
|
||||
or "http://localhost:8000"
|
||||
)
|
||||
default_headers["X-Title"] = os.getenv("OPENROUTER_APP_NAME", "").strip() or "ss-tools"
|
||||
if self.provider_type == LLMProviderType.KILO:
|
||||
default_headers["Authentication"] = f"Bearer {self.api_key}"
|
||||
default_headers["X-API-Key"] = self.api_key
|
||||
@@ -595,6 +861,22 @@ class LLMClient:
|
||||
raise
|
||||
# [/DEF:LLMClient.get_json_completion:Function]
|
||||
|
||||
# [DEF:LLMClient.test_runtime_connection:Function]
|
||||
# @PURPOSE: Validate provider credentials using the same chat completions transport as runtime analysis.
|
||||
# @PRE: Client is initialized with provider credentials and default_model.
|
||||
# @POST: Returns lightweight JSON payload when runtime auth/model path is valid.
|
||||
# @SIDE_EFFECT: Calls external LLM API.
|
||||
async def test_runtime_connection(self) -> Dict[str, Any]:
|
||||
with belief_scope("test_runtime_connection"):
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": 'Return exactly this JSON object and nothing else: {"ok": true}',
|
||||
}
|
||||
]
|
||||
return await self.get_json_completion(messages)
|
||||
# [/DEF:LLMClient.test_runtime_connection:Function]
|
||||
|
||||
# [DEF:LLMClient.analyze_dashboard:Function]
|
||||
# @PURPOSE: Sends dashboard data (screenshot + logs) to LLM for health analysis.
|
||||
# @PRE: screenshot_path exists, logs is a list of strings.
|
||||
@@ -661,9 +943,9 @@ class LLMClient:
|
||||
except Exception as e:
|
||||
logger.error(f"[analyze_dashboard] Failed to get analysis: {str(e)}")
|
||||
return {
|
||||
"status": "FAIL",
|
||||
"status": "UNKNOWN",
|
||||
"summary": f"Failed to get response from LLM: {str(e)}",
|
||||
"issues": [{"severity": "FAIL", "message": "LLM provider returned empty or invalid response"}]
|
||||
"issues": [{"severity": "UNKNOWN", "message": "LLM provider returned empty or invalid response"}]
|
||||
}
|
||||
# [/DEF:LLMClient.analyze_dashboard:Function]
|
||||
# [/DEF:LLMClient:Class]
|
||||
|
||||
Reference in New Issue
Block a user