aiturk-hermes-ide/apps/desktop/assets/codex-bridge/test_jev_settings.py

import copy
import json
import os
import subprocess
import tempfile
import threading
import time
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest.mock import patch

import jev_denetim
import jev_settings


class Settings(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.home = Path(self.tmp.name)
        self.home_patch = patch.object(jev_settings, 'home', return_value=self.home)
        self.home_patch.start()
        self.addCleanup(self.home_patch.stop)
        self.settings = copy.deepcopy(jev_settings.DEFAULTS)

    def save(self, **changes):
        self.settings.update(changes)
        (self.home / 'jev-settings.json').write_text(json.dumps({'settings': self.settings}), encoding='utf-8')

    def test_live_rules_general_override_and_always_pass(self):
        self.save(fallbackModel='B', rejectionRules=[{'from': 'A', 'to': 'C'}, {'from': 'C', 'to': 'D'}], exemptModels=['D'])
        self.assertEqual(jev_denetim.sonrakiler('A'), ['C', 'D'])
        self.assertEqual(jev_denetim.sonrakiler('Other'), ['B'])
        self.assertTrue(jev_denetim.muaf_mi('D'))
        self.assertFalse(jev_denetim.muaf_mi('A'))
        self.save(enabled=False)
        self.assertFalse(jev_denetim.etkin_mi())
        self.assertEqual(jev_denetim.sonrakiler('A'), [])

    def test_cycle_cannot_spin_even_if_file_manually_edited(self):
        self.save(rejectionRules=[{'from': 'A', 'to': 'B'}, {'from': 'B', 'to': 'A'}])
        self.assertEqual(jev_denetim.sonrakiler('A'), ['B'])

    def test_stored_ide_prefix_on_model_names_is_stripped(self):
        # A hand- or agent-edited settings file may store a model name carrying the gateway's own
        # 'ide/' prefix. The bridge adds 'ide/' again when it routes, so a stored prefix would make
        # 'ide/ide/...', an unknown model whose context window the bridge mis-reads before silently
        # dropping the request. read() must strip the stray prefix so routing stays correct.
        self.save(
            fallbackModel='ide/abliteration-abliterated-model-large-v2',
            exemptModels=['ide/claude-claude-opus-5-5', 'agnes-agnes-1.5-flash'],
            prefilter={**self.settings['prefilter'], 'target': 'ide/ide/abliteration-abliterated-model-large-v2'},
        )
        settings, _ = jev_settings.read()
        self.assertEqual(settings['fallbackModel'], 'abliteration-abliterated-model-large-v2')
        self.assertEqual(settings['prefilter']['target'], 'abliteration-abliterated-model-large-v2')
        self.assertEqual(settings['exemptModels'], ['claude-claude-opus-5-5', 'agnes-agnes-1.5-flash'])
        self.assertTrue(jev_denetim.muaf_mi('claude-claude-opus-5-5'))

    def test_strip_ide_leaves_normal_names_and_removes_only_prefix(self):
        self.assertEqual(jev_settings._strip_ide('claude-claude-opus-5-5'), 'claude-claude-opus-5-5')
        self.assertEqual(jev_settings._strip_ide('ide/x'), 'x')
        self.assertEqual(jev_settings._strip_ide('ide/ide/x'), 'x')
        self.assertEqual(jev_settings._strip_ide('provider-ide/x'), 'provider-ide/x')

    def test_prompts_model_criteria_and_thresholds_are_applied(self):
        self.save(model='jev-custom', questions={'ret': 'question A', 'sart': 'question B'}, thresholds={'ret': .9, 'sart': .8}, criteria={'ret': {'true': 'bad', 'false': 'good'}, 'sart': {'true': '', 'false': ''}})
        captured = []
        def send(body, key):
            captured.append(body)
            return {'answers': {'ret': {'noul': .6}, 'sart': {'noul': .2}}}
        result = jev_denetim.sor('request', 'answer', 'test', gonder=send)
        self.assertFalse(jev_denetim.karar(result))
        self.assertEqual(captured[0]['model'], 'jev-custom')
        self.assertEqual(captured[0]['questions']['ret']['instructions'], 'question A')
        self.assertEqual(captured[0]['questions']['ret']['criteria'], {'true': 'bad', 'false': 'good'})
        self.assertNotIn('criteria', captured[0]['questions']['sart'])
        self.save(thresholds={'ret': .1, 'sart': .1})
        self.assertFalse(jev_denetim.karar(result))  # in-flight result retains its own thresholds

    def test_concurrency_is_limited_without_serializing_all_requests(self):
        self.save(maxConcurrent=2)
        count, peak = 0, 0
        lock = threading.Lock()
        def send(body, key):
            nonlocal count, peak
            with lock:
                count += 1
                peak = max(peak, count)
            time.sleep(.04)
            with lock:
                count -= 1
            return {'answers': {'ret': {'noul': 0}, 'sart': {'noul': 0}}}
        with ThreadPoolExecutor(max_workers=8) as pool:
            answers = list(pool.map(lambda _: jev_denetim.sor('r', 'a', 'test', gonder=send), range(8)))
        self.assertEqual(peak, 2)
        self.assertTrue(all(a is not None for a in answers))

    def test_timeout_endpoint_and_failure_behavior(self):
        self.save(endpoint='https://example.test/jev', timeoutSeconds=42, failureAction='next')
        with patch('urllib.request.urlopen', side_effect=TimeoutError) as opening:
            self.assertIsNone(jev_denetim.sor('r', 'a', 'test'))
            self.assertEqual(opening.call_args.kwargs['timeout'], 42)
            self.assertEqual(opening.call_args.args[0].full_url, 'https://example.test/jev')
        self.assertTrue(jev_denetim.hata_gecis())
        self.assertEqual(jev_settings._active, 0)

    def test_queue_wait_expires_and_slot_count_is_preserved(self):
        self.save(maxConcurrent=1, queueWaitSeconds=1)
        settings = jev_settings.read()[0]
        self.assertTrue(jev_settings.acquire(settings))
        try:
            with patch.object(jev_denetim, '_gonder') as send:
                self.assertIsNone(jev_denetim.sor('r', 'a', 'test', gonder=send))
                send.assert_not_called()
            self.assertEqual(jev_settings._active, 1)
        finally:
            jev_settings.release()

    @unittest.skipUnless(os.name == 'nt', 'Windows DPAPI')
    def test_encrypted_key_roundtrip_without_plaintext_on_disk(self):
        secret = 'dummy-key-for-local-test'
        code = 'Add-Type -AssemblyName System.Security; $b=[Text.Encoding]::UTF8.GetBytes([Console]::In.ReadToEnd()); [Console]::Out.Write([Convert]::ToBase64String([Security.Cryptography.ProtectedData]::Protect($b,$null,[Security.Cryptography.DataProtectionScope]::CurrentUser)))'
        encrypted = subprocess.run(['powershell.exe', '-NoProfile', '-NonInteractive', '-Command', code], input=secret, capture_output=True, text=True, check=True, creationflags=subprocess.CREATE_NO_WINDOW).stdout.strip()
        text = json.dumps({'settings': self.settings, 'protectedKey': encrypted})
        (self.home / 'jev-settings.json').write_text(text, encoding='utf-8')
        self.assertNotIn(secret, text)
        self.assertEqual(jev_settings.key(), secret)


if __name__ == '__main__':
    unittest.main()