feat(settings): configure Jev evaluation and model transitions

apps/desktop/assets/codex-bridge/test_jev_settings.py

@@ -0,0 +1,111 @@
+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_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()