fastapi/examples/run_contract_fixture.py
Version 0.142.2.bb1 · MIT. This preview displays packaged text and does not execute code. Treat the contents as untrusted instructions.
← Return to resource and package checksum
"""Run pinned native snippets and separately identified synthetic examples.
No AI client, production endpoint, credentials, real database or external API.
"""
import datetime as dt,faulthandler,hashlib,importlib.metadata as meta,ipaddress,json,platform,re,socket,sys,types
from pathlib import Path
from unittest.mock import patch
from fastapi.exceptions import ResponseValidationError
from fastapi.testclient import TestClient
from public_contract import create_example_app
root=Path(__file__).resolve().parents[1]
observations=[];executed=[];requests=0;loopback_connections=0
def check(name,condition,detail):
if not condition:raise AssertionError(name+': '+detail)
observations.append({'name':name,'status':'passed','observation':detail})
class LocalClient(TestClient):
def request(self,method,url,*args,**kwargs):
global requests
assert str(url).startswith('/') and not str(url).startswith('//')
requests+=1
return super().request(method,url,*args,**kwargs)
def snippet(relative,index,extra=None):
path=root/relative;source=path.read_text(encoding='utf-8')
blocks=re.findall(r'```python\s*\n(.*?)```',source,re.S);code=blocks[index]
name='native_fastapi_fixture_'+str(len(executed));module=types.ModuleType(name);sys.modules[name]=module
module.__dict__.update(extra or {})
exec(compile(code,str(path)+':python-block-'+str(index),'exec'),module.__dict__)
executed.append({'source':relative,'block_index':index,'block_sha256':hashlib.sha256(code.encode()).hexdigest(),'provided_synthetic_symbols':sorted(extra or {})})
return module
def rejection(response,location,error):
return response.status_code==422 and any(item['loc']==location and item['type']==error for item in response.json()['detail'])
def guard(original):
def connect(sock,address,*args,**kwargs):
global loopback_connections
if isinstance(address,tuple):
host=address[0]
try:allowed=ipaddress.ip_address(host).is_loopback
except ValueError:allowed=host=='localhost'
if not allowed:raise AssertionError('External socket connection prohibited')
loopback_connections+=1
else:raise AssertionError('Unexpected socket address')
return original(sock,address,*args,**kwargs)
return connect
def run():
native=snippet('upstream-original/SKILL.md',0)
with LocalClient(native.app) as client:
check('native_annotated_valid_path',client.get('/items/1?q=sample').json()=={'message':'Hello World'},'Pinned Annotated Path/Query example returns its declared response.')
check('native_path_lower_bound',rejection(client.get('/items/0'),['path','item_id'],'greater_than_equal'),'Path zero rejected with 422 and matching lower-bound error.')
check('native_path_type',rejection(client.get('/items/not-an-integer'),['path','item_id'],'int_parsing'),'Invalid integer path rejected at the path field.')
check('native_query_max_length',rejection(client.get('/items/1?q='+'x'*51),['query','q'],'string_too_long'),'Query longer than the native limit is rejected.')
native=snippet('upstream-original/SKILL.md',1)
with LocalClient(native.app) as client:
check('native_dependency_original',client.get('/items/').json()=={'message':'Hello World'},'The upstream endpoint returns a constant message; its synthetic identity is not authentication.')
calls=[]
def local_identity():calls.append('fixture-alice');return {'username':'fixture-alice'}
native.app.dependency_overrides[native.get_current_user]=local_identity
try:
response=client.get('/items/')
check('dependency_override_applied',response.json()=={'message':'Hello World'} and calls==['fixture-alice'],'The function-keyed test override runs; the native endpoint still returns its constant message.')
finally:native.app.dependency_overrides.clear()
response=client.get('/items/')
check('dependency_override_restored',response.json()=={'message':'Hello World'} and not native.app.dependency_overrides and calls==['fixture-alice'],'After cleanup, the overriding provider is not called on the later request.')
native=snippet('references/pydantic.md',0)
with LocalClient(native.app) as client:
check('native_required_query_and_body_valid',client.post('/items/?project_id=2',json={'name':'Fixture','price':1}).json()=={'name':'Fixture','description':None,'price':1.0},'Native typed body and required query work together.')
check('native_required_query_missing',rejection(client.post('/items/',json={'name':'Fixture','price':1}),['query','project_id'],'missing'),'Required query omitted: matching 422 query error.')
check('native_required_body_missing',rejection(client.post('/items/?project_id=2',json={'name':'Fixture'}),['body','price'],'missing'),'Required body value omitted: matching 422 body error.')
check('native_positive_price_constraint',rejection(client.post('/items/?project_id=2',json={'name':'Fixture','price':0}),['body','price'],'greater_than'),'Zero price fails the upstream positive-price constraint.')
native=snippet('references/pydantic.md',2)
with LocalClient(native.app) as client:
check('native_annotated_list',client.post('/items/',json=[1,2]).json()==[1,2],'Native Annotated list validation works without RootModel.')
check('native_empty_list_rejected',rejection(client.post('/items/',json=[]),['body'],'too_short'),'The native minimum list length is enforced.')
check('native_list_item_type_rejected',rejection(client.post('/items/',json=['invalid']),['body',0],'int_parsing'),'Invalid list item reports its exact index.')
native=snippet('references/responses.md',0)
with LocalClient(native.app) as client:
check('native_return_annotation',client.get('/items/me').json()=={'name':'Plumbus','description':'All-purpose home device'},'Upstream return annotation serializes its public Item.')
native=snippet('references/responses.md',1)
with LocalClient(native.app) as client:
check('native_response_model_dictionary',client.get('/items/me').json()=={'name':'Foo','description':'A very nice Item'},'Upstream response_model validates a dictionary return.')
native=snippet('references/responses.md',2)
with LocalClient(native.app) as client:
result=client.get('/items/me')
check('native_internal_field_filtered',result.status_code==200 and result.json()=={'name':'Foo','description':'A very nice Item'} and 'secret_key' not in result.json(),'The original InternalItem field is absent from the public response.')
schema=client.get('/openapi.json').json()
check('native_public_response_schema',set(schema['components']['schemas']['Item']['properties'])=={'name','description'},'Public OpenAPI Item schema excludes the internal field.')
events=[]
class FakeQuery:
def all(self):return [{'name':'Synthetic item'}]
class DBSession:
def __init__(self):events.append('acquire')
def query(self,item):events.append('query');return FakeQuery()
def close(self):events.append('release')
native=snippet('references/dependencies.md',0,{'DBSession':DBSession,'Item':object})
def observe(app):
async def wrapped(scope,receive,send):
async def recording_send(message):
if scope['type']=='http' and message['type'] in {'http.response.start','http.response.body'}:events.append(message['type'])
await send(message)
await app(scope,receive,recording_send)
return wrapped
with LocalClient(observe(native.app)) as client:
check('native_yield_request_result',client.get('/items/').json()==[{'name':'Synthetic item'}],'Original yield dependency uses a synthetic session/query, not a database.')
check('native_request_scope_cleanup',events.count('release')==1 and events.index('release')>events.index('http.response.body'),'Default request-scope release follows the ASGI response body.')
events.clear();native=snippet('references/dependencies.md',1)
# Use the original generator and alias; observe its finally block without changing its code.
native.__dict__['print']=lambda *args:events.append('release')
with LocalClient(observe(native.app)) as client:
check('native_yield_function_result',client.get('/users/me').json()=='Rick','Native function-scoped yield dependency supplies its value.')
check('native_function_scope_cleanup',events.count('release')==1 and events.index('release')<events.index('http.response.start'),'Function-scope release precedes the ASGI response start.')
native=snippet('references/dependencies.md',2)
with LocalClient(native.app) as client:
check('native_function_returns_class_instance',client.get('/items/?offset=2&limit=3&q=local').json()=={'offset':2,'limit':3,'q':'local','items':[]},'Native function dependency constructs the local paginator.')
app=create_example_app()
with LocalClient(app) as client:
check('independent_public_projection',client.post('/products',json={'name':'Fixture','cost':1}).json()=={'name':'Fixture','cost':1.0},'Separate strict example returns only public fields.')
check('independent_unknown_field_rejected',rejection(client.post('/products',json={'name':'Fixture','cost':1,'unexpected':True}),['body','unexpected'],'extra_forbidden'),'Independent extra=forbid input contract rejects an unknown field.')
check('independent_empty_name_rejected',rejection(client.post('/products',json={'name':'','cost':1}),['body','name'],'string_too_short'),'Independent minimum name length is enforced.')
errors=[]
try:client.get('/broken')
except ResponseValidationError as exc:errors=exc.errors()
check('invalid_server_response_raises',any(x['loc']==('response','cost') and x['type']=='missing' for x in errors),'Default TestClient reports a response contract exception, not a client 422.')
result=client.get('/direct-response')
check('direct_response_bypasses_projection',result.status_code==200 and 'internal_note' in result.json(),'Explicit synthetic counterexample: JSONResponse bypasses declared response_model filtering.')
with LocalClient(app,raise_server_exceptions=False) as client:
check('invalid_server_response_is_500',client.get('/broken').status_code==500,'With server exceptions disabled, invalid returned data is a server error.')
if __name__=='__main__':
faulthandler.dump_traceback_later(45,exit=True)
assert meta.version('fastapi')=='0.142.2'
with patch.object(socket.socket,'connect',guard(socket.socket.connect)),patch.object(socket.socket,'connect_ex',guard(socket.socket.connect_ex)):run()
hashes={name:hashlib.sha256((root/name).read_bytes()).hexdigest() for name in ['upstream-original/SKILL.md','references/pydantic.md','references/responses.md','references/dependencies.md','examples/public_contract.py','examples/run_contract_fixture.py']}
versions={name:meta.version(name) for name in ['fastapi','pydantic','starlette','anyio']}
for name in ['httpx','httpx2']:
try:versions[name]=meta.version(name)
except meta.PackageNotFoundError:pass
result={'executed_at':dt.datetime.now(dt.UTC).isoformat(),'python':platform.python_version(),'platform':platform.system(),'versions':versions,'test_transport_module':TestClient.__mro__[1].__module__,'observations':len(observations),'cases':observations,'native_snippets_executed':len(executed),'executed_snippets':executed,'source_hashes':hashes,'in_process_http_requests':requests,'loopback_socket_connections':loopback_connections,'external_network_requests':False,'external_connections_prohibited':True,'runtime_tested':False,'ai_client_executed':False,'production_tested':False,'database_connected':False,'credentials_required':False,'paid_api_used':False}
faulthandler.cancel_dump_traceback_later()
(root/'examples/native-api-evidence.json').write_text(json.dumps(result,indent=2),encoding='utf-8',newline='\n');print(json.dumps(result))