diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..6f94355 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +asyncio_mode = auto diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..6d510e4 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,144 @@ +"""Shared fixtures for magma pytest suite.""" + +import os +import sys +import types +import logging +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import pytest_asyncio + +# --------------------------------------------------------------------------- +# Ensure the project root is on sys.path +# --------------------------------------------------------------------------- +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if PROJECT_ROOT not in sys.path: + sys.path.insert(0, PROJECT_ROOT) + +# --------------------------------------------------------------------------- +# Stub out the Caldera-core imports that magma relies on at import time so +# the test suite can run without a full Caldera installation. +# +# magma's app/ directory is NOT a standalone package — in Caldera, "app" is +# the core framework package. magma_api.py does `from app.service.auth_svc +# import …` expecting Caldera's auth_svc. We create a synthetic "app" +# package whose __path__ includes the real app/ dir so that magma_api and +# magma_svc can be imported, while also providing stubs for the Caldera-core +# sub-modules they reference. +# --------------------------------------------------------------------------- + +def _install_caldera_stubs(): + """Create minimal stand-ins for caldera packages that magma imports.""" + + app_dir = os.path.join(PROJECT_ROOT, "app") + + # --- top-level "app" package with __path__ pointing at real dir -------- + app_pkg = types.ModuleType("app") + app_pkg.__path__ = [app_dir] + app_pkg.__package__ = "app" + + # --- app.utility.base_world ------------------------------------------- + app_utility = types.ModuleType("app.utility") + app_utility.__path__ = [] + app_utility.__package__ = "app.utility" + + base_world_mod = types.ModuleType("app.utility.base_world") + + class _Access: + APP = "app" + RED = "red" + BLUE = "blue" + + class BaseWorld: + Access = _Access + + base_world_mod.BaseWorld = BaseWorld + + # --- app.service.auth_svc --------------------------------------------- + app_service = types.ModuleType("app.service") + app_service.__path__ = [] + app_service.__package__ = "app.service" + + auth_svc_mod = types.ModuleType("app.service.auth_svc") + + def check_authorization(func): + return func + + def for_all_public_methods(decorator): + def wrapper(cls): + return cls + return wrapper + + auth_svc_mod.check_authorization = check_authorization + auth_svc_mod.for_all_public_methods = for_all_public_methods + + # --- register them in sys.modules so later imports resolve ------------ + mods = { + "app": app_pkg, + "app.utility": app_utility, + "app.utility.base_world": base_world_mod, + "app.service": app_service, + "app.service.auth_svc": auth_svc_mod, + } + for name, mod in mods.items(): + sys.modules[name] = mod + + +_install_caldera_stubs() + +# We also need `plugins.magma` to resolve for hook.py's relative import. +_plugins = types.ModuleType("plugins") +_plugins.__path__ = [] +_plugins_magma = types.ModuleType("plugins.magma") +_plugins_magma.__path__ = [PROJECT_ROOT] +_plugins_magma_app = types.ModuleType("plugins.magma.app") +_plugins_magma_app.__path__ = [os.path.join(PROJECT_ROOT, "app")] + +sys.modules["plugins"] = _plugins +sys.modules["plugins.magma"] = _plugins_magma +sys.modules["plugins.magma.app"] = _plugins_magma_app + +# Now import the actual modules under test — the stubs make this safe. +from app.magma_api import MagmaAPI # noqa: E402 +from app.magma_svc import MagmaService # noqa: E402 + +# Patch plugins.magma.app.magma_api so hook.py's import succeeds +_pm_api = types.ModuleType("plugins.magma.app.magma_api") +_pm_api.MagmaAPI = MagmaAPI +_plugins_magma_app.magma_api = _pm_api +sys.modules["plugins.magma.app.magma_api"] = _pm_api + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def mock_services(): + """Return a dict of mock Caldera core services.""" + return { + "auth_svc": MagicMock(name="auth_svc"), + "data_svc": MagicMock(name="data_svc"), + "file_svc": MagicMock(name="file_svc"), + "app_svc": MagicMock(name="app_svc"), + } + + +@pytest.fixture +def magma_api(mock_services): + return MagmaAPI(mock_services) + + +@pytest.fixture +def magma_svc(mock_services): + return MagmaService(mock_services) + + +@pytest.fixture +def mock_app(mock_services): + """Fake aiohttp application (a plain dict is sufficient for router ops).""" + app = MagicMock() + app.router = MagicMock() + mock_services["app_svc"].application = app + return app diff --git a/tests/test_hook.py b/tests/test_hook.py new file mode 100644 index 0000000..98bf927 --- /dev/null +++ b/tests/test_hook.py @@ -0,0 +1,175 @@ +"""Tests for hook.py — plugin metadata and the enable() lifecycle.""" + +import importlib +import sys +import types +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Import hook module (the conftest stubs make this safe) +# --------------------------------------------------------------------------- + +import hook + + +# --------------------------------------------------------------------------- +# Module-level metadata +# --------------------------------------------------------------------------- + +class TestHookMetadata: + + def test_name(self): + assert hook.name == "Magma" + + def test_description_mentions_vue(self): + assert "VueJS" in hook.description + + def test_description_mentions_caldera(self): + assert "Caldera" in hook.description + + def test_description_is_string(self): + assert isinstance(hook.description, str) + + def test_address(self): + assert hook.address == "/plugin/magma/gui" + + def test_address_starts_with_slash(self): + assert hook.address.startswith("/") + + def test_address_contains_plugin(self): + assert "/plugin/" in hook.address + + def test_access_is_app(self): + from app.utility.base_world import BaseWorld + assert hook.access == BaseWorld.Access.APP + + def test_access_value(self): + assert hook.access == "app" + + +# --------------------------------------------------------------------------- +# enable() coroutine +# --------------------------------------------------------------------------- + +class TestEnable: + + @pytest.mark.asyncio + async def test_enable_is_coroutine(self): + import inspect + assert inspect.iscoroutinefunction(hook.enable) + + @pytest.mark.asyncio + async def test_enable_retrieves_app_svc(self, mock_services, mock_app): + await hook.enable(mock_services) + mock_services["app_svc"].application # accessed as attribute + + @pytest.mark.asyncio + async def test_enable_creates_magma_api(self, mock_services, mock_app): + with patch("hook.MagmaAPI") as mock_cls: + await hook.enable(mock_services) + mock_cls.assert_called_once_with(mock_services) + + @pytest.mark.asyncio + async def test_enable_passes_services_to_api(self, mock_services, mock_app): + with patch("hook.MagmaAPI") as mock_cls: + await hook.enable(mock_services) + args, _ = mock_cls.call_args + assert args[0] is mock_services + + @pytest.mark.asyncio + async def test_enable_accesses_application(self, mock_services, mock_app): + await hook.enable(mock_services) + # The enable function reads app_svc.application + _ = mock_services["app_svc"].application + + @pytest.mark.asyncio + async def test_enable_with_minimal_services(self): + """enable() needs app_svc at minimum.""" + app_svc = MagicMock() + app_svc.application = MagicMock() + services = {"app_svc": app_svc} + # Should not raise + await hook.enable(services) + + @pytest.mark.asyncio + async def test_enable_without_app_svc_raises(self): + with pytest.raises((AttributeError, TypeError)): + await hook.enable({}) + + +# --------------------------------------------------------------------------- +# Plugin module attributes expected by Caldera plugin loader +# --------------------------------------------------------------------------- + +class TestPluginContract: + """Caldera expects every plugin hook.py to expose specific module-level + attributes. Verify the contract is satisfied.""" + + def test_has_name(self): + assert hasattr(hook, "name") + + def test_has_description(self): + assert hasattr(hook, "description") + + def test_has_address(self): + assert hasattr(hook, "address") + + def test_has_access(self): + assert hasattr(hook, "access") + + def test_has_enable(self): + assert hasattr(hook, "enable") + + def test_enable_is_callable(self): + assert callable(hook.enable) + + def test_name_is_non_empty_string(self): + assert isinstance(hook.name, str) and len(hook.name) > 0 + + def test_description_is_non_empty_string(self): + assert isinstance(hook.description, str) and len(hook.description) > 0 + + def test_address_is_non_empty_string(self): + assert isinstance(hook.address, str) and len(hook.address) > 0 + + +# --------------------------------------------------------------------------- +# Route / static-file serving expectations +# --------------------------------------------------------------------------- + +class TestRouteRegistration: + """The enable() function should set up the application for serving the + Magma Vue SPA. We test that the expected aiohttp structures are touched.""" + + @pytest.mark.asyncio + async def test_app_obtained_from_app_svc(self, mock_services, mock_app): + await hook.enable(mock_services) + # Confirm application attribute was accessed + assert mock_services["app_svc"].application is mock_app + + @pytest.mark.asyncio + async def test_enable_returns_none(self, mock_services, mock_app): + result = await hook.enable(mock_services) + assert result is None + + +# --------------------------------------------------------------------------- +# Build / dist expectations +# --------------------------------------------------------------------------- + +class TestDistServing: + """Magma's built Vue frontend lives in a dist/ directory. Verify + assumptions about the address where it would be served.""" + + def test_gui_address_ends_with_gui(self): + assert hook.address.endswith("/gui") + + def test_gui_address_has_magma(self): + assert "magma" in hook.address + + def test_gui_address_segments(self): + parts = hook.address.strip("/").split("/") + assert parts == ["plugin", "magma", "gui"] diff --git a/tests/test_magma_api.py b/tests/test_magma_api.py new file mode 100644 index 0000000..e636cb0 --- /dev/null +++ b/tests/test_magma_api.py @@ -0,0 +1,171 @@ +"""Tests for app/magma_api.py — MagmaAPI and its endpoints.""" + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from aiohttp import web +from aiohttp.test_utils import make_mocked_request + +from app.magma_api import MagmaAPI + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_request(body: dict | list | str | None = None, method: str = "POST", path: str = "/"): + """Build a lightweight mock request whose .read() returns *body* as JSON bytes.""" + req = AsyncMock(spec=web.Request) + if body is not None: + payload = json.dumps(body).encode() if not isinstance(body, (bytes, bytearray)) else body + else: + payload = b"{}" + req.read = AsyncMock(return_value=payload) + req.method = method + req.path = path + return req + + +# --------------------------------------------------------------------------- +# Construction +# --------------------------------------------------------------------------- + +class TestMagmaAPIInit: + + def test_stores_services(self, mock_services): + api = MagmaAPI(mock_services) + assert api.services is mock_services + + def test_resolves_auth_svc(self, mock_services): + api = MagmaAPI(mock_services) + assert api.auth_svc is mock_services["auth_svc"] + + def test_resolves_data_svc(self, mock_services): + api = MagmaAPI(mock_services) + assert api.data_svc is mock_services["data_svc"] + + def test_auth_svc_none_when_missing(self): + api = MagmaAPI({}) + assert api.auth_svc is None + + def test_data_svc_none_when_missing(self): + api = MagmaAPI({}) + assert api.data_svc is None + + def test_services_dict_not_mutated(self, mock_services): + original = dict(mock_services) + MagmaAPI(mock_services) + assert mock_services == original + + +# --------------------------------------------------------------------------- +# mirror() endpoint +# --------------------------------------------------------------------------- + +class TestMirrorEndpoint: + + @pytest.mark.asyncio + async def test_mirror_returns_json_response(self, magma_api): + req = _make_request({"key": "value"}) + resp = await magma_api.mirror(req) + assert isinstance(resp, web.Response) + + @pytest.mark.asyncio + async def test_mirror_echoes_dict(self, magma_api): + body = {"hello": "world"} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_echoes_list(self, magma_api): + body = [1, 2, 3] + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_echoes_nested(self, magma_api): + body = {"a": {"b": [1, {"c": True}]}} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_echoes_empty_dict(self, magma_api): + resp = await magma_api.mirror(_make_request({})) + assert json.loads(resp.body) == {} + + @pytest.mark.asyncio + async def test_mirror_echoes_empty_list(self, magma_api): + resp = await magma_api.mirror(_make_request([])) + assert json.loads(resp.body) == [] + + @pytest.mark.asyncio + async def test_mirror_echoes_string_body(self, magma_api): + body = "just a string" + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_echoes_number(self, magma_api): + resp = await magma_api.mirror(_make_request(42)) + assert json.loads(resp.body) == 42 + + @pytest.mark.asyncio + async def test_mirror_echoes_boolean(self, magma_api): + resp = await magma_api.mirror(_make_request(True)) + assert json.loads(resp.body) is True + + @pytest.mark.asyncio + async def test_mirror_echoes_null(self, magma_api): + resp = await magma_api.mirror(_make_request(None)) + # None body -> default "{}" -> echoes {} + assert json.loads(resp.body) == {} + + @pytest.mark.asyncio + async def test_mirror_content_type_json(self, magma_api): + resp = await magma_api.mirror(_make_request({"x": 1})) + assert resp.content_type == "application/json" + + @pytest.mark.asyncio + async def test_mirror_reads_request_body(self, magma_api): + req = _make_request({"a": 1}) + await magma_api.mirror(req) + req.read.assert_awaited_once() + + @pytest.mark.asyncio + async def test_mirror_preserves_unicode(self, magma_api): + body = {"emoji": "\u2728", "cjk": "\u4f60\u597d"} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_large_payload(self, magma_api): + body = {"items": list(range(1000))} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_special_chars_in_keys(self, magma_api): + body = {"key with spaces": 1, "key/with/slashes": 2, "": 3} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body) == body + + @pytest.mark.asyncio + async def test_mirror_invalid_json_raises(self, magma_api): + req = AsyncMock() + req.read = AsyncMock(return_value=b"not json") + with pytest.raises(json.JSONDecodeError): + await magma_api.mirror(req) + + @pytest.mark.asyncio + async def test_mirror_empty_body_raises(self, magma_api): + req = AsyncMock() + req.read = AsyncMock(return_value=b"") + with pytest.raises(json.JSONDecodeError): + await magma_api.mirror(req) + + @pytest.mark.asyncio + async def test_mirror_deeply_nested(self, magma_api): + body = {"a": {"b": {"c": {"d": {"e": "deep"}}}}} + resp = await magma_api.mirror(_make_request(body)) + assert json.loads(resp.body)["a"]["b"]["c"]["d"]["e"] == "deep" diff --git a/tests/test_magma_svc.py b/tests/test_magma_svc.py new file mode 100644 index 0000000..f6b828c --- /dev/null +++ b/tests/test_magma_svc.py @@ -0,0 +1,96 @@ +"""Tests for app/magma_svc.py — MagmaService.""" + +import logging +from unittest.mock import MagicMock + +import pytest + +from app.magma_svc import MagmaService + + +# --------------------------------------------------------------------------- +# Construction / initialisation +# --------------------------------------------------------------------------- + +class TestMagmaServiceInit: + + def test_stores_services_dict(self, mock_services): + svc = MagmaService(mock_services) + assert svc.services is mock_services + + def test_resolves_file_svc(self, mock_services): + svc = MagmaService(mock_services) + assert svc.file_svc is mock_services["file_svc"] + + def test_file_svc_none_when_missing(self): + svc = MagmaService({}) + assert svc.file_svc is None + + def test_creates_logger(self, mock_services): + svc = MagmaService(mock_services) + assert isinstance(svc.log, logging.Logger) + + def test_logger_name(self, mock_services): + svc = MagmaService(mock_services) + assert svc.log.name == "magma_svc" + + def test_services_dict_not_mutated(self, mock_services): + original_keys = set(mock_services.keys()) + MagmaService(mock_services) + assert set(mock_services.keys()) == original_keys + + +# --------------------------------------------------------------------------- +# foo() +# --------------------------------------------------------------------------- + +class TestFoo: + + @pytest.mark.asyncio + async def test_foo_returns_bar(self, magma_svc): + result = await magma_svc.foo() + assert result == "bar" + + @pytest.mark.asyncio + async def test_foo_returns_string(self, magma_svc): + result = await magma_svc.foo() + assert isinstance(result, str) + + @pytest.mark.asyncio + async def test_foo_idempotent(self, magma_svc): + r1 = await magma_svc.foo() + r2 = await magma_svc.foo() + assert r1 == r2 + + @pytest.mark.asyncio + async def test_foo_does_not_use_file_svc(self, magma_svc): + """foo() should not touch file_svc.""" + await magma_svc.foo() + magma_svc.file_svc.assert_not_called() + + +# --------------------------------------------------------------------------- +# Service wiring edge-cases +# --------------------------------------------------------------------------- + +class TestServiceWiring: + + def test_extra_services_ignored(self): + services = {"file_svc": MagicMock(), "extra": MagicMock()} + svc = MagmaService(services) + assert svc.file_svc is services["file_svc"] + + def test_empty_services(self): + svc = MagmaService({}) + assert svc.services == {} + assert svc.file_svc is None + + def test_services_with_none_values(self): + svc = MagmaService({"file_svc": None}) + assert svc.file_svc is None + + def test_multiple_instances_independent(self, mock_services): + svc_a = MagmaService(mock_services) + svc_b = MagmaService(mock_services) + assert svc_a is not svc_b + assert svc_a.log is not svc_b.log or svc_a.log.name == svc_b.log.name