mirror of
https://github.com/rookiestar28/ComfyUI-OpenClaw.git
synced 2026-08-14 08:52:45 +00:00
fix(connector): render bounded jobs summaries
This commit is contained in:
rookiestar28
@@ -0,0 +1,183 @@
|
||||
"""Strict, bounded formatter for the connector's authoritative jobs view."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections import Counter
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
JOBS_CONTRACT_VERSION = 1
|
||||
MAX_RETURNED_JOBS = 200
|
||||
MAX_SNAPSHOT_TOTAL = 10_000
|
||||
MAX_JOB_ID_LENGTH = 128
|
||||
MAX_DISPLAY_JOB_ID_LENGTH = 24
|
||||
MAX_JOBS_SUMMARY_LENGTH = 1_000
|
||||
MAX_QUEUE_REMAINING = 1_000_000
|
||||
MAX_NORMALIZATION_WARNINGS = 2
|
||||
|
||||
JOB_STATUSES = (
|
||||
"pending",
|
||||
"in_progress",
|
||||
"completed",
|
||||
"failed",
|
||||
"cancelled",
|
||||
)
|
||||
STATUS_LABELS = {
|
||||
"pending": "pending",
|
||||
"in_progress": "in progress",
|
||||
"completed": "completed",
|
||||
"failed": "failed",
|
||||
"cancelled": "cancelled",
|
||||
}
|
||||
_SAFE_JOB_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
||||
|
||||
|
||||
class JobsContractError(ValueError):
|
||||
"""Raised when a connector jobs payload is not safe to render."""
|
||||
|
||||
|
||||
def format_jobs_summary(payload: Any) -> str:
|
||||
"""Validate contract version 1 and return a deterministic operator summary."""
|
||||
|
||||
jobs, pagination = _parse_jobs_payload(payload)
|
||||
total = pagination["total"]
|
||||
if total == 0:
|
||||
return "[Jobs] No jobs in the authoritative snapshot."
|
||||
|
||||
counts = Counter(job["status"] for job in jobs)
|
||||
active = counts["pending"] + counts["in_progress"]
|
||||
terminal = counts["completed"] + counts["failed"] + counts["cancelled"]
|
||||
lines = [
|
||||
"[Jobs] Authoritative snapshot",
|
||||
f"Snapshot total: {total}; returned page: {len(jobs)}",
|
||||
(
|
||||
f"Page states: Active {active} (pending {counts['pending']}, "
|
||||
f"in progress {counts['in_progress']}); "
|
||||
f"Terminal {terminal} (completed {counts['completed']}, "
|
||||
f"failed {counts['failed']}, cancelled {counts['cancelled']})"
|
||||
),
|
||||
]
|
||||
if jobs:
|
||||
for job in jobs[:5]:
|
||||
lines.append(
|
||||
f"- {_short_job_id(job['id'])} — {STATUS_LABELS[job['status']]}"
|
||||
)
|
||||
if len(jobs) > 5:
|
||||
lines.append(f"Showing 5 of {len(jobs)} returned jobs.")
|
||||
else:
|
||||
lines.append("No jobs are present on this page.")
|
||||
|
||||
summary = "\n".join(lines)
|
||||
if len(summary) > MAX_JOBS_SUMMARY_LENGTH:
|
||||
raise JobsContractError("jobs summary exceeds the safe display bound")
|
||||
return summary
|
||||
|
||||
|
||||
def format_queue_fallback(response: Any) -> str:
|
||||
"""Render only a bounded coarse queue count from the legacy fallback seam."""
|
||||
|
||||
remaining = _queue_remaining(response)
|
||||
if remaining is None:
|
||||
return "[Jobs fallback] Coarse queue count is unavailable."
|
||||
return (
|
||||
f"[Jobs fallback] Queue remaining: {remaining} "
|
||||
"(coarse count; not an authoritative jobs snapshot)."
|
||||
)
|
||||
|
||||
|
||||
def _parse_jobs_payload(payload: Any) -> tuple[list[dict[str, str]], dict[str, Any]]:
|
||||
if not isinstance(payload, Mapping) or payload.get("ok") is not True:
|
||||
raise JobsContractError("jobs response must be a successful mapping")
|
||||
version = payload.get("contract_version")
|
||||
if isinstance(version, bool) or version != JOBS_CONTRACT_VERSION:
|
||||
raise JobsContractError("unsupported jobs contract version")
|
||||
raw_jobs = payload.get("jobs")
|
||||
pagination = payload.get("pagination")
|
||||
if not isinstance(raw_jobs, list) or len(raw_jobs) > MAX_RETURNED_JOBS:
|
||||
raise JobsContractError("jobs list is malformed or oversized")
|
||||
if not isinstance(pagination, Mapping):
|
||||
raise JobsContractError("jobs pagination is missing")
|
||||
if not isinstance(payload.get("source"), Mapping) or not isinstance(
|
||||
payload.get("scan"), Mapping
|
||||
):
|
||||
raise JobsContractError("jobs source diagnostics are missing")
|
||||
|
||||
parsed_jobs = [_parse_job(item) for item in raw_jobs]
|
||||
parsed_pagination = _parse_pagination(pagination, returned=len(parsed_jobs))
|
||||
return parsed_jobs, parsed_pagination
|
||||
|
||||
|
||||
def _parse_job(item: Any) -> dict[str, str]:
|
||||
if not isinstance(item, Mapping):
|
||||
raise JobsContractError("job summary must be a mapping")
|
||||
job_id = item.get("id")
|
||||
status = item.get("status")
|
||||
if (
|
||||
not isinstance(job_id, str)
|
||||
or not job_id
|
||||
or len(job_id) > MAX_JOB_ID_LENGTH
|
||||
or _SAFE_JOB_ID.fullmatch(job_id) is None
|
||||
):
|
||||
raise JobsContractError("job id is outside the safe display contract")
|
||||
if not isinstance(status, str) or status not in JOB_STATUSES:
|
||||
raise JobsContractError("job status is unsupported")
|
||||
return {"id": job_id, "status": status}
|
||||
|
||||
|
||||
def _parse_pagination(
|
||||
pagination: Mapping[str, Any], *, returned: int
|
||||
) -> dict[str, Any]:
|
||||
offset = _bounded_int(pagination.get("offset"), minimum=0, maximum=10_000)
|
||||
limit = _bounded_int(pagination.get("limit"), minimum=1, maximum=MAX_RETURNED_JOBS)
|
||||
total = _bounded_int(pagination.get("total"), minimum=0, maximum=MAX_SNAPSHOT_TOTAL)
|
||||
has_more = pagination.get("has_more")
|
||||
warnings = pagination.get("warnings")
|
||||
if not isinstance(has_more, bool):
|
||||
raise JobsContractError("jobs has_more must be boolean")
|
||||
if not isinstance(warnings, list) or len(warnings) > MAX_NORMALIZATION_WARNINGS:
|
||||
raise JobsContractError("jobs warnings are malformed")
|
||||
if returned > limit or offset + returned > total:
|
||||
raise JobsContractError("jobs pagination counts are inconsistent")
|
||||
if has_more != (offset + returned < total):
|
||||
raise JobsContractError("jobs has_more is inconsistent")
|
||||
return {
|
||||
"offset": offset,
|
||||
"limit": limit,
|
||||
"total": total,
|
||||
"has_more": has_more,
|
||||
}
|
||||
|
||||
|
||||
def _bounded_int(value: Any, *, minimum: int, maximum: int) -> int:
|
||||
if isinstance(value, bool) or not isinstance(value, int):
|
||||
raise JobsContractError("jobs count must be an integer")
|
||||
if value < minimum or value > maximum:
|
||||
raise JobsContractError("jobs count is outside the safe bound")
|
||||
return value
|
||||
|
||||
|
||||
def _short_job_id(job_id: str) -> str:
|
||||
if len(job_id) <= MAX_DISPLAY_JOB_ID_LENGTH:
|
||||
return job_id
|
||||
return job_id[: MAX_DISPLAY_JOB_ID_LENGTH - 3] + "..."
|
||||
|
||||
|
||||
def _queue_remaining(response: Any) -> int | None:
|
||||
if not isinstance(response, Mapping) or response.get("ok") is not True:
|
||||
return None
|
||||
data = response.get("data")
|
||||
if not isinstance(data, Mapping):
|
||||
return None
|
||||
exec_info = data.get("exec_info")
|
||||
if not isinstance(exec_info, Mapping):
|
||||
return None
|
||||
remaining = exec_info.get("queue_remaining")
|
||||
if (
|
||||
isinstance(remaining, bool)
|
||||
or not isinstance(remaining, int)
|
||||
or remaining < 0
|
||||
or remaining > MAX_QUEUE_REMAINING
|
||||
):
|
||||
return None
|
||||
return remaining
|
||||
@@ -15,7 +15,7 @@ CHAT_SYSTEM_PROMPT = """You are OpenClaw Assistant, an AI helper for controlling
|
||||
**Available Commands (for reference):**
|
||||
- `/run <template_id> [--input key=value ...]` - Execute a workflow template
|
||||
- `/status` - Check system status
|
||||
- `/jobs` - View queue
|
||||
- `/jobs` - View the authoritative jobs summary (admin)
|
||||
- `/approvals` - List pending approvals (admin)
|
||||
- `/approve <id>` - Approve a request (admin)
|
||||
|
||||
|
||||
+39
-15
@@ -5,6 +5,7 @@ Dispatches parsed commands to handlers with AST argument parsing.
|
||||
|
||||
import logging
|
||||
import shlex
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from .config import CommandClass, ConnectorConfig
|
||||
@@ -16,6 +17,7 @@ if False: # Type hinting only
|
||||
from .results_poller import ResultsPoller
|
||||
|
||||
from .command_firewall import CommandFirewall
|
||||
from .jobs_summary import JobsContractError, format_jobs_summary, format_queue_fallback
|
||||
from .llm_client import LLMClient
|
||||
from .prompts import CHAT_STATUS_PROMPT, CHAT_SYSTEM_PROMPT
|
||||
from .rate_limiter import RateLimiter
|
||||
@@ -141,7 +143,7 @@ class CommandRouter:
|
||||
# Phase 3 Introspection
|
||||
("/history", "history"): (self._handle_history, CommandClass.PUBLIC),
|
||||
("/trace", "trace"): (self._handle_trace, CommandClass.ADMIN), # Admin only
|
||||
("/jobs", "jobs", "queue"): (self._handle_jobs, CommandClass.PUBLIC),
|
||||
("/jobs", "jobs", "queue"): (self._handle_jobs, CommandClass.ADMIN),
|
||||
# F30: Chat Assistant
|
||||
("/chat", "chat"): (self._handle_chat, CommandClass.PUBLIC),
|
||||
}
|
||||
@@ -736,8 +738,8 @@ class CommandRouter:
|
||||
"/run <template> [prompt] [k=v] - Run a generation (trusted users auto-exec; others require approval)\n"
|
||||
"/stop [job_id ...] - Cancel jobs by id; no args sends Global Interrupt (Admin)\n"
|
||||
"/history <id> - Job details\n"
|
||||
"/jobs - Queue summary\n"
|
||||
"Admin Only:\n"
|
||||
"/jobs - Authoritative jobs summary\n"
|
||||
"/approvals - List pending approvals\n"
|
||||
"/approve <id>, /reject <id>\n"
|
||||
"/schedules, /schedule run <id>\n"
|
||||
@@ -783,21 +785,44 @@ class CommandRouter:
|
||||
async def _handle_jobs(
|
||||
self, req: CommandRequest, args: List[str]
|
||||
) -> CommandResponse:
|
||||
# Try native /openclaw/jobs first
|
||||
if err := self._require_admin_token_configured():
|
||||
return err
|
||||
|
||||
res = await self.client.get_jobs()
|
||||
if res.get("ok"):
|
||||
# Format nice summary
|
||||
if not isinstance(res, Mapping):
|
||||
return CommandResponse(
|
||||
text=f"Default Jobs View: {sanitize_operator_payload(res.get('data'))}"
|
||||
text="[Jobs] Could not fetch the authoritative jobs snapshot."
|
||||
)
|
||||
if res.get("ok") is True:
|
||||
try:
|
||||
return CommandResponse(text=format_jobs_summary(res.get("data")))
|
||||
except JobsContractError:
|
||||
return CommandResponse(
|
||||
text="[Jobs] Malformed or unsupported jobs response."
|
||||
)
|
||||
|
||||
# Fallback: Queue
|
||||
q = await self.client.get_prompt_queue()
|
||||
if q.get("ok"):
|
||||
rem = q.get("data", {}).get("exec_info", {}).get("queue_remaining", "?")
|
||||
return CommandResponse(text=f"[Fallback] Queue Remaining: {rem}")
|
||||
|
||||
return CommandResponse(text="[Error] Could not fetch jobs or queue.")
|
||||
status = res.get("status")
|
||||
error = res.get("error")
|
||||
access_denied = (
|
||||
isinstance(status, int)
|
||||
and not isinstance(status, bool)
|
||||
and status in {401, 403}
|
||||
)
|
||||
if access_denied:
|
||||
return CommandResponse(
|
||||
text="[Jobs] Access denied. Check connector Admin authorization and token posture."
|
||||
)
|
||||
fallback_allowed = isinstance(error, str) and (
|
||||
(status == 501 and error == "jobs_host_contract_unsupported")
|
||||
or (status == 503 and error == "jobs_backend_unavailable")
|
||||
)
|
||||
if fallback_allowed:
|
||||
return CommandResponse(
|
||||
text=format_queue_fallback(await self.client.get_prompt_queue())
|
||||
)
|
||||
return CommandResponse(
|
||||
text="[Jobs] Could not fetch the authoritative jobs snapshot."
|
||||
)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# F30: Chat LLM Assistant
|
||||
@@ -1020,12 +1045,11 @@ Keep it minimal."""
|
||||
"""Summarize system status using LLM."""
|
||||
# Fetch status data
|
||||
health = await self.client.get_health()
|
||||
jobs = await self.client.get_jobs()
|
||||
queue = await self.client.get_prompt_queue()
|
||||
|
||||
status_data = {
|
||||
"health": health.get("data", {}) if health.get("ok") else "unavailable",
|
||||
"jobs": jobs.get("data", {}) if jobs.get("ok") else "unavailable",
|
||||
"jobs": "admin-only; use /jobs as an authorized operator",
|
||||
"queue": queue.get("data", {}) if queue.get("ok") else "unavailable",
|
||||
}
|
||||
|
||||
|
||||
@@ -54,10 +54,27 @@ class TestCommandRouterPhase3(unittest.TestCase):
|
||||
self.client.get_history.assert_called_with("p1")
|
||||
|
||||
def test_jobs(self):
|
||||
# Public
|
||||
req = self._req("/jobs", sender="123")
|
||||
# Jobs expose cross-job metadata and are connector-admin only.
|
||||
self.client.get_jobs.return_value = {
|
||||
"ok": True,
|
||||
"data": {
|
||||
"ok": True,
|
||||
"contract_version": 1,
|
||||
"jobs": [],
|
||||
"pagination": {
|
||||
"offset": 0,
|
||||
"limit": 50,
|
||||
"total": 0,
|
||||
"has_more": False,
|
||||
"warnings": [],
|
||||
},
|
||||
"source": {},
|
||||
"scan": {},
|
||||
},
|
||||
}
|
||||
req = self._req("/jobs", sender="999")
|
||||
resp = asyncio.run(self.router.handle(req))
|
||||
self.assertIn("5", resp.text)
|
||||
self.assertIn("No jobs in the authoritative snapshot", resp.text)
|
||||
self.client.get_jobs.assert_called_once()
|
||||
|
||||
def test_trace_admin(self):
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
"""Connector jobs v1 parsing, summary, fallback, and authorization contracts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from connector.config import ConnectorConfig
|
||||
from connector.contract import CommandRequest
|
||||
from connector.router import CommandRouter
|
||||
|
||||
|
||||
def _request(text: str = "/jobs", *, sender: str = "admin-user") -> CommandRequest:
|
||||
return CommandRequest(
|
||||
platform="test",
|
||||
sender_id=sender,
|
||||
channel_id="channel-1",
|
||||
username="operator",
|
||||
message_id="message-1",
|
||||
text=text,
|
||||
timestamp=0,
|
||||
)
|
||||
|
||||
|
||||
def _job(job_id: str, status: str, **hostile) -> dict:
|
||||
return {"id": job_id, "status": status, **hostile}
|
||||
|
||||
|
||||
def _success(jobs: list[dict], *, total: int | None = None, version=1) -> dict:
|
||||
body = {
|
||||
"ok": True,
|
||||
"contract_version": version,
|
||||
"jobs": jobs,
|
||||
"pagination": {
|
||||
"offset": 0,
|
||||
"limit": 50,
|
||||
"total": len(jobs) if total is None else total,
|
||||
"has_more": (len(jobs) if total is None else total) > len(jobs),
|
||||
"warnings": [],
|
||||
},
|
||||
"source": {"adapter": "comfy_execution.jobs", "authority": "in_process"},
|
||||
"scan": {
|
||||
"window": 10000,
|
||||
"examined": len(jobs),
|
||||
"excluded": 0,
|
||||
"malformed": 0,
|
||||
"truncated": False,
|
||||
},
|
||||
}
|
||||
return {"ok": True, "status": 200, "data": body}
|
||||
|
||||
|
||||
class TestConnectorJobsCommand(unittest.TestCase):
|
||||
def setUp(self):
|
||||
config = ConnectorConfig()
|
||||
config.admin_users = ["admin-user"]
|
||||
config.admin_token = "configured-admin-token"
|
||||
self.client = MagicMock()
|
||||
self.client.get_prompt_queue = AsyncMock(
|
||||
return_value={
|
||||
"ok": True,
|
||||
"status": 200,
|
||||
"data": {"exec_info": {"queue_remaining": 7}},
|
||||
}
|
||||
)
|
||||
self.client.get_jobs = AsyncMock(return_value=_success([]))
|
||||
self.router = CommandRouter(config, self.client)
|
||||
|
||||
def _run(self, text: str = "/jobs", *, sender: str = "admin-user"):
|
||||
return asyncio.run(self.router.handle(_request(text, sender=sender)))
|
||||
|
||||
def test_active_terminal_mix_is_bounded_and_never_dumps_raw_payload(self):
|
||||
jobs = [
|
||||
_job(
|
||||
"job-pending",
|
||||
"pending",
|
||||
workflow_id="secret-workflow",
|
||||
preview_output={"filename": "secret.png"},
|
||||
error="secret-error",
|
||||
tenant_id="secret-tenant",
|
||||
trace_id="secret-trace",
|
||||
prompt="secret-prompt",
|
||||
reasoning="secret-reasoning",
|
||||
internal="secret-internal",
|
||||
),
|
||||
_job("job-running", "in_progress"),
|
||||
_job("job-completed", "completed"),
|
||||
_job("job-failed", "failed"),
|
||||
_job("job-cancelled", "cancelled"),
|
||||
_job("job-sixth", "completed"),
|
||||
]
|
||||
self.client.get_jobs.return_value = _success(jobs, total=12)
|
||||
|
||||
response = self._run()
|
||||
|
||||
self.assertIn("[Jobs] Authoritative snapshot", response.text)
|
||||
self.assertIn("Snapshot total: 12; returned page: 6", response.text)
|
||||
self.assertIn("Active 2 (pending 1, in progress 1)", response.text)
|
||||
self.assertIn("Terminal 4 (completed 2, failed 1, cancelled 1)", response.text)
|
||||
self.assertIn("- job-pending — pending", response.text)
|
||||
self.assertIn("- job-running — in progress", response.text)
|
||||
self.assertIn("Showing 5 of 6 returned jobs", response.text)
|
||||
self.assertNotIn("job-sixth", response.text)
|
||||
for secret in (
|
||||
"secret-workflow",
|
||||
"secret.png",
|
||||
"secret-error",
|
||||
"secret-tenant",
|
||||
"secret-trace",
|
||||
"secret-prompt",
|
||||
"secret-reasoning",
|
||||
"secret-internal",
|
||||
"workflow_id",
|
||||
"preview_output",
|
||||
):
|
||||
self.assertNotIn(secret, response.text)
|
||||
self.assertNotIn(str(jobs), response.text)
|
||||
self.assertNotIn("{'", response.text)
|
||||
self.assertLessEqual(len(response.text), 1000)
|
||||
self.client.get_prompt_queue.assert_not_called()
|
||||
|
||||
def test_authoritative_empty_is_distinct_and_never_falls_back(self):
|
||||
response = self._run()
|
||||
|
||||
self.assertEqual(response.text, "[Jobs] No jobs in the authoritative snapshot.")
|
||||
self.client.get_prompt_queue.assert_not_called()
|
||||
|
||||
def test_exact_unsupported_and_degraded_errors_use_labeled_coarse_fallback(self):
|
||||
cases = (
|
||||
(501, "jobs_host_contract_unsupported"),
|
||||
(503, "jobs_backend_unavailable"),
|
||||
)
|
||||
for status, error in cases:
|
||||
with self.subTest(status=status):
|
||||
self.client.get_jobs.return_value = {
|
||||
"ok": False,
|
||||
"status": status,
|
||||
"error": error,
|
||||
"data": {"ok": False, "error": error},
|
||||
}
|
||||
self.client.get_prompt_queue.reset_mock()
|
||||
|
||||
response = self._run()
|
||||
|
||||
self.assertEqual(
|
||||
response.text,
|
||||
"[Jobs fallback] Queue remaining: 7 "
|
||||
"(coarse count; not an authoritative jobs snapshot).",
|
||||
)
|
||||
self.client.get_prompt_queue.assert_awaited_once()
|
||||
|
||||
def test_access_denial_never_uses_queue_fallback_or_raw_error(self):
|
||||
self.client.get_jobs.return_value = {
|
||||
"ok": False,
|
||||
"status": 403,
|
||||
"error": "secret-auth-detail",
|
||||
"data": {"error": "secret-auth-detail"},
|
||||
}
|
||||
|
||||
response = self._run()
|
||||
|
||||
self.assertEqual(
|
||||
response.text,
|
||||
"[Jobs] Access denied. Check connector Admin authorization and token posture.",
|
||||
)
|
||||
self.assertNotIn("secret-auth-detail", response.text)
|
||||
self.client.get_prompt_queue.assert_not_called()
|
||||
|
||||
def test_transport_and_arbitrary_errors_never_fallback(self):
|
||||
cases = (
|
||||
{"ok": False, "error": "secret-network-failure"},
|
||||
{"ok": False, "status": 500, "error": "secret-server-failure"},
|
||||
{
|
||||
"ok": False,
|
||||
"status": 501,
|
||||
"error": "some-other-unsupported-secret",
|
||||
},
|
||||
)
|
||||
for payload in cases:
|
||||
with self.subTest(payload=payload.get("status")):
|
||||
self.client.get_jobs.return_value = payload
|
||||
self.client.get_prompt_queue.reset_mock()
|
||||
response = self._run()
|
||||
self.assertEqual(
|
||||
response.text,
|
||||
"[Jobs] Could not fetch the authoritative jobs snapshot.",
|
||||
)
|
||||
self.assertNotIn("secret", response.text)
|
||||
self.client.get_prompt_queue.assert_not_called()
|
||||
|
||||
def test_unknown_or_malformed_success_is_content_free_and_does_not_fallback(self):
|
||||
oversized = _success([])
|
||||
oversized["data"]["jobs"] = [
|
||||
_job(f"job-{index}", "pending") for index in range(201)
|
||||
]
|
||||
bad_total = _success([])
|
||||
bad_total["data"]["pagination"]["total"] = True
|
||||
inconsistent = _success([_job("job", "pending")])
|
||||
inconsistent["data"]["pagination"]["has_more"] = True
|
||||
malformed = (
|
||||
_success([], version=2),
|
||||
{"ok": True, "status": 200, "data": {"contract_version": 1}},
|
||||
{
|
||||
"ok": True,
|
||||
"status": 200,
|
||||
"data": {
|
||||
"ok": True,
|
||||
"contract_version": 1,
|
||||
"jobs": "secret-not-a-list",
|
||||
"pagination": {},
|
||||
},
|
||||
},
|
||||
_success([_job("job", "unknown-secret-status")]),
|
||||
_success([_job("job\nsecret-control", "pending")]),
|
||||
_success([_job("<b>secret-markup</b>", "pending")]),
|
||||
oversized,
|
||||
bad_total,
|
||||
inconsistent,
|
||||
)
|
||||
for payload in malformed:
|
||||
with self.subTest(payload=json.dumps(payload, default=str)[:40]):
|
||||
self.client.get_jobs.return_value = payload
|
||||
self.client.get_prompt_queue.reset_mock()
|
||||
response = self._run()
|
||||
self.assertEqual(
|
||||
response.text,
|
||||
"[Jobs] Malformed or unsupported jobs response.",
|
||||
)
|
||||
self.assertNotIn("secret", response.text)
|
||||
self.client.get_prompt_queue.assert_not_called()
|
||||
|
||||
def test_malformed_queue_count_cannot_escape_fallback_boundary(self):
|
||||
self.client.get_jobs.return_value = {
|
||||
"ok": False,
|
||||
"status": 503,
|
||||
"error": "jobs_backend_unavailable",
|
||||
}
|
||||
for value in (True, -1, 1000001, "secret-unbounded"):
|
||||
with self.subTest(value=value):
|
||||
self.client.get_prompt_queue.return_value = {
|
||||
"ok": True,
|
||||
"data": {"exec_info": {"queue_remaining": value}},
|
||||
}
|
||||
response = self._run()
|
||||
self.assertEqual(
|
||||
response.text,
|
||||
"[Jobs fallback] Coarse queue count is unavailable.",
|
||||
)
|
||||
self.assertNotIn("secret", response.text)
|
||||
|
||||
def test_non_admin_aliases_are_denied_before_backend_call(self):
|
||||
for command in ("/jobs", "jobs", "queue"):
|
||||
with self.subTest(command=command):
|
||||
self.client.get_jobs.reset_mock()
|
||||
response = self._run(command, sender="ordinary-user")
|
||||
self.assertIn("Access Denied", response.text)
|
||||
self.client.get_jobs.assert_not_called()
|
||||
|
||||
def test_admin_aliases_share_the_v1_contract(self):
|
||||
for command in ("/jobs", "jobs", "queue"):
|
||||
with self.subTest(command=command):
|
||||
self.client.get_jobs.reset_mock()
|
||||
response = self._run(command)
|
||||
self.assertEqual(
|
||||
response.text, "[Jobs] No jobs in the authoritative snapshot."
|
||||
)
|
||||
self.client.get_jobs.assert_awaited_once()
|
||||
|
||||
def test_missing_backend_admin_token_fails_before_jobs_request(self):
|
||||
self.router.config.admin_token = None
|
||||
|
||||
response = self._run()
|
||||
|
||||
self.assertIn("Admin token not configured", response.text)
|
||||
self.client.get_jobs.assert_not_called()
|
||||
|
||||
def test_long_safe_job_id_is_display_capped(self):
|
||||
long_id = "job-" + "a" * 80
|
||||
self.client.get_jobs.return_value = _success([_job(long_id, "completed")])
|
||||
|
||||
response = self._run()
|
||||
|
||||
rendered_id = response.text.split("- ", 1)[1].split(" —", 1)[0]
|
||||
self.assertEqual(len(rendered_id), 24)
|
||||
self.assertTrue(rendered_id.endswith("..."))
|
||||
self.assertNotIn(long_id, response.text)
|
||||
|
||||
def test_help_places_jobs_in_admin_section(self):
|
||||
response = self._run("/help", sender="ordinary-user")
|
||||
public, admin = response.text.split("Admin Only:", 1)
|
||||
self.assertNotIn("/jobs", public)
|
||||
self.assertIn("/jobs - Authoritative jobs summary", admin)
|
||||
|
||||
def test_public_chat_status_does_not_fetch_or_forward_admin_jobs(self):
|
||||
self.client.get_health = AsyncMock(return_value={"ok": True, "data": {}})
|
||||
llm = MagicMock()
|
||||
llm.chat = AsyncMock(return_value="bounded status")
|
||||
|
||||
response = asyncio.run(self.router._chat_status(llm))
|
||||
|
||||
self.assertEqual(response.text, "bounded status")
|
||||
self.client.get_jobs.assert_not_called()
|
||||
user_prompt = llm.chat.await_args.args[1]
|
||||
self.assertIn("admin-only; use /jobs as an authorized operator", user_prompt)
|
||||
self.assertNotIn("contract_version", user_prompt)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
@@ -20,6 +20,7 @@
|
||||
"tests.test_connector_callback_contract",
|
||||
"tests.test_s100_jobs_security_contract",
|
||||
"tests.test_r213_jobs_endpoint",
|
||||
"tests.connector.test_r214_jobs_command",
|
||||
"tests.test_s70_ssrf_pinning_regression"
|
||||
],
|
||||
"no_skip_module_metadata": {
|
||||
@@ -99,6 +100,10 @@
|
||||
"reason": "The authoritative jobs read model protects tenant-filtered bounded job metadata and must remain non-skippable.",
|
||||
"review_after": "2026-10-31"
|
||||
},
|
||||
"tests.connector.test_r214_jobs_command": {
|
||||
"reason": "Connector jobs summaries protect the Admin jobs boundary from raw payload and fallback leakage and must remain non-skippable.",
|
||||
"review_after": "2026-10-31"
|
||||
},
|
||||
"tests.test_s70_ssrf_pinning_regression": {
|
||||
"reason": "SSRF pinning regression is a security-critical boundary test and must never degrade to skip coverage.",
|
||||
"review_after": "2026-10-31"
|
||||
|
||||
Reference in New Issue
Block a user