|
6 | 6 | import unittest |
7 | 7 | from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
8 | 8 | from pathlib import Path |
| 9 | +from unittest.mock import patch |
| 10 | +from urllib.error import URLError |
9 | 11 |
|
10 | | -from cas_evals.reference_product import ReferenceProductError, evaluate_reference_suite |
| 12 | +from cas_evals.reference_product import ( |
| 13 | + ReferenceProductError, |
| 14 | + _http_transport, |
| 15 | + evaluate_reference_suite, |
| 16 | +) |
11 | 17 |
|
12 | 18 | ROOT = Path(__file__).parents[1] |
13 | 19 | GOLDEN = ROOT / "benchmarks/reference-product/v0.1/golden.json" |
@@ -105,6 +111,18 @@ def test_invalid_endpoint_and_timeout_fail_closed(self): |
105 | 111 | with self.assertRaises(ReferenceProductError): |
106 | 112 | evaluate_reference_suite(GOLDEN, timeout_seconds=0) |
107 | 113 |
|
| 114 | + def test_network_failure_preserves_root_cause(self): |
| 115 | + failure = URLError("network unavailable") |
| 116 | + transport = _http_transport("http://127.0.0.1:8080/api/v1/workflows", 1.0) |
| 117 | + |
| 118 | + with patch("cas_evals.reference_product.urlopen", side_effect=failure): |
| 119 | + with self.assertRaisesRegex( |
| 120 | + ReferenceProductError, "reference product is unavailable" |
| 121 | + ) as caught: |
| 122 | + transport({"kind": "PromptEnvelope"}) |
| 123 | + |
| 124 | + self.assertIs(caught.exception.__cause__, failure) |
| 125 | + |
108 | 126 | def test_http_endpoint_is_executable(self): |
109 | 127 | server = ThreadingHTTPServer(("127.0.0.1", 0), ReferenceHandler) |
110 | 128 | thread = threading.Thread(target=server.serve_forever, daemon=True) |
|
0 commit comments