diff --git a/app/config.py.example b/app/config.py.example
index 5b499142..b165b5f1 100644
--- a/app/config.py.example
+++ b/app/config.py.example
@@ -1,7 +1,17 @@
import os
from fastapi_mail import ConnectionConfig
-# flake8: noqa
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+ app_name: str = "PyLander"
+ bot_api: str = "BOT_API"
+ webhook_url: str = "WEBHOOK_URL"
+
+ class Config:
+ env_file = ".env"
+
# general
DOMAIN = 'Our-Domain'
diff --git a/app/database/models.py b/app/database/models.py
index 91fc16b2..4d243fef 100644
--- a/app/database/models.py
+++ b/app/database/models.py
@@ -32,6 +32,7 @@ class User(Base):
full_name = Column(String)
description = Column(String, default="Happy new user!")
avatar = Column(String, default="profile.png")
+ telegram_id = Column(String, unique=True)
is_active = Column(Boolean, default=False)
events = relationship("UserEvent", back_populates="participants")
diff --git a/app/dependencies.py b/app/dependencies.py
index 79ae18c5..bf3fe2ef 100644
--- a/app/dependencies.py
+++ b/app/dependencies.py
@@ -1,3 +1,4 @@
+from functools import lru_cache
import os
from fastapi.templating import Jinja2Templates
@@ -11,3 +12,8 @@
TEMPLATES_PATH = os.path.join(APP_PATH, "templates")
templates = Jinja2Templates(directory=TEMPLATES_PATH)
+
+
+@lru_cache()
+def get_settings():
+ return config.Settings()
diff --git a/app/main.py b/app/main.py
index 796ade3e..9f562d34 100644
--- a/app/main.py
+++ b/app/main.py
@@ -6,8 +6,9 @@
from app.database.database import engine
from app.dependencies import (
MEDIA_PATH, STATIC_PATH, templates)
-from app.routers import (agenda, dayview, email, event, invitation, profile,
- search)
+from app.routers import (
+ agenda, dayview, email, event, invitation, profile, search, telegram)
+from app.telegram.bot import telegram_bot
def create_tables(engine, psql_environment):
@@ -29,11 +30,14 @@ def create_tables(engine, psql_environment):
app.include_router(profile.router)
app.include_router(event.router)
app.include_router(agenda.router)
+app.include_router(telegram.router)
app.include_router(dayview.router)
app.include_router(email.router)
app.include_router(invitation.router)
app.include_router(search.router)
+telegram_bot.set_webhook()
+
@app.get("/")
async def home(request: Request):
diff --git a/app/media/fake_user.png b/app/media/fake_user.png
deleted file mode 100644
index bd856aaa..00000000
Binary files a/app/media/fake_user.png and /dev/null differ
diff --git a/app/routers/profile.py b/app/routers/profile.py
index 40ac1073..3e70e2cf 100644
--- a/app/routers/profile.py
+++ b/app/routers/profile.py
@@ -26,6 +26,7 @@ def get_placeholder_user():
email='my@email.po',
password='1a2s3d4f5g6',
full_name='My Name',
+ telegram_id=''
)
@@ -63,8 +64,7 @@ async def update_user_fullname(
session.commit()
url = router.url_path_for("profile")
- response = RedirectResponse(url=url, status_code=HTTP_302_FOUND)
- return response
+ return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
@router.post("/update_user_email")
@@ -110,14 +110,27 @@ async def upload_user_photo(
# Save to database
user.avatar = await process_image(pic, user)
session.commit()
-
finally:
- session.close()
-
url = router.url_path_for("profile")
return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
+@router.post("/update_telegram_id")
+async def update_telegram_id(
+ request: Request, session=Depends(get_db)):
+
+ user = session.query(User).filter_by(id=1).first()
+ data = await request.form()
+ new_telegram_id = data['telegram_id']
+
+ # Update database
+ user.telegram_id = new_telegram_id
+ session.commit()
+
+ url = router.url_path_for("profile")
+ return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
+
+
async def process_image(image, user):
img = Image.open(io.BytesIO(image))
width, height = img.size
diff --git a/app/routers/telegram.py b/app/routers/telegram.py
new file mode 100644
index 00000000..5c5e9d7b
--- /dev/null
+++ b/app/routers/telegram.py
@@ -0,0 +1,33 @@
+from fastapi import APIRouter, Body, Depends, Request
+
+from app.database.database import get_db
+from app.database.models import User
+from app.telegram.handlers import MessageHandler, reply_unknown_user
+from app.telegram.models import Chat
+
+
+router = APIRouter(
+ prefix="/telegram",
+ tags=["telegram"],
+ responses={404: {"description": "Not found"}},
+)
+
+
+@router.get("/")
+async def telegram(request: Request, session=Depends(get_db)):
+
+ # todo: Add templating
+ return "Start using PyLander telegram bot!"
+
+
+@router.post("/")
+async def bot_client(req: dict = Body(...), session=Depends(get_db)):
+ chat = Chat(req)
+
+ # Check if current chatter is registered to use the bot
+ user = session.query(User).filter_by(telegram_id=chat.user_id).first()
+ if user is None:
+ return await reply_unknown_user(chat)
+
+ message = MessageHandler(chat, user)
+ return await message.process_callback()
diff --git a/app/telegram/__init__.py b/app/telegram/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/app/telegram/bot.py b/app/telegram/bot.py
new file mode 100644
index 00000000..30aeaf6c
--- /dev/null
+++ b/app/telegram/bot.py
@@ -0,0 +1,11 @@
+from app import config
+from app.dependencies import get_settings
+from .models import Bot
+
+
+settings: config.Settings = get_settings()
+
+BOT_API = settings.bot_api
+WEBHOOK_URL = settings.webhook_url
+
+telegram_bot = Bot(BOT_API, WEBHOOK_URL)
diff --git a/app/telegram/handlers.py b/app/telegram/handlers.py
new file mode 100644
index 00000000..c522418a
--- /dev/null
+++ b/app/telegram/handlers.py
@@ -0,0 +1,115 @@
+import datetime
+
+from .keyboards import (
+ DATE_FORMAT, gen_inline_keyboard, get_this_week_buttons, show_events_kb)
+from .models import Chat
+from .bot import telegram_bot
+from app.database.models import User
+
+
+class MessageHandler:
+ def __init__(self, chat: Chat, user: User):
+ self.chat = chat
+ self.user = user
+ self.handlers = {}
+ self.handlers['/start'] = self.start_handler
+ self.handlers['/show_events'] = self.show_events_handler
+ self.handlers['Today'] = self.today_handler
+ self.handlers['This week'] = self.this_week_handler
+
+ # Add next 6 days to handlers dict
+ for row in get_this_week_buttons():
+ for button in row:
+ self.handlers[button['text']] = self.chosen_day_handler
+
+ async def process_callback(self):
+ if self.chat.message in self.handlers:
+ return await self.handlers[self.chat.message]()
+ return await self.default_handler()
+
+ async def default_handler(self):
+ answer = "Unknown command."
+ await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
+ return answer
+
+ async def start_handler(self):
+ answer = f'''Hello, {self.chat.first_name}!
+Welcome to Pylander telegram client!'''
+ await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
+ return answer
+
+ async def show_events_handler(self):
+ answer = 'Choose events day.'
+ await telegram_bot.send_message(
+ chat_id=self.chat.user_id,
+ text=answer,
+ reply_markup=show_events_kb)
+ return answer
+
+ async def today_handler(self):
+ today = datetime.datetime.today()
+ events = [
+ _.events for _ in self.user.events
+ if _.events.start <= today <= _.events.end]
+
+ answer = f"{today.strftime('%B %d')}, {today.strftime('%A')} Events:\n"
+
+ if not events:
+ answer = "There're no events today."
+
+ for event in events:
+ answer += f'''
+From {event.start.strftime('%d/%m %H:%M')} \
+to {event.end.strftime('%d/%m %H:%M')}: {event.title}.\n'''
+
+ await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
+ return answer
+
+ async def this_week_handler(self):
+ answer = 'Choose a day.'
+ this_week_kb = gen_inline_keyboard(get_this_week_buttons())
+
+ await telegram_bot.send_message(
+ chat_id=self.chat.user_id,
+ text=answer,
+ reply_markup=this_week_kb)
+ return answer
+
+ async def chosen_day_handler(self):
+ # Convert chosen day (string) to datetime format
+ chosen_date = datetime.datetime.strptime(
+ self.chat.message, DATE_FORMAT)
+
+ events = [
+ _.events for _ in self.user.events
+ if _.events.start <= chosen_date <= _.events.end]
+
+ answer = f"{chosen_date.strftime('%B %d')}, \
+{chosen_date.strftime('%A')} Events:\n"
+
+ if not events:
+ answer = f"There're no events on {chosen_date.strftime('%B %d')}."
+
+ for event in events:
+ answer += f'''
+From {event.start.strftime('%d/%m %H:%M')} \
+to {event.end.strftime('%d/%m %H:%M')}: {event.title}.\n'''
+
+ await telegram_bot.send_message(chat_id=self.chat.user_id, text=answer)
+ return answer
+
+
+async def reply_unknown_user(chat):
+ answer = f'''
+Hello, {chat.first_name}!
+
+To use PyLander Bot you have to register
+your Telegram Id in your profile page.
+
+Your Id is {chat.user_id}
+Keep it secret!
+
+https://calendar.pythonic.guru/profile/
+'''
+ await telegram_bot.send_message(chat_id=chat.user_id, text=answer)
+ return answer
diff --git a/app/telegram/keyboards.py b/app/telegram/keyboards.py
new file mode 100644
index 00000000..0cbdd66f
--- /dev/null
+++ b/app/telegram/keyboards.py
@@ -0,0 +1,47 @@
+import datetime
+import json
+from typing import Any, Dict, List
+
+
+show_events_buttons = [
+ [
+ {'text': 'Today', 'callback_data': 'Today'},
+ {'text': 'This week', 'callback_data': 'This week'}
+ ]
+]
+
+DATE_FORMAT = '%d %b %Y'
+
+
+def get_this_week_buttons() -> List[List[Any]]:
+ today = datetime.datetime.today()
+ buttons = []
+ for day in range(1, 7):
+ day = today + datetime.timedelta(days=day)
+ buttons.append(day.strftime(DATE_FORMAT))
+
+ return [
+ [
+ {'text': buttons[0],
+ 'callback_data': buttons[0]},
+ {'text': buttons[1],
+ 'callback_data': buttons[1]},
+ {'text': buttons[2],
+ 'callback_data': buttons[2]}
+ ],
+ [
+ {'text': buttons[3],
+ 'callback_data': buttons[3]},
+ {'text': buttons[4],
+ 'callback_data': buttons[4]},
+ {'text': buttons[5],
+ 'callback_data': buttons[5]}
+ ]
+ ]
+
+
+def gen_inline_keyboard(buttons: List[List[Any]]) -> Dict[str, Any]:
+ return {'reply_markup': json.dumps({'inline_keyboard': buttons})}
+
+
+show_events_kb = gen_inline_keyboard(show_events_buttons)
diff --git a/app/telegram/models.py b/app/telegram/models.py
new file mode 100644
index 00000000..b34b1a6e
--- /dev/null
+++ b/app/telegram/models.py
@@ -0,0 +1,57 @@
+from typing import Any, Dict, Optional
+
+from httpx import AsyncClient
+import requests
+
+
+class Chat:
+ def __init__(self, data: Dict):
+ self.message = self._get_message_content(data)
+ self.user_id = self._get_user_id(data)
+ self.first_name = self._get_first_name(data)
+
+ def _get_message_content(self, data: Dict) -> str:
+ if 'callback_query' in data:
+ return data['callback_query']['data']
+ return data['message']['text']
+
+ def _get_user_id(self, data: Dict) -> str:
+ if 'callback_query' in data:
+ return data['callback_query']['from']['id']
+ return data['message']['from']['id']
+
+ def _get_first_name(self, data: Dict) -> str:
+ if 'callback_query' in data:
+ return data['callback_query']['from']['first_name']
+ return data['message']['from']['first_name']
+
+
+class Bot:
+ def __init__(self, bot_api: str, webhook_url: str):
+ self.base = self._set_base_url(bot_api)
+ self.webhook_setter_url = self._set_webhook_setter_url(webhook_url)
+
+ def _set_base_url(self, bot_api: str) -> str:
+ return f'https://api.telegram.org/bot{bot_api}/'
+
+ def _set_webhook_setter_url(self, webhook_url: str) -> str:
+ return f'{self.base}setWebhook?url={webhook_url}/telegram/'
+
+ def set_webhook(self):
+ return requests.get(self.webhook_setter_url)
+
+ def drop_webhook(self):
+ data = {'drop_pending_updates': True}
+ return requests.get(url=f'{self.base}deleteWebhook', data=data)
+
+ async def send_message(
+ self, chat_id: str,
+ text: str,
+ reply_markup: Optional[Dict[str, Any]] = None):
+ async with AsyncClient(base_url=self.base) as ac:
+ message = {
+ 'chat_id': chat_id,
+ 'text': text}
+ if reply_markup:
+ message.update(reply_markup)
+ return await ac.post('sendMessage', data=message)
diff --git a/app/templates/profile.html b/app/templates/profile.html
index aecb5164..2fadafb8 100644
--- a/app/templates/profile.html
+++ b/app/templates/profile.html
@@ -21,25 +21,31 @@
+
+
+
@@ -131,6 +137,32 @@ Upload photo
+
+
+
@@ -158,14 +190,18 @@ {{ user.full_name }}
- Features
+ Explore more features
+ {% if not user.telegram_id %}
-
- Export my calendar
+
+ Try PyLander bot
+
+ {% endif %}
-
- Your feature
+ Export my calendar
-
Your feature
diff --git a/requirements.txt b/requirements.txt
index 69f01ab2..d1cf56d5 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,6 +4,7 @@ aiosmtpd==1.2.2
aiosmtplib==1.1.4
apipkg==1.5
arrow==0.17.0
+asyncio==3.4.3
async-timeout==3.0.1
atomicwrites==1.4.0
atpublic==2.1.2
diff --git a/tests/asyncio_fixture.py b/tests/asyncio_fixture.py
new file mode 100644
index 00000000..fbc79108
--- /dev/null
+++ b/tests/asyncio_fixture.py
@@ -0,0 +1,63 @@
+from datetime import datetime, timedelta
+
+from httpx import AsyncClient
+import pytest
+
+from app.database.database import Base
+from app.database.models import User
+from app.main import app
+from app.routers import telegram
+from app.routers.event import create_event
+from tests.conftest import test_engine, get_test_db
+
+
+@pytest.fixture
+async def telegram_client():
+ Base.metadata.create_all(bind=test_engine)
+ app.dependency_overrides[telegram.get_db] = get_test_db
+ async with AsyncClient(app=app, base_url="http://test") as ac:
+ yield ac
+ app.dependency_overrides = {}
+ Base.metadata.drop_all(bind=test_engine)
+
+
+session = get_test_db()
+today_date = datetime.today().replace(hour=0, minute=0, second=0)
+
+
+def get_test_placeholder_user():
+ return User(
+ username='fake_user',
+ email='fake@mail.fake',
+ password='123456fake',
+ full_name='FakeName',
+ telegram_id='666666'
+ )
+
+
+@pytest.fixture
+def fake_user_events():
+ Base.metadata.create_all(bind=test_engine)
+ user = get_test_placeholder_user()
+ session.add(user)
+ session.commit()
+ create_event(
+ db=session,
+ title='Cool today event',
+ start=today_date,
+ end=today_date + timedelta(days=2),
+ content='test event',
+ owner_id=user.id,
+ location="Here",
+ )
+ create_event(
+ db=session,
+ title='Cool (somewhen in two days) event',
+ start=today_date + timedelta(days=1),
+ end=today_date + timedelta(days=3),
+ content='this week test event',
+ owner_id=user.id,
+ location="Here",
+ )
+ yield user
+ Base.metadata.drop_all(bind=test_engine)
diff --git a/tests/client_fixture.py b/tests/client_fixture.py
index c99b5fe5..edc24889 100644
--- a/tests/client_fixture.py
+++ b/tests/client_fixture.py
@@ -4,7 +4,7 @@
from app.database.models import User
from app.main import app
from app.database.database import Base
-from app.routers import profile, agenda, invitation
+from app.routers import agenda, invitation, profile
from tests.conftest import test_engine, get_test_db
@@ -56,5 +56,6 @@ def get_test_placeholder_user():
username='fake_user',
email='fake@mail.fake',
password='123456fake',
- full_name='FakeName'
+ full_name='FakeName',
+ telegram_id='666666'
)
diff --git a/tests/conftest.py b/tests/conftest.py
index a5e0defd..1100fc70 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -11,6 +11,7 @@
'tests.invitation_fixture',
'tests.association_fixture',
'tests.client_fixture',
+ 'tests.asyncio_fixture',
'smtpdfix',
]
diff --git a/tests/test_a_telegram_asyncio.py b/tests/test_a_telegram_asyncio.py
new file mode 100644
index 00000000..37d832ac
--- /dev/null
+++ b/tests/test_a_telegram_asyncio.py
@@ -0,0 +1,260 @@
+from datetime import timedelta
+
+from fastapi import status
+import pytest
+
+from .asyncio_fixture import today_date
+from .client_fixture import get_test_placeholder_user
+from app.telegram.handlers import MessageHandler, reply_unknown_user
+from app.telegram.keyboards import DATE_FORMAT
+from app.telegram.models import Bot, Chat
+
+
+def gen_message(text):
+ return {
+ 'update_id': 10000000,
+ 'message': {
+ 'message_id': 2434,
+ 'from': {
+ 'id': 666666,
+ 'is_bot': False,
+ 'first_name': 'Moshe',
+ 'username': 'banana',
+ 'language_code': 'en'
+ },
+ 'chat': {
+ 'id': 666666,
+ 'first_name': 'Moshe',
+ 'username': 'banana',
+ 'type': 'private'
+ },
+ 'date': 1611240725,
+ 'text': f'{text}'
+ }
+ }
+
+
+def gen_callback(text):
+ return {
+ 'update_id': 568265,
+ 'callback_query': {
+ 'id': '546565356486',
+ 'from': {
+ 'id': 666666,
+ 'is_bot': False,
+ 'first_name': 'Moshe',
+ 'username': 'banana',
+ 'language_code': 'en'
+ }, 'message': {
+ 'message_id': 838,
+ 'from': {
+ 'id': 2566252,
+ 'is_bot': True,
+ 'first_name': 'PyLandar',
+ 'username': 'pylander_bot'
+ }, 'chat': {
+ 'id': 666666,
+ 'first_name': 'Moshe',
+ 'username': 'banana',
+ 'type': 'private'
+ },
+ 'date': 161156,
+ 'text': 'Choose events day.',
+ 'reply_markup': {
+ 'inline_keyboard': [
+ [
+ {
+ 'text': 'Today',
+ 'callback_data': 'Today'
+ },
+ {
+ 'text': 'This week',
+ 'callback_data': 'This week'
+ }
+ ]
+ ]
+ }
+ },
+ 'chat_instance': '-154494',
+ 'data': f'{text}'}}
+
+
+class TestChatModel:
+
+ @staticmethod
+ def test_private_message():
+ chat = Chat(gen_message('Cool message'))
+ assert chat.message == 'Cool message'
+ assert chat.user_id == 666666
+ assert chat.first_name == 'Moshe'
+
+ @staticmethod
+ def test_callback_message():
+ chat = Chat(gen_callback('Callback Message'))
+ assert chat.message == 'Callback Message'
+ assert chat.user_id == 666666
+ assert chat.first_name == 'Moshe'
+
+
+@pytest.mark.asyncio
+async def test_bot_model():
+ bot = Bot("fake bot id", "https://google.com")
+ assert bot.base == 'https://api.telegram.org/botfake bot id/'
+ assert bot.webhook_setter_url == 'https://api.telegram.org/botfake \
+bot id/setWebhook?url=https://google.com/telegram/'
+
+ assert bot.base == bot._set_base_url("fake bot id")
+ assert bot.webhook_setter_url == bot._set_webhook_setter_url(
+ "https://google.com")
+
+ set_request = bot.set_webhook()
+ assert set_request.status_code == status.HTTP_404_NOT_FOUND
+ assert set_request.json() == {
+ 'ok': False,
+ 'error_code': 404,
+ 'description': 'Not Found'
+ }
+
+ drop_request = bot.drop_webhook()
+ assert drop_request.status_code == status.HTTP_404_NOT_FOUND
+ assert drop_request.json() == {
+ 'ok': False,
+ 'error_code': 404,
+ 'description': 'Not Found'
+ }
+
+ send_request = await bot.send_message("654654645", "hello")
+ assert send_request.status_code == status.HTTP_404_NOT_FOUND
+ assert send_request.json() == {
+ 'ok': False,
+ 'error_code': 404,
+ 'description': 'Not Found'
+ }
+
+
+class TestHandlers:
+ TEST_USER = get_test_placeholder_user()
+
+ @pytest.mark.asyncio
+ async def test_start_handlers(self):
+ chat = Chat(gen_message('/start'))
+ message = MessageHandler(chat, self.TEST_USER)
+
+ assert '/start' in message.handlers
+ assert await message.process_callback() == '''Hello, Moshe!
+Welcome to Pylander telegram client!'''
+
+ @pytest.mark.asyncio
+ async def test_default_handlers(self):
+ wrong_start = MessageHandler(
+ Chat(gen_message('start')), self.TEST_USER)
+ wrong_show_events = MessageHandler(
+ Chat(gen_message('show_events')), self.TEST_USER)
+ message = MessageHandler(
+ Chat(gen_message('hello')), self.TEST_USER)
+
+ assert await wrong_start.process_callback() == "Unknown command."
+ assert await wrong_show_events.process_callback() == "Unknown command."
+ assert await message.process_callback() == "Unknown command."
+
+ @pytest.mark.asyncio
+ async def test_show_events_handler(self):
+ chat = Chat(gen_message('/show_events'))
+ message = MessageHandler(chat, self.TEST_USER)
+ assert await message.process_callback() == 'Choose events day.'
+
+ @pytest.mark.asyncio
+ async def test_no_today_events_handler(self):
+ chat = Chat(gen_callback('Today'))
+ message = MessageHandler(chat, self.TEST_USER)
+ assert await message.process_callback() == "There're no events today."
+
+ @pytest.mark.asyncio
+ async def test_today_handler(self, fake_user_events):
+ chat = Chat(gen_callback('Today'))
+ message = MessageHandler(chat, fake_user_events)
+ assert await message.process_callback() == f'''\
+{today_date.strftime('%B %d')}, {today_date.strftime('%A')} Events:
+
+From {today_date.strftime('%d/%m %H:%M')} \
+to {(today_date + timedelta(days=2)).strftime('%d/%m %H:%M')}: \
+Cool today event.\n'''
+
+ @pytest.mark.asyncio
+ async def test_this_week_handler(self):
+ chat = Chat(gen_callback('This week'))
+ message = MessageHandler(chat, self.TEST_USER)
+ assert await message.process_callback() == 'Choose a day.'
+
+ @pytest.mark.asyncio
+ async def test_no_chosen_day_handler(self):
+ chat = Chat(gen_callback('10 Feb 2021'))
+ message = MessageHandler(chat, self.TEST_USER)
+ message.handlers['10 Feb 2021'] = message.chosen_day_handler
+ assert await message.process_callback() == \
+ "There're no events on February 10."
+
+ @pytest.mark.asyncio
+ async def test_chosen_day_handler(self, fake_user_events):
+ chosen_date = today_date + timedelta(days=2)
+ button = str(chosen_date.strftime(DATE_FORMAT))
+ chat = Chat(gen_callback(button))
+ message = MessageHandler(chat, fake_user_events)
+ message.handlers[button] = message.chosen_day_handler
+ assert await message.chosen_day_handler() == f'''\
+{chosen_date.strftime('%B %d')}, {chosen_date.strftime('%A')} Events:
+
+From {today_date.strftime('%d/%m %H:%M')} \
+to {(today_date + timedelta(days=2)).strftime('%d/%m %H:%M')}: \
+Cool today event.
+
+From {(chosen_date + timedelta(days=-1)).strftime('%d/%m %H:%M')} \
+to {(chosen_date + timedelta(days=1)).strftime('%d/%m %H:%M')}: \
+Cool (somewhen in two days) event.\n'''
+
+
+@pytest.mark.asyncio
+async def test_reply_unknown_user():
+ chat = Chat(gen_message('/show_events'))
+ answer = await reply_unknown_user(chat)
+ assert answer == '''
+Hello, Moshe!
+
+To use PyLander Bot you have to register
+your Telegram Id in your profile page.
+
+Your Id is 666666
+Keep it secret!
+
+https://calendar.pythonic.guru/profile/
+'''
+
+
+class TestBotClient:
+
+ @staticmethod
+ @pytest.mark.asyncio
+ async def test_user_not_registered(telegram_client):
+ response = await telegram_client.post(
+ '/telegram/', json=gen_message('/start'))
+ assert response.status_code == status.HTTP_200_OK
+ assert b'Hello, Moshe!' in response.content
+ assert b'To use PyLander Bot you have to register' \
+ in response.content
+
+ @staticmethod
+ @pytest.mark.asyncio
+ async def test_user_registered(telegram_client, session):
+ session.add(get_test_placeholder_user())
+ session.commit()
+ response = await telegram_client.post(
+ '/telegram/', json=gen_message('/start'))
+ assert response.status_code == status.HTTP_200_OK
+ assert b'Welcome to Pylander telegram client!' in response.content
+
+ @staticmethod
+ @pytest.mark.asyncio
+ async def test_telegram_router(telegram_client):
+ response = await telegram_client.get('/telegram')
+ assert response.status_code == status.HTTP_200_OK
+ assert b"Start using PyLander telegram bot!" in response.content
diff --git a/tests/test_profile.py b/tests/test_profile.py
index 4fbb33b9..1b3a8d79 100644
--- a/tests/test_profile.py
+++ b/tests/test_profile.py
@@ -1,5 +1,6 @@
import os
+from fastapi import status
from PIL import Image
import pytest
@@ -31,7 +32,7 @@ def test_get_image_crop_area(width, height, result):
def test_profile_page(profile_test_client):
profile = profile_test_client.get('/profile')
data = profile.content
- assert profile.status_code == 200
+ assert profile.ok
assert b'profile.png' in data
assert b'FakeName' in data
assert b'Happy new user!' in data
@@ -47,7 +48,7 @@ def test_update_user_fullname(profile_test_client):
# Post new data
profile = profile_test_client.post(
'/profile/update_user_fullname', data=new_name_data)
- assert profile.status_code == 302
+ assert profile.status_code == status.HTTP_302_FOUND
# Get updated data
data = profile_test_client.get('/profile').content
@@ -64,7 +65,7 @@ def test_update_user_email(profile_test_client):
# Post new data
profile = profile_test_client.post(
'/profile/update_user_email', data=new_email)
- assert profile.status_code == 302
+ assert profile.status_code == status.HTTP_302_FOUND
# Get updated data
data = profile_test_client.get('/profile').content
@@ -81,13 +82,30 @@ def test_update_user_description(profile_test_client):
# Post new data
profile = profile_test_client.post(
'/profile/update_user_description', data=new_description)
- assert profile.status_code == 302
+ assert profile.status_code == status.HTTP_302_FOUND
# Get updated data
data = profile_test_client.get('/profile').content
assert b"FastAPI Developer" in data
+def test_update_telegram_id(profile_test_client):
+ new_telegram_id = {
+ 'telegram_id': "12345"
+ }
+ # Get profile page and initialize database
+ profile = profile_test_client.get('/profile')
+
+ # Post new data
+ profile = profile_test_client.post(
+ '/profile/update_telegram_id', data=new_telegram_id)
+ assert profile.status_code == status.HTTP_302_FOUND
+
+ # Get updated data
+ data = profile_test_client.get('/profile').content
+ assert b"12345" in data
+
+
def test_upload_user_photo(profile_test_client):
example_new_photo = f"{MEDIA_PATH}/example.png"
@@ -99,7 +117,7 @@ def test_upload_user_photo(profile_test_client):
'/profile/upload_user_photo',
files={'file': (
"filename", open(example_new_photo, "rb"), "image/png")})
- assert profile.status_code == 302
+ assert profile.status_code == status.HTTP_302_FOUND
# Validate new picture saved in media directory
assert 'fake_user.png' in os.listdir(MEDIA_PATH)
@@ -107,3 +125,4 @@ def test_upload_user_photo(profile_test_client):
# Validate new picture size
new_avatar_path = os.path.join(MEDIA_PATH, 'fake_user.png')
assert Image.open(new_avatar_path).size == config.AVATAR_SIZE
+ os.remove(new_avatar_path)