tegwick
047a6a2d31
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
"""No provider calls or response-body disclosure in readiness checks."""
|
|
import io
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib import error
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
|
|
from inference_preflight import NoRedirect, probe
|
|
|
|
|
|
class Response(io.BytesIO):
|
|
status = 200
|
|
|
|
|
|
class Opener:
|
|
def __init__(self, body=b'{"status":"ok"}', exception=None):
|
|
self.body, self.exception, self.requests = body, exception, []
|
|
|
|
def open(self, req, timeout):
|
|
self.requests.append((req, timeout))
|
|
if self.exception:
|
|
raise self.exception
|
|
return Response(self.body)
|
|
|
|
|
|
def test_success_is_one_get_with_no_prompt_or_reservation():
|
|
opener = Opener()
|
|
result = probe('http://127.0.0.1:18081', opener=opener)
|
|
req, timeout = opener.requests[0]
|
|
assert req.full_url == 'http://127.0.0.1:18081/health'
|
|
assert req.get_method() == 'GET' and req.data is None
|
|
assert len(opener.requests) == 1 and timeout == 5
|
|
assert result['ready'] and result['inference_calls'] == result['attempts_reserved'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize('body', [b'sensitive-invalid-body', b'{"status":"bad","secret":"sentinel"}', b'[]', b'x' * 4097])
|
|
def test_malformed_body_blocks_without_echo(body):
|
|
result = probe('http://localhost:8080', opener=Opener(body))
|
|
assert not result['ready'] and result['reason'] == 'invalid_health_response'
|
|
assert 'sentinel' not in json.dumps(result) and 'sensitive' not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.parametrize('exception,reason', [
|
|
(error.URLError('sensitive transport detail'), 'health_unreachable'),
|
|
(TimeoutError('sensitive timeout detail'), 'health_unreachable'),
|
|
(error.HTTPError('http://localhost', 503, 'sensitive error', {}, None), 'health_http_error'),
|
|
])
|
|
def test_failure_has_no_retry_and_no_error_detail(exception, reason):
|
|
opener = Opener(exception=exception)
|
|
result = probe('http://localhost:8080', opener=opener)
|
|
assert not result['ready'] and result['reason'] == reason
|
|
assert len(opener.requests) == 1 and 'sensitive' not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.parametrize('endpoint', ['file:///tmp/x', 'http://user:password@localhost', 'http://localhost/execute', 'http://localhost?token=x', 'http://localhost#x'])
|
|
def test_invalid_origin_never_requests(endpoint):
|
|
opener = Opener()
|
|
with pytest.raises(ValueError):
|
|
probe(endpoint, opener=opener)
|
|
assert not opener.requests
|
|
|
|
|
|
def test_redirect_is_not_followed():
|
|
assert NoRedirect().redirect_request(None, None, 302, '', {}, 'http://elsewhere') is None
|