pyappengine/tests/test_commands_loader.py

import pytest
import logging
from configparser import ConfigParser
from threading import Event

from appengine import CommandsLoader


SIMPLE_MODULE = """\
from appengine import Commands

class {classname}(Commands):
    def cmd_hello(self):
        \"\"\"Say hello.\"\"\"
        return "hello"
"""

THREADED_MODULE = """\
from appengine import Commands

class {classname}(Commands):
    def __init__(self, config, log):
        super().__init__(config, log)
        self.threaded = True

    def run(self):
        while not self.stopped:
            pass

    def cmd_ping(self):
        return "pong"
"""


@pytest.fixture
def log():
    return logging.getLogger("test_loader")


@pytest.fixture
def basic_config():
    config = ConfigParser()
    config["general"] = {}
    return config


def make_loader(tmp_path, config, log):
    stop_event = Event()
    loader = CommandsLoader(stop_event, config, log, str(tmp_path))
    return loader, stop_event


class TestLoadCommands:
    def test_loads_matching_file(self, tmp_path, basic_config, log):
        (tmp_path / "cmds_hello.py").write_text(SIMPLE_MODULE.format(classname="Hello"))
        loader, _ = make_loader(tmp_path, basic_config, log)
        assert "hello" in loader.cmods

    def test_ignores_non_prefixed_files(self, tmp_path, basic_config, log):
        (tmp_path / "not_a_module.py").write_text(SIMPLE_MODULE.format(classname="NotA"))
        loader, _ = make_loader(tmp_path, basic_config, log)
        assert len(loader.cmods) == 0

    def test_nickname_overrides_filename(self, tmp_path, log):
        (tmp_path / "cmds_mymod.py").write_text("""\
from appengine import Commands

class MyMod(Commands):
    def __init__(self, config, log):
        super().__init__(config, log)
        self.nickname = "custom"
""")
        config = ConfigParser()
        config["general"] = {}
        loader, _ = make_loader(tmp_path, config, log)
        assert "custom" in loader.cmods

    def test_loads_multiple_modules(self, tmp_path, basic_config, log):
        (tmp_path / "cmds_alpha.py").write_text(SIMPLE_MODULE.format(classname="Alpha"))
        (tmp_path / "cmds_beta.py").write_text(SIMPLE_MODULE.format(classname="Beta"))
        loader, _ = make_loader(tmp_path, basic_config, log)
        assert "alpha" in loader.cmods
        assert "beta" in loader.cmods

    def test_module_config_section_passed(self, tmp_path, log):
        (tmp_path / "cmds_configured.py").write_text("""\
from appengine import Commands

class Configured(Commands):
    def cmd_get_setting(self):
        return self.config.get("my_key")
""")
        config = ConfigParser()
        config["general"] = {}
        config["configured"] = {"my_key": "my_value"}
        loader, _ = make_loader(tmp_path, config, log)
        assert loader.cmods["configured"].config["my_key"] == "my_value"


class TestSetCmods:
    def test_cmods_shared_across_all_modules(self, tmp_path, basic_config, log):
        (tmp_path / "cmds_a.py").write_text(SIMPLE_MODULE.format(classname="A"))
        (tmp_path / "cmds_b.py").write_text(SIMPLE_MODULE.format(classname="B"))
        loader, _ = make_loader(tmp_path, basic_config, log)
        for v in loader.cmods.values():
            assert set(v.cmods.keys()) == {"a", "b"}


class TestLoadDependencies:
    def test_list_dependency_injected(self, tmp_path, log):
        (tmp_path / "cmds_provider.py").write_text(SIMPLE_MODULE.format(classname="Provider"))
        (tmp_path / "cmds_consumer.py").write_text("""\
from appengine import Commands

class Consumer(Commands):
    dependencies = ["provider"]

    def cmd_test(self):
        return "ok"
""")
        config = ConfigParser()
        config["general"] = {}
        loader, _ = make_loader(tmp_path, config, log)
        loader._load_dependencies()
        assert hasattr(loader.cmods["consumer"], "provider")
        assert loader.cmods["consumer"].provider is loader.cmods["provider"]

    def test_dict_dependency_injected(self, tmp_path, log):
        (tmp_path / "cmds_svc.py").write_text(SIMPLE_MODULE.format(classname="Svc"))
        (tmp_path / "cmds_client.py").write_text("""\
from appengine import Commands

class Client(Commands):
    dependencies = {"my_svc": "svc"}

    def cmd_test(self):
        return "ok"
""")
        config = ConfigParser()
        config["general"] = {}
        loader, _ = make_loader(tmp_path, config, log)
        loader._load_dependencies()
        assert hasattr(loader.cmods["client"], "my_svc")
        assert loader.cmods["client"].my_svc is loader.cmods["svc"]

    def test_missing_dependency_logs_error(self, tmp_path, log, caplog):
        (tmp_path / "cmds_orphan.py").write_text("""\
from appengine import Commands

class Orphan(Commands):
    dependencies = ["nonexistent"]

    def cmd_test(self):
        return "ok"
""")
        config = ConfigParser()
        config["general"] = {}
        loader, _ = make_loader(tmp_path, config, log)
        with caplog.at_level(logging.ERROR, logger="test_loader"):
            loader._load_dependencies()
        assert "nonexistent" in caplog.text


class TestLifecycle:
    def test_start_and_stop_threaded_module(self, tmp_path, log):
        (tmp_path / "cmds_worker.py").write_text(THREADED_MODULE.format(classname="Worker"))
        config = ConfigParser()
        config["general"] = {}
        loader, stop_event = make_loader(tmp_path, config, log)
        loader.start()
        assert loader.cmods["worker"].is_alive()
        loader.stop()
        loader.join()
        assert not loader.cmods["worker"].is_alive()

    def test_free_does_not_raise(self, tmp_path, basic_config, log):
        (tmp_path / "cmds_simple.py").write_text(SIMPLE_MODULE.format(classname="Simple"))
        loader, _ = make_loader(tmp_path, basic_config, log)
        loader.free()  # must not raise