from pathlib import Path import subprocess import unittest ROOT = Path(__file__).resolve().parents[1] class WorkerInstallContractTests(unittest.TestCase): def test_worker_scripts_parse_as_bash(self) -> None: for relative in ('deploy/worker/install_worker.sh', 'deploy/worker/verify_worker.sh'): subprocess.run( ['bash', '-n'], input=(ROOT / relative).read_bytes().replace(b'\r\n', b'\n'), check=True, ) def test_worker_packaging_forces_unix_line_endings(self) -> None: attributes = (ROOT / '.gitattributes').read_text(encoding='utf-8') self.assertIn('*.sh text eol=lf', attributes) self.assertIn('*.service text eol=lf', attributes) def test_worker_scripts_are_executable_in_git_archives(self) -> None: output = subprocess.run( ['git', 'ls-files', '-s', '--', 'deploy/worker/install_worker.sh', 'deploy/worker/verify_worker.sh'], cwd=ROOT, check=True, capture_output=True, text=True, ).stdout.splitlines() self.assertEqual(len(output), 2) self.assertTrue(all(line.startswith('100755 ') for line in output)) def test_installer_stays_provisioning_and_uses_loopback_contract(self) -> None: source = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') self.assertIn('set -euo pipefail', source) self.assertIn('provisioning', source) self.assertIn('BEE_TEST_API_URL=http://127.0.0.1:8080', source) self.assertIn('verify_worker.sh', source) self.assertIn('runtime-manifest.json', source) self.assertIn('BOT_RUNTIME_MANIFEST=', source) self.assertIn('BOT_RUN_COMMAND=', source) self.assertIn('BEE_RUNTIME_SHA256=', source) self.assertIn('BEE_WORKER_GENERATION=', source) self.assertIn('BEE_MTLS_CERT_FILE=', source) self.assertIn('BEE_MTLS_KEY_FILE=', source) self.assertIn('BEE_MTLS_CA_FILE=', source) self.assertIn('BEE_MTLS_CERT_PATH=', source) self.assertIn('BEE_MTLS_KEY_PATH=', source) self.assertIn('BEE_MTLS_CLIENT_CA_PATH=', source) self.assertIn('BEE_MTLS_CERT_PATH', source) self.assertIn('BEE_MTLS_KEY_PATH', source) self.assertIn('BEE_MTLS_CLIENT_CA_PATH', source) self.assertIn('API_CORS_ALLOWED_URLS=', source) self.assertIn('WORKER_API_CORS_ALLOWED_URLS', source) self.assertIn('WORKER_BEE_URL:-http://127.0.0.1:8080/api', source) self.assertIn('WORKER_BEE_INDICATORS_URL:-http://127.0.0.1:8080/api/ind', source) self.assertIn('BEE_URL=$worker_bee_url', source) self.assertIn('BEE_INDICATORS_URL=$worker_bee_indicators_url', source) self.assertIn('TEST_LOG_FILES_LOCATION=', source) self.assertIn('BOT_PYTHON_EXECUTABLE=', source) self.assertIn('/opt/razgar-bot-venv/bin/python', source) self.assertIn('/var/log/bee_logs', source) self.assertIn('web_root="${BEE_WEB_ROOT:-/var/www/html}"', source) self.assertIn('$web_root must be a symlink', source) self.assertIn('"$web_root.next"', source) self.assertIn('mv -Tf "$web_root.next" "$web_root"', source) self.assertIn('systemctl daemon-reload', source) self.assertIn('install -d -m 0755 /run/razgar', source) self.assertIn('install -d -m 0750 -o root -g www-data /etc/razgar', source) self.assertIn('chown root:www-data /etc/razgar/bee-worker.env', source) self.assertIn('chown root:www-data "$bot_release/runtime-manifest.json"', source) self.assertIn('chmod 0640 "$bot_release/runtime-manifest.json"', source) self.assertIn('BEE_SERVICE_HMAC_KEY', source) self.assertIn('BOT_SERVICE_HMAC_KEY', source) self.assertIn('FERNET_KEY', source) self.assertIn('from cryptography.fernet import Fernet', source) self.assertIn("$'\\r'", source) def test_installer_persists_a_private_tls_health_target(self) -> None: source = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') self.assertIn( 'worker_private_health_url="${WORKER_PRIVATE_HEALTH_URL:-https://bee-app.razgar.io/api/private/health.php}"', source, ) self.assertIn( 'worker_private_health_resolve="${WORKER_PRIVATE_HEALTH_RESOLVE:?WORKER_PRIVATE_HEALTH_RESOLVE is required}"', source, ) self.assertIn('BEE_PRIVATE_HEALTH_URL=$worker_private_health_url', source) self.assertIn('BEE_PRIVATE_HEALTH_RESOLVE=$worker_private_health_resolve', source) def test_installer_creates_a_local_only_bee_api_vhost(self) -> None: installer = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') vhost = (ROOT / 'deploy/worker/apache/bee-worker-loopback.conf').read_text( encoding='utf-8' ) self.assertIn('bee-worker-loopback.conf', installer) self.assertIn('a2ensite bee-worker-loopback.conf', installer) self.assertIn('apache2ctl configtest', installer) self.assertIn('systemctl reload apache2', installer) self.assertIn('Listen 127.0.0.1:8080', vhost) self.assertIn('', vhost) self.assertIn('DocumentRoot /var/www/html', vhost) self.assertIn('Require local', vhost) self.assertIn('RewriteEngine On', vhost) self.assertIn('RewriteCond %{DOCUMENT_ROOT}%{REQUEST_URI}.php -f', vhost) self.assertIn('RewriteRule ^(.*)$ $1.php [L]', vhost) def test_worker_services_load_the_installed_identity_environment(self) -> None: for relative in ( 'deploy/systemd/bee_data_handler.service', 'deploy/systemd/bee_websocket_server.service', 'deploy/systemd/bot_management_api.service', ): source = (ROOT / relative).read_text(encoding='utf-8') self.assertIn('EnvironmentFile=-/etc/razgar/bee-worker.env', source) unit_section = source.split('[Service]', 1)[0] self.assertIn('StartLimitIntervalSec=500', unit_section) self.assertIn('StartLimitBurst=5', unit_section) def test_worker_installs_loopback_bot_management_from_canonical_release(self) -> None: installer = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') service = (ROOT / 'deploy/systemd/bot_management_api.service').read_text(encoding='utf-8') self.assertIn('/opt/razgar-bot.next', installer) self.assertIn('mv -Tf /opt/razgar-bot.next /opt/razgar-bot', installer) self.assertIn('bot_management_api.service', installer) self.assertNotIn( 'systemctl restart bee_data_handler.service bee_websocket_server.service bot_management_api.service', installer, ) self.assertIn('WorkingDirectory=/opt/razgar-bot', service) self.assertIn('EnvironmentFile=-/etc/razgar/bee-worker.env', service) self.assertIn('--app-dir /opt/razgar-bot/bot/endpoints/bot_management', service) self.assertIn('--host 127.0.0.1 --port 8001', service) def test_installer_installs_log_maintenance_without_runtime_restarts(self) -> None: installer = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') self.assertIn('deploy_root="$(cd "$(dirname "$0")/.." && pwd)"', installer) self.assertIn('install -d -m 0755 /etc/logrotate.d', installer) self.assertIn('install -d -m 0700 /var/lib/razgar/logrotate-backups', installer) self.assertIn('install -m 0600 /etc/logrotate.d/razgar-bee', installer) self.assertIn('/var/lib/razgar/logrotate-backups/razgar-bee.', installer) self.assertIn( 'install -m 0644 "$deploy_root/logrotate/razgar-bee" /etc/logrotate.d/razgar-bee', installer, ) self.assertIn( 'install -m 0644 "$deploy_root/logrotate/apache2" /etc/logrotate.d/apache2', installer, ) self.assertIn( 'install -m 0755 "$deploy_root/maintenance/prune_bee_diagnostic_logs.sh" /usr/local/sbin/prune_bee_diagnostic_logs.sh', installer, ) for unit in ( 'razgar-logrotate.service', 'razgar-logrotate.timer', 'razgar-bee-log-maintenance.service', 'razgar-bee-log-maintenance.timer', ): self.assertIn( f'install -m 0644 "$deploy_root/systemd/{unit}" /etc/systemd/system/{unit}', installer, ) self.assertIn('systemctl daemon-reload', installer) self.assertIn('systemctl enable --now razgar-bee-log-maintenance.timer', installer) self.assertIn('systemctl enable --now razgar-logrotate.timer', installer) self.assertIn('command -v logrotate >/dev/null 2>&1', installer) self.assertIn('apt-get install -y logrotate', installer) self.assertIn('[[ -x /usr/sbin/logrotate ]]', installer) self.assertLess( installer.index('apt-get install -y logrotate'), installer.index('[[ -x /usr/sbin/logrotate ]]'), ) self.assertIn( 'package_timer="$(systemctl list-unit-files --type=timer --no-legend --no-pager logrotate.timer)"', installer, ) self.assertIn('systemctl disable --now logrotate.timer', installer) self.assertIn( 'if [[ "$package_timer" == logrotate.timer* ]]; then\n systemctl disable --now logrotate.timer\nfi', installer, ) self.assertNotIn('systemctl disable --now logrotate.timer >/dev/null 2>&1 || true', installer) self.assertIn('dpkg-divert --local --rename --add /etc/cron.daily/logrotate', installer) self.assertIn('[[ -e /etc/cron.daily/logrotate ]]', installer) self.assertIn( 'if [[ -e /etc/cron.daily/logrotate ]]; then\n dpkg-divert --local --rename --add /etc/cron.daily/logrotate\nfi', installer, ) self.assertNotIn('rm -f /etc/cron.daily/logrotate', installer) self.assertNotIn('chmod a-x /etc/cron.daily/logrotate', installer) self.assertEqual(1, installer.count('dpkg-divert --local --rename --add /etc/cron.daily/logrotate')) self.assertNotIn('logrotate.timer.d/razgar-six-hourly.conf', installer) self.assertNotIn('rm -f /etc/logrotate.d', installer) self.assertLess( installer.index('systemctl daemon-reload'), installer.index('systemctl enable --now razgar-bee-log-maintenance.timer'), ) self.assertLess( installer.index('systemctl daemon-reload'), installer.index('systemctl enable --now razgar-logrotate.timer'), ) def test_installer_accepts_only_verify_or_explicit_runtime_restart_mode(self) -> None: installer = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') self.assertIn('restart_runtime=false', installer) self.assertIn('case "$#" in', installer) self.assertIn('--verify)', installer) self.assertIn('--restart-runtime)', installer) self.assertIn('restart_runtime=true', installer) self.assertIn('Usage: $0 [--verify|--restart-runtime]', installer) self.assertIn('exit 2', installer) self.assertLess( installer.index('case "$#" in'), installer.index('require_file "$bee_release/init.inc.php"'), ) self.assertLess( installer.index('case "$#" in'), installer.index('worker_id="${BEE_WORKER_ID:?BEE_WORKER_ID is required}"'), ) def test_default_mode_preserves_readiness_and_opt_in_restarts_runtime_in_order(self) -> None: installer = (ROOT / 'deploy/worker/install_worker.sh').read_text(encoding='utf-8') function_start = installer.index('restart_worker_runtime() {') function_end = installer.index('\n}', function_start) activation = installer[function_start:function_end] marker_reset = 'rm -f /run/razgar/bee-data-handler.ready /run/razgar/bee-ws.ready' self.assertIn(marker_reset, activation) restart_order = ( 'systemctl restart bee_data_handler.service', 'systemctl restart bee_websocket_server.service', 'systemctl restart bot_management_api.service', ) for command in restart_order: self.assertIn(command, activation) self.assertLess(activation.index(restart_order[0]), activation.index(restart_order[1])) self.assertLess(activation.index(restart_order[1]), activation.index(restart_order[2])) self.assertEqual(1, installer.count(marker_reset)) self.assertEqual(2, installer.count('restart_worker_runtime')) opt_in = 'if [[ "$restart_runtime" == true ]]; then\n restart_worker_runtime\nfi' self.assertIn(opt_in, installer) self.assertEqual(1, installer.count('systemctl enable --now razgar-bee-log-maintenance.timer')) self.assertLess( installer.index('systemctl enable --now razgar-bee-log-maintenance.timer'), installer.index(opt_in), ) self.assertNotRegex(installer, r'systemctl\s+(?:restart|start|stop)\s+(?:redis|test-)') verifier = (ROOT / 'deploy/worker/verify_worker.sh').read_text(encoding='utf-8') self.assertIn('/run/razgar/bee-data-handler.ready', verifier) self.assertIn('/run/razgar/bee-ws.ready', verifier) def test_verifier_is_one_shot_and_never_enables_scheduling(self) -> None: source = (ROOT / 'deploy/worker/verify_worker.sh').read_text(encoding='utf-8') self.assertIn('https://127.0.0.1/api/private/health.php', source) self.assertIn("health_body='{}'", source) self.assertIn('--data "$health_body"', source) for contract in ('BEE_PRIVATE_HEALTH_URL', 'BEE_PRIVATE_HEALTH_RESOLVE', 'BEE_MTLS_CERT_FILE', 'BEE_MTLS_KEY_FILE', 'BEE_MTLS_CA_FILE', 'json_module_ready', 'database.status', 'active_test_count', 'systemctl --failed'): self.assertIn(contract, source) for contract in ('BEE_WORKER_ID', 'BEE_WORKER_GENERATION', 'BEE_RUNTIME_MANIFEST', 'sha256sum', 'bee_runtime_sha256', 'bot_runtime_sha256', 'timing_policy_version', 'systemctl is-active --quiet bee_data_handler.service', 'systemctl is-active --quiet bee_websocket_server.service'): self.assertIn(contract, source) for contract in ('BEE_SERVICE_HMAC_KEY', 'BEE_HEALTH_MESSAGE', '/private/health.php', 'X-Razgar-Timestamp', 'X-Razgar-Nonce', 'X-Razgar-Signature'): self.assertIn(contract, source) self.assertIn('for _ in {1..30}', source) self.assertIn('http://127.0.0.1:8001/health', source) self.assertIn('Bot management API did not become reachable.', source) self.assertIn('failed_units="$(systemctl --failed --no-legend --plain)"', source) self.assertIn('[[ -z "$failed_units" ]]', source) self.assertIn('systemctl is-active --quiet bot_management_api.service', source) self.assertIn('BOT_SERVICE_HMAC_KEY', source) self.assertIn('/host_status', source) self.assertIn(".active_live_units' <<<\"$bot_status\")\" == '0'", source) self.assertIn(".active_bee_units' <<<\"$bot_status\")\" == '0'", source) self.assertNotIn('state":"ready', source) def test_verifier_requires_the_bot_manifest_and_tree_identities(self) -> None: source = (ROOT / 'deploy/worker/verify_worker.sh').read_text(encoding='utf-8') self.assertIn("bot_tree_sha256=\"$(jq -r '.tree_sha256' \"$bot_manifest\")\"", source) self.assertIn('.result.bot_tree_sha256', source) self.assertIn(".tree_sha256' <<<\"$bot_status\")", source) def test_verifier_loads_worker_env_before_private_health_defaults(self) -> None: source = (ROOT / 'deploy/worker/verify_worker.sh').read_text(encoding='utf-8') self.assertLess( source.index('[[ -f "$worker_env" ]] && source "$worker_env"'), source.index('health_url="${BEE_PRIVATE_HEALTH_URL:-https://127.0.0.1/api/private/health.php}"'), ) def test_verifier_authenticates_redis_without_putting_the_secret_on_cli(self) -> None: source = (ROOT / 'deploy/worker/verify_worker.sh').read_text(encoding='utf-8') self.assertIn( 'REDISCLI_AUTH="${REDIS_PASS:?REDIS_PASS is required}" redis-cli --raw ping', source, ) self.assertNotIn('redis-cli -a', source) def test_apache_config_requires_client_certificate_and_proxies_ws(self) -> None: source = (ROOT / 'deploy/worker/apache/bee-worker.conf').read_text(encoding='utf-8') self.assertIn('SSLVerifyClient require', source) self.assertIn('SSLCACertificateFile', source) self.assertIn('', source) self.assertIn('ProxyPass /private/bot/ http://127.0.0.1:8001/', source) self.assertIn('ProxyPass /ws ws://127.0.0.1:8765/', source) if __name__ == '__main__': unittest.main()