|
7 | 7 | import sys |
8 | 8 | import tempfile |
9 | 9 | import unittest |
| 10 | +import urllib.error |
10 | 11 | from contextlib import redirect_stdout |
11 | 12 | from pathlib import Path |
12 | 13 | from unittest.mock import patch |
@@ -283,6 +284,32 @@ def fake_urlopen(request, timeout): |
283 | 284 | }, |
284 | 285 | ) |
285 | 286 |
|
| 287 | + def test_publish_reports_only_http_status_for_a_rejected_request(self): |
| 288 | + envelope = publisher.build_m0_research_publisher_envelope( |
| 289 | + source_snapshot=self._snapshot(), |
| 290 | + source_artifact=self._artifact("f" * 64), |
| 291 | + producer_repository="QuantStrategyLab/QuantRuntimeSettings", |
| 292 | + producer_revision="e" * 40, |
| 293 | + now="2026-08-21T12:00:00Z", |
| 294 | + ) |
| 295 | + rejected = urllib.error.HTTPError( |
| 296 | + "https://research-console.example/api/internal/sync-m0-research-ledger", |
| 297 | + 409, |
| 298 | + "Conflict", |
| 299 | + hdrs=None, |
| 300 | + fp=io.BytesIO(b'{"error":"must-not-be-exposed"}'), |
| 301 | + ) |
| 302 | + with patch.object(publisher.urllib.request, "urlopen", side_effect=rejected): |
| 303 | + with self.assertRaisesRegex(publisher.M0ResearchPublisherEnvelopeError, "m0_publish_http_409") as caught: |
| 304 | + publisher.publish_m0_research_publisher_envelope( |
| 305 | + envelope, |
| 306 | + environ={ |
| 307 | + publisher.PUBLISH_URL_ENV: "https://research-console.example/api/internal/sync-m0-research-ledger", |
| 308 | + publisher.PUBLISH_TOKEN_ENV: "dedicated-publisher-token", |
| 309 | + }, |
| 310 | + ) |
| 311 | + self.assertNotIn("must-not-be-exposed", str(caught.exception)) |
| 312 | + |
286 | 313 |
|
287 | 314 | if __name__ == "__main__": |
288 | 315 | unittest.main() |
0 commit comments