wehub-resource-sync
b4fbd6fe9f
Deploy Site / deploy-vercel (push) Has been skipped
Deploy Site / deploy-docs (push) Has been skipped
Build Skills Index / build-index (push) Has been skipped
Build Skills Index / trigger-deploy (push) Waiting to run
CI / Deny unrelated histories (push) Has been skipped
CI / CI timing report (push) Blocked by required conditions
CI / Detect affected areas (push) Successful in 27m35s
CI / OSV scan (push) Failing after 4s
CI / Build&Test Docker image (push) Successful in 9s
CI / Supply-chain scan (push) Has been skipped
CI / Lint Docker scripts (push) Failing after 5m13s
CI / Check contributors (push) Failing after 12m8s
CI / Docs Site (push) Failing after 12m8s
CI / TypeScript (push) Failing after 12m8s
CI / Python lints (push) Failing after 12m9s
CI / Python tests (push) Failing after 12m9s
CI / Check uv.lock (push) Failing after 23m22s
CI / All required checks pass (push) Waiting to run
228 lines
8.3 KiB
Python
228 lines
8.3 KiB
Python
"""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()
|