-
-
Notifications
You must be signed in to change notification settings - Fork 407
/
test_connector.py
120 lines (90 loc) · 3.91 KB
/
test_connector.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import unittest
import asyncio
import asynctest
import asynctest.mock as amock
from opsdroid.cli.start import configure_lang
from opsdroid.core import OpsDroid
from opsdroid.connector import Connector, register_event
from opsdroid.events import Event, Message, Reaction
from opsdroid.cli.start import configure_lang
class TestConnectorBaseClass(unittest.TestCase):
"""Test the opsdroid connector base class."""
def setUp(self):
self.loop = asyncio.new_event_loop()
configure_lang({})
def test_init(self):
config = {"example_item": "test"}
connector = Connector(config, opsdroid=OpsDroid())
self.assertEqual(None, connector.default_target)
self.assertEqual("", connector.name)
self.assertEqual("test", connector.config["example_item"])
def test_property(self):
opsdroid = amock.CoroutineMock()
connector = Connector({"name": "shell"}, opsdroid=opsdroid)
self.assertEqual("shell", connector.configuration.get("name"))
def test_connect(self):
connector = Connector({}, opsdroid=OpsDroid())
with self.assertRaises(NotImplementedError):
self.loop.run_until_complete(connector.connect())
def test_listen(self):
connector = Connector({}, opsdroid=OpsDroid())
with self.assertRaises(NotImplementedError):
self.loop.run_until_complete(connector.listen())
def test_respond(self):
connector = Connector({}, opsdroid=OpsDroid())
with self.assertRaises(TypeError):
self.loop.run_until_complete(connector.send(Message("")))
def test_unsupported_event(self):
connector = Connector({}, opsdroid=OpsDroid())
with self.assertRaises(TypeError):
self.loop.run_until_complete(connector.send(Reaction("emoji")))
def test_incorrect_event(self):
class NotanEvent:
pass
class MyConnector(Connector):
@register_event(NotanEvent)
def send_my_event(self, event):
pass
with self.assertRaises(TypeError):
MyConnector()
def test_event_subclasses(self):
class MyEvent(Message):
pass
class MyConnector(Connector):
@register_event(Message, include_subclasses=True)
def send_my_event(self, event):
pass
c = MyConnector({})
assert MyEvent in c.events
class TestConnectorAsync(asynctest.TestCase):
"""Test the async methods of the opsdroid connector base class."""
async def setup(self):
configure_lang({})
async def test_disconnect(self):
connector = Connector({}, opsdroid=OpsDroid())
res = await connector.disconnect()
assert res is None
async def test_send_incorrect_event(self):
connector = Connector({"name": "shell"})
with self.assertRaises(TypeError):
await connector.send(object())
async def test_dep_respond(self):
connector = Connector({"name": "shell"})
with amock.patch("opsdroid.connector.Connector.send") as patched_send:
with self.assertWarns(DeprecationWarning):
await connector.respond("hello", room="bob")
patched_send.call_count == 1
async def test_dep_react(self):
connector = Connector({"name": "shell"})
with amock.patch("opsdroid.events.Message.respond") as patched_respond:
with self.assertWarns(DeprecationWarning):
await connector.react(Message("ori"), "hello")
patched_respond.call_count == 1
async def test_depreacted_properties(self):
connector = Connector({"name": "shell"})
connector.default_target = "spam"
with self.assertWarns(DeprecationWarning):
assert connector.default_room == "spam"
with self.assertWarns(DeprecationWarning):
connector.default_room = "eggs"
assert connector.default_target == "eggs"