kopia lustrzana https://github.com/Yakifo/amqtt
71 wiersze
1.6 KiB
Python
71 wiersze
1.6 KiB
Python
import logging
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from amqtt.plugins.base import BasePlugin, BaseAuthPlugin, BaseTopicPlugin
|
|
from amqtt.contexts import BaseContext, Action
|
|
|
|
from amqtt.session import Session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TestSimplePlugin(BasePlugin):
|
|
|
|
def __init__(self, context: BaseContext):
|
|
super().__init__(context)
|
|
|
|
|
|
class TestConfigPlugin(BasePlugin):
|
|
|
|
def __init__(self, context: BaseContext):
|
|
super().__init__(context)
|
|
|
|
@dataclass
|
|
class Config:
|
|
option1: int
|
|
option2: str
|
|
option3: int = 20
|
|
|
|
|
|
class TestCoroErrorPlugin(BaseAuthPlugin):
|
|
|
|
def authenticate(self, *, session: Session) -> bool | None:
|
|
return True
|
|
|
|
|
|
class TestAuthPlugin(BaseAuthPlugin):
|
|
|
|
async def authenticate(self, *, session: Session) -> bool | None:
|
|
return True
|
|
|
|
|
|
class TestNoAuthPlugin(BaseAuthPlugin):
|
|
|
|
|
|
async def authenticate(self, *, session: Session) -> bool | None:
|
|
return False
|
|
|
|
|
|
class TestAllowTopicPlugin(BaseTopicPlugin):
|
|
|
|
def __init__(self, context: BaseContext):
|
|
super().__init__(context)
|
|
|
|
async def topic_filtering(
|
|
self, *, session: Session | None = None, topic: str | None = None, action: Action | None = None
|
|
) -> bool:
|
|
return True
|
|
|
|
|
|
class TestBlockTopicPlugin(BaseTopicPlugin):
|
|
|
|
def __init__(self, context: BaseContext):
|
|
super().__init__(context)
|
|
|
|
async def topic_filtering(
|
|
self, *, session: Session | None = None, topic: str | None = None, action: Action | None = None
|
|
) -> bool:
|
|
logger.debug("topic filtering plugin is returning false")
|
|
return False
|