bots: Add bot for uploading files to Zulip server.
Add file_uploader bot as an example of using ExternalBotHandler's methods for uploading files to Zulip server.
This commit is contained in:
parent
e5239c5c54
commit
7fc1ff5e0e
21
zulip_bots/zulip_bots/bots/file_uploader/doc.md
Normal file
21
zulip_bots/zulip_bots/bots/file_uploader/doc.md
Normal file
|
@ -0,0 +1,21 @@
|
|||
# File Uploader Bot
|
||||
|
||||
This bot allows the user to upload a file with a given path to the Zulip server.
|
||||
|
||||
## Usage
|
||||
|
||||
Use this bot with any of the following commands:
|
||||
|
||||
- `@uploader <local_file_path>` : Upload a file, where `<local_file_path>` is the path to the file
|
||||
- `@uploader help` : Display help message
|
||||
|
||||
### Usage examples
|
||||
|
||||
The following command will upload the file `/tmp/image.png` to the Zulip server:
|
||||
```
|
||||
@**uploader** /tmp/image.png
|
||||
```
|
||||
|
||||
Here's an example response:
|
||||
|
||||
> [image.png](https://server.zulipchat.com/user_uploads/3787/RgoZReSsfMjlQSzvVxjIgAQy/image.png)
|
41
zulip_bots/zulip_bots/bots/file_uploader/file_uploader.py
Normal file
41
zulip_bots/zulip_bots/bots/file_uploader/file_uploader.py
Normal file
|
@ -0,0 +1,41 @@
|
|||
from typing import Any, Dict
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
class FileUploaderHandler(object):
|
||||
def usage(self) -> str:
|
||||
return (
|
||||
'This interactive bot is used to upload files (such as images) to the Zulip server:'
|
||||
'\n- @uploader <local_file_path> : Upload a file, where <local_file_path> is the path to the file'
|
||||
'\n- @uploader help : Display help message'
|
||||
)
|
||||
|
||||
def handle_message(self, message: Dict[str, str], bot_handler: Any) -> None:
|
||||
HELP_STR = (
|
||||
'Use this bot with any of the following commands:'
|
||||
'\n* `@uploader <local_file_path>` : Upload a file, where `<local_file_path>` is the path to the file'
|
||||
'\n* `@uploader help` : Display help message'
|
||||
)
|
||||
|
||||
content = message['content'].strip()
|
||||
if content == 'help':
|
||||
bot_handler.send_reply(message, HELP_STR)
|
||||
return
|
||||
|
||||
path = Path(os.path.expanduser(content))
|
||||
if not path.is_file():
|
||||
bot_handler.send_reply(message, 'File `{}` not found'.format(content))
|
||||
return
|
||||
|
||||
path = path.resolve()
|
||||
upload = bot_handler.upload_file_from_path(str(path))
|
||||
if upload['result'] != 'success':
|
||||
msg = upload['msg']
|
||||
bot_handler.send_reply(message, 'Failed to upload `{}` file: {}'.format(path, msg))
|
||||
return
|
||||
|
||||
uploaded_file_reply = '[{}]({})'.format(path.name, upload['uri'])
|
||||
bot_handler.send_reply(message, uploaded_file_reply)
|
||||
|
||||
handler_class = FileUploaderHandler
|
|
@ -0,0 +1,39 @@
|
|||
from unittest.mock import patch, MagicMock, Mock
|
||||
|
||||
from zulip_bots.test_lib import (
|
||||
get_bot_message_handler,
|
||||
StubBotHandler,
|
||||
BotTestCase,
|
||||
DefaultTests,
|
||||
)
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
class TestFileUploaderBot(BotTestCase, DefaultTests):
|
||||
bot_name = "file_uploader"
|
||||
|
||||
@patch('pathlib.Path.is_file', return_value=False)
|
||||
def test_file_not_found(self, is_file: Mock) -> None:
|
||||
self.verify_reply('file.txt', 'File `file.txt` not found')
|
||||
|
||||
@patch('pathlib.Path.resolve', return_value=Path('/file.txt'))
|
||||
@patch('pathlib.Path.is_file', return_value=True)
|
||||
def test_file_upload_failed(self, is_file: Mock, resolve: Mock) -> None:
|
||||
server_reply = dict(result='', msg='error')
|
||||
with patch('zulip_bots.test_lib.StubBotHandler.upload_file_from_path',
|
||||
return_value=server_reply) as m:
|
||||
self.verify_reply('file.txt', 'Failed to upload `/file.txt` file: error')
|
||||
|
||||
@patch('pathlib.Path.resolve', return_value=Path('/file.txt'))
|
||||
@patch('pathlib.Path.is_file', return_value=True)
|
||||
def test_file_upload_success(self, is_file: Mock, resolve: Mock) -> None:
|
||||
server_reply = dict(result='success', uri='https://file/uri')
|
||||
with patch('zulip_bots.test_lib.StubBotHandler.upload_file_from_path',
|
||||
return_value=server_reply) as m:
|
||||
self.verify_reply('file.txt', '[file.txt](https://file/uri)')
|
||||
|
||||
def test_help(self):
|
||||
self.verify_reply('help',
|
||||
('Use this bot with any of the following commands:'
|
||||
'\n* `@uploader <local_file_path>` : Upload a file, where `<local_file_path>` is the path to the file'
|
||||
'\n* `@uploader help` : Display help message'))
|
|
@ -2,6 +2,7 @@ import configparser
|
|||
import sys
|
||||
|
||||
from zulip_bots.lib import BotIdentity
|
||||
from uuid import uuid4
|
||||
|
||||
class SimpleStorage:
|
||||
def __init__(self):
|
||||
|
@ -32,6 +33,9 @@ class SimpleMessageServer:
|
|||
def update(self, message):
|
||||
self.messages[message['message_id']] = message
|
||||
|
||||
def upload_file(self, file):
|
||||
return dict(result='success', msg='', uri='https://server/user_uploads/{}'.format(uuid4()))
|
||||
|
||||
class TerminalBotHandler:
|
||||
def __init__(self, bot_config_file):
|
||||
self.bot_config_file = bot_config_file
|
||||
|
@ -68,6 +72,13 @@ class TerminalBotHandler:
|
|||
{}
|
||||
'''.format(message['message_id'], message['content']))
|
||||
|
||||
def upload_file_from_path(self, file_path):
|
||||
with open(file_path) as file:
|
||||
return self.upload_file(file)
|
||||
|
||||
def upload_file(self, file):
|
||||
return self.message_server.upload_file(file)
|
||||
|
||||
def get_config_info(self, bot_name, optional=False):
|
||||
if self.bot_config_file is None:
|
||||
if optional:
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
import unittest
|
||||
|
||||
from typing import List, Dict, Any, Tuple, Optional
|
||||
from typing import List, Dict, Any, Tuple, Optional, IO
|
||||
|
||||
from zulip_bots.custom_exceptions import (
|
||||
ConfigValidationError,
|
||||
|
@ -53,6 +53,15 @@ class StubBotHandler:
|
|||
def update_message(self, message: Dict[str, Any]) -> None:
|
||||
self.message_server.update(message)
|
||||
|
||||
def upload_file_from_path(self, file_path):
|
||||
# type: (str) -> Dict[str, Any]
|
||||
with open(file_path, 'rb') as file:
|
||||
return self.message_server.upload_file(file)
|
||||
|
||||
def upload_file(self, file):
|
||||
# type: (IO[Any]) -> Dict[str, Any]
|
||||
return self.message_server.upload_file(file)
|
||||
|
||||
class BotQuitException(Exception):
|
||||
pass
|
||||
|
||||
|
|
Loading…
Reference in a new issue