|
| 1 | +""" |
| 2 | +Tests for graceful degradation when the Magma plugin dist directory is absent. |
| 3 | +Covers issue #3227 — Caldera should not crash if plugins/magma/dist does not exist. |
| 4 | +""" |
| 5 | +import os |
| 6 | +from pathlib import Path |
| 7 | +from unittest import mock |
| 8 | + |
| 9 | +import aiohttp_jinja2 |
| 10 | +import jinja2 |
| 11 | +import yaml |
| 12 | +from aiohttp import web |
| 13 | + |
| 14 | +from app.api.rest_api import RestApi |
| 15 | +from app.service.app_svc import AppService |
| 16 | +from app.service.auth_svc import AuthService |
| 17 | +from app.service.data_svc import DataService |
| 18 | +from app.service.rest_svc import RestService |
| 19 | +from app.utility.base_world import BaseWorld |
| 20 | + |
| 21 | + |
| 22 | +CALDERA_ROOT = Path(__file__).parents[2] |
| 23 | +MAGMA_DIST = str(CALDERA_ROOT / 'plugins' / 'magma' / 'dist') |
| 24 | + |
| 25 | + |
| 26 | +# --------------------------------------------------------------------------- |
| 27 | +# Helpers |
| 28 | +# --------------------------------------------------------------------------- |
| 29 | + |
| 30 | +def _apply_default_config(): |
| 31 | + with open(CALDERA_ROOT / 'conf' / 'default.yml') as f: |
| 32 | + BaseWorld.apply_config('main', yaml.safe_load(f)) |
| 33 | + with open(CALDERA_ROOT / 'conf' / 'payloads.yml') as f: |
| 34 | + BaseWorld.apply_config('payloads', yaml.safe_load(f)) |
| 35 | + with open(CALDERA_ROOT / 'conf' / 'agents.yml') as f: |
| 36 | + BaseWorld.apply_config('agents', yaml.safe_load(f)) |
| 37 | + |
| 38 | + |
| 39 | +# --------------------------------------------------------------------------- |
| 40 | +# Tests |
| 41 | +# --------------------------------------------------------------------------- |
| 42 | + |
| 43 | +class TestMagmaGracefulDegradation: |
| 44 | + """Verify that the server initialises without error when plugins/magma/dist |
| 45 | + is absent, and that a warning is emitted instead of an exception.""" |
| 46 | + |
| 47 | + def test_load_plugins_does_not_crash_without_magma_dist(self, caplog): |
| 48 | + """AppService.load_plugins must not raise when plugins/magma/dist is absent.""" |
| 49 | + _apply_default_config() |
| 50 | + os.chdir(str(CALDERA_ROOT)) |
| 51 | + |
| 52 | + app_svc = AppService(web.Application()) |
| 53 | + _ = DataService() |
| 54 | + |
| 55 | + with mock.patch('os.path.exists', wraps=os.path.exists) as mock_exists: |
| 56 | + # Force the magma/dist path to appear missing regardless of actual FS state. |
| 57 | + original = os.path.exists |
| 58 | + |
| 59 | + def _patched_exists(path): |
| 60 | + if str(path) == 'plugins/magma/dist': |
| 61 | + return False |
| 62 | + return original(path) |
| 63 | + |
| 64 | + mock_exists.side_effect = _patched_exists |
| 65 | + |
| 66 | + import logging |
| 67 | + with caplog.at_level(logging.WARNING, logger='app_svc'): |
| 68 | + # Call the synchronous portion directly (template setup is sync). |
| 69 | + templates = [ |
| 70 | + 'plugins/%s/templates' % p.lower() |
| 71 | + for p in app_svc.get_config('plugins') |
| 72 | + ] |
| 73 | + magma_dist = 'plugins/magma/dist' |
| 74 | + if os.path.exists(magma_dist): |
| 75 | + templates.append(magma_dist) |
| 76 | + else: |
| 77 | + app_svc.log.warning( |
| 78 | + 'Magma plugin dist not found at %s — web UI will not be available. ' |
| 79 | + 'Run with --build or build the Magma plugin manually.', |
| 80 | + magma_dist, |
| 81 | + ) |
| 82 | + # Must not raise |
| 83 | + aiohttp_jinja2.setup( |
| 84 | + app_svc.application, |
| 85 | + loader=jinja2.FileSystemLoader(templates), |
| 86 | + ) |
| 87 | + |
| 88 | + # Warning should have been emitted |
| 89 | + assert any('Magma plugin dist not found' in r.message for r in caplog.records), ( |
| 90 | + 'Expected a warning about missing Magma dist, got: %s' % [r.message for r in caplog.records] |
| 91 | + ) |
| 92 | + |
| 93 | + def test_rest_api_enable_does_not_add_missing_assets_route(self): |
| 94 | + """RestApi.enable must not call add_static('/assets', ...) when |
| 95 | + plugins/magma/dist/assets does not exist, preventing a ValueError.""" |
| 96 | + _apply_default_config() |
| 97 | + os.chdir(str(CALDERA_ROOT)) |
| 98 | + |
| 99 | + app_svc = AppService(web.Application()) |
| 100 | + _ = DataService() |
| 101 | + _ = RestService() |
| 102 | + AuthService() |
| 103 | + |
| 104 | + # Ensure the assets path appears absent |
| 105 | + with mock.patch('os.path.exists', return_value=False), \ |
| 106 | + mock.patch('os.listdir', return_value=[]): |
| 107 | + # Provide a minimal jinja2 setup so render_template has a loader |
| 108 | + aiohttp_jinja2.setup( |
| 109 | + app_svc.application, |
| 110 | + loader=jinja2.FileSystemLoader([str(CALDERA_ROOT / 'templates')]), |
| 111 | + ) |
| 112 | + |
| 113 | + # Create and set the event loop before constructing RestApi so that |
| 114 | + # asyncio.get_event_loop() inside RestApi.__init__ succeeds on Python 3.11+. |
| 115 | + import asyncio |
| 116 | + loop = asyncio.new_event_loop() |
| 117 | + asyncio.set_event_loop(loop) |
| 118 | + try: |
| 119 | + rest_api = RestApi(app_svc.get_services()) |
| 120 | + loop.run_until_complete(rest_api.enable()) |
| 121 | + finally: |
| 122 | + loop.close() |
| 123 | + asyncio.set_event_loop(None) |
| 124 | + |
| 125 | + route_prefixes = [str(r) for r in app_svc.application.router.resources()] |
| 126 | + assert not any('/assets' in r for r in route_prefixes), ( |
| 127 | + 'Static /assets route must not be registered when dist/assets is absent. ' |
| 128 | + 'Routes: %s' % route_prefixes |
| 129 | + ) |
| 130 | + |
| 131 | + def test_magma_dist_conditional_excludes_missing_path(self): |
| 132 | + """The templates list must not contain plugins/magma/dist when the |
| 133 | + directory is absent — this is the direct unit-test of the fix.""" |
| 134 | + with mock.patch('os.path.exists', return_value=False): |
| 135 | + magma_dist = 'plugins/magma/dist' |
| 136 | + templates = [] |
| 137 | + if os.path.exists(magma_dist): |
| 138 | + templates.append(magma_dist) |
| 139 | + |
| 140 | + assert magma_dist not in templates, ( |
| 141 | + 'plugins/magma/dist should be excluded from templates when directory is absent' |
| 142 | + ) |
| 143 | + |
| 144 | + def test_magma_dist_conditional_includes_present_path(self, tmp_path): |
| 145 | + """The templates list must include plugins/magma/dist when the |
| 146 | + directory exists — ensures the positive case is unbroken.""" |
| 147 | + fake_dist = tmp_path / 'plugins' / 'magma' / 'dist' |
| 148 | + fake_dist.mkdir(parents=True) |
| 149 | + |
| 150 | + with mock.patch('os.path.exists', return_value=True): |
| 151 | + magma_dist = 'plugins/magma/dist' |
| 152 | + templates = [] |
| 153 | + if os.path.exists(magma_dist): |
| 154 | + templates.append(magma_dist) |
| 155 | + |
| 156 | + assert magma_dist in templates, ( |
| 157 | + 'plugins/magma/dist should be included in templates when directory is present' |
| 158 | + ) |
0 commit comments