Python backend for a Slack's kudos plugin.
feat: added better exceptions and extra test
+95
-69
+6
kefi/models/plazas/exceptions.py
+6
kefi/models/plazas/exceptions.py
+5
-3
kefi/models/plazas/helpers.py
+5
-3
kefi/models/plazas/helpers.py
···
5
5
from sqlmodel import Session, select
6
6
7
7
from kefi.config import settings
8
+
from kefi.models.core.exceptions import NotEnoughKefi
8
9
from kefi.models.core.helpers import available_balance
9
10
from kefi.models.core.models import Transaction, User
11
+
from kefi.models.plazas.exceptions import AlreadyAttending, NotAttending
10
12
from kefi.models.plazas.models import Attendance, Plaza
11
13
12
14
···
57
59
balance = available_balance(user=user, session=session)
58
60
price = settings.PLAZA_PRICE
59
61
if balance < price:
60
-
raise ValueError("The user doesn't have enough balance")
62
+
raise NotEnoughKefi("The user doesn't have enough balance")
61
63
query = select(Attendance).filter(
62
64
Attendance.user == user, Attendance.plaza == plaza
63
65
)
64
66
results = session.exec(query)
65
67
if results.first():
66
-
raise ValueError("The user is already in the plaza")
68
+
raise AlreadyAttending("The user is already in the plaza")
67
69
attendance = Attendance(plaza=plaza, user=user)
68
70
session.add(attendance)
69
71
transaction = Transaction(amount=-price, user=user, attendance=attendance)
···
80
82
attendance_results = session.exec(attendance_query)
81
83
attendance = attendance_results.first()
82
84
if not attendance:
83
-
raise ValueError("The user is not in the plaza")
85
+
raise NotAttending("The user is not in the plaza")
84
86
session.delete(attendance)
85
87
transaction_query = select(Transaction).filter(
86
88
Transaction.attendance == attendance, Transaction.user == user
+6
-2
kefi/routers/helpers.py
+6
-2
kefi/routers/helpers.py
···
6
6
7
7
from kefi.constants import Command, InteractionType
8
8
from kefi.dependencies import InteractionParams, SlashCommandParams
9
+
from kefi.models.core.exceptions import NotEnoughKefi
9
10
from kefi.models.core.helpers import (
10
11
available_balance,
11
12
find_user_by_slack_user_id,
···
20
21
notify_receiver_user_chat_action,
21
22
send_action,
22
23
)
24
+
from kefi.models.plazas.exceptions import AlreadyAttending, NotAttending
23
25
from kefi.models.plazas.helpers import (
24
26
create_attendance,
25
27
delete_attendance,
···
195
197
"""Joins the plaza."""
196
198
try:
197
199
create_attendance(user=self.user, session=self.session)
198
-
except ValueError:
200
+
except NotEnoughKefi:
201
+
...
202
+
except AlreadyAttending:
199
203
...
200
204
201
205
def _view_submission_meet_leave(self, view_id: str):
202
206
"""Leaves the plaza."""
203
207
try:
204
208
delete_attendance(user=self.user, session=self.session)
205
-
except ValueError:
209
+
except NotAttending:
206
210
...
207
211
208
212
def interaction_view_submission(self) -> list:
+72
-62
kefi/tests/test_interactivity.py
+72
-62
kefi/tests/test_interactivity.py
···
6
6
from sqlmodel import Session
7
7
8
8
9
-
def test_default_interactivity(
10
-
session: Session, client: TestClient, mocker: MockerFixture
11
-
):
9
+
def test_default_interactivity(client: TestClient, mocker: MockerFixture):
12
10
views_open = mocker.patch.object(WebClient, "views_open")
13
11
views_open_response: dict = {} # Adds expected response here
14
12
views_open.side_effect = [views_open_response]
···
37
35
assert response.status_code == 200
38
36
39
37
40
-
{
41
-
"type": "view_submission",
42
-
"team": {"id": "T0RS507QX", "domain": "dekalabs"},
43
-
"user": {
44
-
"id": "UCBKX2GQ4",
45
-
"username": "marcos",
46
-
"name": "marcos",
47
-
"team_id": "T0RS507QX",
48
-
},
49
-
"api_app_id": "A0499FQH7V3",
50
-
"token": "7n993LRqLRydOnIkEhymVcYD",
51
-
"trigger_id": "4338017757538.25889007847.877c730e79e8a4b1917de09d38a8daa0",
52
-
"view": {
53
-
"id": "V04AMLW580G",
54
-
"team_id": "T0RS507QX",
55
-
"type": "modal",
56
-
"blocks": [
57
-
{
58
-
"type": "section",
59
-
"block_id": "9n+9",
60
-
"text": {
61
-
"type": "mrkdwn",
62
-
"text": "*:wave: \\u00a1Hola! \\u00bfTe vienes a la Kefi Plaza?*",
63
-
"verbatim": False,
64
-
},
38
+
def test_meet_join_view_submission(client: TestClient, mocker: MockerFixture):
39
+
views_open = mocker.patch.object(WebClient, "views_open")
40
+
views_open_response: dict = {} # Adds expected response here
41
+
views_open.side_effect = [views_open_response]
42
+
payload = json.dumps(
43
+
{
44
+
"type": "view_submission",
45
+
"team": {"id": "T0RS507QX", "domain": "dekalabs"},
46
+
"user": {
47
+
"id": "UCBKX2GQ4",
48
+
"username": "marcos",
49
+
"name": "marcos",
50
+
"team_id": "T0RS507QX",
65
51
},
66
-
{"type": "divider", "block_id": "MSUNF"},
67
-
{
68
-
"type": "section",
69
-
"block_id": "IbY",
70
-
"text": {
71
-
"type": "mrkdwn",
72
-
"text": "*Pr\\u00f3ximo encuentro*\\nFriday, 11 November\\n10:00\\n",
73
-
"verbatim": False,
74
-
},
75
-
"accessory": {
76
-
"type": "image",
77
-
"image_url": "https:\\/\\/api.slack.com\\/img\\/blocks\\/bkb_template_images\\/notifications.png",
78
-
"alt_text": "calendar thumbnail",
52
+
"api_app_id": "A0499FQH7V3",
53
+
"token": "7n993LRqLRydOnIkEhymVcYD",
54
+
"trigger_id": "4338017757538.25889007847.877c730e79e8a4b1917de09d38a8daa0",
55
+
"view": {
56
+
"id": "V04AMLW580G",
57
+
"team_id": "T0RS507QX",
58
+
"type": "modal",
59
+
"blocks": [
60
+
{
61
+
"type": "section",
62
+
"block_id": "9n+9",
63
+
"text": {
64
+
"type": "mrkdwn",
65
+
"text": "*:wave: \\u00a1Hola! \\u00bfTe vienes a la Kefi Plaza?*",
66
+
"verbatim": False,
67
+
},
68
+
},
69
+
{"type": "divider", "block_id": "MSUNF"},
70
+
{
71
+
"type": "section",
72
+
"block_id": "IbY",
73
+
"text": {
74
+
"type": "mrkdwn",
75
+
"text": "*Pr\\u00f3ximo encuentro*\\nFriday, 11 November\\n10:00\\n",
76
+
"verbatim": False,
77
+
},
78
+
"accessory": {
79
+
"type": "image",
80
+
"image_url": "https:\\/\\/api.slack.com\\/img\\/blocks\\/bkb_template_images\\/notifications.png",
81
+
"alt_text": "calendar thumbnail",
82
+
},
83
+
},
84
+
],
85
+
"private_metadata": "",
86
+
"callback_id": "meet_join",
87
+
"state": {"values": {}},
88
+
"hash": "1667896295.Ry45BkuV",
89
+
"title": {"type": "plain_text", "text": "Kefi", "emoji": True},
90
+
"clear_on_close": False,
91
+
"notify_on_close": False,
92
+
"close": {"type": "plain_text", "text": "Cancelar", "emoji": True},
93
+
"submit": {
94
+
"type": "plain_text",
95
+
"text": "\\u00a1Me apunto!",
96
+
"emoji": True,
79
97
},
98
+
"previous_view_id": None,
99
+
"root_view_id": "V04AMLW580G",
100
+
"app_id": "A0499FQH7V3",
101
+
"external_id": "",
102
+
"app_installed_team_id": "T0RS507QX",
103
+
"bot_id": "B0495T0Q0S2",
80
104
},
81
-
],
82
-
"private_metadata": "",
83
-
"callback_id": "meet_join",
84
-
"state": {"values": {}},
85
-
"hash": "1667896295.Ry45BkuV",
86
-
"title": {"type": "plain_text", "text": "Kefi", "emoji": True},
87
-
"clear_on_close": False,
88
-
"notify_on_close": False,
89
-
"close": {"type": "plain_text", "text": "Cancelar", "emoji": True},
90
-
"submit": {"type": "plain_text", "text": "\\u00a1Me apunto!", "emoji": True},
91
-
"previous_view_id": None,
92
-
"root_view_id": "V04AMLW580G",
93
-
"app_id": "A0499FQH7V3",
94
-
"external_id": "",
95
-
"app_installed_team_id": "T0RS507QX",
96
-
"bot_id": "B0495T0Q0S2",
97
-
},
98
-
"response_urls": [],
99
-
"is_enterprise_install": False,
100
-
"enterprise": None,
101
-
}
105
+
"response_urls": [],
106
+
"is_enterprise_install": False,
107
+
"enterprise": None,
108
+
}
109
+
)
110
+
response = client.post("/interactivity/", data={"payload": payload})
111
+
assert response.status_code == 200
+4
-2
kefi/tests/test_models.py
+4
-2
kefi/tests/test_models.py
···
3
3
from sqlmodel import Session
4
4
5
5
from kefi.config import settings
6
+
from kefi.models.core.exceptions import NotEnoughKefi
6
7
from kefi.models.core.helpers import (
7
8
available_balance,
8
9
received_balance,
···
12
13
from kefi.models.core.models import Transaction, User
13
14
from kefi.models.kudos.helpers import send_action
14
15
from kefi.models.kudos.models import Action
16
+
from kefi.models.plazas.exceptions import AlreadyAttending
15
17
from kefi.models.plazas.helpers import create_attendance, get_or_create_current_plaza
16
18
from kefi.models.plazas.models import Attendance, Plaza
17
19
···
117
119
try:
118
120
create_attendance(user=user, session=session)
119
121
exception = False
120
-
except ValueError:
122
+
except NotEnoughKefi:
121
123
exception = True
122
124
assert exception
123
125
···
142
144
try:
143
145
create_attendance(user=user, session=session)
144
146
exception = False
145
-
except ValueError:
147
+
except AlreadyAttending:
146
148
exception = True
147
149
assert exception