84 lines
4.1 KiB
Python
84 lines
4.1 KiB
Python
"""Native switch regressions using the real HTTP splitter and offline peers."""
|
|
import copy
|
|
import gzip
|
|
import json
|
|
import unittest
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
import bridge
|
|
import gecis_context as gc
|
|
import ozet_kopru
|
|
import test_ayirici as fixtures
|
|
|
|
|
|
class NativeSwitch(unittest.TestCase):
|
|
setUp = fixtures.Ayirici.setUp
|
|
tearDown = fixtures.Ayirici.tearDown
|
|
iste = fixtures.Ayirici.iste
|
|
|
|
def test_unknown_native_capacity_is_not_invented(self):
|
|
tree = ast.parse(Path(__file__).with_name('kopru.py').read_text(encoding='utf-8'))
|
|
method = next(node for node in ast.walk(tree)
|
|
if isinstance(node, ast.FunctionDef) and node.name == '_transfer_window')
|
|
ns = {'ek_model_mu': lambda _: False, 'LAST_MODELS': {}, 'gecis_context': gc, 'json': json}
|
|
exec(compile(ast.Module(body=[method], type_ignores=[]), 'window', 'exec'), ns)
|
|
from types import SimpleNamespace
|
|
with self.assertRaises(gc.TransferError):
|
|
ns['_transfer_window'](SimpleNamespace(headers={}), 'gpt-6-astra')
|
|
|
|
def test_opaque_native_switch_with_persisted_foreign_residue(self):
|
|
self.ozgun.cevap = fixtures.duz(200, b'data: {"type":"response.completed"}\n\n', 'text/event-stream')
|
|
compact = {'type': 'compaction', 'id': 'cmp_native',
|
|
'encrypted_content': 'gAAAAABp' + 'AqZ09_-jX' * 10000}
|
|
req = {'model': 'gpt-6.1-sol', 'prompt_cache_key': 'restart', 'input': [
|
|
ozet_kopru.ozet_ogesi('DECISION-4827 must survive'), compact,
|
|
{'type': 'custom_tool_call', 'id': 'fc_123', 'call_id': 'call_123',
|
|
'name': 'apply_patch', 'input': 'patch text'},
|
|
{'type': 'custom_tool_call_output', 'id': 'fco_123',
|
|
'call_id': 'call_123', 'output': 'complete'},
|
|
{'role': 'user', 'content': 'continue'}]}
|
|
before = copy.deepcopy(req)
|
|
for cold, model in ((True, 'gpt-6.1-sol'), (False, 'gpt-6-astra'), (True, 'gpt-6-astra')):
|
|
if cold:
|
|
bridge.TRANSFERS.clear()
|
|
req['model'] = model
|
|
body = gzip.compress(json.dumps(req).encode())
|
|
self.assertEqual(self.iste('POST', '/backend-api/codex/responses', body,
|
|
{'Content-Encoding': 'gzip'})[0], 200)
|
|
sent = json.loads(self.ozgun.gelen[-1]['govde'])
|
|
self.assertEqual(sent['input'][1], compact)
|
|
self.assertEqual(sent['input'][2]['id'], 'ctc_123')
|
|
self.assertEqual(sent['input'][3]['id'], 'ctco_123')
|
|
self.assertEqual(sent['input'][3]['call_id'], 'call_123')
|
|
self.assertIn('DECISION-4827', json.dumps(sent))
|
|
self.assertEqual(self.kopru.gelen, [])
|
|
self.assertEqual(req['input'], before['input'])
|
|
|
|
def test_clean_native_body_is_byte_identical_even_after_external_source(self):
|
|
req = {'model': 'ide/test', 'prompt_cache_key': 'return', 'input': []}
|
|
self.iste('POST', '/backend-api/codex/responses', json.dumps(req).encode())
|
|
self.kopru.gelen.clear()
|
|
req.update(model='gpt-6-astra', input=[{'type': 'compaction',
|
|
'encrypted_content': 'gAAAAABp' + 'X' * 100000}])
|
|
raw = gzip.compress(json.dumps(req, indent=3).encode())
|
|
self.iste('POST', '/backend-api/codex/responses', raw, {'Content-Encoding': 'gzip'})
|
|
self.assertEqual(self.ozgun.gelen[-1]['govde'], raw)
|
|
self.assertEqual(self.kopru.gelen, [])
|
|
|
|
def test_cold_native_tool_id_repair_does_not_require_semantic_budget(self):
|
|
req = {'model': 'gpt-6-astra', 'input': [
|
|
{'type': 'custom_tool_call', 'id': 'fc_old', 'call_id': 'call_old',
|
|
'name': 'apply_patch', 'input': 'x' * 100000},
|
|
{'type': 'custom_tool_call_output', 'id': 'fco_old',
|
|
'call_id': 'call_old', 'output': 'complete'}]}
|
|
self.iste('POST', '/backend-api/codex/responses', json.dumps(req).encode())
|
|
self.assertEqual(len(self.kopru.gelen), 0)
|
|
sent = json.loads(self.ozgun.gelen[-1]['govde'])
|
|
self.assertEqual(sent, gc.normalize(req))
|
|
self.assertEqual(req['input'][0]['id'], 'fc_old')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|