-
-
Notifications
You must be signed in to change notification settings - Fork 407
/
test_database_sqlite.py
115 lines (88 loc) · 3.5 KB
/
test_database_sqlite.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
"""Tests for the DatabaseSqlite class."""
import asyncio
import unittest
import asynctest
import asynctest.mock as amock
from opsdroid.database.sqlite import DatabaseSqlite
from opsdroid.cli.start import configure_lang
class TestDatabaseSqlite(unittest.TestCase):
"""A database test class.
Test the opsdroid sqlite database class.
"""
def setUp(self):
self.loop = asyncio.new_event_loop()
configure_lang({})
def test_init(self):
"""Test initialisation of database class.
This method will test the initialisation of the database
class. It will assert if the database class properties are
declared.
"""
database = DatabaseSqlite({"path": "sqlite.db"})
self.assertEqual(None, database.client)
self.assertEqual("sqlite.db", database.db_file)
self.assertEqual("opsdroid", database.table)
self.assertEqual({"isolation_level": None}, database.conn_args)
class TestDatabaseSqliteAsync(asynctest.TestCase):
"""A async database test class.
Test the async methods of the opsdroid sqlite database class.
"""
async def test_connect(self):
"""Test database connection.
This method will test the database connection of sqlite database.
As the database is created `opsdroid` table is created first.
"""
database = DatabaseSqlite({"path": "sqlite.db"})
opsdroid = amock.CoroutineMock()
opsdroid.eventloop = self.loop
try:
await database.connect()
table = database.table
client = type(database.client).__name__
await database.disconnect()
except NotImplementedError:
raise Exception
else:
self.assertEqual("opsdroid", table)
self.assertEqual("Connection", client)
async def test_disconnect(self):
"""Test of database disconnection.
This method will test the database disconnection of sqlite database.
"""
database = DatabaseSqlite({"path": "sqlite.db"})
opsdroid = amock.CoroutineMock()
opsdroid.eventloop = self.loop
try:
await database.connect()
await database.disconnect()
except NotImplementedError:
raise Exception
else:
pass
async def test_get_put_and_delete(self):
"""Test get, put and delete functions of database
This method will test the get, put and delete functions which help to read
and write data from the database. The function `put` a value with
key and asserts the same value after the `get` operation is completed
followed by the `delete` operation which deletes the key.
"""
database = DatabaseSqlite({"path": "sqlite.db"})
opsdroid = amock.CoroutineMock()
opsdroid.eventloop = self.loop
try:
await database.connect()
table = database.table
client = type(database.client).__name__
await database.put("hello", {})
data = await database.get("hello")
await database.delete("hello")
await database.disconnect()
except NotImplementedError:
raise Exception
else:
self.assertEqual("opsdroid", table)
self.assertEqual({}, data)
self.assertEqual("Connection", client)
async def test_deprecated_path(self):
database = DatabaseSqlite({"file": "sqlite.db"})
assert database.db_file == "sqlite.db"