mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-11 12:48:54 +08:00
Compare commits
3 Commits
dependabot
...
feat/debug
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
db9a9d8e18 | ||
|
|
086dd4c28b | ||
|
|
bb6474cc51 |
2
cli.py
2
cli.py
@@ -7272,7 +7272,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
||||
elif canonical == "copy":
|
||||
self._handle_copy_command(cmd_original)
|
||||
elif canonical == "debug":
|
||||
self._handle_debug_command()
|
||||
self._handle_debug_command(cmd_original)
|
||||
elif canonical == "update":
|
||||
if self._handle_update_command():
|
||||
return False
|
||||
|
||||
@@ -2090,12 +2090,25 @@ class CLICommandsMixin:
|
||||
else:
|
||||
_cprint(f" {_ACCENT}✓ {feature_name} set to {label} (session only){_RST}")
|
||||
|
||||
def _handle_debug_command(self):
|
||||
"""Handle /debug — upload debug report + logs and print paste URLs."""
|
||||
def _handle_debug_command(self, cmd_original: str = ""):
|
||||
"""Handle /debug — upload debug report + logs and print share URLs.
|
||||
|
||||
Accepts optional destination words after the command:
|
||||
|
||||
- ``/debug`` → upload to the public paste service (default)
|
||||
- ``/debug nous`` → upload to Nous-internal storage (private, staff-only)
|
||||
- ``/debug local`` → render the report to stdout, no upload
|
||||
|
||||
``nous`` and ``local`` are mutually exclusive; if both are given,
|
||||
``local`` wins (it never touches the network).
|
||||
"""
|
||||
from hermes_cli.debug import run_debug_share
|
||||
from types import SimpleNamespace
|
||||
|
||||
args = SimpleNamespace(lines=200, expire=7, local=False)
|
||||
words = {w.lower() for w in cmd_original.split()[1:]}
|
||||
local = "local" in words
|
||||
nous = "nous" in words and not local
|
||||
args = SimpleNamespace(lines=200, expire=7, local=local, nous=nous)
|
||||
run_debug_share(args)
|
||||
|
||||
def _handle_update_command(self) -> bool:
|
||||
|
||||
@@ -217,7 +217,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
|
||||
cli_only=True, args_hint="<path>"),
|
||||
CommandDef("update", "Update Hermes Agent to the latest version", "Info"),
|
||||
CommandDef("version", "Show Hermes Agent version", "Info", aliases=("v",)),
|
||||
CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info"),
|
||||
CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info",
|
||||
args_hint="[nous|local]"),
|
||||
|
||||
# Exit
|
||||
CommandDef("quit", "Exit the CLI (use --delete to also remove session history)", "Exit",
|
||||
|
||||
@@ -9,8 +9,16 @@ Currently supports:
|
||||
``~/.hermes/logs/*.log`` are not leaked into
|
||||
the public paste service. Pass ``--no-redact``
|
||||
to disable.
|
||||
Pass ``--nous`` to upload instead to Nous-internal
|
||||
storage (AWS S3) via a signed URL minted by the
|
||||
Nous account service: the bundle is private
|
||||
(viewable only by Nous staff / allowlisted mods via
|
||||
a Google-login-gated viewer) and auto-deletes after
|
||||
14 days, rather than going to a public paste.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import gzip
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
@@ -581,6 +589,97 @@ def collect_debug_report(
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared bundle collection (used by both the paste.rs and Nous-S3 paths)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Bundle format identifier embedded in the Nous-S3 JSON envelope. The
|
||||
# discord-support viewer keys off this string to parse the bundle.
|
||||
_NOUS_BUNDLE_FORMAT = "hermes-debug-share/1"
|
||||
|
||||
|
||||
def collect_share_bundle(
|
||||
log_lines: int = 200,
|
||||
redact: bool = True,
|
||||
) -> dict[str, str]:
|
||||
"""Collect the debug report + full logs as a label→text mapping.
|
||||
|
||||
Returns ``{"report": ..., "agent.log": ..., "gateway.log": ...,
|
||||
"desktop.log": ...}`` where each value is the already-redacted (when
|
||||
``redact`` is True) text that would be uploaded. Keys for logs that are
|
||||
absent/empty are simply omitted.
|
||||
|
||||
This is the single source of collection + redaction shared by both
|
||||
destinations: the paste.rs path (:func:`build_debug_share`) and the
|
||||
Nous-S3 path (``--nous``). Centralising it guarantees the Nous bundle is
|
||||
built from the *same* force-redacted snapshots as the public paste path —
|
||||
redaction is the safety boundary, so the Nous path must never see raw
|
||||
logs.
|
||||
|
||||
The dump header is prepended to each full log (mirroring the historical
|
||||
paste behaviour) so every file is self-contained, and the redaction
|
||||
banner is prepended when ``redact`` is True.
|
||||
"""
|
||||
dump_text = _capture_dump()
|
||||
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
|
||||
|
||||
report = collect_debug_report(
|
||||
log_lines=log_lines,
|
||||
dump_text=dump_text,
|
||||
log_snapshots=log_snapshots,
|
||||
)
|
||||
agent_log = log_snapshots["agent"].full_text
|
||||
gateway_log = log_snapshots["gateway"].full_text
|
||||
desktop_log = log_snapshots["desktop"].full_text
|
||||
|
||||
# Prepend dump header to each full log so every file is self-contained.
|
||||
if agent_log:
|
||||
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
|
||||
|
||||
# Visible banner so reviewers know redaction was applied at upload time.
|
||||
if redact:
|
||||
report = _REDACTION_BANNER + report
|
||||
if agent_log:
|
||||
agent_log = _REDACTION_BANNER + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = _REDACTION_BANNER + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = _REDACTION_BANNER + desktop_log
|
||||
|
||||
bundle: dict[str, str] = {"report": report}
|
||||
if agent_log:
|
||||
bundle["agent.log"] = agent_log
|
||||
if gateway_log:
|
||||
bundle["gateway.log"] = gateway_log
|
||||
if desktop_log:
|
||||
bundle["desktop.log"] = desktop_log
|
||||
return bundle
|
||||
|
||||
|
||||
def build_nous_bundle(bundle: dict[str, str], redact: bool = True) -> bytes:
|
||||
"""Gzip-compress a :func:`collect_share_bundle` mapping into the Nous envelope.
|
||||
|
||||
The JSON shape is what the discord-support viewer (Repo 3) parses::
|
||||
|
||||
{"format": "hermes-debug-share/1",
|
||||
"redacted": <bool>,
|
||||
"created": <iso8601>,
|
||||
"files": {"report": ..., "agent.log": ..., ...}}
|
||||
"""
|
||||
created = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
||||
envelope = {
|
||||
"format": _NOUS_BUNDLE_FORMAT,
|
||||
"redacted": bool(redact),
|
||||
"created": created,
|
||||
"files": bundle,
|
||||
}
|
||||
return gzip.compress(json.dumps(envelope).encode("utf-8"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI entry points
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -620,45 +719,18 @@ def build_debug_share(
|
||||
"""
|
||||
_best_effort_sweep_expired_pastes()
|
||||
|
||||
# Capture dump once — prepended to every paste for context.
|
||||
# The dump is already redacted at extract time via dump.py:_redact;
|
||||
# log_snapshots are redacted by _capture_default_log_snapshots when
|
||||
# redact=True so credentials never reach the public paste service.
|
||||
dump_text = _capture_dump()
|
||||
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
|
||||
# Collect the report + full logs (force-redacted when redact=True) via the
|
||||
# shared collector so the paste.rs and Nous-S3 paths build identical,
|
||||
# identically-redacted bundles. The dump header + redaction banner are
|
||||
# applied inside collect_share_bundle.
|
||||
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
|
||||
|
||||
if redact:
|
||||
logger.info(
|
||||
"hermes debug share: applied force-mode redaction to log snapshots before upload"
|
||||
)
|
||||
|
||||
report = collect_debug_report(
|
||||
log_lines=log_lines,
|
||||
dump_text=dump_text,
|
||||
log_snapshots=log_snapshots,
|
||||
)
|
||||
agent_log = log_snapshots["agent"].full_text
|
||||
gateway_log = log_snapshots["gateway"].full_text
|
||||
desktop_log = log_snapshots["desktop"].full_text
|
||||
|
||||
# Prepend dump header to each full log so every paste is self-contained.
|
||||
if agent_log:
|
||||
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
|
||||
|
||||
# Visible banner so reviewers reading the public paste know redaction
|
||||
# was applied at upload time. Banner is omitted under --no-redact.
|
||||
if redact:
|
||||
report = _REDACTION_BANNER + report
|
||||
if agent_log:
|
||||
agent_log = _REDACTION_BANNER + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = _REDACTION_BANNER + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = _REDACTION_BANNER + desktop_log
|
||||
report = bundle["report"]
|
||||
|
||||
urls: dict[str, str] = {}
|
||||
failures: list[str] = []
|
||||
@@ -667,11 +739,8 @@ def build_debug_share(
|
||||
urls["Report"] = upload_to_pastebin(report, expiry_days=expiry)
|
||||
|
||||
# 2-4. Full logs (optional — failures are collected, not raised)
|
||||
for label, content in (
|
||||
("agent.log", agent_log),
|
||||
("gateway.log", gateway_log),
|
||||
("desktop.log", desktop_log),
|
||||
):
|
||||
for label in ("agent.log", "gateway.log", "desktop.log"):
|
||||
content = bundle.get(label)
|
||||
if not content:
|
||||
continue
|
||||
try:
|
||||
@@ -696,43 +765,23 @@ def run_debug_share(args):
|
||||
log_lines = getattr(args, "lines", 200)
|
||||
expiry = getattr(args, "expire", 7)
|
||||
local_only = getattr(args, "local", False)
|
||||
nous = getattr(args, "nous", False)
|
||||
redact = not getattr(args, "no_redact", False)
|
||||
|
||||
if local_only:
|
||||
# Local-only path never uploads — render the report to stdout and bail
|
||||
# before any network I/O. Mirrors the upload path's collection logic.
|
||||
# before any network I/O. Reuses the shared collector so the rendered
|
||||
# output matches exactly what would be uploaded.
|
||||
_best_effort_sweep_expired_pastes()
|
||||
print("Collecting debug report...")
|
||||
dump_text = _capture_dump()
|
||||
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
|
||||
report = collect_debug_report(
|
||||
log_lines=log_lines,
|
||||
dump_text=dump_text,
|
||||
log_snapshots=log_snapshots,
|
||||
)
|
||||
agent_log = log_snapshots["agent"].full_text
|
||||
gateway_log = log_snapshots["gateway"].full_text
|
||||
desktop_log = log_snapshots["desktop"].full_text
|
||||
if agent_log:
|
||||
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
|
||||
if redact:
|
||||
report = _REDACTION_BANNER + report
|
||||
if agent_log:
|
||||
agent_log = _REDACTION_BANNER + agent_log
|
||||
if gateway_log:
|
||||
gateway_log = _REDACTION_BANNER + gateway_log
|
||||
if desktop_log:
|
||||
desktop_log = _REDACTION_BANNER + desktop_log
|
||||
print(report)
|
||||
for title, body in (
|
||||
("FULL agent.log", agent_log),
|
||||
("FULL gateway.log", gateway_log),
|
||||
("FULL desktop.log", desktop_log),
|
||||
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
|
||||
print(bundle["report"])
|
||||
for title, label in (
|
||||
("FULL agent.log", "agent.log"),
|
||||
("FULL gateway.log", "gateway.log"),
|
||||
("FULL desktop.log", "desktop.log"),
|
||||
):
|
||||
body = bundle.get(label)
|
||||
if body:
|
||||
print(f"\n\n{'=' * 60}")
|
||||
print(title)
|
||||
@@ -740,6 +789,10 @@ def run_debug_share(args):
|
||||
print(body)
|
||||
return
|
||||
|
||||
if nous:
|
||||
_run_debug_share_nous(log_lines=log_lines, redact=redact)
|
||||
return
|
||||
|
||||
print(_PRIVACY_NOTICE)
|
||||
print("Collecting debug report...")
|
||||
print("Uploading...")
|
||||
@@ -773,6 +826,80 @@ def run_debug_share(args):
|
||||
print(f"\nShare these links with the Hermes team for support.")
|
||||
|
||||
|
||||
_NOUS_PRIVACY_NOTICE = """\
|
||||
⚠️ --nous: This uploads your debug bundle to Nous-INTERNAL storage (AWS S3),
|
||||
NOT a public paste service. The following is included:
|
||||
• System info (OS, Python/Hermes version, provider, which API keys are
|
||||
configured — NOT the actual keys)
|
||||
• Full agent.log, gateway.log, and desktop.log (up to 512 KB each — likely
|
||||
contains conversation content, tool outputs, and file paths)
|
||||
|
||||
• The bundle is viewable only by Nous staff (and allowlisted Discord mods)
|
||||
via a Google-login-gated viewer.
|
||||
• It is NOT a public paste — there is no public URL to the contents.
|
||||
• It auto-deletes after 14 days.
|
||||
"""
|
||||
|
||||
|
||||
def _run_debug_share_nous(*, log_lines: int, redact: bool) -> None:
|
||||
"""Handle ``hermes debug share --nous``: upload the bundle to Nous-S3.
|
||||
|
||||
Collects the same force-redacted bundle as the paste path, gzips it into
|
||||
the Nous envelope, requests a signed URL from NAS, uploads, and prints the
|
||||
private viewer link. On any failure falls back to a clear error that
|
||||
suggests ``--local``.
|
||||
"""
|
||||
from hermes_cli.diagnostics_upload import share_to_nous
|
||||
|
||||
print(_NOUS_PRIVACY_NOTICE)
|
||||
if not redact:
|
||||
print(
|
||||
"⚠️ --no-redact is set: secrets in your logs will NOT be redacted "
|
||||
"before upload.\n"
|
||||
)
|
||||
print("Collecting debug report...")
|
||||
_best_effort_sweep_expired_pastes()
|
||||
|
||||
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
|
||||
if redact:
|
||||
logger.info(
|
||||
"hermes debug share --nous: applied force-mode redaction before upload"
|
||||
)
|
||||
blob = build_nous_bundle(bundle, redact=redact)
|
||||
|
||||
print("Uploading to Nous diagnostics storage...")
|
||||
try:
|
||||
res = share_to_nous(blob)
|
||||
except Exception as exc:
|
||||
print(
|
||||
f"\nNous upload failed: {exc}\n"
|
||||
"\nThe Nous diagnostics service may be unavailable or not yet "
|
||||
"provisioned.\n"
|
||||
"Run `hermes debug share --local` to print the report instead, "
|
||||
"or `hermes debug share` to upload to a public paste service.\n",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
view_url = res.get("viewUrl") or res.get("view_url")
|
||||
print("\nDebug bundle uploaded to Nous (private):")
|
||||
if view_url:
|
||||
print(f" View URL {view_url}")
|
||||
else:
|
||||
print(f" (no view URL returned; upload id: {res.get('id', '?')})")
|
||||
|
||||
expires_at = res.get("expiresAt") or res.get("expires_at")
|
||||
if expires_at:
|
||||
print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention).")
|
||||
else:
|
||||
print("\n⏱ Auto-deletes after 14 days.")
|
||||
|
||||
print(
|
||||
"\nShare this private link with the Nous team — only Nous staff "
|
||||
"(via Google login) can open it."
|
||||
)
|
||||
|
||||
|
||||
def run_debug_delete(args):
|
||||
"""Delete one or more paste URLs uploaded by /debug."""
|
||||
urls = getattr(args, "urls", [])
|
||||
@@ -823,6 +950,8 @@ def run_debug(args):
|
||||
print(" --lines N Number of log lines to include (default: 200)")
|
||||
print(" --expire N Paste expiry in days (default: 7)")
|
||||
print(" --local Print report locally instead of uploading")
|
||||
print(" --nous Upload to Nous-internal storage (private, staff-only,")
|
||||
print(" auto-deletes in 14 days) instead of a public paste")
|
||||
print(" --no-redact Disable upload-time secret redaction (default: redact)")
|
||||
print()
|
||||
print("Options (delete):")
|
||||
|
||||
138
hermes_cli/diagnostics_upload.py
Normal file
138
hermes_cli/diagnostics_upload.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""Client for uploading ``hermes debug share`` bundles to Nous-internal S3.
|
||||
|
||||
This is the opt-in (``--nous``) destination for ``hermes debug share``.
|
||||
Unlike the public paste.rs path, bundles uploaded here go to a Nous-owned
|
||||
S3 bucket via a short-lived signed URL minted by the Nous account service
|
||||
(NAS). The bucket auto-expires objects after 14 days, and the contents are
|
||||
only viewable by Nous staff (and allowlisted Discord mods) through a
|
||||
Google-OAuth-gated viewer.
|
||||
|
||||
Flow:
|
||||
|
||||
1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...}
|
||||
(the request body carries ``sizeBytes``; NAS signs it into the presigned
|
||||
URL's ``ContentLength``, so the PUT must send exactly that many bytes)
|
||||
2. PUT <uploadUrl> (the gzipped bundle, Content-Type application/gzip)
|
||||
|
||||
NAS is stateless — the object's existence in S3 is the only state, so there is
|
||||
no confirm/callback step.
|
||||
|
||||
Uses stdlib ``urllib`` only, matching ``debug.py`` style — no third-party deps.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
|
||||
# Base URL of the Nous account service that mints the signed upload URL.
|
||||
# Overridable via env so the feature can be pointed at staging / a local dev
|
||||
# NAS instance during testing.
|
||||
NAS_BASE = os.environ.get(
|
||||
"HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com"
|
||||
)
|
||||
|
||||
# Network timeout for each request (seconds). The upload itself can be larger
|
||||
# (a gzipped log bundle), so the PUT gets a more generous window.
|
||||
_REQUEST_TIMEOUT = 30
|
||||
_UPLOAD_TIMEOUT = 120
|
||||
|
||||
_USER_AGENT = "hermes-agent/debug-share"
|
||||
|
||||
|
||||
def request_upload_url(
|
||||
content_type: str = "application/gzip",
|
||||
size_bytes: int | None = None,
|
||||
) -> dict:
|
||||
"""Ask NAS to mint a presigned PUT URL for a diagnostics bundle.
|
||||
|
||||
POSTs a small JSON body to ``{NAS_BASE}/api/diagnostics/upload-url`` and
|
||||
returns the parsed JSON response, expected to contain at least
|
||||
``uploadUrl``, ``viewUrl`` and ``id`` (plus optional ``expiresAt`` /
|
||||
``uploadExpiresInSeconds``).
|
||||
|
||||
Raises on non-2xx responses or unparseable JSON.
|
||||
"""
|
||||
payload: dict = {"contentType": content_type}
|
||||
if size_bytes is not None:
|
||||
payload["sizeBytes"] = int(size_bytes)
|
||||
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
f"{NAS_BASE}/api/diagnostics/upload-url",
|
||||
data=data,
|
||||
method="POST",
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
"User-Agent": _USER_AGENT,
|
||||
},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp:
|
||||
status = getattr(resp, "status", None)
|
||||
if status is None:
|
||||
status = resp.getcode()
|
||||
if not (200 <= status < 300):
|
||||
raise RuntimeError(
|
||||
f"diagnostics upload-url request failed: HTTP {status}"
|
||||
)
|
||||
body = resp.read().decode("utf-8")
|
||||
|
||||
try:
|
||||
result = json.loads(body)
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
raise RuntimeError(
|
||||
f"diagnostics upload-url returned non-JSON response: {body[:200]}"
|
||||
) from exc
|
||||
|
||||
if not isinstance(result, dict) or not result.get("uploadUrl"):
|
||||
raise RuntimeError(
|
||||
"diagnostics upload-url response missing 'uploadUrl': "
|
||||
f"{body[:200]}"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def put_bundle(
|
||||
upload_url: str,
|
||||
data: bytes,
|
||||
content_type: str = "application/gzip",
|
||||
) -> None:
|
||||
"""PUT the gzipped *data* bundle to a presigned *upload_url*.
|
||||
|
||||
Sets the ``Content-Type`` header (must match what NAS pinned when signing
|
||||
the URL, otherwise S3 rejects the signature). Raises on non-2xx.
|
||||
"""
|
||||
req = urllib.request.Request(
|
||||
upload_url,
|
||||
data=data,
|
||||
method="PUT",
|
||||
headers={
|
||||
"Content-Type": content_type,
|
||||
"User-Agent": _USER_AGENT,
|
||||
},
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=_UPLOAD_TIMEOUT) as resp:
|
||||
status = getattr(resp, "status", None)
|
||||
if status is None:
|
||||
status = resp.getcode()
|
||||
if not (200 <= status < 300):
|
||||
raise RuntimeError(f"diagnostics bundle PUT failed: HTTP {status}")
|
||||
|
||||
|
||||
def share_to_nous(report_bundle: bytes) -> dict:
|
||||
"""Orchestrate the full Nous-S3 upload of a gzipped *report_bundle*.
|
||||
|
||||
Two steps: mint a presigned PUT URL (sending the exact ``sizeBytes`` NAS
|
||||
signs into the URL's ``ContentLength``), then PUT the bundle. NAS is
|
||||
stateless — the object's existence in S3 is the only state, so there is no
|
||||
confirm/callback step. Returns the dict from :func:`request_upload_url`
|
||||
(which carries ``viewUrl`` / ``id`` / expiry metadata) so the caller can
|
||||
print the viewer link. Raises on any failure of either step.
|
||||
"""
|
||||
size_bytes = len(report_bundle)
|
||||
info = request_upload_url(
|
||||
content_type="application/gzip", size_bytes=size_bytes
|
||||
)
|
||||
put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
|
||||
|
||||
return info
|
||||
@@ -29,6 +29,7 @@ Examples:
|
||||
hermes debug share --expire 30 Keep paste for 30 days
|
||||
hermes debug share --local Print report locally (no upload)
|
||||
hermes debug share --no-redact Disable upload-time secret redaction
|
||||
hermes debug share --nous Upload to Nous-internal storage (private)
|
||||
hermes debug delete <url> Delete a previously uploaded paste
|
||||
""",
|
||||
)
|
||||
@@ -64,6 +65,17 @@ Examples:
|
||||
"into the public paste service."
|
||||
),
|
||||
)
|
||||
share_parser.add_argument(
|
||||
"--nous",
|
||||
action="store_true",
|
||||
help=(
|
||||
"Upload the debug bundle to Nous-internal storage (AWS S3) instead "
|
||||
"of a public paste service. The bundle is private — viewable only "
|
||||
"by Nous staff (and allowlisted Discord mods) via a Google-login-"
|
||||
"gated viewer — and auto-deletes after 14 days. Still force-redacts "
|
||||
"secrets unless --no-redact is also passed."
|
||||
),
|
||||
)
|
||||
delete_parser = debug_sub.add_parser(
|
||||
"delete",
|
||||
help="Delete a paste uploaded by 'hermes debug share'",
|
||||
|
||||
@@ -491,6 +491,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), \
|
||||
patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \
|
||||
@@ -509,6 +510,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), \
|
||||
patch(
|
||||
@@ -529,6 +531,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = True
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
run_debug_share(args)
|
||||
@@ -546,6 +549,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
call_count = [0]
|
||||
uploaded_content = []
|
||||
@@ -599,6 +603,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
uploaded_content = []
|
||||
|
||||
@@ -644,6 +649,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
call_count = [0]
|
||||
def _mock_upload(content, expiry_days=7):
|
||||
@@ -668,6 +674,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
call_count = [0]
|
||||
def _mock_upload(content, expiry_days=7):
|
||||
@@ -694,6 +701,7 @@ class TestRunDebugShare:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), \
|
||||
patch("hermes_cli.debug.upload_to_pastebin",
|
||||
@@ -742,6 +750,7 @@ class TestRunDebugShareRedaction:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
args.no_redact = False
|
||||
|
||||
captured: list[str] = []
|
||||
@@ -772,6 +781,7 @@ class TestRunDebugShareRedaction:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
args.no_redact = False
|
||||
|
||||
captured: list[str] = []
|
||||
@@ -800,6 +810,7 @@ class TestRunDebugShareRedaction:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
args.no_redact = True
|
||||
|
||||
captured: list[str] = []
|
||||
@@ -850,6 +861,7 @@ class TestRunDebug:
|
||||
args.lines = 200
|
||||
args.expire = 7
|
||||
args.local = True
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
run_debug(args)
|
||||
@@ -1228,6 +1240,7 @@ class TestShareIncludesAutoDelete:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), \
|
||||
patch("hermes_cli.debug.upload_to_pastebin",
|
||||
@@ -1250,6 +1263,7 @@ class TestShareIncludesAutoDelete:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = False
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), \
|
||||
patch("hermes_cli.debug.upload_to_pastebin",
|
||||
@@ -1267,6 +1281,7 @@ class TestShareIncludesAutoDelete:
|
||||
args.lines = 50
|
||||
args.expire = 7
|
||||
args.local = True
|
||||
args.nous = False
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
run_debug_share(args)
|
||||
@@ -1380,3 +1395,220 @@ class TestBuildDebugShare:
|
||||
), patch("hermes_cli.debug._schedule_auto_delete"):
|
||||
with pytest.raises(RuntimeError, match="all paste services down"):
|
||||
build_debug_share(log_lines=50, redact=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared bundle collection + Nous-S3 path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCollectShareBundle:
|
||||
def test_returns_report_and_logs(self, hermes_home):
|
||||
from hermes_cli.debug import collect_share_bundle
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
bundle = collect_share_bundle(log_lines=50, redact=True)
|
||||
|
||||
assert "report" in bundle
|
||||
assert "agent.log" in bundle
|
||||
assert "gateway.log" in bundle
|
||||
assert "desktop.log" in bundle
|
||||
# Banner is prepended under redact=True.
|
||||
assert "redacted at upload time" in bundle["report"]
|
||||
assert "session started" in bundle["agent.log"]
|
||||
|
||||
def test_no_redact_omits_banner(self, hermes_home):
|
||||
from hermes_cli.debug import collect_share_bundle
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
bundle = collect_share_bundle(log_lines=50, redact=False)
|
||||
|
||||
assert "redacted at upload time" not in bundle["report"]
|
||||
|
||||
def test_redaction_keeps_secrets_out(self, hermes_home):
|
||||
from hermes_cli.debug import collect_share_bundle
|
||||
|
||||
secret = "sk-proj-abcdefghijklmnopqrstuvwxyz1234567890"
|
||||
(hermes_home / "logs" / "agent.log").write_text(
|
||||
f"line one\nOPENAI_API_KEY={secret}\nline three\n"
|
||||
)
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
redacted = collect_share_bundle(log_lines=50, redact=True)
|
||||
unredacted = collect_share_bundle(log_lines=50, redact=False)
|
||||
|
||||
# Sanity: without redaction the secret is present in the bundle.
|
||||
assert secret in "\n".join(unredacted.values())
|
||||
# With redaction it must be scrubbed everywhere.
|
||||
assert secret not in "\n".join(redacted.values())
|
||||
|
||||
|
||||
def test_build_debug_share_uses_collector(self, hermes_home):
|
||||
# build_debug_share must produce the same report text the collector does
|
||||
# (i.e. the refactor preserved paste.rs behaviour).
|
||||
from hermes_cli.debug import build_debug_share, collect_share_bundle
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"):
|
||||
expected = collect_share_bundle(log_lines=50, redact=True)["report"]
|
||||
|
||||
uploaded = []
|
||||
|
||||
def _upload(content, expiry_days=7):
|
||||
uploaded.append(content)
|
||||
return "https://paste.rs/x"
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), patch(
|
||||
"hermes_cli.debug.upload_to_pastebin", side_effect=_upload
|
||||
), patch("hermes_cli.debug._schedule_auto_delete"):
|
||||
result = build_debug_share(log_lines=50, redact=True)
|
||||
|
||||
assert result.urls["Report"] == "https://paste.rs/x"
|
||||
# The report uploaded should match the collector's report.
|
||||
assert uploaded[0] == expected
|
||||
|
||||
|
||||
class TestBuildNousBundle:
|
||||
def test_envelope_shape_and_gzip(self, hermes_home):
|
||||
import gzip
|
||||
import json as _json
|
||||
|
||||
from hermes_cli.debug import build_nous_bundle
|
||||
|
||||
files = {"report": "hello", "agent.log": "log line"}
|
||||
blob = build_nous_bundle(files, redact=True)
|
||||
|
||||
# It's gzip — magic bytes.
|
||||
assert blob[:2] == b"\x1f\x8b"
|
||||
envelope = _json.loads(gzip.decompress(blob).decode())
|
||||
assert envelope["format"] == "hermes-debug-share/1"
|
||||
assert envelope["redacted"] is True
|
||||
assert envelope["files"] == files
|
||||
assert "created" in envelope
|
||||
|
||||
def test_redacted_false_recorded(self):
|
||||
import gzip
|
||||
import json as _json
|
||||
|
||||
from hermes_cli.debug import build_nous_bundle
|
||||
|
||||
blob = build_nous_bundle({"report": "x"}, redact=False)
|
||||
envelope = _json.loads(gzip.decompress(blob).decode())
|
||||
assert envelope["redacted"] is False
|
||||
|
||||
|
||||
class TestRunDebugShareNous:
|
||||
def _args(self, **over):
|
||||
class _A:
|
||||
lines = 50
|
||||
expire = 7
|
||||
local = False
|
||||
nous = True
|
||||
no_redact = False
|
||||
|
||||
a = _A()
|
||||
for k, v in over.items():
|
||||
setattr(a, k, v)
|
||||
return a
|
||||
|
||||
def test_nous_success_prints_view_url(self, hermes_home, capsys):
|
||||
from hermes_cli.debug import run_debug_share
|
||||
|
||||
res = {
|
||||
"id": "id-1",
|
||||
"viewUrl": "https://support.example.com/diagnostics/id-1",
|
||||
"expiresAt": "2026-06-20T00:00:00Z",
|
||||
}
|
||||
with patch("hermes_cli.dump.run_dump"), patch(
|
||||
"hermes_cli.diagnostics_upload.share_to_nous", return_value=res
|
||||
) as share:
|
||||
run_debug_share(self._args())
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Nous-INTERNAL" in out
|
||||
assert "https://support.example.com/diagnostics/id-1" in out
|
||||
assert "2026-06-20T00:00:00Z" in out
|
||||
# The blob passed to share_to_nous must be gzip bytes.
|
||||
blob = share.call_args[0][0]
|
||||
assert isinstance(blob, (bytes, bytearray)) and blob[:2] == b"\x1f\x8b"
|
||||
|
||||
def test_nous_failure_suggests_local(self, hermes_home, capsys):
|
||||
from hermes_cli.debug import run_debug_share
|
||||
|
||||
with patch("hermes_cli.dump.run_dump"), patch(
|
||||
"hermes_cli.diagnostics_upload.share_to_nous",
|
||||
side_effect=RuntimeError("service down"),
|
||||
):
|
||||
with pytest.raises(SystemExit) as exc:
|
||||
run_debug_share(self._args())
|
||||
assert exc.value.code == 1
|
||||
err = capsys.readouterr().err
|
||||
assert "Nous upload failed" in err
|
||||
assert "--local" in err
|
||||
|
||||
def test_nous_does_not_touch_pastebin(self, hermes_home):
|
||||
from hermes_cli.debug import run_debug_share
|
||||
|
||||
res = {"id": "id-1", "viewUrl": "https://v"}
|
||||
with patch("hermes_cli.dump.run_dump"), patch(
|
||||
"hermes_cli.diagnostics_upload.share_to_nous", return_value=res
|
||||
), patch("hermes_cli.debug.upload_to_pastebin") as paste:
|
||||
run_debug_share(self._args())
|
||||
paste.assert_not_called()
|
||||
|
||||
|
||||
class TestDebugSlashCommand:
|
||||
"""`/debug [nous|local]` parsing in the CLI/TUI handler.
|
||||
|
||||
The classic CLI and the TUI slash worker both dispatch through
|
||||
``HermesCLI.process_command`` → ``_handle_debug_command(cmd_original)``,
|
||||
which parses an optional destination word and builds the args namespace
|
||||
handed to ``run_debug_share``.
|
||||
"""
|
||||
|
||||
def _handler(self):
|
||||
from hermes_cli.cli_commands_mixin import CLICommandsMixin
|
||||
|
||||
class _Stub(CLICommandsMixin):
|
||||
pass
|
||||
|
||||
return _Stub()._handle_debug_command
|
||||
|
||||
def _captured(self, cmd_original):
|
||||
captured = {}
|
||||
|
||||
def _fake_run(args):
|
||||
captured.update(vars(args))
|
||||
|
||||
with patch("hermes_cli.debug.run_debug_share", _fake_run):
|
||||
self._handler()(cmd_original)
|
||||
return captured
|
||||
|
||||
def test_bare_debug_defaults_to_paste(self):
|
||||
c = self._captured("/debug")
|
||||
assert c["nous"] is False and c["local"] is False
|
||||
assert c["lines"] == 200 and c["expire"] == 7
|
||||
|
||||
def test_nous_word_sets_nous(self):
|
||||
c = self._captured("/debug nous")
|
||||
assert c["nous"] is True and c["local"] is False
|
||||
|
||||
def test_local_word_sets_local(self):
|
||||
c = self._captured("/debug local")
|
||||
assert c["local"] is True and c["nous"] is False
|
||||
|
||||
def test_word_parsing_is_case_insensitive(self):
|
||||
c = self._captured("/debug NOUS")
|
||||
assert c["nous"] is True
|
||||
|
||||
def test_local_wins_over_nous(self):
|
||||
# local never touches the network, so it takes precedence.
|
||||
c = self._captured("/debug nous local")
|
||||
assert c["local"] is True and c["nous"] is False
|
||||
|
||||
def test_unknown_word_falls_back_to_default(self):
|
||||
c = self._captured("/debug paste")
|
||||
assert c["nous"] is False and c["local"] is False
|
||||
|
||||
def test_no_arg_default_keyword(self):
|
||||
# Calling with no cmd_original (legacy callers) must still work.
|
||||
c = self._captured("")
|
||||
assert c["nous"] is False and c["local"] is False
|
||||
|
||||
|
||||
227
tests/hermes_cli/test_diagnostics_upload.py
Normal file
227
tests/hermes_cli/test_diagnostics_upload.py
Normal file
@@ -0,0 +1,227 @@
|
||||
"""Tests for ``hermes_cli.diagnostics_upload`` — the Nous-S3 upload client.
|
||||
|
||||
All network I/O is mocked at ``urllib.request.urlopen``; no real requests
|
||||
are made.
|
||||
"""
|
||||
|
||||
import io
|
||||
import json
|
||||
import urllib.error
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _resp(*, status=200, body=b""):
|
||||
"""Build a context-manager mock mimicking ``urllib.request.urlopen``."""
|
||||
m = MagicMock()
|
||||
m.status = status
|
||||
m.getcode.return_value = status
|
||||
m.read.return_value = body
|
||||
m.__enter__ = lambda s: s
|
||||
m.__exit__ = MagicMock(return_value=False)
|
||||
return m
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# request_upload_url
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRequestUploadUrl:
|
||||
def test_happy_path_posts_json_and_returns_dict(self):
|
||||
from hermes_cli.diagnostics_upload import request_upload_url
|
||||
|
||||
payload = {
|
||||
"success": True,
|
||||
"id": "abc-123",
|
||||
"uploadUrl": "https://bucket.s3.amazonaws.com/uploads/abc-123.json.gz?sig",
|
||||
"viewUrl": "https://support.example.com/diagnostics/abc-123",
|
||||
"uploadExpiresInSeconds": 900,
|
||||
}
|
||||
resp = _resp(status=200, body=json.dumps(payload).encode())
|
||||
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
) as urlopen:
|
||||
result = request_upload_url(content_type="application/gzip", size_bytes=512)
|
||||
|
||||
assert result == payload
|
||||
|
||||
# The request object passed to urlopen carries our JSON body + headers.
|
||||
req = urlopen.call_args[0][0]
|
||||
assert req.method == "POST"
|
||||
assert req.full_url.endswith("/api/diagnostics/upload-url")
|
||||
sent = json.loads(req.data.decode())
|
||||
assert sent["contentType"] == "application/gzip"
|
||||
assert sent["sizeBytes"] == 512
|
||||
# urllib lower-cases header keys.
|
||||
assert req.headers["Content-type"] == "application/json"
|
||||
|
||||
def test_non_2xx_raises(self):
|
||||
from hermes_cli.diagnostics_upload import request_upload_url
|
||||
|
||||
resp = _resp(status=500, body=b"boom")
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
):
|
||||
with pytest.raises(RuntimeError):
|
||||
request_upload_url()
|
||||
|
||||
def test_missing_upload_url_raises(self):
|
||||
from hermes_cli.diagnostics_upload import request_upload_url
|
||||
|
||||
resp = _resp(status=200, body=json.dumps({"id": "x"}).encode())
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
):
|
||||
with pytest.raises(RuntimeError):
|
||||
request_upload_url()
|
||||
|
||||
def test_non_json_raises(self):
|
||||
from hermes_cli.diagnostics_upload import request_upload_url
|
||||
|
||||
resp = _resp(status=200, body=b"<html>not json</html>")
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
):
|
||||
with pytest.raises(RuntimeError):
|
||||
request_upload_url()
|
||||
|
||||
def test_base_url_env_override(self, monkeypatch):
|
||||
# NAS_BASE is read at import time; re-import the module under the
|
||||
# patched env to confirm the override is honoured.
|
||||
import importlib
|
||||
|
||||
monkeypatch.setenv("HERMES_DIAGNOSTICS_BASE_URL", "https://staging.example.com")
|
||||
import hermes_cli.diagnostics_upload as mod
|
||||
|
||||
mod = importlib.reload(mod)
|
||||
try:
|
||||
assert mod.NAS_BASE == "https://staging.example.com"
|
||||
resp = _resp(
|
||||
status=200,
|
||||
body=json.dumps({"uploadUrl": "u", "id": "i", "viewUrl": "v"}).encode(),
|
||||
)
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
) as urlopen:
|
||||
mod.request_upload_url()
|
||||
req = urlopen.call_args[0][0]
|
||||
assert req.full_url == "https://staging.example.com/api/diagnostics/upload-url"
|
||||
finally:
|
||||
monkeypatch.delenv("HERMES_DIAGNOSTICS_BASE_URL", raising=False)
|
||||
importlib.reload(mod)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# put_bundle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestPutBundle:
|
||||
def test_put_sends_exact_body_and_content_type(self):
|
||||
from hermes_cli.diagnostics_upload import put_bundle
|
||||
|
||||
data = b"\x1f\x8b\x08gzipped-bytes"
|
||||
resp = _resp(status=200, body=b"")
|
||||
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
) as urlopen:
|
||||
put_bundle("https://bucket.s3.amazonaws.com/uploads/x.json.gz?sig", data)
|
||||
|
||||
req = urlopen.call_args[0][0]
|
||||
assert req.method == "PUT"
|
||||
# PUT body must be the bundle bytes, unchanged.
|
||||
assert req.data == data
|
||||
assert req.headers["Content-type"] == "application/gzip"
|
||||
|
||||
def test_custom_content_type(self):
|
||||
from hermes_cli.diagnostics_upload import put_bundle
|
||||
|
||||
resp = _resp(status=204, body=b"")
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
) as urlopen:
|
||||
put_bundle("https://u", b"data", content_type="application/json")
|
||||
req = urlopen.call_args[0][0]
|
||||
assert req.headers["Content-type"] == "application/json"
|
||||
|
||||
def test_non_2xx_raises(self):
|
||||
from hermes_cli.diagnostics_upload import put_bundle
|
||||
|
||||
resp = _resp(status=403, body=b"AccessDenied")
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
return_value=resp,
|
||||
):
|
||||
with pytest.raises(RuntimeError):
|
||||
put_bundle("https://u", b"data")
|
||||
|
||||
def test_http_error_propagates(self):
|
||||
from hermes_cli.diagnostics_upload import put_bundle
|
||||
|
||||
err = urllib.error.HTTPError("https://u", 500, "err", {}, io.BytesIO(b""))
|
||||
with patch(
|
||||
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
|
||||
side_effect=err,
|
||||
):
|
||||
with pytest.raises(urllib.error.HTTPError):
|
||||
put_bundle("https://u", b"data")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# share_to_nous (orchestration)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestShareToNous:
|
||||
def test_orchestrates_request_then_put(self):
|
||||
from hermes_cli import diagnostics_upload as mod
|
||||
|
||||
info = {
|
||||
"id": "id-9",
|
||||
"uploadUrl": "https://bucket/uploads/id-9.json.gz?sig",
|
||||
"viewUrl": "https://support/diagnostics/id-9",
|
||||
"expiresAt": "2026-06-20T00:00:00Z",
|
||||
}
|
||||
blob = b"gzipped-bundle"
|
||||
|
||||
with patch.object(mod, "request_upload_url", return_value=info) as req, \
|
||||
patch.object(mod, "put_bundle") as put:
|
||||
result = mod.share_to_nous(blob)
|
||||
|
||||
assert result == info
|
||||
req.assert_called_once()
|
||||
# request was told the real byte size (NAS signs it into ContentLength)
|
||||
assert req.call_args.kwargs["size_bytes"] == len(blob)
|
||||
# PUT got the signed URL + the exact blob
|
||||
put.assert_called_once_with(
|
||||
info["uploadUrl"], blob, content_type="application/gzip"
|
||||
)
|
||||
|
||||
def test_put_failure_propagates(self):
|
||||
from hermes_cli import diagnostics_upload as mod
|
||||
|
||||
info = {"id": "id-9", "uploadUrl": "https://u", "viewUrl": "v"}
|
||||
with patch.object(mod, "request_upload_url", return_value=info), \
|
||||
patch.object(mod, "put_bundle", side_effect=RuntimeError("PUT failed")):
|
||||
with pytest.raises(RuntimeError):
|
||||
mod.share_to_nous(b"data")
|
||||
|
||||
def test_share_succeeds_without_id_in_response(self):
|
||||
from hermes_cli import diagnostics_upload as mod
|
||||
|
||||
# NAS is stateless and there is no confirm step, so the share must
|
||||
# succeed regardless of whether the response carries an ``id``.
|
||||
info = {"uploadUrl": "https://u", "viewUrl": "v"} # no id
|
||||
with patch.object(mod, "request_upload_url", return_value=info), \
|
||||
patch.object(mod, "put_bundle") as put:
|
||||
result = mod.share_to_nous(b"data")
|
||||
assert result == info
|
||||
put.assert_called_once()
|
||||
Reference in New Issue
Block a user