96 lines
5.1 KiB
Python
96 lines
5.1 KiB
Python
"""Guard the updater's platform boundary and failure behaviour without a live server."""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
# Reuse pack's macOS Python 3.11+ selection before importing tomllib users.
|
|
import pack
|
|
import server_runtime
|
|
|
|
|
|
class ServerUpdateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
(self.root / '.tools').mkdir()
|
|
(self.root / '.tools/tool.jar').write_bytes(b'verified-tool')
|
|
self.platform = {'versions': {'minecraft': '26.3', 'fabric': '0.19.5'},
|
|
'sha256': {'tool.jar': hashlib.sha256(b'verified-tool').hexdigest()}}
|
|
self.platform['pack_url'] = 'https://pack.example/pack.toml'
|
|
(self.root / '.tools/platform.json').write_text(json.dumps(self.platform))
|
|
|
|
def manifest(self, minecraft='26.3', loader='0.19.5'):
|
|
return io.BytesIO(f'[versions]\nminecraft="{minecraft}"\nfabric="{loader}"\n'.encode())
|
|
|
|
def test_headless_server_update_uses_pinned_tools_and_instance(self):
|
|
with patch('server_runtime.urlopen', return_value=self.manifest()), patch('server_runtime.subprocess.run') as run:
|
|
server_runtime.update(self.root, self.platform, 'https://pack.example/pack.toml', '/java path/java')
|
|
args = run.call_args.args[0]
|
|
self.assertEqual(args[0], '/java path/java')
|
|
self.assertEqual(args[args.index('-s') + 1], 'server')
|
|
self.assertIn('-g', args)
|
|
self.assertIn('--bootstrap-no-update', args)
|
|
self.assertEqual(args[args.index('--pack-folder') + 1], str(self.root))
|
|
self.assertTrue(run.call_args.kwargs['check'])
|
|
|
|
def test_platform_changes_stop_before_packwiz(self):
|
|
for minecraft, loader in [('26.4', '0.19.5'), ('26.3', '0.20.0')]:
|
|
with self.subTest(minecraft=minecraft, loader=loader), patch('server_runtime.urlopen', return_value=self.manifest(minecraft, loader)), patch('server_runtime.subprocess.run') as run:
|
|
with self.assertRaisesRegex(SystemExit, 'explicit migration'):
|
|
server_runtime.update(self.root, self.platform, 'https://pack.example/pack.toml', 'java')
|
|
run.assert_not_called()
|
|
|
|
def test_tool_hash_mismatch_stops_before_packwiz(self):
|
|
(self.root / '.tools/tool.jar').write_bytes(b'changed')
|
|
with patch('server_runtime.urlopen', return_value=self.manifest()), patch('server_runtime.subprocess.run') as run:
|
|
with self.assertRaisesRegex(SystemExit, 'hash mismatch'):
|
|
server_runtime.update(self.root, self.platform, 'https://pack.example/pack.toml', 'java')
|
|
run.assert_not_called()
|
|
|
|
def test_update_failure_propagates_and_personal_files_are_untouched(self):
|
|
settings = self.root / 'server.properties'
|
|
settings.write_bytes(b'online-mode=true\n')
|
|
with patch('server_runtime.urlopen', return_value=self.manifest()), patch('server_runtime.subprocess.run', side_effect=subprocess.CalledProcessError(1, 'installer')):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
server_runtime.update(self.root, self.platform, 'https://pack.example/pack.toml', 'java')
|
|
self.assertEqual(settings.read_bytes(), b'online-mode=true\n')
|
|
|
|
def test_failed_update_prevents_minecraft_start(self):
|
|
with patch('server_runtime.__file__', str(self.root / '.tools/server.py')), patch('sys.argv', ['server.py', 'start']), patch('server_runtime.update', side_effect=OSError('unreachable')), patch('server_runtime.subprocess.run') as run:
|
|
with self.assertRaises(OSError):
|
|
server_runtime.main()
|
|
run.assert_not_called()
|
|
|
|
def test_running_server_prevents_an_update(self):
|
|
with (self.root / '.sanctuary-server.lock').open('a') as lock:
|
|
server_runtime.fcntl.flock(lock, server_runtime.fcntl.LOCK_EX)
|
|
with patch('server_runtime.__file__', str(self.root / '.tools/server.py')), patch('sys.argv', ['server.py', 'update']), patch('server_runtime.update') as update:
|
|
with self.assertRaisesRegex(SystemExit, 'already in progress'):
|
|
server_runtime.main()
|
|
update.assert_not_called()
|
|
|
|
def test_shutdown_waits_for_java_before_releasing_the_lock(self):
|
|
handlers = {}
|
|
def install(signum, handler):
|
|
handlers[signum] = handler
|
|
return 'original-handler'
|
|
with patch('server_runtime.signal.signal', side_effect=install), patch('server_runtime.subprocess.Popen') as popen:
|
|
process = popen.return_value
|
|
def wait():
|
|
handlers[server_runtime.signal.SIGTERM](server_runtime.signal.SIGTERM, None)
|
|
return 0
|
|
process.wait.side_effect = wait
|
|
self.assertEqual(server_runtime.run_minecraft(self.root, 'java', '8G'), 0)
|
|
process.send_signal.assert_called_once_with(server_runtime.signal.SIGTERM)
|
|
self.assertTrue(all(h == 'original-handler' for h in handlers.values()))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|