business-assistant/scripts/tests/test_memory_service.py

from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import Mock, patch

from scripts.memory_service import render, unit_name
from scripts.deploy import Deployment, MODULE


class MemoryServiceTests(unittest.TestCase):
    def test_render_limits_children_swap_and_restart_storms(self):
        content = render('/opt/business-assistant', 'assistant', 4096)
        for expected in ('MemoryMax=4096M', 'MemoryHigh=3584M', 'MemorySwapMax=0',
                         'OOMPolicy=kill', 'KillMode=control-group', 'Restart=on-failure',
                         'StartLimitBurst=3', 'API_REQUIRE_MEMORY_CGROUP=1',
                         'API_MEMORY_HARD_LIMIT_MB=4096', 'API_MEMORY_BUDGET_MB=3072',
                         'ExecStart=/bin/sh /opt/business-assistant/deploy.sh serve'):
            self.assertIn(expected, content)
        self.assertNotIn('EnvironmentFile', content)
        self.assertEqual(unit_name('/opt/a'), unit_name('/opt/a'))
        self.assertNotEqual(unit_name('/opt/a'), unit_name('/opt/b'))

    def test_render_rejects_unsafe_parameters(self):
        for args in [('/opt/a\nMemoryMax=infinity', 'assistant', 4096),
                     ('/opt/%a', 'assistant', 4096), ('/opt/a', 'root', 4096),
                     ('/opt/a', 'assistant', 100), ('/opt/a', 'a\nUser=root', 4096)]:
            with self.assertRaises(ValueError):
                render(*args)

    def test_render_uses_offline_entry_for_extracted_release(self):
        content = render('/opt/business-assistant', 'assistant', 4096, offline=True)
        self.assertIn('ExecStart=/bin/sh /opt/business-assistant/deploy-offline.sh serve', content)

    def test_installed_unit_routes_start_stop_without_background_fallback(self):
        with TemporaryDirectory() as directory:
            deployment = Deployment(Path(directory))
            deployment.systemd_managed = lambda: True
            deployment.systemd_pid = lambda: 1
            deployment.settings = Mock(return_value=Mock(host='127.0.0.1', port=8000))
            deployment.processes = Mock(return_value=[{'pid': 1}])
            deployment.healthy = Mock(return_value=True)
            deployment.run = Mock()
            with patch('subprocess.Popen') as popen:
                deployment.start()
                popen.assert_not_called()
            deployment.run.assert_called_once_with('systemctl', 'start', unit_name(deployment.root))
            deployment.run.reset_mock()
            deployment.processes.return_value = []
            deployment.stop()
            deployment.run.assert_called_once_with('systemctl', 'stop', unit_name(deployment.root))
            deployment.run.side_effect = RuntimeError('synthetic permission denied')
            with patch('subprocess.Popen') as popen:
                with self.assertRaises(RuntimeError):
                    deployment.start()
                popen.assert_not_called()

    def test_foreground_resolves_selected_runtime_and_requires_guard(self):
        with TemporaryDirectory() as directory:
            deployment = Deployment(Path(directory))
            deployment.processes = Mock(return_value=[])
            with patch.dict('os.environ', {}, clear=True):
                with self.assertRaises(RuntimeError):
                    deployment.serve()
            with patch.dict('os.environ', {'API_REQUIRE_MEMORY_CGROUP': '1'}), patch('os.execv') as execute:
                deployment.serve()
                execute.assert_called_once_with(str(deployment.python), [str(deployment.python), '-m', MODULE])