207 lines
9.0 KiB
Python
207 lines
9.0 KiB
Python
"""Exercise an isolated copy of the init script; never launch real daemons."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
NODE = shutil.which('node', path='/usr/bin:/bin')
|
|
|
|
@unittest.skipUnless(NODE, 'Node is required for the fake Forever module')
|
|
class InitServiceTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='emalibre-test-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.app = self.root / 'app'
|
|
self.module = self.app / 'node_modules/forever'
|
|
(self.module / 'bin').mkdir(parents=True)
|
|
(self.app / 'gui_web').mkdir()
|
|
(self.app / 'gui_web/express.js').touch()
|
|
(self.app / 'conf').mkdir()
|
|
self.config = self.app / 'conf/ConfiguracionGeneral.js'
|
|
self.config.write_text('module.exports = {ConfGeneral: {inicio_automatico: false}};')
|
|
self.state = self.root / 'state.json'
|
|
self.state.write_text('[]')
|
|
self.calls = self.root / 'calls.jsonl'
|
|
self.pid = self.root / 'emalibre.pid'
|
|
self.mode = self.root / 'mode'
|
|
self.mode.write_text('web\n')
|
|
self.module.joinpath('index.js').write_text('''
|
|
const fs = require('fs');
|
|
exports.stop = uid => {
|
|
const emitter = new (require('events').EventEmitter)();
|
|
process.nextTick(() => {
|
|
if (process.env.COMMAND_FAIL === 'stop') return emitter.emit('error', Error('stop failed'));
|
|
const state = JSON.parse(fs.readFileSync(process.env.TEST_STATE));
|
|
fs.writeFileSync(process.env.TEST_STATE, JSON.stringify(state.filter(p => p.uid !== uid)));
|
|
emitter.emit('stop');
|
|
});
|
|
return emitter;
|
|
};
|
|
exports.list = (_, cb) => cb(process.env.LIST_FAIL ? Error('query failed') : null,
|
|
JSON.parse(fs.readFileSync(process.env.TEST_STATE)));
|
|
''')
|
|
self.module.joinpath('bin/forever').write_text('''
|
|
const fs = require('fs');
|
|
const args = process.argv.slice(2);
|
|
fs.appendFileSync(process.env.TEST_CALLS, JSON.stringify(args) + '\\n');
|
|
if (process.env.COMMAND_FAIL === args[0]) process.exit(7);
|
|
let state = JSON.parse(fs.readFileSync(process.env.TEST_STATE));
|
|
if (args[0] === 'start') {
|
|
// Match Forever's tryStart: path.join(sourceDir, script), not path.resolve.
|
|
const script = require('path').join(args[args.indexOf('--sourceDir') + 1], args[args.length - 2]);
|
|
if (!fs.existsSync(script)) {
|
|
console.error('script ' + script + ' does not exist');
|
|
process.exit(255);
|
|
}
|
|
state.push({uid: 'EMA_Web'});
|
|
fs.writeFileSync(args[args.indexOf('--pidFile') + 1], '123456');
|
|
}
|
|
if (args[0] === 'stop') state = state.filter(p => p.uid !== args[1]);
|
|
fs.writeFileSync(process.env.TEST_STATE, JSON.stringify(state));
|
|
''')
|
|
wrapper = self.app / 'node_modules/node/bin/node'
|
|
wrapper.parent.mkdir(parents=True)
|
|
# Allow tests on older host Node; production still enforces Node >= 20.
|
|
wrapper.write_text('#!/bin/sh\nif [ "$1" = -e ]; then exit "${VERSION_FAIL:-0}"; fi\nexec ' + NODE + ' "$@"\n')
|
|
wrapper.chmod(0o755)
|
|
script = (REPO / 'build/Deb64/etc/init.d/emalibre').read_text()
|
|
for old, new in {
|
|
'/usr/share/emalibre': str(self.app),
|
|
'/run/emalibre.pid': str(self.pid),
|
|
'/var/log/emalibre': str(self.root / 'logs'),
|
|
'/etc/emalibre_modo': str(self.mode),
|
|
'/root/.forever': str(self.root / 'forever'),
|
|
'HOME=/root': 'HOME=' + str(self.root),
|
|
'/run/lock/emalibre.lock': str(self.root / 'lock'),
|
|
'/run/systemd/system': str(self.root / 'systemd'),
|
|
'$(id -u)': '${TEST_UID:-0}',
|
|
}.items():
|
|
script = script.replace(old, new)
|
|
# Both service managers are mocks; no system configuration is touched.
|
|
script = script.replace('set_boot() {', '''
|
|
systemctl() { echo "$*" >> "${TEST_BOOT}"; return "${BOOT_FAIL:-0}"; }
|
|
update-rc.d() { echo "$*" >> "${TEST_BOOT}"; return "${BOOT_FAIL:-0}"; }
|
|
set_boot() {''')
|
|
self.script = self.root / 'service'
|
|
self.script.write_text(script)
|
|
self.env = dict(os.environ, TEST_STATE=str(self.state), TEST_CALLS=str(self.calls),
|
|
TEST_BOOT=str(self.root / 'boot'))
|
|
|
|
def run_service(self, action, expected=0, **env):
|
|
result = subprocess.run(['bash', str(self.script), action], env=dict(self.env, **env),
|
|
capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, expected, result.stdout + result.stderr)
|
|
return result
|
|
|
|
def commands(self):
|
|
return [json.loads(line) for line in self.calls.read_text().splitlines()] if self.calls.exists() else []
|
|
|
|
def test_start_and_idempotency(self):
|
|
self.run_service('start')
|
|
self.run_service('start')
|
|
self.assertEqual(len(self.commands()), 1)
|
|
self.assertEqual(self.commands()[0][-1], 'iniciar')
|
|
args = self.commands()[0]
|
|
self.assertEqual(args[args.index('-c') + 1], str(self.app / 'node_modules/node/bin/node'))
|
|
|
|
def test_stale_pid_does_not_block_start(self):
|
|
self.pid.write_text('123456')
|
|
self.run_service('start')
|
|
self.assertEqual(self.commands()[0][0], 'start')
|
|
|
|
def test_start_error_is_preserved(self):
|
|
self.run_service('start', 7, COMMAND_FAIL='start')
|
|
|
|
def test_stop_only_ema_and_without_pid(self):
|
|
self.state.write_text('[{"uid":"EMA_Web"},{"uid":"other"}]')
|
|
self.run_service('stop')
|
|
self.assertEqual(self.commands(), [['stop', 'EMA_Web']])
|
|
self.assertEqual(json.loads(self.state.read_text()), [{'uid': 'other'}])
|
|
|
|
def test_stop_idempotency_and_stale_pid(self):
|
|
self.pid.write_text('123456')
|
|
self.run_service('stop')
|
|
self.run_service('stop')
|
|
self.assertFalse(self.pid.exists())
|
|
self.assertEqual(self.commands(), [])
|
|
|
|
def test_stop_error_is_preserved(self):
|
|
self.state.write_text('[{"uid":"EMA_Web"}]')
|
|
self.run_service('stop', 7, COMMAND_FAIL='stop')
|
|
|
|
def test_restart_and_stop_failure(self):
|
|
self.state.write_text('[{"uid":"EMA_Web"}]')
|
|
self.run_service('restart', 7, COMMAND_FAIL='stop')
|
|
self.assertEqual(len(self.commands()), 1)
|
|
self.run_service('restart')
|
|
self.assertEqual([a[0] for a in self.commands()], ['stop', 'stop', 'start'])
|
|
|
|
def test_status(self):
|
|
self.run_service('status', 3)
|
|
self.state.write_text('[{"uid":"EMA_Web"}]')
|
|
self.run_service('status')
|
|
self.run_service('status', 4, LIST_FAIL='1')
|
|
|
|
def test_query_failure_prevents_start(self):
|
|
self.run_service('start', 4, LIST_FAIL='1')
|
|
self.assertEqual(self.commands(), [])
|
|
|
|
def test_mode_validation_and_default(self):
|
|
for value in ['', 'consola', 'web invalid']:
|
|
self.mode.write_text(value)
|
|
self.run_service('start', 2)
|
|
self.mode.unlink()
|
|
self.run_service('start')
|
|
self.assertEqual(self.mode.read_text(), 'web\n')
|
|
|
|
def test_missing_bundled_node_does_not_use_system_node(self):
|
|
(self.app / 'node_modules/node/bin/node').unlink()
|
|
self.run_service('start', 5)
|
|
self.assertEqual(self.commands(), [])
|
|
|
|
def test_missing_runtime_and_app(self):
|
|
self.run_service('start', 5, VERSION_FAIL='1')
|
|
(self.app / 'gui_web/express.js').unlink()
|
|
self.run_service('start', 5)
|
|
|
|
def test_usage_and_permissions(self):
|
|
self.run_service('invalid', 2)
|
|
self.run_service('start', 4, TEST_UID='1000')
|
|
|
|
def test_boot_independent_from_collection(self):
|
|
before = self.config.read_text()
|
|
self.run_service('set-inicio-enabled')
|
|
self.assertEqual((self.root / 'boot').read_text(), 'emalibre defaults\nemalibre enable\n')
|
|
(self.root / 'systemd').mkdir()
|
|
self.run_service('set-inicio-disable')
|
|
self.run_service('set-inicio-enabled', 9, BOOT_FAIL='9')
|
|
self.assertEqual(self.config.read_text(), before)
|
|
|
|
def test_migration_stops_only_ema(self):
|
|
source = (REPO / 'build/Deb64/DEBIAN/preinst').read_text()
|
|
js = source.split("<<'JS'\n", 1)[1].split('\nJS', 1)[0]
|
|
js = js.replace('/usr/share/emalibre', str(self.app))
|
|
for initial, expected, code, extra in [
|
|
([{'uid': 'EMA_Web'}, {'uid': 'other'}], [{'uid': 'other'}], 0, {}),
|
|
([{'uid': 'other'}], [{'uid': 'other'}], 0, {}),
|
|
([{'uid': 'EMA_Web'}], [{'uid': 'EMA_Web'}], 1, {'COMMAND_FAIL': 'stop'}),
|
|
]:
|
|
self.state.write_text(json.dumps(initial))
|
|
result = subprocess.run([NODE, '-'], input=js, text=True, capture_output=True,
|
|
env=dict(self.env, **extra), timeout=20)
|
|
self.assertEqual(result.returncode, code, result.stderr)
|
|
self.assertEqual(json.loads(self.state.read_text()), expected)
|
|
|
|
def test_view_does_not_create_mode(self):
|
|
self.mode.unlink()
|
|
self.run_service('view-conf', 1)
|
|
self.assertFalse(self.mode.exists())
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|