kopia lustrzana https://github.com/Yakifo/amqtt
plugins.topic_checking tests: Fix issues identified by flake8
Stuart Longland
Florian Ludwig
rodzic
bf5279911b
commit
fed6d2c31c
|
|
@ -130,7 +130,7 @@ async def test_taboo_empty_config(logdog):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_taboo_not_taboo_topic(logdog):
|
async def test_taboo_disabled(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicTabooPlugin returns true if checking disabled.
|
Check TopicTabooPlugin returns true if checking disabled.
|
||||||
"""
|
"""
|
||||||
|
|
@ -278,7 +278,7 @@ def test_topic_ac_match_exact():
|
||||||
assert TopicAccessControlListPlugin.topic_ac("exact/topic", "exact/topic") is True
|
assert TopicAccessControlListPlugin.topic_ac("exact/topic", "exact/topic") is True
|
||||||
|
|
||||||
|
|
||||||
def test_topic_ac_match_hash():
|
def test_topic_ac_match_plus():
|
||||||
"""
|
"""
|
||||||
Test TopicAccessControlListPlugin.topic_ac correctly handles '+' wildcard.
|
Test TopicAccessControlListPlugin.topic_ac correctly handles '+' wildcard.
|
||||||
"""
|
"""
|
||||||
|
|
@ -330,19 +330,18 @@ async def test_taclp_true_disabled(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns true if topic checking is disabled.
|
Check TopicAccessControlListPlugin returns true if topic checking is disabled.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {"topic-check": {"enabled": False}}
|
||||||
context.config = {"topic-check": {"enabled": False}}
|
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.publish, session=session, topic="a/topic"
|
action=Action.publish, session=session, topic="a/topic"
|
||||||
)
|
)
|
||||||
assert authorised is True
|
assert authorised is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -351,19 +350,18 @@ async def test_taclp_true_no_pub_acl(logdog):
|
||||||
Check TopicAccessControlListPlugin returns true if action=publish and no publish-acl given.
|
Check TopicAccessControlListPlugin returns true if action=publish and no publish-acl given.
|
||||||
(This is for backward-compatibility with existing installations.)
|
(This is for backward-compatibility with existing installations.)
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {"topic-check": {"enabled": True}}
|
||||||
context.config = {"topic-check": {"enabled": True}}
|
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.publish, session=session, topic="a/topic"
|
action=Action.publish, session=session, topic="a/topic"
|
||||||
)
|
)
|
||||||
assert authorised is True
|
assert authorised is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -371,24 +369,23 @@ async def test_taclp_false_sub_no_topic(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns false user there is no topic.
|
Check TopicAccessControlListPlugin returns false user there is no topic.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"acl": {"anotheruser": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"acl": {"anotheruser": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.subscribe, session=session, topic=""
|
action=Action.subscribe, session=session, topic=""
|
||||||
)
|
)
|
||||||
assert authorised is False
|
assert authorised is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -396,24 +393,23 @@ async def test_taclp_false_sub_unknown_user(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns false user is not listed in ACL.
|
Check TopicAccessControlListPlugin returns false user is not listed in ACL.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"acl": {"anotheruser": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"acl": {"anotheruser": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.subscribe, session=session, topic="allowed/topic"
|
action=Action.subscribe, session=session, topic="allowed/topic"
|
||||||
)
|
)
|
||||||
assert authorised is False
|
assert authorised is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -421,24 +417,23 @@ async def test_taclp_false_sub_no_permission(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns false if "acl" does not list allowed topic.
|
Check TopicAccessControlListPlugin returns false if "acl" does not list allowed topic.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.subscribe, session=session, topic="forbidden/topic"
|
action=Action.subscribe, session=session, topic="forbidden/topic"
|
||||||
)
|
)
|
||||||
assert authorised is False
|
assert authorised is False
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -446,24 +441,23 @@ async def test_taclp_true_sub_permission(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns true if "acl" lists allowed topic.
|
Check TopicAccessControlListPlugin returns true if "acl" lists allowed topic.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.subscribe, session=session, topic="allowed/topic"
|
action=Action.subscribe, session=session, topic="allowed/topic"
|
||||||
)
|
)
|
||||||
assert authorised is True
|
assert authorised is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -471,24 +465,23 @@ async def test_taclp_true_pub_permission(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin returns true if "publish-acl" lists allowed topic for publish action.
|
Check TopicAccessControlListPlugin returns true if "publish-acl" lists allowed topic for publish action.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"publish-acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"publish-acl": {"user": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = "user"
|
session.username = "user"
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.publish, session=session, topic="allowed/topic"
|
action=Action.publish, session=session, topic="allowed/topic"
|
||||||
)
|
)
|
||||||
assert authorised is True
|
assert authorised is True
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
@ -496,21 +489,20 @@ async def test_taclp_true_anon_sub_permission(logdog):
|
||||||
"""
|
"""
|
||||||
Check TopicAccessControlListPlugin handles anonymous users.
|
Check TopicAccessControlListPlugin handles anonymous users.
|
||||||
"""
|
"""
|
||||||
with logdog() as pile:
|
context = BaseContext()
|
||||||
context = BaseContext()
|
context.logger = logging.getLogger("testlog")
|
||||||
context.logger = logging.getLogger("testlog")
|
context.config = {
|
||||||
context.config = {
|
"topic-check": {
|
||||||
"topic-check": {
|
"enabled": True,
|
||||||
"enabled": True,
|
"acl": {"anonymous": ["allowed/topic", "another/allowed/topic/#"]},
|
||||||
"acl": {"anonymous": ["allowed/topic", "another/allowed/topic/#"]},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
session = Session()
|
session = Session()
|
||||||
session.username = None
|
session.username = None
|
||||||
|
|
||||||
plugin = TopicAccessControlListPlugin(context)
|
plugin = TopicAccessControlListPlugin(context)
|
||||||
authorised = await plugin.topic_filtering(
|
authorised = await plugin.topic_filtering(
|
||||||
action=Action.subscribe, session=session, topic="allowed/topic"
|
action=Action.subscribe, session=session, topic="allowed/topic"
|
||||||
)
|
)
|
||||||
assert authorised is True
|
assert authorised is True
|
||||||
|
|
|
||||||
Ładowanie…
Reference in New Issue