56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import unittest
|
|
|
|
from app.services.settings_service import (
|
|
_normalize_process_count,
|
|
_normalize_thread_count,
|
|
resolve_process_count,
|
|
resolve_thread_count,
|
|
)
|
|
|
|
|
|
class SettingsServiceTests(unittest.TestCase):
|
|
def test_normalize_thread_count_allows_values_above_256(self) -> None:
|
|
self.assertEqual(512, _normalize_thread_count(512))
|
|
|
|
def test_normalize_thread_count_rejects_non_positive_values(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
_normalize_thread_count(0)
|
|
|
|
def test_resolve_thread_count_uses_large_node_override(self) -> None:
|
|
payload = {
|
|
"thread_count": 8,
|
|
"node_thread_counts": {
|
|
"mainland-controller-01": 512,
|
|
},
|
|
}
|
|
|
|
resolved = resolve_thread_count(node_code="mainland-controller-01", settings_payload=payload)
|
|
|
|
self.assertEqual(8, resolved["default_thread_count"])
|
|
self.assertEqual(512, resolved["override_thread_count"])
|
|
self.assertEqual(512, resolved["effective_thread_count"])
|
|
self.assertEqual("node_override", resolved["source"])
|
|
|
|
def test_normalize_process_count_rejects_non_positive_values(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
_normalize_process_count(0)
|
|
|
|
def test_resolve_process_count_uses_large_node_override(self) -> None:
|
|
payload = {
|
|
"process_count": 80,
|
|
"node_process_counts": {
|
|
"mainland-controller-01": 96,
|
|
},
|
|
}
|
|
|
|
resolved = resolve_process_count(node_code="mainland-controller-01", settings_payload=payload)
|
|
|
|
self.assertEqual(80, resolved["default_process_count"])
|
|
self.assertEqual(96, resolved["override_process_count"])
|
|
self.assertEqual(96, resolved["effective_process_count"])
|
|
self.assertEqual("node_override", resolved["source"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|