ticketfrei/ticketfrei3
8
1
Fork 0
Code Issues 11 Pull requests 2 Projects 1 Releases Wiki Activity

[misc] Use double-quotes (black default)

Browse source
This commit is contained in:
Thomas Lindner 2023-03-18 18:47:02 +01:00
parent fd09b381a6
commit 13c20ca245
35 changed files with 464 additions and 464 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/setup.cfg
View file

@ -55,7 +55,7 @@ deps =
mypy
types-requests
commands =
black -S --check --diff src tests
black --check --diff src tests
flake8 src tests
# not yet
#mypy --ignore-missing-imports src tests

20
backend/src/kibicara/config.py
View file

@ -12,16 +12,16 @@ from nacl.utils import random
The default configuration gets overwritten by a configuration file if one exists.
"""
config = {
'database_connection': 'sqlite:////tmp/kibicara.sqlite',
'frontend_url': 'http://localhost:4200', # url of frontend, change in prod
'secret': random(SecretBox.KEY_SIZE).hex(), # generate with: openssl rand -hex 32
"database_connection": "sqlite:////tmp/kibicara.sqlite",
"frontend_url": "http://localhost:4200", # url of frontend, change in prod
"secret": random(SecretBox.KEY_SIZE).hex(), # generate with: openssl rand -hex 32
# production params
'frontend_path': None, # required, path to frontend html/css/js files
'production': True,
'behind_proxy': False,
'keyfile': None, # optional for ssl
'certfile': None, # optional for ssl
"frontend_path": None, # required, path to frontend html/css/js files
"production": True,
"behind_proxy": False,
"keyfile": None, # optional for ssl
"certfile": None, # optional for ssl
# dev params
'root_url': 'http://localhost:8000', # url of backend
'cors_allow_origin': 'http://localhost:4200',
"root_url": "http://localhost:8000", # url of backend
"cors_allow_origin": "http://localhost:4200",
}

10
backend/src/kibicara/email.py
View file

@ -15,7 +15,7 @@ from socket import getfqdn
logger = getLogger(__name__)
def send_email(to, subject, sender='kibicara', body=''):
def send_email(to, subject, sender="kibicara", body=""):
"""E-Mail sender.
Sends an E-Mail to a specified recipient with a body
@ -33,10 +33,10 @@ def send_email(to, subject, sender='kibicara', body=''):
body (str): The body of the e-mail
"""
msg = MIMEMultipart()
msg['From'] = 'Kibicara <{0}@{1}>'.format(sender, getfqdn())
msg['To'] = to
msg['Subject'] = '[Kibicara] {0}'.format(subject)
msg["From"] = "Kibicara <{0}@{1}>".format(sender, getfqdn())
msg["To"] = to
msg["Subject"] = "[Kibicara] {0}".format(subject)
msg.attach(MIMEText(body))
with SMTP('localhost') as smtp:
with SMTP("localhost") as smtp:
smtp.send_message(msg)

52
backend/src/kibicara/kibicara.py
View file

@ -35,17 +35,17 @@ class Main:
def __init__(self):
parser = ArgumentParser()
parser.add_argument(
'-f',
'--config',
dest='configfile',
default='/etc/kibicara.conf',
help='path to config file',
"-f",
"--config",
dest="configfile",
default="/etc/kibicara.conf",
help="path to config file",
)
parser.add_argument(
'-v',
'--verbose',
action='count',
help='Raise verbosity level',
"-v",
"--verbose",
action="count",
help="Raise verbosity level",
)
args = parser.parse_args()
@ -63,9 +63,9 @@ class Main:
}
basicConfig(
level=LOGLEVELS.get(args.verbose, DEBUG),
format='%(asctime)s %(name)s %(message)s',
format="%(asctime)s %(name)s %(message)s",
)
getLogger('aiosqlite').setLevel(WARNING)
getLogger("aiosqlite").setLevel(WARNING)
Mapping.create_all()
asyncio_run(self.__run())
@ -78,31 +78,31 @@ class Main:
async def get_response(self, path, scope):
response = await super().get_response(path, scope)
if response.status_code == 404:
response = await super().get_response('.', scope)
response = await super().get_response(".", scope)
return response
app = FastAPI()
server_config = Config()
server_config.accesslog = '-'
server_config.behind_proxy = config['behind_proxy']
server_config.keyfile = config['keyfile']
server_config.certfile = config['certfile']
if config['production']:
server_config.bind = ['0.0.0.0:8000', '[::]:8000']
server_config.accesslog = "-"
server_config.behind_proxy = config["behind_proxy"]
server_config.keyfile = config["keyfile"]
server_config.certfile = config["certfile"]
if config["production"]:
server_config.bind = ["0.0.0.0:8000", "[::]:8000"]
api = FastAPI()
api.include_router(router)
app.mount('/api', api)
if not config['production'] and config['cors_allow_origin']:
app.mount("/api", api)
if not config["production"] and config["cors_allow_origin"]:
app.add_middleware(
CORSMiddleware,
allow_origins=config['cors_allow_origin'],
allow_origins=config["cors_allow_origin"],
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'],
allow_methods=["*"],
allow_headers=["*"],
)
if config['frontend_path'] is not None:
if config["frontend_path"] is not None:
app.mount(
'/',
app=SinglePageApplication(directory=config['frontend_path'], html=True),
"/",
app=SinglePageApplication(directory=config["frontend_path"], html=True),
)
await serve(app, server_config)

12
backend/src/kibicara/model.py
View file

@ -14,7 +14,7 @@ from kibicara.config import config
class Mapping:
database = Database(config['database_connection'])
database = Database(config["database_connection"])
metadata = MetaData()
@classmethod
@ -34,7 +34,7 @@ class Admin(Model):
passhash: Text()
class Mapping(Mapping):
table_name = 'admins'
table_name = "admins"
class Hood(Model):
@ -44,7 +44,7 @@ class Hood(Model):
email_enabled: Boolean() = True
class Mapping(Mapping):
table_name = 'hoods'
table_name = "hoods"
class AdminHoodRelation(Model):
@ -53,7 +53,7 @@ class AdminHoodRelation(Model):
hood: ForeignKey(Hood)
class Mapping(Mapping):
table_name = 'admin_hood_relations'
table_name = "admin_hood_relations"
class Trigger(Model):
@ -62,7 +62,7 @@ class Trigger(Model):
pattern: Text()
class Mapping(Mapping):
table_name = 'triggers'
table_name = "triggers"
class BadWord(Model):
@ -71,4 +71,4 @@ class BadWord(Model):
pattern: Text()
class Mapping(Mapping):
table_name = 'badwords'
table_name = "badwords"

8
backend/src/kibicara/platformapi.py
View file

@ -94,7 +94,7 @@ class Censor:
async def __run(self):
await self.hood.load()
self.__task.set_name('{0} {1}'.format(self.__class__.__name__, self.hood.name))
self.__task.set_name("{0} {1}".format(self.__class__.__name__, self.hood.name))
try:
self.status = BotStatus.RUNNING
await self.run()
@ -143,17 +143,17 @@ class Censor:
for badword in await BadWord.objects.filter(hood=self.hood).all():
if search(badword.pattern, message.text, IGNORECASE):
logger.debug(
'Matched bad word - dropped message: {0}'.format(message.text)
"Matched bad word - dropped message: {0}".format(message.text)
)
return False
for trigger in await Trigger.objects.filter(hood=self.hood).all():
if search(trigger.pattern, message.text, IGNORECASE):
logger.debug(
'Matched trigger - passed message: {0}'.format(message.text)
"Matched trigger - passed message: {0}".format(message.text)
)
return True
logger.debug(
'Did not match any trigger - dropped message: {0}'.format(message.text)
"Did not match any trigger - dropped message: {0}".format(message.text)
)
return False

16
backend/src/kibicara/platforms/email/bot.py
View file

@ -36,7 +36,7 @@ class EmailBot(Censor):
while True:
message = await self.receive()
logger.debug(
'Received message from censor ({0}): {1}'.format(
"Received message from censor ({0}): {1}".format(
self.hood.name, message.text
)
)
@ -45,19 +45,19 @@ class EmailBot(Censor):
).all():
token = to_token(email=subscriber.email, hood=self.hood.id)
body = (
'{0}\n\n--\n'
+ 'If you want to stop receiving these mails,'
+ 'follow this link: {1}/hoods/{2}/email-unsubscribe?token={3}'
).format(message.text, config['frontend_url'], self.hood.id, token)
"{0}\n\n--\n"
+ "If you want to stop receiving these mails,"
+ "follow this link: {1}/hoods/{2}/email-unsubscribe?token={3}"
).format(message.text, config["frontend_url"], self.hood.id, token)
try:
logger.debug('Trying to send: \n{0}'.format(body))
logger.debug("Trying to send: \n{0}".format(body))
email.send_email(
subscriber.email,
'Kibicara {0}'.format(self.hood.name),
"Kibicara {0}".format(self.hood.name),
body=body,
)
except (ConnectionRefusedError, SMTPException):
logger.exception('Sending email to subscriber failed.')
logger.exception("Sending email to subscriber failed.")
spawner = Spawner(Hood, EmailBot)

50
backend/src/kibicara/platforms/email/mda.py
View file

@ -29,14 +29,14 @@ class Main:
def __init__(self):
parser = ArgumentParser()
parser.add_argument(
'-f',
'--config',
dest='configfile',
default='/etc/kibicara.conf',
help='path to config file',
"-f",
"--config",
dest="configfile",
default="/etc/kibicara.conf",
help="path to config file",
)
# the MDA passes the recipient address as command line argument
parser.add_argument('recipient')
parser.add_argument("recipient")
args = parser.parse_args()
try:
@ -55,54 +55,54 @@ class Main:
try:
email = await Email.objects.get(name=email_name)
except NoMatch:
logger.error('No recipient with this name')
logger.error("No recipient with this name")
exit(1)
# read mail from STDIN and parse to EmailMessage object
message = BytesParser(policy=default).parsebytes(stdin.buffer.read())
sender = ''
if message.get('sender'):
sender = message.get('sender')
elif message.get('from'):
sender = message.get('from')
sender = ""
if message.get("sender"):
sender = message.get("sender")
elif message.get("from"):
sender = message.get("from")
else:
logger.error('No Sender of From header')
logger.error("No Sender of From header")
exit(1)
sender = parseaddr(sender)[1]
if not sender:
logger.error('Could not parse sender')
logger.error("Could not parse sender")
exit(1)
maybe_subscriber = await EmailSubscribers.objects.filter(email=sender).all()
if len(maybe_subscriber) != 1 or maybe_subscriber[0].hood.id != email.hood.id:
logger.error('Not a subscriber')
logger.error("Not a subscriber")
exit(1)
# extract relevant data from mail
text = sub(
r'<[^>]*>',
'',
message.get_body(preferencelist=('plain', 'html')).get_content(),
r"<[^>]*>",
"",
message.get_body(preferencelist=("plain", "html")).get_content(),
)
response = post(
'{0}/api/hoods/{1}/email/messages/'.format(
config['root_url'], email.hood.pk
"{0}/api/hoods/{1}/email/messages/".format(
config["root_url"], email.hood.pk
),
json={'text': text, 'secret': email.secret},
json={"text": text, "secret": email.secret},
)
if response.status_code == status.HTTP_201_CREATED:
exit(0)
elif response.status_code == status.HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS:
logger.error('Message was\'t accepted: {0}'.format(text))
logger.error("Message was't accepted: {0}".format(text))
elif response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY:
logger.error('Malformed request: {0}'.format(response.json()))
logger.error("Malformed request: {0}".format(response.json()))
elif response.status_code == status.HTTP_401_UNAUTHORIZED:
logger.error('Wrong API secret. kibicara_mda seems to be misconfigured')
logger.error("Wrong API secret. kibicara_mda seems to be misconfigured")
else:
logger.error(
'REST-API failed with response status {0}'.format(response.status_code)
"REST-API failed with response status {0}".format(response.status_code)
)
exit(1)

4
backend/src/kibicara/platforms/email/model.py
View file

@ -19,7 +19,7 @@ class Email(Model):
secret: Text()
class Mapping(Mapping):
table_name = 'email'
table_name = "email"
class EmailSubscribers(Model):
@ -30,4 +30,4 @@ class EmailSubscribers(Model):
email: Text(unique=True)
class Mapping(Mapping):
table_name = 'email_subscribers'
table_name = "email_subscribers"

94
backend/src/kibicara/platforms/email/webapi.py
View file

@ -29,10 +29,10 @@ logger = getLogger(__name__)
class BodyEmail(BaseModel):
name: str
@validator('name')
@validator("name")
def valid_prefix(cls, value):
if not value.startswith('kibicara-'):
raise ValueError('Recipient address didn\'t start with kibicara-')
if not value.startswith("kibicara-"):
raise ValueError("Recipient address didn't start with kibicara-")
return value
@ -79,9 +79,9 @@ router = APIRouter()
@router.get(
'/public',
"/public",
# TODO response_model
operation_id='get_emails_public',
operation_id="get_emails_public",
)
async def email_read_all_public(hood=Depends(get_hood_unauthorized)):
if hood.email_enabled:
@ -91,19 +91,19 @@ async def email_read_all_public(hood=Depends(get_hood_unauthorized)):
@router.get(
'/',
"/",
# TODO response_model
operation_id='get_emails',
operation_id="get_emails",
)
async def email_read_all(hood=Depends(get_hood)):
return await Email.objects.filter(hood=hood).select_related('hood').all()
return await Email.objects.filter(hood=hood).select_related("hood").all()
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model
operation_id='create_email',
operation_id="create_email",
)
async def email_create(values: BodyEmail, response: Response, hood=Depends(get_hood)):
"""Create an Email bot. Call this when creating a hood.
@ -115,27 +115,27 @@ async def email_create(values: BodyEmail, response: Response, hood=Depends(get_h
email = await Email.objects.create(
hood=hood, secret=urandom(32).hex(), **values.__dict__
)
response.headers['Location'] = str(hood.id)
response.headers["Location"] = str(hood.id)
return email
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.get(
'/status',
"/status",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='status_email',
operation_id="status_email",
)
async def email_status(hood=Depends(get_hood)):
return {'status': spawner.get(hood).status.name}
return {"status": spawner.get(hood).status.name}
@router.post(
'/start',
"/start",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='start_email',
operation_id="start_email",
)
async def email_start(hood=Depends(get_hood)):
await hood.update(email_enabled=True)
@ -144,10 +144,10 @@ async def email_start(hood=Depends(get_hood)):
@router.post(
'/stop',
"/stop",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='stop_email',
operation_id="stop_email",
)
async def email_stop(hood=Depends(get_hood)):
await hood.update(email_enabled=False)
@ -156,16 +156,16 @@ async def email_stop(hood=Depends(get_hood)):
@router.get(
'/{email_id}',
"/{email_id}",
# TODO response_model
operation_id='get_email',
operation_id="get_email",
)
async def email_read(email=Depends(get_email)):
return email
@router.delete(
'/{email_id}', status_code=status.HTTP_204_NO_CONTENT, operation_id='delete_email'
"/{email_id}", status_code=status.HTTP_204_NO_CONTENT, operation_id="delete_email"
)
async def email_delete(email=Depends(get_email)):
"""Delete an Email bot.
@ -179,9 +179,9 @@ async def email_delete(email=Depends(get_email)):
@router.post(
'/subscribe/',
"/subscribe/",
status_code=status.HTTP_202_ACCEPTED,
operation_id='subscribe',
operation_id="subscribe",
response_model=BaseModel,
)
async def email_subscribe(
@ -194,8 +194,8 @@ async def email_subscribe(
:return: Returns status code 200 after sending confirmation email.
"""
token = to_token(hood=hood.id, email=subscriber.email)
confirm_link = '{0}/hoods/{1}/email-confirm?token={2}'.format(
config['frontend_url'],
confirm_link = "{0}/hoods/{1}/email-confirm?token={2}".format(
config["frontend_url"],
hood.id,
token,
)
@ -205,26 +205,26 @@ async def email_subscribe(
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
email.send_email(
subscriber.email,
'Subscribe to Kibicara {0}'.format(hood.name),
body='To confirm your subscription, follow this link: {0}'.format(
"Subscribe to Kibicara {0}".format(hood.name),
body="To confirm your subscription, follow this link: {0}".format(
confirm_link
),
)
return {}
except ConnectionRefusedError:
logger.info(token)
logger.error('Sending subscription confirmation email failed.', exc_info=True)
logger.error("Sending subscription confirmation email failed.", exc_info=True)
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY)
except SMTPException:
logger.info(token)
logger.error('Sending subscription confirmation email failed.', exc_info=True)
logger.error("Sending subscription confirmation email failed.", exc_info=True)
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY)
@router.post(
'/subscribe/confirm/{token}',
"/subscribe/confirm/{token}",
status_code=status.HTTP_201_CREATED,
operation_id='confirm_subscriber',
operation_id="confirm_subscriber",
response_model=BaseModel,
)
async def email_subscribe_confirm(token, hood=Depends(get_hood_unauthorized)):
@ -236,19 +236,19 @@ async def email_subscribe_confirm(token, hood=Depends(get_hood_unauthorized)):
"""
payload = from_token(token)
# If token.hood and url.hood are different, raise an error:
if hood.id is not payload['hood']:
if hood.id is not payload["hood"]:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
try:
await EmailSubscribers.objects.create(hood=hood.id, email=payload['email'])
await EmailSubscribers.objects.create(hood=hood.id, email=payload["email"])
return {}
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.delete(
'/unsubscribe/{token}',
"/unsubscribe/{token}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='unsubscribe',
operation_id="unsubscribe",
)
async def email_unsubscribe(token, hood=Depends(get_hood_unauthorized)):
"""Remove a subscriber from the database when they click on an unsubscribe link.
@ -257,13 +257,13 @@ async def email_unsubscribe(token, hood=Depends(get_hood_unauthorized)):
:param hood: Hood the Email bot belongs to.
"""
try:
logger.warning('token is: {0}'.format(token))
logger.warning("token is: {0}".format(token))
payload = from_token(token)
# If token.hood and url.hood are different, raise an error:
if hood.id is not payload['hood']:
if hood.id is not payload["hood"]:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
subscriber = await EmailSubscribers.objects.filter(
hood=payload['hood'], email=payload['email']
hood=payload["hood"], email=payload["email"]
).get()
await subscriber.delete()
return Response(status_code=status.HTTP_204_NO_CONTENT)
@ -274,28 +274,28 @@ async def email_unsubscribe(token, hood=Depends(get_hood_unauthorized)):
@router.get(
'/subscribers/',
"/subscribers/",
# TODO response_model
operation_id='get_subscribers',
operation_id="get_subscribers",
)
async def subscribers_read_all(hood=Depends(get_hood)):
return await EmailSubscribers.objects.filter(hood=hood).all()
@router.get(
'/subscribers/{subscriber_id}',
"/subscribers/{subscriber_id}",
# TODO response_model
operation_id='get_subscriber',
operation_id="get_subscriber",
)
async def subscribers_read(subscriber=Depends(get_subscriber)):
return subscriber
@router.post(
'/messages/',
"/messages/",
status_code=status.HTTP_201_CREATED,
# TODO response_model
operation_id='send_message',
operation_id="send_message",
)
async def email_message_create(
message: BodyMessage, hood=Depends(get_hood_unauthorized)
@ -310,14 +310,14 @@ async def email_message_create(
if message.secret == receiver.secret:
# pass message.text to bot.py
if await spawner.get(hood).publish(Message(message.text)):
logger.warning('Message was accepted: {0}'.format(message.text))
logger.warning("Message was accepted: {0}".format(message.text))
return {}
else:
logger.warning('Message wasn\'t accepted: {0}'.format(message.text))
logger.warning("Message wasn't accepted: {0}".format(message.text))
raise HTTPException(
status_code=status.HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS
)
logger.warning(
'Someone is trying to submit an email without the correct API secret'
"Someone is trying to submit an email without the correct API secret"
)
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

42
backend/src/kibicara/platforms/telegram/bot.py
View file

@ -32,9 +32,9 @@ class TelegramBot(Censor):
def _create_dispatcher(self):
dp = Dispatcher(self.bot)
dp.register_message_handler(self._send_welcome, commands=['start'])
dp.register_message_handler(self._remove_user, commands=['stop'])
dp.register_message_handler(self._send_help, commands=['help'])
dp.register_message_handler(self._send_welcome, commands=["start"])
dp.register_message_handler(self._remove_user, commands=["stop"])
dp.register_message_handler(self._send_help, commands=["help"])
dp.register_message_handler(self._receive_message)
return dp
@ -42,30 +42,30 @@ class TelegramBot(Censor):
try:
self.bot = Bot(token=self.telegram_model.api_token)
self.dp = self._create_dispatcher()
logger.debug('Bot {0} starting.'.format(self.telegram_model.hood.name))
logger.debug("Bot {0} starting.".format(self.telegram_model.hood.name))
user = await self.bot.get_me()
if user.username:
await self.telegram_model.update(username=user.username)
await gather(self.dp.start_polling(), self._push())
except CancelledError:
logger.debug(
'Bot {0} received Cancellation.'.format(self.telegram_model.hood.name)
"Bot {0} received Cancellation.".format(self.telegram_model.hood.name)
)
self.dp = None
raise
except exceptions.ValidationError:
logger.debug(
'Bot {0} has invalid auth token.'.format(self.telegram_model.hood.name)
"Bot {0} has invalid auth token.".format(self.telegram_model.hood.name)
)
await self.telegram_model.update(enabled=False)
finally:
logger.debug('Bot {0} stopped.'.format(self.telegram_model.hood.name))
logger.debug("Bot {0} stopped.".format(self.telegram_model.hood.name))
async def _push(self):
while True:
message = await self.receive()
logger.debug(
'Received message from censor ({0}): {1}'.format(
"Received message from censor ({0}): {1}".format(
self.telegram_model.hood.name, message.text
)
)
@ -79,34 +79,34 @@ class TelegramBot(Censor):
await self.bot.send_message(user_id, message, disable_notification=False)
except exceptions.BotBlocked:
logger.error(
'Target [ID:{0}] ({1}): blocked by user'.format(
"Target [ID:{0}] ({1}): blocked by user".format(
user_id, self.telegram_model.hood.name
)
)
except exceptions.ChatNotFound:
logger.error(
'Target [ID:{0}] ({1}): invalid user ID'.format(
"Target [ID:{0}] ({1}): invalid user ID".format(
user_id, self.telegram_model.hood.name
)
)
except exceptions.RetryAfter as e:
logger.error(
'Target [ID:{0}] ({1}): Flood limit is exceeded.'.format(
"Target [ID:{0}] ({1}): Flood limit is exceeded.".format(
user_id, self.telegram_model.hood.name
)
+ 'Sleep {0} seconds.'.format(e.timeout)
+ "Sleep {0} seconds.".format(e.timeout)
)
await sleep(e.timeout)
return await self._send_message(user_id, message)
except exceptions.UserDeactivated:
logger.error(
'Target [ID:{0}] ({1}): user is deactivated'.format(
"Target [ID:{0}] ({1}): user is deactivated".format(
user_id, self.telegram_model.hood.name
)
)
except exceptions.TelegramAPIError:
logger.exception(
'Target [ID:{0}] ({1}): failed'.format(
"Target [ID:{0}] ({1}): failed".format(
user_id, self.telegram_model.hood.name
)
)
@ -114,14 +114,14 @@ class TelegramBot(Censor):
async def _send_welcome(self, message: types.Message):
try:
if message.from_user.is_bot:
await message.reply('Error: Bots can not join here.')
await message.reply("Error: Bots can not join here.")
return
await TelegramUser.objects.create(
user_id=message.from_user.id, bot=self.telegram_model
)
await message.reply(self.telegram_model.welcome_message)
except IntegrityError:
await message.reply('Error: You are already registered.')
await message.reply("Error: You are already registered.")
async def _remove_user(self, message: types.Message):
try:
@ -129,19 +129,19 @@ class TelegramBot(Censor):
user_id=message.from_user.id, bot=self.telegram_model
)
await telegram_user.delete()
await message.reply('You were removed successfully from this bot.')
await message.reply("You were removed successfully from this bot.")
except NoMatch:
await message.reply('Error: You are not subscribed to this bot.')
await message.reply("Error: You are not subscribed to this bot.")
async def _send_help(self, message: types.Message):
if message.from_user.is_bot:
await message.reply('Error: Bots can\'t be helped.')
await message.reply("Error: Bots can't be helped.")
return
await message.reply('Send messages here to broadcast them to your hood')
await message.reply("Send messages here to broadcast them to your hood")
async def _receive_message(self, message: types.Message):
if not message.text:
await message.reply('Error: Only text messages are allowed.')
await message.reply("Error: Only text messages are allowed.")
return
await self.publish(Message(message.text))

4
backend/src/kibicara/platforms/telegram/model.py
View file

@ -17,7 +17,7 @@ class Telegram(Model):
enabled: Boolean() = True
class Mapping(Mapping):
table_name = 'telegrambots'
table_name = "telegrambots"
class TelegramUser(Model):
@ -27,4 +27,4 @@ class TelegramUser(Model):
bot: ForeignKey(Telegram)
class Mapping(Mapping):
table_name = 'telegramusers'
table_name = "telegramusers"

44
backend/src/kibicara/platforms/telegram/webapi.py
View file

@ -21,9 +21,9 @@ logger = getLogger(__name__)
class BodyTelegram(BaseModel):
api_token: str
welcome_message: str = 'Welcome!'
welcome_message: str = "Welcome!"
@validator('api_token')
@validator("api_token")
def valid_api_token(cls, value):
try:
check_token(value)
@ -48,9 +48,9 @@ telegram_callback_router = APIRouter()
@router.get(
'/public',
"/public",
# TODO response_model,
operation_id='get_telegrams_public',
operation_id="get_telegrams_public",
)
async def telegram_read_all_public(hood=Depends(get_hood_unauthorized)):
telegrambots = await Telegram.objects.filter(hood=hood).all()
@ -62,27 +62,27 @@ async def telegram_read_all_public(hood=Depends(get_hood_unauthorized)):
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_telegrams',
operation_id="get_telegrams",
)
async def telegram_read_all(hood=Depends(get_hood)):
return await Telegram.objects.filter(hood=hood).all()
@router.get(
'/{telegram_id}',
"/{telegram_id}",
# TODO response_model,
operation_id='get_telegram',
operation_id="get_telegram",
)
async def telegram_read(telegram=Depends(get_telegram)):
return telegram
@router.delete(
'/{telegram_id}',
"/{telegram_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='delete_telegram',
operation_id="delete_telegram",
)
async def telegram_delete(telegram=Depends(get_telegram)):
spawner.stop(telegram)
@ -93,10 +93,10 @@ async def telegram_delete(telegram=Depends(get_telegram)):
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model,
operation_id='create_telegram',
operation_id="create_telegram",
)
async def telegram_create(
response: Response, values: BodyTelegram, hood=Depends(get_hood)
@ -104,17 +104,17 @@ async def telegram_create(
try:
telegram = await Telegram.objects.create(hood=hood, **values.__dict__)
spawner.start(telegram)
response.headers['Location'] = str(telegram.id)
response.headers["Location"] = str(telegram.id)
return telegram
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.put(
'/{telegram_id}',
"/{telegram_id}",
status_code=status.HTTP_202_ACCEPTED,
# TODO response_model,
operation_id='update_telegram',
operation_id="update_telegram",
)
async def telegram_update(values: BodyTelegram, telegram=Depends(get_telegram)):
try:
@ -127,20 +127,20 @@ async def telegram_update(values: BodyTelegram, telegram=Depends(get_telegram)):
@router.get(
'/{telegram_id}/status',
"/{telegram_id}/status",
status_code=status.HTTP_200_OK,
# TODO response_model,
operation_id='status_telegram',
operation_id="status_telegram",
)
async def telegram_status(telegram=Depends(get_telegram)):
return {'status': spawner.get(telegram).status.name}
return {"status": spawner.get(telegram).status.name}
@router.post(
'/{telegram_id}/start',
"/{telegram_id}/start",
status_code=status.HTTP_200_OK,
# TODO response_model,
operation_id='start_telegram',
operation_id="start_telegram",
)
async def telegram_start(telegram=Depends(get_telegram)):
await telegram.update(enabled=True)
@ -149,10 +149,10 @@ async def telegram_start(telegram=Depends(get_telegram)):
@router.post(
'/{telegram_id}/stop',
"/{telegram_id}/stop",
status_code=status.HTTP_200_OK,
# TODO response_model,
operation_id='stop_telegram',
operation_id="stop_telegram",
)
async def telegram_stop(telegram=Depends(get_telegram)):
await telegram.update(enabled=False)

2
backend/src/kibicara/platforms/test/model.py
View file

@ -13,4 +13,4 @@ class Test(Model):
hood: ForeignKey(Hood)
class Mapping(Mapping):
table_name = 'testapi'
table_name = "testapi"

14
backend/src/kibicara/platforms/test/webapi.py
View file

@ -30,39 +30,39 @@ async def get_test(test_id: int, hood=Depends(get_hood)):
router = APIRouter()
@router.get('/')
@router.get("/")
async def test_read_all(hood=Depends(get_hood)):
return await Test.objects.filter(hood=hood).all()
@router.post('/', status_code=status.HTTP_201_CREATED)
@router.post("/", status_code=status.HTTP_201_CREATED)
async def test_create(response: Response, hood=Depends(get_hood)):
try:
test = await Test.objects.create(hood=hood)
spawner.start(test)
response.headers['Location'] = str(test.id)
response.headers["Location"] = str(test.id)
return test
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.get('/{test_id}')
@router.get("/{test_id}")
async def test_read(test=Depends(get_test)):
return test
@router.delete('/{test_id}', status_code=status.HTTP_204_NO_CONTENT)
@router.delete("/{test_id}", status_code=status.HTTP_204_NO_CONTENT)
async def test_delete(test=Depends(get_test)):
spawner.stop(test)
await test.delete()
@router.get('/{test_id}/messages/')
@router.get("/{test_id}/messages/")
async def test_message_read_all(test=Depends(get_test)):
return spawner.get(test).messages
@router.post('/{test_id}/messages/')
@router.post("/{test_id}/messages/")
async def test_message_create(message: BodyMessage, test=Depends(get_test)):
await spawner.get(test).publish(Message(message.text))
return {}

30
backend/src/kibicara/platforms/twitter/bot.py
View file

@ -33,52 +33,52 @@ class TwitterBot(Censor):
async def run(self):
try:
if not self.twitter_model.verified:
raise ValueError('Oauth Handshake not completed')
raise ValueError("Oauth Handshake not completed")
self.client = PeonyClient(
consumer_key=config['twitter']['consumer_key'],
consumer_secret=config['twitter']['consumer_secret'],
consumer_key=config["twitter"]["consumer_key"],
consumer_secret=config["twitter"]["consumer_secret"],
access_token=self.twitter_model.access_token,
access_token_secret=self.twitter_model.access_token_secret,
)
if self.twitter_model.mentions_since_id is None:
logger.debug('since_id is None in model, fetch newest mention id')
logger.debug("since_id is None in model, fetch newest mention id")
await self._poll_mentions()
if self.twitter_model.dms_since_id is None:
logger.debug('since_id is None in model, fetch newest dm id')
logger.debug("since_id is None in model, fetch newest dm id")
await self._poll_direct_messages()
user = await self.client.user
if user.screen_name:
await self.twitter_model.update(username=user.screen_name)
logger.debug(
'Starting Twitter bot: {0}'.format(self.twitter_model.__dict__)
"Starting Twitter bot: {0}".format(self.twitter_model.__dict__)
)
await gather(self.poll(), self.push())
except CancelledError:
logger.debug(
'Bot {0} received Cancellation.'.format(self.twitter_model.hood.name)
"Bot {0} received Cancellation.".format(self.twitter_model.hood.name)
)
except exceptions.Unauthorized:
logger.debug(
'Bot {0} has invalid auth token.'.format(self.twitter_model.hood.name)
"Bot {0} has invalid auth token.".format(self.twitter_model.hood.name)
)
await self.twitter_model.update(enabled=False)
self.enabled = self.twitter_model.enabled
except (KeyError, ValueError, exceptions.NotAuthenticated):
logger.warning('Missing consumer_keys for Twitter in your configuration.')
logger.warning("Missing consumer_keys for Twitter in your configuration.")
await self.twitter_model.update(enabled=False)
self.enabled = self.twitter_model.enabled
finally:
logger.debug('Bot {0} stopped.'.format(self.twitter_model.hood.name))
logger.debug("Bot {0} stopped.".format(self.twitter_model.hood.name))
async def poll(self):
while True:
dms = await self._poll_direct_messages()
logger.debug(
'Polled dms ({0}): {1}'.format(self.twitter_model.hood.name, str(dms))
"Polled dms ({0}): {1}".format(self.twitter_model.hood.name, str(dms))
)
mentions = await self._poll_mentions()
logger.debug(
'Polled mentions ({0}): {1}'.format(
"Polled mentions ({0}): {1}".format(
self.twitter_model.hood.name, str(mentions)
)
)
@ -135,7 +135,7 @@ class TwitterBot(Censor):
remove_indices.update(range(url.indices[0], url.indices[1] + 1))
for symbol in entities.symbols:
remove_indices.update(range(symbol.indices[0], symbol.indices[1] + 1))
filtered_text = ''
filtered_text = ""
for index, character in enumerate(text):
if index not in remove_indices:
filtered_text += character
@ -145,11 +145,11 @@ class TwitterBot(Censor):
while True:
message = await self.receive()
logger.debug(
'Received message from censor ({0}): {1}'.format(
"Received message from censor ({0}): {1}".format(
self.twitter_model.hood.name, message.text
)
)
if hasattr(message, 'twitter_mention_id'):
if hasattr(message, "twitter_mention_id"):
await self._retweet(message.twitter_mention_id)
else:
await self._post_tweet(message.text)

2
backend/src/kibicara/platforms/twitter/model.py
View file

@ -20,4 +20,4 @@ class Twitter(Model):
enabled: Boolean() = False
class Mapping(Mapping):
table_name = 'twitterbots'
table_name = "twitterbots"

62
backend/src/kibicara/platforms/twitter/webapi.py
View file

@ -36,9 +36,9 @@ twitter_callback_router = APIRouter()
@router.get(
'/public',
"/public",
# TODO response_model,
operation_id='get_twitters_public',
operation_id="get_twitters_public",
)
async def twitter_read_all_public(hood=Depends(get_hood_unauthorized)):
twitterbots = await Twitter.objects.filter(hood=hood).all()
@ -50,28 +50,28 @@ async def twitter_read_all_public(hood=Depends(get_hood_unauthorized)):
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_twitters',
operation_id="get_twitters",
)
async def twitter_read_all(hood=Depends(get_hood)):
return await Twitter.objects.filter(hood=hood).all()
@router.get(
'/{twitter_id}',
"/{twitter_id}",
# TODO response_model
operation_id='get_twitter',
operation_id="get_twitter",
)
async def twitter_read(twitter=Depends(get_twitter)):
return twitter
@router.delete(
'/{twitter_id}',
"/{twitter_id}",
status_code=status.HTTP_204_NO_CONTENT,
# TODO response_model
operation_id='delete_twitter',
operation_id="delete_twitter",
)
async def twitter_delete(twitter=Depends(get_twitter)):
spawner.stop(twitter)
@ -80,20 +80,20 @@ async def twitter_delete(twitter=Depends(get_twitter)):
@router.get(
'/{twitter_id}/status',
"/{twitter_id}/status",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='status_twitter',
operation_id="status_twitter",
)
async def twitter_status(twitter=Depends(get_twitter)):
return {'status': spawner.get(twitter).status.name}
return {"status": spawner.get(twitter).status.name}
@router.post(
'/{twitter_id}/start',
"/{twitter_id}/start",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='start_twitter',
operation_id="start_twitter",
)
async def twitter_start(twitter=Depends(get_twitter)):
await twitter.update(enabled=True)
@ -102,10 +102,10 @@ async def twitter_start(twitter=Depends(get_twitter)):
@router.post(
'/{twitter_id}/stop',
"/{twitter_id}/stop",
status_code=status.HTTP_200_OK,
# TODO response_model
operation_id='stop_twitter',
operation_id="stop_twitter",
)
async def twitter_stop(twitter=Depends(get_twitter)):
await twitter.update(enabled=False)
@ -114,10 +114,10 @@ async def twitter_stop(twitter=Depends(get_twitter)):
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model
operation_id='create_twitter',
operation_id="create_twitter",
)
async def twitter_create(response: Response, hood=Depends(get_hood)):
"""
@ -129,20 +129,20 @@ async def twitter_create(response: Response, hood=Depends(get_hood)):
await corpse.delete()
# Create Twitter
request_token = await get_oauth_token(
config['twitter']['consumer_key'],
config['twitter']['consumer_secret'],
callback_uri='{0}/dashboard/twitter-callback?hood={1}'.format(
config['frontend_url'], hood.id
config["twitter"]["consumer_key"],
config["twitter"]["consumer_secret"],
callback_uri="{0}/dashboard/twitter-callback?hood={1}".format(
config["frontend_url"], hood.id
),
)
if request_token['oauth_callback_confirmed'] != 'true':
if request_token["oauth_callback_confirmed"] != "true":
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE)
twitter = await Twitter.objects.create(
hood=hood,
access_token=request_token['oauth_token'],
access_token_secret=request_token['oauth_token_secret'],
access_token=request_token["oauth_token"],
access_token_secret=request_token["oauth_token_secret"],
)
response.headers['Location'] = str(twitter.id)
response.headers["Location"] = str(twitter.id)
return twitter
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@ -151,9 +151,9 @@ async def twitter_create(response: Response, hood=Depends(get_hood)):
@twitter_callback_router.get(
'/callback',
"/callback",
# TODO response_model
operation_id='callback_twitter',
operation_id="callback_twitter",
)
async def twitter_read_callback(
oauth_token: str, oauth_verifier: str, hood=Depends(get_hood)
@ -161,15 +161,15 @@ async def twitter_read_callback(
try:
twitter = await Twitter.objects.filter(access_token=oauth_token).get()
access_token = await get_access_token(
config['twitter']['consumer_key'],
config['twitter']['consumer_secret'],
config["twitter"]["consumer_key"],
config["twitter"]["consumer_secret"],
twitter.access_token,
twitter.access_token_secret,
oauth_verifier,
)
await twitter.update(
access_token=access_token['oauth_token'],
access_token_secret=access_token['oauth_token_secret'],
access_token=access_token["oauth_token"],
access_token_secret=access_token["oauth_token_secret"],
verified=True,
enabled=True,
)

18
backend/src/kibicara/webapi/__init__.py
View file

@ -23,20 +23,20 @@ from kibicara.webapi.hoods.badwords import router as badwords_router
from kibicara.webapi.hoods.triggers import router as triggers_router
router = APIRouter()
router.include_router(admin_router, prefix='/admin', tags=['admin'])
router.include_router(admin_router, prefix="/admin", tags=["admin"])
hoods_router.include_router(
triggers_router, prefix='/{hood_id}/triggers', tags=['triggers']
triggers_router, prefix="/{hood_id}/triggers", tags=["triggers"]
)
hoods_router.include_router(
badwords_router, prefix='/{hood_id}/badwords', tags=['badwords']
badwords_router, prefix="/{hood_id}/badwords", tags=["badwords"]
)
hoods_router.include_router(test_router, prefix='/{hood_id}/test', tags=['test'])
hoods_router.include_router(test_router, prefix="/{hood_id}/test", tags=["test"])
hoods_router.include_router(
telegram_router, prefix='/{hood_id}/telegram', tags=['telegram']
telegram_router, prefix="/{hood_id}/telegram", tags=["telegram"]
)
hoods_router.include_router(
twitter_router, prefix='/{hood_id}/twitter', tags=['twitter']
twitter_router, prefix="/{hood_id}/twitter", tags=["twitter"]
)
router.include_router(twitter_callback_router, prefix='/twitter', tags=['twitter'])
hoods_router.include_router(email_router, prefix='/{hood_id}/email', tags=['email'])
router.include_router(hoods_router, prefix='/hoods')
router.include_router(twitter_callback_router, prefix="/twitter", tags=["twitter"])
hoods_router.include_router(email_router, prefix="/{hood_id}/email", tags=["email"])
router.include_router(hoods_router, prefix="/hoods")

84
backend/src/kibicara/webapi/admin.py
View file

@ -37,10 +37,10 @@ class BodyEmail(BaseModel):
class BodyPassword(BaseModel):
password: str
@validator('password')
@validator("password")
def valid_password(cls, value):
if len(value) < 8:
raise ValueError('Password is too short')
raise ValueError("Password is too short")
return value
@ -50,22 +50,22 @@ class BodyAdmin(BodyEmail, BodyPassword):
class BodyAccessToken(BaseModel):
access_token: str
token_type: str = 'bearer'
token_type: str = "bearer"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/admin/login')
secret_box = SecretBox(bytes.fromhex(config['secret']))
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/admin/login")
secret_box = SecretBox(bytes.fromhex(config["secret"]))
def to_token(**kwargs):
return secret_box.encrypt(dumps(kwargs), encoder=URLSafeBase64Encoder).decode(
'ascii'
"ascii"
)
def from_token(token):
return loads(
secret_box.decrypt(token.encode('ascii'), encoder=URLSafeBase64Encoder)
secret_box.decrypt(token.encode("ascii"), encoder=URLSafeBase64Encoder)
)
@ -85,8 +85,8 @@ async def get_admin(access_token=Depends(oauth2_scheme)):
except (CryptoError, ValueError):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail='Invalid authentication credentials',
headers={'WWW-Authenticate': 'Bearer'},
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return admin
@ -95,10 +95,10 @@ router = APIRouter()
@router.post(
'/register/',
"/register/",
status_code=status.HTTP_202_ACCEPTED,
response_model=BaseModel,
operation_id='register',
operation_id="register",
)
async def admin_register(values: BodyAdmin):
"""Sends an email with a confirmation link.
@ -107,28 +107,28 @@ async def admin_register(values: BodyAdmin):
- **password**: Password of new hood admin
"""
register_token = to_token(**values.__dict__)
logger.debug('register_token={0}'.format(register_token))
logger.debug("register_token={0}".format(register_token))
try:
admin = await Admin.objects.filter(email=values.email).all()
if admin:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
body = '{0}/confirm?token={1}'.format(config['frontend_url'], register_token)
body = "{0}/confirm?token={1}".format(config["frontend_url"], register_token)
logger.debug(body)
email.send_email(
to=values.email,
subject='Confirm Account',
subject="Confirm Account",
body=body,
)
except (ConnectionRefusedError, SMTPException):
logger.exception('Email sending failed')
logger.exception("Email sending failed")
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY)
return {}
@router.post(
'/confirm/{register_token}',
"/confirm/{register_token}",
response_model=BodyAccessToken,
operation_id='confirm',
operation_id="confirm",
)
async def admin_confirm(register_token: str):
"""Registration confirmation and account creation.
@ -137,17 +137,17 @@ async def admin_confirm(register_token: str):
"""
try:
values = from_token(register_token)
passhash = argon2.hash(values['password'])
await Admin.objects.create(email=values['email'], passhash=passhash)
passhash = argon2.hash(values["password"])
await Admin.objects.create(email=values["email"], passhash=passhash)
return BodyAccessToken(access_token=register_token)
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.post(
'/login/',
"/login/",
response_model=BodyAccessToken,
operation_id='login',
operation_id="login",
)
async def admin_login(form_data: OAuth2PasswordRequestForm = Depends()):
"""Get an access token.
@ -160,17 +160,17 @@ async def admin_login(form_data: OAuth2PasswordRequestForm = Depends()):
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Incorrect email or password',
detail="Incorrect email or password",
)
token = to_token(email=form_data.username, password=form_data.password)
return BodyAccessToken(access_token=token)
@router.post(
'/reset/',
"/reset/",
status_code=status.HTTP_202_ACCEPTED,
response_model=BaseModel,
operation_id='reset',
operation_id="reset",
)
async def admin_reset_password(values: BodyEmail):
"""Sends an email with a password reset link.
@ -179,41 +179,41 @@ async def admin_reset_password(values: BodyEmail):
- **password**: Password of new hood admin
"""
register_token = to_token(datetime=datetime.now().isoformat(), **values.__dict__)
logger.debug('register_token={0}'.format(register_token))
logger.debug("register_token={0}".format(register_token))
try:
admin = await Admin.objects.filter(email=values.email).all()
if not admin:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
body = '{0}/password-reset?token={1}'.format(
config['frontend_url'], register_token
body = "{0}/password-reset?token={1}".format(
config["frontend_url"], register_token
)
logger.debug(body)
email.send_email(
to=values.email,
subject='Reset your password',
subject="Reset your password",
body=body,
)
except (ConnectionRefusedError, SMTPException):
logger.exception('Email sending failed')
logger.exception("Email sending failed")
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY)
return {}
@router.post(
'/reset/{reset_token}',
"/reset/{reset_token}",
response_model=BodyAccessToken,
operation_id='confirm_reset',
operation_id="confirm_reset",
)
async def admin_confirm_reset(reset_token: str, values: BodyPassword):
try:
token_values = from_token(reset_token)
if (
datetime.fromisoformat(token_values['datetime']) + timedelta(hours=3)
datetime.fromisoformat(token_values["datetime"]) + timedelta(hours=3)
< datetime.now()
):
raise HTTPException(status_code=status.HTTP_410_GONE)
passhash = argon2.hash(values.password)
admins = await Admin.objects.filter(email=token_values['email']).all()
admins = await Admin.objects.filter(email=token_values["email"]).all()
if len(admins) != 1:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
await admins[0].update(passhash=passhash)
@ -225,21 +225,21 @@ async def admin_confirm_reset(reset_token: str, values: BodyPassword):
@router.get(
'/hoods/',
"/hoods/",
# TODO response_model,
operation_id='get_hoods_admin',
operation_id="get_hoods_admin",
)
async def admin_hood_read_all(admin=Depends(get_admin)):
"""Get a list of all hoods of a given admin."""
return (
await AdminHoodRelation.objects.select_related('hood').filter(admin=admin).all()
await AdminHoodRelation.objects.select_related("hood").filter(admin=admin).all()
)
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_admin',
operation_id="get_admin",
)
async def admin_read(admin=Depends(get_admin)):
"""Get a list of all hoods of a given admin."""
@ -250,17 +250,17 @@ async def admin_read(admin=Depends(get_admin)):
@router.delete(
'/',
"/",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='delete_admin',
operation_id="delete_admin",
)
async def admin_delete(admin=Depends(get_admin)):
hood_relations = (
await AdminHoodRelation.objects.select_related('hood').filter(admin=admin).all()
await AdminHoodRelation.objects.select_related("hood").filter(admin=admin).all()
)
for hood in hood_relations:
admins = (
await AdminHoodRelation.objects.select_related('admin')
await AdminHoodRelation.objects.select_related("admin")
.filter(hood=hood.id)
.all()
)

40
backend/src/kibicara/webapi/hoods/__init__.py
View file

@ -20,9 +20,9 @@ from kibicara.webapi.utils import delete_hood
class BodyHood(BaseModel):
name: str
landingpage: str = '''
landingpage: str = """
Default Landing Page
'''
"""
async def get_hood_unauthorized(hood_id: int):
@ -39,7 +39,7 @@ async def get_hood(hood=Depends(get_hood_unauthorized), admin=Depends(get_admin)
except NoMatch:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
headers={'WWW-Authenticate': 'Bearer'},
headers={"WWW-Authenticate": "Bearer"},
)
return hood
@ -48,10 +48,10 @@ router = APIRouter()
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_hoods',
tags=['hoods'],
operation_id="get_hoods",
tags=["hoods"],
)
async def hood_read_all():
"""Get all existing hoods."""
@ -59,11 +59,11 @@ async def hood_read_all():
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model,
operation_id='create_hood',
tags=['hoods'],
operation_id="create_hood",
tags=["hoods"],
)
async def hood_create(values: BodyHood, response: Response, admin=Depends(get_admin)):
"""Creates a hood.
@ -77,19 +77,19 @@ async def hood_create(values: BodyHood, response: Response, admin=Depends(get_ad
spawner.start(hood)
# Initialize Triggers to match all
await Trigger.objects.create(hood=hood, pattern='.')
await Trigger.objects.create(hood=hood, pattern=".")
response.headers['Location'] = str(hood.id)
response.headers["Location"] = str(hood.id)
return hood
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@router.get(
'/{hood_id}',
"/{hood_id}",
# TODO response_model,
operation_id='get_hood',
tags=['hoods'],
operation_id="get_hood",
tags=["hoods"],
)
async def hood_read(hood=Depends(get_hood_unauthorized)):
"""Get hood with id **hood_id**."""
@ -97,10 +97,10 @@ async def hood_read(hood=Depends(get_hood_unauthorized)):
@router.put(
'/{hood_id}',
"/{hood_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='update_hood',
tags=['hoods'],
operation_id="update_hood",
tags=["hoods"],
)
async def hood_update(values: BodyHood, hood=Depends(get_hood)):
"""Updates hood with id **hood_id**.
@ -113,10 +113,10 @@ async def hood_update(values: BodyHood, hood=Depends(get_hood)):
@router.delete(
'/{hood_id}',
"/{hood_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='delete_hood',
tags=['hoods'],
operation_id="delete_hood",
tags=["hoods"],
)
async def hood_delete(hood=Depends(get_hood)):
"""Deletes hood with id **hood_id**."""

22
backend/src/kibicara/webapi/hoods/badwords.py
View file

@ -38,9 +38,9 @@ router = APIRouter()
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_badwords',
operation_id="get_badwords",
)
async def badword_read_all(hood=Depends(get_hood)):
"""Get all badwords of hood with id **hood_id**."""
@ -48,10 +48,10 @@ async def badword_read_all(hood=Depends(get_hood)):
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model,
operation_id='create_badword',
operation_id="create_badword",
)
async def badword_create(
values: BodyBadWord, response: Response, hood=Depends(get_hood)
@ -63,7 +63,7 @@ async def badword_create(
try:
regex_compile(values.pattern)
badword = await BadWord.objects.create(hood=hood, **values.__dict__)
response.headers['Location'] = str(badword.id)
response.headers["Location"] = str(badword.id)
return badword
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@ -72,9 +72,9 @@ async def badword_create(
@router.get(
'/{badword_id}',
"/{badword_id}",
# TODO response_model,
operation_id='get_badword',
operation_id="get_badword",
)
async def badword_read(badword=Depends(get_badword)):
"""Reads badword with id **badword_id** for hood with id **hood_id**."""
@ -82,9 +82,9 @@ async def badword_read(badword=Depends(get_badword)):
@router.put(
'/{badword_id}',
"/{badword_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='update_badword',
operation_id="update_badword",
)
async def badword_update(values: BodyBadWord, badword=Depends(get_badword)):
"""Updates badword with id **badword_id** for hood with id **hood_id**.
@ -96,9 +96,9 @@ async def badword_update(values: BodyBadWord, badword=Depends(get_badword)):
@router.delete(
'/{badword_id}',
"/{badword_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='delete_badword',
operation_id="delete_badword",
)
async def badword_delete(badword=Depends(get_badword)):
"""Deletes badword with id **badword_id** for hood with id **hood_id**."""

22
backend/src/kibicara/webapi/hoods/triggers.py
View file

@ -39,9 +39,9 @@ router = APIRouter()
@router.get(
'/',
"/",
# TODO response_model,
operation_id='get_triggers',
operation_id="get_triggers",
)
async def trigger_read_all(hood=Depends(get_hood)):
"""Get all triggers of hood with id **hood_id**."""
@ -49,10 +49,10 @@ async def trigger_read_all(hood=Depends(get_hood)):
@router.post(
'/',
"/",
status_code=status.HTTP_201_CREATED,
# TODO response_model,
operation_id='create_trigger',
operation_id="create_trigger",
)
async def trigger_create(
values: BodyTrigger, response: Response, hood=Depends(get_hood)
@ -64,7 +64,7 @@ async def trigger_create(
try:
regex_compile(values.pattern)
trigger = await Trigger.objects.create(hood=hood, **values.__dict__)
response.headers['Location'] = str(trigger.id)
response.headers["Location"] = str(trigger.id)
return trigger
except IntegrityError:
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
@ -73,9 +73,9 @@ async def trigger_create(
@router.get(
'/{trigger_id}',
"/{trigger_id}",
# TODO response_model,
operation_id='get_trigger',
operation_id="get_trigger",
)
async def trigger_read(trigger=Depends(get_trigger)):
"""Reads trigger with id **trigger_id** for hood with id **hood_id**."""
@ -83,9 +83,9 @@ async def trigger_read(trigger=Depends(get_trigger)):
@router.put(
'/{trigger_id}',
"/{trigger_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='update_trigger',
operation_id="update_trigger",
)
async def trigger_update(values: BodyTrigger, trigger=Depends(get_trigger)):
"""Updates trigger with id **trigger_id** for hood with id **hood_id**.
@ -97,9 +97,9 @@ async def trigger_update(values: BodyTrigger, trigger=Depends(get_trigger)):
@router.delete(
'/{trigger_id}',
"/{trigger_id}",
status_code=status.HTTP_204_NO_CONTENT,
operation_id='delete_trigger',
operation_id="delete_trigger",
)
async def trigger_delete(trigger=Depends(get_trigger)):
"""Deletes trigger with id **trigger_id** for hood with id **hood_id**."""

68
backend/tests/conftest.py
View file

@ -15,16 +15,16 @@ from kibicara.model import Mapping
from kibicara.webapi import router
@fixture(scope='module')
@fixture(scope="module")
def client():
Mapping.drop_all()
Mapping.create_all()
app = FastAPI()
app.include_router(router, prefix='/api')
app.include_router(router, prefix="/api")
return TestClient(app)
@fixture(scope='module')
@fixture(scope="module")
def monkeymodule():
from _pytest.monkeypatch import MonkeyPatch
@ -33,96 +33,96 @@ def monkeymodule():
mpatch.undo()
@fixture(scope='module')
@fixture(scope="module")
def receive_email(monkeymodule):
mailbox = []
def mock_send_email(to, subject, sender='kibicara', body=''):
def mock_send_email(to, subject, sender="kibicara", body=""):
mailbox.append(dict(to=to, subject=subject, sender=sender, body=body))
def mock_receive_email():
return mailbox.pop()
monkeymodule.setattr(email, 'send_email', mock_send_email)
monkeymodule.setattr(email, "send_email", mock_send_email)
return mock_receive_email
@fixture(scope='module')
@fixture(scope="module")
def register_token(client, receive_email):
response = client.post(
'/api/admin/register/', json={'email': 'user', 'password': 'password'}
"/api/admin/register/", json={"email": "user", "password": "password"}
)
assert response.status_code == status.HTTP_202_ACCEPTED
return urlparse(receive_email()['body']).query.split('=', 1)[1]
return urlparse(receive_email()["body"]).query.split("=", 1)[1]
@fixture(scope='module')
@fixture(scope="module")
def register_confirmed(client, register_token):
response = client.post('/api/admin/confirm/{0}'.format(register_token))
response = client.post("/api/admin/confirm/{0}".format(register_token))
assert response.status_code == status.HTTP_200_OK
@fixture(scope='module')
@fixture(scope="module")
def access_token(client, register_confirmed):
response = client.post(
'/api/admin/login/', data={'username': 'user', 'password': 'password'}
"/api/admin/login/", data={"username": "user", "password": "password"}
)
assert response.status_code == status.HTTP_200_OK
return response.json()['access_token']
return response.json()["access_token"]
@fixture(scope='module')
@fixture(scope="module")
def auth_header(access_token):
return {'Authorization': 'Bearer {0}'.format(access_token)}
return {"Authorization": "Bearer {0}".format(access_token)}
@fixture(scope='function')
@fixture(scope="function")
def hood_id(client, auth_header):
response = client.post('/api/hoods/', json={'name': 'hood'}, headers=auth_header)
response = client.post("/api/hoods/", json={"name": "hood"}, headers=auth_header)
assert response.status_code == status.HTTP_201_CREATED
hood_id = int(response.headers['Location'])
hood_id = int(response.headers["Location"])
yield hood_id
client.delete('/api/hoods/{0}'.format(hood_id), headers=auth_header)
client.delete("/api/hoods/{0}".format(hood_id), headers=auth_header)
@fixture(scope='function')
@fixture(scope="function")
def trigger_id(client, hood_id, auth_header):
response = client.post(
'/api/hoods/{0}/triggers/'.format(hood_id),
json={'pattern': 'test'},
"/api/hoods/{0}/triggers/".format(hood_id),
json={"pattern": "test"},
headers=auth_header,
)
assert response.status_code == status.HTTP_201_CREATED
trigger_id = int(response.headers['Location'])
trigger_id = int(response.headers["Location"])
yield trigger_id
client.delete(
'/api/hoods/{0}/triggers/{1}'.format(hood_id, trigger_id), headers=auth_header
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id), headers=auth_header
)
@fixture(scope='function')
@fixture(scope="function")
def badword_id(client, hood_id, auth_header):
response = client.post(
'/api/hoods/{0}/badwords/'.format(hood_id),
json={'pattern': ''},
"/api/hoods/{0}/badwords/".format(hood_id),
json={"pattern": ""},
headers=auth_header,
)
assert response.status_code == status.HTTP_201_CREATED
badword_id = int(response.headers['Location'])
badword_id = int(response.headers["Location"])
yield badword_id
client.delete(
'/api/hoods/{0}/badwords/{1}'.format(hood_id, badword_id), headers=auth_header
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id), headers=auth_header
)
@fixture(scope='function')
@fixture(scope="function")
def test_id(client, hood_id, auth_header):
response = client.post(
'/api/hoods/{0}/test/'.format(hood_id), json={}, headers=auth_header
"/api/hoods/{0}/test/".format(hood_id), json={}, headers=auth_header
)
assert response.status_code == status.HTTP_201_CREATED
test_id = int(response.headers['Location'])
test_id = int(response.headers["Location"])
yield test_id
client.delete(
'/api/hoods/{0}/test/{1}'.format(hood_id, test_id), headers=auth_header
"/api/hoods/{0}/test/{1}".format(hood_id, test_id), headers=auth_header
)

4
backend/tests/tests_core/test_api_admin.py
View file

@ -6,10 +6,10 @@ from fastapi import status
def test_hoods_unauthorized(client):
response = client.get('/api/admin/hoods/')
response = client.get("/api/admin/hoods/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hoods_success(client, auth_header):
response = client.get('/api/admin/hoods/', headers=auth_header)
response = client.get("/api/admin/hoods/", headers=auth_header)
assert response.status_code == status.HTTP_200_OK

30
backend/tests/tests_core/test_api_hoods.py
View file

@ -8,75 +8,75 @@ from fastapi import status
def test_hood_read_all(client):
response = client.get('/api/hoods/')
response = client.get("/api/hoods/")
assert response.status_code == status.HTTP_200_OK
def test_hood_create_unauthorized(client, hood_id):
response = client.post('/api/hoods/')
response = client.post("/api/hoods/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hood_read(client, hood_id):
response = client.get('/api/hoods/{0}'.format(hood_id))
response = client.get("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_200_OK
def test_hood_update_unauthorized(client, hood_id):
response = client.put('/api/hoods/{0}'.format(hood_id))
response = client.put("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hood_delete_unauthorized(client, hood_id):
response = client.delete('/api/hoods/{0}'.format(hood_id))
response = client.delete("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_read_all_unauthorized(client, hood_id):
response = client.get('/api/hoods/{0}/triggers/'.format(hood_id))
response = client.get("/api/hoods/{0}/triggers/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_create_unauthorized(client, hood_id):
response = client.post('/api/hoods/{0}/triggers/'.format(hood_id))
response = client.post("/api/hoods/{0}/triggers/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_read_unauthorized(client, hood_id, trigger_id):
response = client.get('/api/hoods/{0}/triggers/{1}'.format(hood_id, trigger_id))
response = client.get("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_update_unauthorized(client, hood_id, trigger_id):
response = client.put('/api/hoods/{0}/triggers/{1}'.format(hood_id, trigger_id))
response = client.put("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_delete_unauthorized(client, hood_id, trigger_id):
response = client.delete('/api/hoods/{0}/triggers/{1}'.format(hood_id, trigger_id))
response = client.delete("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_read_all_unauthorized(client, hood_id):
response = client.get('/api/hoods/{0}/badwords/'.format(hood_id))
response = client.get("/api/hoods/{0}/badwords/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_create_unauthorized(client, hood_id):
response = client.post('/api/hoods/{0}/badwords/'.format(hood_id))
response = client.post("/api/hoods/{0}/badwords/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_read_unauthorized(client, hood_id, badword_id):
response = client.get('/api/hoods/{0}/badwords/{1}'.format(hood_id, badword_id))
response = client.get("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_update_unauthorized(client, hood_id, badword_id):
response = client.put('/api/hoods/{0}/badwords/{1}'.format(hood_id, badword_id))
response = client.put("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_delete_unauthorized(client, hood_id, badword_id):
response = client.delete('/api/hoods/{0}/badwords/{1}'.format(hood_id, badword_id))
response = client.delete("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED

10
backend/tests/tests_email/conftest.py
View file

@ -9,16 +9,16 @@ from fastapi import status
from pytest import fixture
@fixture(scope='function')
@fixture(scope="function")
def email_row(client, hood_id, auth_header):
response = client.post(
'/api/hoods/{0}/email/'.format(hood_id),
json={'name': 'kibicara-test'},
"/api/hoods/{0}/email/".format(hood_id),
json={"name": "kibicara-test"},
headers=auth_header,
)
assert response.status_code == status.HTTP_201_CREATED
email_id = int(response.headers['Location'])
email_id = int(response.headers["Location"])
yield response.json()
client.delete(
'/api/hoods/{0}/email/{1}'.format(hood_id, email_id), headers=auth_header
"/api/hoods/{0}/email/{1}".format(hood_id, email_id), headers=auth_header
)

32
backend/tests/tests_email/test_api_email_happy_path.py
View file

@ -15,49 +15,49 @@ from kibicara.webapi.admin import to_token
def test_email_subscribe_unsubscribe(client, hood_id, receive_email):
response = client.post(
'/api/hoods/{0}/email/subscribe/'.format(hood_id),
json={'email': 'test@localhost'},
"/api/hoods/{0}/email/subscribe/".format(hood_id),
json={"email": "test@localhost"},
)
assert response.status_code == status.HTTP_202_ACCEPTED
mail = receive_email()
body = mail['body']
body = mail["body"]
confirm_url = findall(
r'http[s]?://'
+ r'(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+',
r"http[s]?://"
+ r"(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+",
body,
)[0]
start = len('token=')
start = len("token=")
response = client.post(
'/api/hoods/{0}/email/subscribe/confirm/{1}'.format(
"/api/hoods/{0}/email/subscribe/confirm/{1}".format(
hood_id, urlparse(confirm_url).query[start:]
)
)
assert response.status_code == status.HTTP_201_CREATED
response = client.post(
'/api/hoods/{0}/email/subscribe/confirm/{1}'.format(
"/api/hoods/{0}/email/subscribe/confirm/{1}".format(
hood_id, urlparse(confirm_url).query[start:]
)
)
assert response.status_code == status.HTTP_409_CONFLICT
token = to_token(email=mail['to'], hood=hood_id)
token = to_token(email=mail["to"], hood=hood_id)
response = client.delete(
'/api/hoods/{0}/email/unsubscribe/{1}'.format(hood_id, token)
"/api/hoods/{0}/email/unsubscribe/{1}".format(hood_id, token)
)
assert response.status_code == status.HTTP_204_NO_CONTENT
def test_email_message(client, hood_id, trigger_id, email_row):
body = {
'text': 'test',
'author': 'test@localhost',
'secret': email_row['secret'],
"text": "test",
"author": "test@localhost",
"secret": email_row["secret"],
}
response = client.post('/api/hoods/{0}/email/messages/'.format(hood_id), json=body)
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
assert response.status_code == status.HTTP_201_CREATED
def test_email_send_mda(trigger_id, email_row):
skip('Only works if kibicara is listening on port 8000, and only sometimes')
skip("Only works if kibicara is listening on port 8000, and only sometimes")
mail = """From test@example.com Tue Jun 16 15:33:19 2020
Return-path: <test@example.com>
Envelope-to: hood@localhost
@ -85,6 +85,6 @@ test
--AqNPlAX243a8sip3B7kXv8UKD8wuti--
"""
proc = subprocess.run(
['kibicara_mda', 'hood'], stdout=subprocess.PIPE, input=mail, encoding='ascii'
["kibicara_mda", "hood"], stdout=subprocess.PIPE, input=mail, encoding="ascii"
)
assert proc.returncode == 0

8
backend/tests/tests_email/test_api_email_unauthorized.py
View file

@ -8,18 +8,18 @@ from fastapi import status
def test_email_create_unauthorized(client, hood_id):
response = client.post('/api/hoods/{0}/email/'.format(hood_id))
response = client.post("/api/hoods/{0}/email/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_email_delete_unauthorized(client, hood_id, email_row):
response = client.delete(
'/api/hoods/{0}/email/{1}'.format(hood_id, email_row['id'])
"/api/hoods/{0}/email/{1}".format(hood_id, email_row["id"])
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_email_message_unauthorized(client, hood_id, email_row):
body = {'text': 'test', 'author': 'author', 'secret': 'wrong'}
response = client.post('/api/hoods/{0}/email/messages/'.format(hood_id), json=body)
body = {"text": "test", "author": "author", "secret": "wrong"}
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

24
backend/tests/tests_email/test_api_email_wrong.py
View file

@ -8,15 +8,15 @@ from nacl.exceptions import CryptoError
def test_email_subscribe_empty(client, hood_id):
response = client.post('/api/hoods/{0}/email/subscribe/'.format(hood_id))
response = client.post("/api/hoods/{0}/email/subscribe/".format(hood_id))
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_email_subscribe_confirm_wrong_token(client, hood_id):
try:
response = client.post(
'/api/hoods/{0}/email/subscribe/confirm/'.format(hood_id)
+ 'asdfasdfasdfasdfasdfasdfasdfasdf'
"/api/hoods/{0}/email/subscribe/confirm/".format(hood_id)
+ "asdfasdfasdfasdfasdfasdfasdfasdf"
)
assert response.status_code is not status.HTTP_201_CREATED
except CryptoError:
@ -25,25 +25,25 @@ def test_email_subscribe_confirm_wrong_token(client, hood_id):
def test_email_subscribe_confirm_wrong_hood(client):
response = client.delete(
'/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf'
"/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf"
)
assert response.json()['detail'] == 'Not Found'
assert response.json()["detail"] == "Not Found"
def test_email_message_wrong(client, hood_id, email_row):
body = {
'text': '',
'author': 'test@localhost',
'secret': email_row['secret'],
"text": "",
"author": "test@localhost",
"secret": email_row["secret"],
}
response = client.post('/api/hoods/{0}/email/messages/'.format(hood_id), json=body)
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
assert response.status_code == status.HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS
def test_email_unsubscribe_wrong_token(client, hood_id):
try:
client.delete(
'/api/hoods/{0}/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf'.format(
"/api/hoods/{0}/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf".format(
hood_id
)
)
@ -53,6 +53,6 @@ def test_email_unsubscribe_wrong_token(client, hood_id):
def test_email_unsubscribe_wrong_hood(client):
response = client.delete(
'/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf'
"/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf"
)
assert response.json()['detail'] == 'Not Found'
assert response.json()["detail"] == "Not Found"

6
backend/tests/tests_telegram/conftest.py
View file

@ -9,13 +9,13 @@ from kibicara.model import Hood
from kibicara.platforms.telegram.model import Telegram
@fixture(scope='function')
@fixture(scope="function")
def telegram(event_loop, hood_id, bot):
hood = event_loop.run_until_complete(Hood.objects.get(id=hood_id))
return event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token=bot['api_token'],
welcome_message=bot['welcome_message'],
api_token=bot["api_token"],
welcome_message=bot["welcome_message"],
)
)

30
backend/tests/tests_telegram/test_api_telegram_create_bot.py
View file

@ -10,16 +10,16 @@ from kibicara.platforms import telegram
from kibicara.platforms.telegram.model import Telegram
@fixture(scope='function')
@fixture(scope="function")
def disable_spawner(monkeypatch):
class DoNothing:
def start(self, bot):
assert bot is not None
monkeypatch.setattr(telegram.webapi, 'spawner', DoNothing())
monkeypatch.setattr(telegram.webapi, "spawner", DoNothing())
@mark.parametrize('body', [{'api_token': 'string', 'welcome_message': 'string'}])
@mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
def test_telegram_create_bot(
event_loop,
client,
@ -32,27 +32,27 @@ def test_telegram_create_bot(
def check_token_mock(token):
return True
monkeypatch.setattr(telegram.webapi, 'check_token', check_token_mock)
monkeypatch.setattr(telegram.webapi, "check_token", check_token_mock)
response = client.post(
'/api/hoods/{0}/telegram/'.format(hood_id),
"/api/hoods/{0}/telegram/".format(hood_id),
json=body,
headers=auth_header,
)
assert response.status_code == status.HTTP_201_CREATED
bot_id = response.json()['id']
bot_id = response.json()["id"]
telegram_obj = event_loop.run_until_complete(Telegram.objects.get(id=bot_id))
assert response.json()['api_token'] == body['api_token'] == telegram_obj.api_token
assert response.json()["api_token"] == body["api_token"] == telegram_obj.api_token
assert (
response.json()['welcome_message']
== body['welcome_message']
response.json()["welcome_message"]
== body["welcome_message"]
== telegram_obj.welcome_message
)
assert response.json()['hood']['id'] == telegram_obj.hood.id
assert response.json()["hood"]["id"] == telegram_obj.hood.id
assert telegram_obj.enabled
@mark.parametrize('body', [{'api_token': 'string', 'welcome_message': 'string'}])
@mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
def test_telegram_invalid_api_token(
event_loop,
client,
@ -63,7 +63,7 @@ def test_telegram_invalid_api_token(
body,
):
response = client.post(
'/api/hoods/{0}/telegram/'.format(hood_id),
"/api/hoods/{0}/telegram/".format(hood_id),
json=body,
headers=auth_header,
)
@ -71,12 +71,12 @@ def test_telegram_invalid_api_token(
def test_telegram_create_telegram_invalid_id(client, auth_header):
response = client.post('/api/hoods/1337/telegram/', headers=auth_header)
response = client.post("/api/hoods/1337/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.post('/api/hoods/wrong/telegram/', headers=auth_header)
response = client.post("/api/hoods/wrong/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_telegram_create_unauthorized(client, hood_id):
response = client.post('/api/hoods/{hood_id}/telegram/')
response = client.post("/api/hoods/{hood_id}/telegram/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED

16
backend/tests/tests_telegram/test_api_telegram_delete_bot.py
View file

@ -10,7 +10,7 @@ from pytest import mark, raises
from kibicara.platforms.telegram.model import Telegram, TelegramUser
@mark.parametrize('bot', [{'api_token': 'apitoken123', 'welcome_message': 'msg'}])
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_delete_bot(client, event_loop, bot, telegram, auth_header):
event_loop.run_until_complete(
TelegramUser.objects.create(user_id=1234, bot=telegram.id)
@ -19,7 +19,7 @@ def test_telegram_delete_bot(client, event_loop, bot, telegram, auth_header):
TelegramUser.objects.create(user_id=5678, bot=telegram.id)
)
response = client.delete(
'/api/hoods/{0}/telegram/{1}'.format(telegram.hood.id, telegram.id),
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id),
headers=auth_header,
)
assert response.status_code == status.HTTP_204_NO_CONTENT
@ -30,23 +30,23 @@ def test_telegram_delete_bot(client, event_loop, bot, telegram, auth_header):
def test_telegram_delete_bot_invalid_id(client, auth_header, hood_id):
response = client.delete('/api/hoods/1337/telegram/123', headers=auth_header)
response = client.delete("/api/hoods/1337/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete('/api/hoods/wrong/telegram/123', headers=auth_header)
response = client.delete("/api/hoods/wrong/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = client.delete(
'/api/hoods/{0}/telegram/7331'.format(hood_id), headers=auth_header
"/api/hoods/{0}/telegram/7331".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete(
'/api/hoods/{0}/telegram/wrong'.format(hood_id), headers=auth_header
"/api/hoods/{0}/telegram/wrong".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@mark.parametrize('bot', [{'api_token': 'apitoken123', 'welcome_message': 'msg'}])
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_delete_bot_unauthorized(client, bot, telegram):
response = client.delete(
'/api/hoods/{0}/telegram/{1}'.format(telegram.hood.id, telegram.id)
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

22
backend/tests/tests_telegram/test_api_telegram_get_bot.py
View file

@ -7,36 +7,36 @@ from fastapi import status
from pytest import mark
@mark.parametrize('bot', [{'api_token': 'apitoken123', 'welcome_message': 'msg'}])
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_get_bot(client, auth_header, event_loop, bot, telegram):
response = client.get(
'/api/hoods/{0}/telegram/{1}'.format(telegram.hood.id, telegram.id),
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id),
headers=auth_header,
)
assert response.status_code == status.HTTP_200_OK
assert response.json()['id'] == telegram.id
assert response.json()['api_token'] == telegram.api_token
assert response.json()['welcome_message'] == telegram.welcome_message
assert response.json()["id"] == telegram.id
assert response.json()["api_token"] == telegram.api_token
assert response.json()["welcome_message"] == telegram.welcome_message
def test_telegram_get_bot_invalid_id(client, auth_header, hood_id):
response = client.get('/api/hoods/1337/telegram/123', headers=auth_header)
response = client.get("/api/hoods/1337/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get('/api/hoods/wrong/telegram/123', headers=auth_header)
response = client.get("/api/hoods/wrong/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = client.get(
'/api/hoods/{0}/telegram/7331'.format(hood_id), headers=auth_header
"/api/hoods/{0}/telegram/7331".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get(
'/api/hoods/{0}/telegram/wrong'.format(hood_id), headers=auth_header
"/api/hoods/{0}/telegram/wrong".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@mark.parametrize('bot', [{'api_token': 'apitoken456', 'welcome_message': 'msg'}])
@mark.parametrize("bot", [{"api_token": "apitoken456", "welcome_message": "msg"}])
def test_telegram_get_bot_unauthorized(client, bot, telegram):
response = client.get(
'/api/hoods/{0}/telegram/{1}'.format(telegram.hood.id, telegram.id)
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

24
backend/tests/tests_telegram/test_api_telegram_get_bots.py
View file

@ -14,34 +14,34 @@ def test_telegram_get_bots(client, auth_header, event_loop, hood_id):
telegram0 = event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token='api_token123',
welcome_message='welcome_message123',
api_token="api_token123",
welcome_message="welcome_message123",
)
)
telegram1 = event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token='api_token456',
welcome_message='welcome_message123',
api_token="api_token456",
welcome_message="welcome_message123",
)
)
response = client.get(
'/api/hoods/{0}/telegram'.format(telegram0.hood.id), headers=auth_header
"/api/hoods/{0}/telegram".format(telegram0.hood.id), headers=auth_header
)
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]['id'] == telegram0.id
assert response.json()[0]['api_token'] == telegram0.api_token
assert response.json()[1]['id'] == telegram1.id
assert response.json()[1]['api_token'] == telegram1.api_token
assert response.json()[0]["id"] == telegram0.id
assert response.json()[0]["api_token"] == telegram0.api_token
assert response.json()[1]["id"] == telegram1.id
assert response.json()[1]["api_token"] == telegram1.api_token
def test_telegram_get_bots_invalid_id(client, auth_header, hood_id):
response = client.get('/api/hoods/1337/telegram', headers=auth_header)
response = client.get("/api/hoods/1337/telegram", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get('/api/hoods/wrong/telegram', headers=auth_header)
response = client.get("/api/hoods/wrong/telegram", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_telegram_get_bots_unauthorized(client, hood_id):
response = client.get('/api/hoods/{0}/telegram'.format(hood_id))
response = client.get("/api/hoods/{0}/telegram".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED