gemini-api-dev/examples/run_fixture.py

Version 832c8f94114d.bb1 · Apache-2.0. This preview displays packaged text and does not execute code. Treat the contents as untrusted instructions.

← Return to resource and package checksum

"""Execute six preserved Python examples with native SDK and mock HTTP only.

Constructor binding, missing client prefixes, canned responses and bounded sleep
are explicit fixture changes. No real model, managed agent or paid API is used.
"""
from pathlib import Path
import contextlib,datetime as dt,hashlib,importlib.metadata,io,json,re,socket
from unittest import mock
import httpx
from google import genai

b=Path(__file__).parent
packaged_manifest=b.parent/'SOURCE-MANIFEST.json'
source=b.parent/'upstream-original' if packaged_manifest.exists() else b.parent/'curation-gemini-v54'
manifest=json.loads((packaged_manifest if packaged_manifest.exists() else source/'source-manifest.json').read_text())
path=source/('skills/gemini-api-dev/SKILL.md' if packaged_manifest.exists() else 'upstream/skills/gemini-api-dev/SKILL.md');raw=path.read_bytes()
record=next(r for r in manifest['files'] if r['path']=='skills/gemini-api-dev/SKILL.md')
assert hashlib.sha256(raw).hexdigest()==record['sha256']
blocks=re.findall(r'```([^\n]*)\n(.*?)```',raw.decode(),re.S)
python_blocks={i:code for i,(lang,code) in enumerate(blocks) if lang=='python'}
assert list(python_blocks)==[0,2,4,6,8,10]
NativeClient=genai.Client;observations=[];socket_attempts=[]
class PollingLimit(RuntimeError):pass
def deny(*args,**kwargs):socket_attempts.append('denied');raise RuntimeError('External network denied in fixture')
def interaction(identity='synthetic-interaction',status='completed',text='Synthetic result.'):
    return {'id':identity,'status':status,'steps':[{'type':'model_output','content':[{'type':'text','text':text}]}],'environment_id':'synthetic-environment','usage':{'total_tokens':7}}
def execute(index,mode='completed'):
    requests=[];polls=0;sleeps=[]
    def handler(req):
        nonlocal polls
        assert req.url.host=='fixture.invalid'
        body=json.loads(req.content) if req.content else None
        requests.append({'method':req.method,'path':req.url.path,'query':str(req.url.query.decode()),'body':body})
        if req.method=='GET':
            polls+=1
            if mode=='cancelled' and polls>2:raise PollingLimit('Original polling loop did not terminate on cancelled after two mocked responses')
            status='in_progress' if mode=='completed' and polls==1 else mode
            payload=interaction(status=status)
            if status=='failed':payload['error']={'code':'synthetic_failure','message':'Synthetic failure'}
        elif req.url.path.endswith('/agents'):
            payload={'id':'code-reviewer','base_agent':'antigravity-preview-09-2026'}
        elif body and body.get('stream'):
            events=[{'event_type':'step.delta','index':0,'delta':{'type':'text','text':'Synthetic stream.'}},{'event_type':'interaction.completed','interaction':interaction()}]
            content=''.join('event: '+e['event_type']+'\ndata: '+json.dumps(e)+'\n\n' for e in events)
            return httpx.Response(200,headers={'content-type':'text/event-stream'},content=content.encode(),request=req)
        else:
            payload=interaction('synthetic-'+str(len(requests)),status='in_progress' if index==4 else 'completed')
        return httpx.Response(200,json=payload,request=req)
    transport=httpx.MockTransport(handler)
    with httpx.Client(transport=transport,trust_env=False) as http:
        client=NativeClient(vertexai=False,enterprise=False,api_key='synthetic-not-a-real-api-key',http_options={'base_url':'https://fixture.invalid','httpx_client':http})
        namespace={'client':client};output=io.StringIO();counterexample=False
        try:
            with mock.patch.object(genai,'Client',return_value=client),mock.patch('time.sleep',side_effect=lambda seconds:sleeps.append(seconds)),mock.patch.object(socket.socket,'connect',deny),mock.patch.object(socket,'create_connection',deny),mock.patch.object(socket,'getaddrinfo',deny),contextlib.redirect_stdout(output):
                exec(compile(python_blocks[index],f'preserved-SKILL-block-{index}.py','exec'),namespace)
        except PollingLimit:
            assert index==4 and mode=='cancelled';counterexample=True
        finally:client.close()
    result={'block_index':index,'source_block_sha256':hashlib.sha256(python_blocks[index].encode()).hexdigest(),'mode':mode,'requests':requests,'stdout':output.getvalue(),'sleep_seconds_recorded_not_slept':sleeps,'bounded_cancelled_loop_counterexample':counterexample,'constructor_injected':index in [0,6],'missing_client_prefix_supplied':index not in [0,6]}
    observations.append(result)
    return result

quick=execute(0);assert len(quick['requests'])==1 and quick['stdout']=='Synthetic result.\n'
stateful=execute(2);assert len(stateful['requests'])==2 and stateful['requests'][1]['body']['previous_interaction_id']=='synthetic-1'
research=execute(4);assert [r['method'] for r in research['requests']]==['POST','GET','GET'] and research['sleep_seconds_recorded_not_slept']==[10]
failed=execute(4,'failed');assert failed['stdout'].startswith('Failed:') and not failed['sleep_seconds_recorded_not_slept']
cancelled=execute(4,'cancelled');assert cancelled['bounded_cancelled_loop_counterexample'] and cancelled['sleep_seconds_recorded_not_slept']==[10,10]
remote=execute(6);assert remote['requests'][0]['body']['environment']=='remote' and 'synthetic-environment' in remote['stdout']
custom=execute(8);assert len(custom['requests'])==2 and custom['requests'][0]['path'].endswith('/agents') and custom['requests'][1]['body']['agent']=='code-reviewer'
stream=execute(10);assert 'Synthetic stream.' in stream['stdout'] and 'Total Tokens: 7' in stream['stdout']
assert not socket_attempts
evidence={'verified_at':dt.datetime.now(dt.UTC).isoformat(),'source_commit':manifest['commit'],'original_python_blocks':6,'original_typescript_blocks_not_executed':6,'recorded_runs':len(observations),'python_sdk':importlib.metadata.version('google-genai'),'httpx':importlib.metadata.version('httpx'),'pydantic':importlib.metadata.version('pydantic'),'transport':'httpx.MockTransport with native Google GenAI request serialization and response parsing','network_attempts':len(socket_attempts),'real_credentials_used':False,'paid_api_calls':0,'real_models_or_managed_agents_executed':False,'model_ids_are_unverified_example_strings':True,'full_skill_runtime_tested':False,'upstream_examples_not_rewritten':True,'runs':observations,'limits':['Constructor is injected; other snippets receive a client prefix.','Synthetic responses do not verify model availability, generated quality, billing, auth or server-side persistence.','Sleep is recorded, not performed; the cancelled-loop counterexample is forcibly bounded.','No TypeScript, multimodal, TTS, real-time Live API, remote execution or repository access was exercised.']}
result_dir=b/'output' if packaged_manifest.exists() else b
result_dir.mkdir(exist_ok=True)
(result_dir/'fixture-evidence.json').write_text(json.dumps(evidence,indent=2)+'\n',encoding='utf-8',newline='\n')
locked=sorted((d.metadata['Name']+'=='+d.version) for d in importlib.metadata.distributions())
(result_dir/'installed-distributions.txt').write_text('\n'.join(locked)+'\n',encoding='utf-8',newline='\n')
print(json.dumps({k:v for k,v in evidence.items() if k not in ['runs','limits']}))