"""Protocol-level S3 contract check with botocore's validation stub (no live S3).""" import io import boto3 from botocore.response import StreamingBody from botocore.stub import Stubber from app.storage import Storage def test_s3_private_round_trip_contract(tmp_path,monkeypatch): from app.config import settings from app.storage import storage monkeypatch.setenv('STORAGE_DIR',str(tmp_path)); monkeypatch.setenv('S3_BUCKET','test-private') settings.cache_clear(); storage.cache_clear() client=boto3.client('s3',region_name='us-east-1',aws_access_key_id='test',aws_secret_access_key='test') monkeypatch.setattr(boto3,'client',lambda *a,**kw:client) key='a'*48+'/input.SchDoc'; payload=b'actual bytes' with Stubber(client) as stub: stub.add_response('put_object',{}, {'Bucket':'test-private','Key':key,'Body':payload,'ContentType':'application/octet-stream'}) stub.add_response('get_object',{'Body':StreamingBody(io.BytesIO(payload),len(payload))},{'Bucket':'test-private','Key':key}) stub.add_response('delete_object',{}, {'Bucket':'test-private','Key':key}) service=Storage();service.put(key,payload);assert service.get(key)==payload;service.delete(key) stub.assert_no_pending_responses() settings.cache_clear(); storage.cache_clear()