yilsemandClaude Opus 5.5
7c4ef18df0
Owner request, 14 October 2026: use the decision model (Jev / Celeris) at several stages, let Dify supply the rules, code it without new bugs. Every new stage is switched on in Settings > Kademeli Gecis and starts disabled, so the upgrade changes no behaviour by itself. - Settings schema 2 (Python and TypeScript validate the same rules); a schema 1 file is filled from the defaults. Default chain is now Katman-1 -> Katman-2 -> Katman-3 with Katman-3, Chat-Model and bekci-sesli-bekci always passing (owner decision of the same day). - watchModels: the Bekci correction channel follows a settings list instead of the hard-coded Sesli-Hizli-Zeki. - prefilter: a predicted refusal starts the step at the chosen layer. - autoRoute: virtual "Otomatik" model; one decision per user turn picks the combo, tool steps of the same turn keep it. - finishCheck: an answer that stops early goes back to the same model once. - A layer whose window cannot hold the request is skipped. - policy: stage rules from a Dify workflow (karar_politikasi.py), fetched in the background, validated, last good copy kept; turns never wait. - karar_sor.py: verifierCheck for the dynamic-workflow acceptance step. - Bridge version 19, IDE 1.0.0-beta.39. Tests: 299 bridge tests, 36 electron tests; live run on a side bridge (port 8797) with Celeris and the AITURK gateway for every stage. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
419 lines
20 KiB
Python
419 lines
20 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 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
|
|
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,
|
|
'log': self.logs.append}
|
|
exec(compile(ast.Module(body=[fonksiyon('karar_asamasi')], 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()
|
|
req = self.istek('ide/Otomatik', user('yap'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/Otomatik'), 'ide/Otomatik')
|
|
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
|
|
req3 = self.istek('ide/Otomatik', user('kaynaklari karsilastir'), arac, 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'))
|
|
self.assertEqual(self.karar(1, req, 'sohbet-1', 'ide/Otomatik'), 'ide/Otomatik')
|
|
self.assertEqual(self.kararlar, [])
|
|
|
|
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 test_a_layer_too_small_for_the_request_is_skipped(self):
|
|
logs = []
|
|
env = {'IDE_PREFIX': 'ide/', 'IDE_BILGI': {'ide/Katman-2': {'pencere': 1000}, 'ide/Katman-3': {'pencere': 10 ** 6}},
|
|
'pencere': pencere, 'log': logs.append}
|
|
exec(compile(ast.Module(body=[fonksiyon('sigan_katmanlar')], type_ignores=[]), 'kopru.py', 'exec'), env)
|
|
req = {'input': [user('x' * 20000)]}
|
|
self.assertEqual(env['sigan_katmanlar'](['Katman-2', 'Katman-3', 'Bilinmeyen'], req, '[1]'), ['Katman-3', 'Bilinmeyen'])
|
|
self.assertIn('Katman-2', logs[0])
|
|
|
|
|
|
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_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))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
|
|
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)
|