mirror
/
amqtt
kopia lustrzana https://github.com/Yakifo/amqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

provide plugin manager context so that it identifies auth and topic check plugins correctly

pull/244/head
Andrew Mirsky 2025-07-02 12:24:41 -04:00
rodzic ee7250c720
commit 03bf1c34d9
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: A98E67635CDF2C39
1 zmienionych plików z 8 dodań i 2 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

10
tests/plugins/test_manager.py
Wyświetl plik

@ -84,8 +84,11 @@ class TestPluginManager(unittest.TestCase):
assert plugin.test_close_flag assert plugin.test_close_flag
def test_plugin_auth_coro(self) -> None: def test_plugin_auth_coro(self) -> None:
# provide context that activates auth plugins
context = BaseContext()
context.config = {'auth':{}}
manager = PluginManager("amqtt.test.plugins", context=None) manager = PluginManager("amqtt.test.plugins", context=context)
self.loop.run_until_complete(manager.map_plugin_auth(session=Session())) self.loop.run_until_complete(manager.map_plugin_auth(session=Session()))
self.loop.run_until_complete(asyncio.sleep(0.5)) self.loop.run_until_complete(asyncio.sleep(0.5))
plugin = manager.get_plugin("EventTestPlugin") plugin = manager.get_plugin("EventTestPlugin")
@ -93,8 +96,11 @@ class TestPluginManager(unittest.TestCase):
assert plugin.test_auth_flag assert plugin.test_auth_flag
def test_plugin_topic_coro(self) -> None: def test_plugin_topic_coro(self) -> None:
# provide context that activates topic check plugins
context = BaseContext()
context.config = {'topic-check':{}}
manager = PluginManager("amqtt.test.plugins", context=None) manager = PluginManager("amqtt.test.plugins", context=context)
self.loop.run_until_complete(manager.map_plugin_topic(session=Session(), topic="test", action=Action.PUBLISH)) self.loop.run_until_complete(manager.map_plugin_topic(session=Session(), topic="test", action=Action.PUBLISH))
self.loop.run_until_complete(asyncio.sleep(0.5)) self.loop.run_until_complete(asyncio.sleep(0.5))
plugin = manager.get_plugin("EventTestPlugin") plugin = manager.get_plugin("EventTestPlugin")