yilsemandClaude Opus 5.5
a7ba300448
- One decision per real turn: a turn starts with a user message after the previous final answer; tool steps, messages typed mid-turn, a lone "." and compaction keep the combo. Works without prompt_cache_key too. - The virtual Otomatik name never reaches the gateway (stage off or error -> last choice or targets.zor); prefilter.target may not be it. - finishCheck: at most once per turn; a broken, empty or refused retry delivers the already approved first answer. - Layer window filter: only trusted windows, never the last layer. - Correction channel only on always-pass targets while checks are on; Sesli-Hizli-Zeki stays always-pass by default; a >256 KB SSE frame passes unfiltered instead of breaking the stream. - Turn-start questions wait in the local queue no longer than their timeout; decision-turn memory under a lock with refreshed order. - Dify fetch refuses redirects (the app key never moves host); the rules file write retries a Windows sharing violation. - An unreadable settings file no longer breaks every request. - TypeScript accepts only https:// URLs the bridge accepts; finishCheck.timeoutSeconds (unused) removed. Tests: 310 bridge tests, 36 electron tests; side-bridge live run for the turn, prefilter (incl. decision timeout) and finish-check paths. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
551 lines
27 KiB
Python
551 lines
27 KiB
Python
"""Decision stages (14 Oct 2026): settings schema 2, turn-start decisions, layer window filter, finish check
|
|
and Dify rules. Every new stage is off by default; the tests turn them on explicitly."""
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import threading
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import jev_denetim
|
|
import jev_settings
|
|
import karar_politikasi
|
|
import model_gecis
|
|
import pencere
|
|
import yarim
|
|
from test_jev_denetim import Yanit, method, sse, user
|
|
|
|
SOURCE = Path(__file__).with_name("kopru.py").read_text(encoding="utf-8")
|
|
TREE = ast.parse(SOURCE)
|
|
|
|
|
|
def fonksiyon(name):
|
|
return next(n for n in TREE.body if isinstance(n, ast.FunctionDef) and n.name == name)
|
|
|
|
|
|
class Ev(unittest.TestCase):
|
|
"""A temporary bridge home with a settings file."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.home = Path(self.tmp.name)
|
|
p = patch.object(jev_settings, 'home', return_value=self.home)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
self.settings = copy.deepcopy(jev_settings.DEFAULTS)
|
|
|
|
def save(self, settings=None, **extra):
|
|
data = {'settings': settings if settings is not None else self.settings}
|
|
data.update(extra)
|
|
(self.home / 'jev-settings.json').write_text(json.dumps(data), encoding='utf-8')
|
|
|
|
|
|
class Ayarlar(Ev):
|
|
def test_a_schema_1_file_gets_every_new_stage_disabled(self):
|
|
eski = {k: v for k, v in self.settings.items() if k not in jev_settings.SECTIONS}
|
|
self.save(eski)
|
|
s, _ = jev_settings.read()
|
|
for bolum in ('prefilter', 'autoRoute', 'finishCheck', 'verifierCheck'):
|
|
self.assertFalse(s[bolum]['enabled'], bolum)
|
|
self.assertEqual(s['policy']['source'], 'local')
|
|
self.assertEqual(s['watchModels'], jev_settings.DEFAULTS['watchModels'])
|
|
|
|
def test_a_partial_section_keeps_written_values_and_fills_the_rest(self):
|
|
self.settings['prefilter'] = {'enabled': True, 'threshold': 0.8}
|
|
self.save()
|
|
p = jev_settings.read()[0]['prefilter']
|
|
self.assertEqual((p['enabled'], p['threshold']), (True, 0.8))
|
|
self.assertEqual(p['target'], jev_settings.DEFAULTS['prefilter']['target'])
|
|
|
|
def test_bad_values_are_rejected(self):
|
|
for bolum, alan, deger in (('prefilter', 'threshold', 1.5), ('prefilter', 'timeoutSeconds', 0),
|
|
('autoRoute', 'model', ''), ('finishCheck', 'note', ''),
|
|
('policy', 'endpoint', 'http://dify.example'), ('policy', 'source', 'web')):
|
|
with self.subTest(bolum=bolum, alan=alan):
|
|
s = copy.deepcopy(self.settings)
|
|
s[bolum][alan] = deger
|
|
self.save(s)
|
|
with self.assertRaises(ValueError):
|
|
jev_settings.read()
|
|
|
|
def test_automatic_model_cannot_target_itself(self):
|
|
self.settings['autoRoute']['targets']['zor'] = 'Otomatik'
|
|
self.save()
|
|
with self.assertRaises(ValueError):
|
|
jev_settings.read()
|
|
|
|
def test_watch_channel_is_an_exact_name_from_settings(self):
|
|
self.settings['watchModels'] = ['Chat-Model']
|
|
self.save()
|
|
self.assertTrue(jev_denetim.izleme_mi('Chat-Model'))
|
|
for ad in ('chat-model', 'Chat-Model ', 'ide/Chat-Model', 'Sesli-Hizli-Zeki'):
|
|
self.assertFalse(jev_denetim.izleme_mi(ad), ad)
|
|
|
|
|
|
class Politika(Ev):
|
|
def test_dify_rules_apply_only_when_the_source_is_dify_and_they_are_valid(self):
|
|
(self.home / jev_settings.POLICY_FILE).write_text(json.dumps(
|
|
{'ok': True, 'overrides': {'prefilter': {'enabled': True, 'target': 'Katman-3'}}}), encoding='utf-8')
|
|
self.save()
|
|
self.assertFalse(jev_settings.read()[0]['prefilter']['enabled']) # source local
|
|
self.settings['policy']['source'] = 'dify'
|
|
self.save()
|
|
p = jev_settings.read()[0]['prefilter']
|
|
self.assertEqual((p['enabled'], p['target']), (True, 'Katman-3'))
|
|
self.assertFalse(jev_settings.read(overlay=False)[0]['prefilter']['enabled'])
|
|
(self.home / jev_settings.POLICY_FILE).write_text(json.dumps(
|
|
{'ok': True, 'overrides': {'prefilter': {'threshold': 7}}}), encoding='utf-8')
|
|
self.assertFalse(jev_settings.read()[0]['prefilter']['enabled']) # invalid rules: local in force
|
|
|
|
def test_rules_cannot_change_the_chain_or_unknown_sections(self):
|
|
s = jev_settings.read(overlay=False)[0]
|
|
with self.assertRaises(ValueError):
|
|
karar_politikasi.ayikla({'exemptModels': {'x': 1}}, s)
|
|
with self.assertRaises(ValueError):
|
|
karar_politikasi.ayikla({'autoRoute': {'targets': {'zor': 'Otomatik'}}}, s)
|
|
with self.assertRaises(ValueError):
|
|
karar_politikasi.ayikla('[1, 2]', s)
|
|
self.assertEqual(karar_politikasi.ayikla('{"autoRoute": {"enabled": true}}', s), {'autoRoute': {'enabled': True}})
|
|
|
|
def test_fetch_reads_the_kurallar_output_and_never_raises(self):
|
|
s = jev_settings.read(overlay=False)[0]
|
|
|
|
def yanit(govde):
|
|
r = Mock()
|
|
r.__enter__ = lambda self_: self_
|
|
r.__exit__ = lambda *a: False
|
|
r.read = lambda n=-1: json.dumps(govde).encode()
|
|
return r
|
|
istekler = []
|
|
|
|
def urlopen(req, timeout):
|
|
istekler.append(req)
|
|
return yanit({'data': {'outputs': {'kurallar': {'finishCheck': {'enabled': True}}}}})
|
|
sonuc = karar_politikasi.cek(s, 'app-key', urlopen=urlopen)
|
|
self.assertEqual(sonuc, {'ok': True, 'overrides': {'finishCheck': {'enabled': True}}})
|
|
self.assertTrue(istekler[0].full_url.endswith('/workflows/run'))
|
|
self.assertEqual(istekler[0].get_header('Authorization'), 'Bearer app-key')
|
|
self.assertFalse(karar_politikasi.cek(s, '', urlopen=urlopen)['ok'])
|
|
self.assertFalse(karar_politikasi.cek(s, 'k', urlopen=Mock(side_effect=OSError('kapali')))['ok'])
|
|
bos = karar_politikasi.cek(s, 'k', urlopen=lambda req, timeout: yanit({'data': {'outputs': {}}}))
|
|
self.assertEqual(bos['ok'], False)
|
|
|
|
def test_a_failed_fetch_keeps_the_last_good_rules(self):
|
|
jev_settings.write_policy({'ok': True, 'overrides': {'finishCheck': {'enabled': True}}})
|
|
kayit = jev_settings.write_policy({'ok': False, 'error': 'Dify kapali'})
|
|
self.assertTrue(kayit['ok'])
|
|
self.assertEqual(kayit['overrides'], {'finishCheck': {'enabled': True}})
|
|
self.assertEqual(kayit['error'], 'Dify kapali')
|
|
self.settings['policy']['source'] = 'dify'
|
|
self.save()
|
|
self.assertTrue(jev_settings.read()[0]['finishCheck']['enabled'])
|
|
|
|
|
|
class OnKarar(Ev):
|
|
def gonder(self, cevap):
|
|
self.govdeler = []
|
|
|
|
def f(govde, key):
|
|
self.govdeler.append(govde)
|
|
if isinstance(cevap, Exception):
|
|
raise cevap
|
|
return {'answers': {k: {'noul': v} for k, v in cevap.items()}}
|
|
return f
|
|
|
|
def test_automatic_routing_picks_research_then_hard_then_simple(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
t = self.settings['autoRoute']['targets']
|
|
for puan, hedef in (({'zor': 0.9, 'arastirma': 0.8}, t['arastirma']), ({'zor': 0.9, 'arastirma': 0.1}, t['zor']),
|
|
({'zor': 0.1, 'arastirma': 0.1}, t['basit'])):
|
|
sonuc = jev_denetim.on_karar('istek', True, False, 'k', gonder=self.gonder(puan))
|
|
self.assertEqual(sonuc['hedef'], hedef)
|
|
self.assertEqual(set(self.govdeler[0]['questions']), {'zor', 'arastirma'})
|
|
self.assertNotIn('assistant_reply', self.govdeler[0]['state'])
|
|
|
|
def test_no_decision_falls_to_the_strong_target(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
sonuc = jev_denetim.on_karar('istek', True, False, 'k', gonder=self.gonder(OSError('yok')))
|
|
self.assertEqual((sonuc['hedef'], sonuc['ret']), (self.settings['autoRoute']['targets']['zor'], None))
|
|
|
|
def test_refusal_question_goes_in_the_same_call(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
sonuc = jev_denetim.on_karar('istek', True, True, 'k', gonder=self.gonder({'zor': 0.1, 'arastirma': 0.1, 'ret': 0.9}))
|
|
self.assertEqual(len(self.govdeler), 1)
|
|
self.assertEqual(set(self.govdeler[0]['questions']), {'zor', 'arastirma', 'ret'})
|
|
self.assertTrue(sonuc['ret'])
|
|
|
|
def test_a_missing_answer_is_no_decision(self):
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
sonuc = jev_denetim.on_karar('istek', False, True, 'k', gonder=self.gonder({}))
|
|
self.assertIsNone(sonuc['ret'])
|
|
|
|
|
|
class TurBasi(Ev):
|
|
"""kopru.karar_asamasi through its own source."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.kararlar = []
|
|
self.logs = []
|
|
sahte_denetim = Mock(wraps=jev_denetim)
|
|
sahte_denetim.anahtar = lambda: 'k'
|
|
sahte_denetim.on_karar = self.on_karar
|
|
sahte_denetim.muaf_mi = jev_denetim.muaf_mi
|
|
sahte_denetim.sonrakiler = jev_denetim.sonrakiler
|
|
sahte_denetim.son_istek = jev_denetim.son_istek
|
|
self.cevap = {'hedef': None, 'ret': None, 'puan': None}
|
|
sahte_denetim.son_istek_sirasi = jev_denetim.son_istek_sirasi
|
|
sahte_denetim.istek_metni = jev_denetim.istek_metni
|
|
self.env = {'IDE_PREFIX': 'ide/', 'ON_KARAR': {}, 'ON_KARAR_SINIR': 3, 'jev_settings': jev_settings,
|
|
'jev_denetim': sahte_denetim, 'model_gecis': model_gecis, 'hashlib': hashlib, 'time': time,
|
|
'json': json, 'KARAR_KILIDI': threading.Lock(), 'BITIS_TURLARI': {}, 'log': self.logs.append}
|
|
govde = [fonksiyon(a) for a in ('tur_bilgisi', 'karar_asamasi', 'bitis_kullan')]
|
|
exec(compile(ast.Module(body=govde, type_ignores=[]), 'kopru.py', 'exec'), self.env)
|
|
self.karar = self.env['karar_asamasi']
|
|
|
|
def on_karar(self, istek, oto, ret, key):
|
|
self.kararlar.append((istek, oto, ret))
|
|
return dict(self.cevap)
|
|
|
|
def istek(self, model, *girdi):
|
|
return {'model': model, 'prompt_cache_key': 'sohbet-1', 'input': list(girdi)}
|
|
|
|
def test_everything_off_changes_nothing_and_asks_nothing(self):
|
|
self.save()
|
|
for model in ('ide/Ajan', 'ide/codex-gpt-6-astra', 'ide/Chat-Model'):
|
|
req = self.istek(model, user('yap'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', model), model)
|
|
self.assertEqual(req['model'], model)
|
|
self.assertEqual(self.kararlar, [])
|
|
|
|
def test_automatic_choice_is_kept_for_the_tool_steps_of_the_same_turn(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
self.cevap = {'hedef': 'Arastirma', 'ret': None, 'puan': {'zor': 0.2, 'arastirma': 0.9}}
|
|
req = self.istek('ide/Otomatik', user('kaynaklari karsilastir'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/Otomatik'), 'ide/Arastirma')
|
|
self.assertEqual(req['model'], 'ide/Arastirma')
|
|
arac = {'type': 'function_call_output', 'call_id': 'c1', 'output': 'ok'}
|
|
req2 = self.istek('ide/Otomatik', user('kaynaklari karsilastir'), {'type': 'function_call', 'call_id': 'c1'}, arac)
|
|
self.assertEqual(self.karar(2, req2, 'sohbet-1', 'ide/Otomatik'), 'ide/Arastirma')
|
|
self.assertEqual(len(self.kararlar), 1) # same turn: no second question
|
|
son_cevap = {'type': 'message', 'role': 'assistant', 'content': [{'type': 'output_text', 'text': 'Rapor hazir.'}]}
|
|
req3 = self.istek('ide/Otomatik', user('kaynaklari karsilastir'), arac, son_cevap, user('simdi kodu duzelt'))
|
|
self.cevap = {'hedef': 'Ajan', 'ret': None, 'puan': {'zor': 0.9, 'arastirma': 0.1}}
|
|
self.assertEqual(self.karar(3, req3, 'sohbet-1', 'ide/Otomatik'), 'ide/Ajan')
|
|
self.assertEqual(len(self.kararlar), 2) # new user message: new decision
|
|
|
|
def test_predicted_refusal_starts_at_the_configured_layer(self):
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
self.cevap = {'hedef': None, 'ret': True, 'puan': {'ret': 0.9}}
|
|
req = self.istek('ide/codex-gpt-6-astra', user('sert bir roast yaz'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/codex-gpt-6-astra'), 'ide/Katman-2')
|
|
self.assertEqual(self.kararlar, [('sert bir roast yaz', False, True)])
|
|
|
|
def test_exempt_and_later_layers_are_not_asked(self):
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
for model in ('ide/Chat-Model', 'ide/Katman-2', 'ide/Katman-3', 'codex/gpt-6-astra'):
|
|
req = self.istek(model, user('yap'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', model), model)
|
|
self.assertEqual(self.kararlar, [])
|
|
|
|
def test_no_refusal_or_no_decision_keeps_the_model(self):
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
for cevap in ({'hedef': None, 'ret': False, 'puan': {'ret': 0.1}}, {'hedef': None, 'ret': None, 'puan': None}):
|
|
self.env['ON_KARAR'].clear()
|
|
self.cevap = cevap
|
|
req = self.istek('ide/Ajan', user('yap'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/Ajan'), 'ide/Ajan')
|
|
self.assertEqual(req['model'], 'ide/Ajan')
|
|
|
|
def test_master_switch_off_disables_every_stage(self):
|
|
self.settings['enabled'] = False
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
req = self.istek('ide/Otomatik', user('yap'))
|
|
# The virtual name never leaves the bridge; nothing is asked while the main switch is off.
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/Otomatik'), 'ide/' + self.settings['autoRoute']['targets']['zor'])
|
|
self.assertEqual(self.kararlar, [])
|
|
|
|
def test_a_message_typed_mid_turn_or_a_dot_keeps_the_choice(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
self.cevap = {'hedef': 'Arastirma', 'ret': None, 'puan': {'zor': 0.2, 'arastirma': 0.9}}
|
|
bas = user('kaynaklari karsilastir')
|
|
cagri = {'type': 'function_call', 'call_id': 'c1', 'name': 'shell', 'arguments': '{}'}
|
|
cikti = {'type': 'function_call_output', 'call_id': 'c1', 'output': 'ok'}
|
|
self.assertEqual(self.karar(1, self.istek('ide/Otomatik', bas), 'sohbet-1', 'ide/Otomatik'), 'ide/Arastirma')
|
|
self.cevap = {'hedef': 'Workflow-Kesif', 'ret': None, 'puan': {'zor': 0.1, 'arastirma': 0.1}}
|
|
for ek in (user('.'), user('bir de sunu ekle'), {'type': 'compaction_trigger'}):
|
|
req = self.istek('ide/Otomatik', bas, cagri, cikti, ek)
|
|
self.assertEqual(self.karar(2, req, 'sohbet-1', 'ide/Otomatik'), 'ide/Arastirma', ek)
|
|
self.assertEqual(len(self.kararlar), 1)
|
|
|
|
def test_a_disabled_automatic_model_never_reaches_the_gateway(self):
|
|
self.save() # autoRoute off
|
|
req = self.istek('ide/Otomatik', user('yap'))
|
|
secilen = self.karar(1, req, 'sohbet-1', 'ide/Otomatik')
|
|
self.assertEqual(secilen, 'ide/' + self.settings['autoRoute']['targets']['zor'])
|
|
self.assertEqual(req['model'], secilen)
|
|
self.assertEqual(self.kararlar, [])
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.settings['enabled'] = False # main switch off
|
|
self.save()
|
|
self.assertNotEqual(self.karar(2, self.istek('ide/Otomatik', user('yap')), 'sohbet-1', 'ide/Otomatik'), 'ide/Otomatik')
|
|
|
|
def test_without_a_conversation_key_the_tool_steps_still_keep_the_choice(self):
|
|
self.settings['autoRoute']['enabled'] = True
|
|
self.save()
|
|
self.cevap = {'hedef': 'Ajan', 'ret': None, 'puan': {'zor': 0.9, 'arastirma': 0.1}}
|
|
bas = user('kodu duzelt')
|
|
self.assertEqual(self.karar(1, {'model': 'ide/Otomatik', 'input': [bas]}, '', 'ide/Otomatik'), 'ide/Ajan')
|
|
cikti = {'type': 'function_call_output', 'call_id': 'c1', 'output': 'ok'}
|
|
req = {'model': 'ide/Otomatik', 'input': [bas, {'type': 'function_call', 'call_id': 'c1'}, cikti]}
|
|
self.assertEqual(self.karar(2, req, '', 'ide/Otomatik'), 'ide/Ajan')
|
|
self.assertEqual(len(self.kararlar), 1)
|
|
|
|
def test_the_finish_retry_is_used_once_per_turn(self):
|
|
self.assertTrue(self.env['bitis_kullan']('tur-1'))
|
|
self.assertFalse(self.env['bitis_kullan']('tur-1'))
|
|
self.assertTrue(self.env['bitis_kullan']('tur-2'))
|
|
|
|
def test_turn_memory_is_bounded(self):
|
|
self.settings['prefilter']['enabled'] = True
|
|
self.save()
|
|
self.cevap = {'hedef': None, 'ret': False, 'puan': {'ret': 0.1}}
|
|
for i in range(6):
|
|
self.karar(i, {'model': 'ide/Ajan', 'prompt_cache_key': f's{i}', 'input': [user('yap')]}, f's{i}', 'ide/Ajan')
|
|
self.assertLessEqual(len(self.env['ON_KARAR']), 3)
|
|
|
|
|
|
class KatmanPenceresi(unittest.TestCase):
|
|
def suzgec(self, bilgi):
|
|
logs = []
|
|
env = {'IDE_PREFIX': 'ide/', 'IDE_BILGI': bilgi, 'pencere': pencere, 'log': logs.append}
|
|
exec(compile(ast.Module(body=[fonksiyon('sigan_katmanlar')], type_ignores=[]), 'kopru.py', 'exec'), env)
|
|
return env['sigan_katmanlar'], logs
|
|
|
|
def test_a_layer_too_small_for_the_request_is_skipped(self):
|
|
f, logs = self.suzgec({'ide/Katman-2': {'pencere': 1000, 'kaynak': 'katalog'}, 'ide/Katman-3': {'pencere': 10 ** 6}})
|
|
req = {'input': [user('x' * 20000)]}
|
|
self.assertEqual(f(['Katman-2', 'Katman-3', 'Bilinmeyen'], req, '[1]'), ['Katman-3', 'Bilinmeyen'])
|
|
self.assertIn('Katman-2', logs[0])
|
|
|
|
def test_the_last_layer_and_untrusted_windows_are_never_skipped(self):
|
|
req = {'input': [user('x' * 20000)]}
|
|
f, _ = self.suzgec({'ide/Katman-2': {'pencere': 1000, 'kaynak': 'katalog'}, 'ide/Katman-3': {'pencere': 1000, 'kaynak': 'katalog'}})
|
|
self.assertEqual(f(['Katman-2', 'Katman-3'], req, '[1]'), ['Katman-3']) # last one stays
|
|
f, _ = self.suzgec({'ide/Katman-2': {'pencere': 1000, 'kaynak': 'varsayilan'}})
|
|
self.assertEqual(f(['Katman-2', 'Katman-3'], req, '[1]'), ['Katman-2', 'Katman-3']) # placeholder window
|
|
|
|
|
|
class BitisDenetimi(Ev):
|
|
"""_denetimli_akis with finishCheck: an answer that stops early goes back to the same model once."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.settings['finishCheck']['enabled'] = True
|
|
self.save()
|
|
self.asked, self.logs, self.devamlar = [], [], []
|
|
self.puanlar = []
|
|
sahte = Mock()
|
|
sahte.anahtar = lambda: 'k'
|
|
sahte.sor = self.sor
|
|
sahte.karar = jev_denetim.karar
|
|
sahte.bitmedi_mi = jev_denetim.bitmedi_mi
|
|
sahte.hazirlik_mi = jev_denetim.hazirlik_mi
|
|
sahte.muaf_mi = jev_denetim.muaf_mi
|
|
sahte.hata_gecis = jev_denetim.hata_gecis
|
|
env = {'jev_denetim': sahte, 'yarim': yarim, 'time': time, 'http': __import__('http.client'),
|
|
'log': self.logs.append, 'RESP_HOP': set(), 'json': json}
|
|
adlar = ('_akis_basligi', '_denetimli_akis', '_stream')
|
|
exec(compile(ast.Module(body=[method(a) for a in adlar], type_ignores=[]), 'kopru.py', 'exec'), env)
|
|
self.h = type('H', (), {a: env[a] for a in adlar})()
|
|
self.h.wfile = io.BytesIO()
|
|
self.h.send_response = self.h.send_header = self.h.end_headers = Mock()
|
|
|
|
def sor(self, istek, metin, key, bitis=False):
|
|
self.asked.append((metin, bitis))
|
|
return self.puanlar.pop(0) if self.puanlar else {'ret': 0.0, 'sart': 0.0}
|
|
|
|
def devam(self, ad, onceki):
|
|
self.devamlar.append((ad, onceki))
|
|
return Yanit(sse('gercekten bitti, cikti: OK')), None, f'[{ad} devam]'
|
|
|
|
def calistir(self, body):
|
|
self.h._denetimli_akis(Yanit(body), '[1]', None, None, 'yap', ['Katman-1'], Mock(), hedef='Ajan', devam=self.devam)
|
|
return self.gonderilen()
|
|
|
|
def gonderilen(self):
|
|
raw, out = self.h.wfile.getvalue(), b''
|
|
while raw:
|
|
size, _, rest = raw.partition(b'\r\n')
|
|
n = int(size, 16)
|
|
if n == 0:
|
|
break
|
|
out += rest[:n]
|
|
raw = rest[n + 2:]
|
|
return out
|
|
|
|
def test_an_early_stop_goes_back_once_and_the_retry_is_delivered(self):
|
|
self.puanlar = [{'ret': 0.0, 'sart': 0.0, 'bitmedi': 0.9}, {'ret': 0.0, 'sart': 0.0, 'bitmedi': 0.9}]
|
|
out = self.calistir(sse('Bitti.'))
|
|
self.assertEqual(self.devamlar, [('Ajan', 'Bitti.')])
|
|
self.assertIn(b'gercekten bitti', out)
|
|
self.assertNotIn(b'"Bitti."', out)
|
|
self.assertEqual([b for _, b in self.asked], [True, False]) # the retry is not asked again
|
|
|
|
def test_a_finished_answer_is_delivered_unchanged(self):
|
|
self.puanlar = [{'ret': 0.0, 'sart': 0.0, 'bitmedi': 0.1}]
|
|
out = self.calistir(sse('Tamam, test ciktisi: 3 passed.'))
|
|
self.assertEqual(self.devamlar, [])
|
|
self.assertIn(b'3 passed', out)
|
|
|
|
def test_a_refusal_still_goes_down_the_chain_not_back(self):
|
|
self.puanlar = [{'ret': 0.9, 'sart': 0.0, 'bitmedi': 0.9}]
|
|
ac = Mock(return_value=(Yanit(sse('katman')), None, '[K1]'))
|
|
self.h._denetimli_akis(Yanit(sse('Yapamam.')), '[1]', None, None, 'yap', ['Katman-1'], ac, hedef='Ajan', devam=self.devam)
|
|
self.assertEqual(self.devamlar, [])
|
|
ac.assert_called_once_with('Katman-1')
|
|
|
|
def test_a_failed_retry_delivers_the_first_answer(self):
|
|
self.puanlar = [{'ret': 0.0, 'sart': 0.0, 'bitmedi': 0.9}]
|
|
|
|
def bozuk(ad, onceki):
|
|
raise OSError('acilamadi')
|
|
self.h._denetimli_akis(Yanit(sse('Bitti.')), '[1]', None, None, 'yap', [], Mock(), hedef='Ajan', devam=bozuk)
|
|
self.assertIn(b'Bitti.', self.gonderilen())
|
|
|
|
def test_a_broken_or_refused_retry_delivers_the_approved_first_answer(self):
|
|
for ikinci in (sse(), sse('Yapamam.')):
|
|
with self.subTest(ikinci=ikinci[:40]):
|
|
self.setUp()
|
|
self.puanlar = [{'ret': 0.0, 'sart': 0.0, 'bitmedi': 0.9}, {'ret': 0.9, 'sart': 0.0}]
|
|
ac = Mock(return_value=(Yanit(sse('katman')), None, '[K1]'))
|
|
self.h._denetimli_akis(Yanit(sse('Ilk cevap.')), '[1]', None, None, 'yap', ['Katman-1'], ac,
|
|
hedef='Ajan', devam=lambda ad, onceki: (Yanit(ikinci), None, '[devam]'))
|
|
self.assertIn(b'Ilk cevap.', self.gonderilen())
|
|
ac.assert_not_called()
|
|
|
|
def test_without_the_stage_the_finish_question_is_not_sent(self):
|
|
govdeler = []
|
|
|
|
def gonder(govde, key):
|
|
govdeler.append(govde)
|
|
return {'answers': {'ret': {'noul': 0.1}, 'sart': {'noul': 0.1}}}
|
|
puan = jev_denetim.sor('i', 'c', 'k', gonder=gonder)
|
|
self.assertNotIn('bitmedi', govdeler[0]['questions'])
|
|
self.assertNotIn('bitmedi', puan)
|
|
puan = jev_denetim.sor('i', 'c', 'k', gonder=gonder, bitis=True)
|
|
self.assertIn('bitmedi', govdeler[1]['questions'])
|
|
self.assertEqual(set(puan), {'ret', 'sart'}) # unanswered extra question: main verdict stands
|
|
self.assertFalse(jev_denetim.bitmedi_mi(puan))
|
|
|
|
|
|
|
|
class DogrulayiciElemesi(Ev):
|
|
def test_off_by_default_and_asked_only_when_enabled(self):
|
|
import io as _io
|
|
import karar_sor
|
|
self.save()
|
|
out = _io.StringIO()
|
|
self.assertEqual(karar_sor.main(['dogrulayici'], _io.StringIO('{"user_request": "x", "assistant_reply": "kabul"}'), out), 0)
|
|
self.assertEqual(json.loads(out.getvalue())['enabled'], False)
|
|
self.settings['verifierCheck']['enabled'] = True
|
|
self.save()
|
|
govdeler = []
|
|
|
|
def gonder(govde, key):
|
|
govdeler.append(govde)
|
|
return {'answers': {'kanitsiz': {'noul': 0.92}}}
|
|
sonuc = karar_sor.dogrulayici({'user_request': 'olcut', 'assistant_reply': '{"karar":"kabul","bulgular":[]}'}, 'k', gonder)
|
|
self.assertEqual((sonuc['enabled'], sonuc['kanitsiz'], sonuc['puan']), (True, True, 0.92))
|
|
self.assertEqual(set(govdeler[0]['questions']), {'kanitsiz'})
|
|
bozuk = karar_sor.dogrulayici({'user_request': 'o', 'assistant_reply': 'r'}, 'k', Mock(side_effect=OSError('yok')))
|
|
self.assertIsNone(bozuk['kanitsiz'])
|
|
|
|
def test_wrong_usage_exits_2(self):
|
|
import io as _io
|
|
import karar_sor
|
|
self.assertEqual(karar_sor.main(['baska'], _io.StringIO('{}'), _io.StringIO()), 2)
|
|
|
|
|
|
class Inceleme(Ev):
|
|
def test_the_correction_channel_needs_an_always_pass_target_while_checks_are_on(self):
|
|
self.settings['watchModels'] = ['Chat-Model', 'Ajan']
|
|
self.save()
|
|
self.assertTrue(jev_denetim.izleme_mi('Chat-Model'))
|
|
self.assertFalse(jev_denetim.izleme_mi('Ajan')) # checked target: no sideband
|
|
self.settings['enabled'] = False
|
|
self.save()
|
|
self.assertTrue(jev_denetim.izleme_mi('Ajan'))
|
|
|
|
def test_a_turn_start_question_does_not_wait_in_the_queue_beyond_its_timeout(self):
|
|
self.settings['maxConcurrent'] = 1
|
|
self.settings['queueWaitSeconds'] = 3600
|
|
self.save()
|
|
s = jev_settings.read()[0]
|
|
self.assertTrue(jev_settings.acquire(s)) # the only slot is taken
|
|
try:
|
|
t0 = time.monotonic()
|
|
self.assertIsNone(jev_denetim.sorular({'user_request': 'x'}, {'zor': 'q'}, 'k', 1, gonder=lambda g, k: {}))
|
|
self.assertLess(time.monotonic() - t0, 3)
|
|
finally:
|
|
jev_settings.release()
|
|
|
|
def test_the_prefilter_target_cannot_be_the_automatic_model(self):
|
|
self.settings['prefilter']['target'] = self.settings['autoRoute']['model']
|
|
self.save()
|
|
with self.assertRaises(ValueError):
|
|
jev_settings.read()
|
|
|
|
def test_dify_redirects_are_refused_so_the_key_never_moves(self):
|
|
import urllib.error
|
|
with self.assertRaises(urllib.error.HTTPError):
|
|
karar_politikasi._YonlendirmeYok().redirect_request(
|
|
Mock(full_url='https://dify.example/v1/workflows/run'), None, 302, 'Found', {}, 'http://evil.example/')
|
|
|
|
|
|
class IzlemeSuzgeci(unittest.TestCase):
|
|
def test_a_huge_frame_passes_through_instead_of_breaking_the_stream(self):
|
|
tree_sinif = next(n for n in TREE.body if isinstance(n, ast.ClassDef) and n.name == 'WatchResponse')
|
|
from bekci_watch import MetadataFilter
|
|
logs = []
|
|
env = {'MetadataFilter': MetadataFilter, 'log': logs.append}
|
|
exec(compile(ast.Module(body=[tree_sinif], type_ignores=[]), 'kopru.py', 'exec'), env)
|
|
buyuk = b'data: ' + b'x' * 300000
|
|
parcalar = [buyuk[:200000], buyuk[200000:], b'\n\n', b'']
|
|
|
|
class Ham:
|
|
def read1(self, n):
|
|
return parcalar.pop(0)
|
|
w = env['WatchResponse'](Ham(), lambda m: None)
|
|
cikti = b''
|
|
while True:
|
|
b = w.read1(65536)
|
|
if not b and w.done or not parcalar and not b:
|
|
break
|
|
cikti += b
|
|
self.assertIn(b'x' * 1000, cikti)
|
|
self.assertEqual(cikti.count(b'x'), 300000)
|
|
self.assertTrue(logs)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|