"""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()