Establish dependency returns and block gated workplans
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
This commit is contained in:
parent
70f0408240
commit
047a6a2d31
13 changed files with 331 additions and 14 deletions
67
tests/test_inference_preflight.py
Normal file
67
tests/test_inference_preflight.py
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
"""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
|
||||
Loading…
Add table
Add a link
Reference in a new issue