112 lines
5.4 KiB
Python
112 lines
5.4 KiB
Python
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()
|