missytake/team-bot
1
0
Fork 1
Code Issues 3 Pull requests Packages Projects Releases 4 Wiki Activity

Compare commits

base: missytake:main
Branches Tags
missytake:main
missytake:dev-dependencies
missytake:invite-codes-1
link2xt:link2xt/mvbox-move-start
link2xt:main
missytake:1.2.1
missytake:1.2.0
missytake:1.1.0
missytake:1.0.0
..
compare: link2xt:main
Branches Tags
link2xt:link2xt/mvbox-move-start
link2xt:main
missytake:main
missytake:dev-dependencies
missytake:invite-codes-1
missytake:1.2.1
missytake:1.2.0
missytake:1.1.0
missytake:1.0.0

No commits in common. "main" and "main" have entirely different histories.
main ... main

15 changed files with 151 additions and 296 deletions
Show stats Expand all files Collapse all files

23
.github/workflows/ci.yml vendored
View file

@ -1,23 +0,0 @@
name: CI
on:
pull_request:
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.8', '3.11']
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install '.[dev]'
- run: tox

2
.gitignore vendored
View file

@ -3,8 +3,6 @@
__pycache__/ __pycache__/
/.tox/ /.tox/
/build/ /build/
/dist/
/venv/ /venv/
bot.db/ bot.db/
teams_bot_data/ teams_bot_data/
team_bot_data/

25
README.md
View file

@ -1,4 +1,4 @@
# Team Bot # Teams Bot
This bot connects your team to the outside This bot connects your team to the outside
and makes it addressable. and makes it addressable.
@ -22,8 +22,8 @@ To install this bot,
run: run:
``` ```
git clone https://github.com/deltachat-bot/team-bot git clone https://git.0x90.space/missytake/teams-bot
cd team-bot cd teams-bot
pip install . pip install .
``` ```
@ -32,19 +32,19 @@ with an email address
you want to use as a team: you want to use as a team:
``` ```
team-bot init --email helpdesk@example.org --password p455w0rD teams-bot init --email helpdesk@example.org --password p455w0rD
``` ```
This command will show a QR code; This command will show a QR code;
scan it with Delta Chat scan it with Delta Chat
to become part of the "team", to become part of the "team",
the verified group which manages the Team Bot. the verified group which manages the Teams Bot.
Now to run it, Now to run it,
simply execute: simply execute:
``` ```
team-bot run -v teams-bot run -v
``` ```
The bot only works as long as this command is running. The bot only works as long as this command is running.
@ -59,9 +59,9 @@ you can deploy this bot with it.
Just import it into your [deploy.py file](https://docs.pyinfra.com/en/2.x/getting-started.html#create-a-deploy) like this: Just import it into your [deploy.py file](https://docs.pyinfra.com/en/2.x/getting-started.html#create-a-deploy) like this:
``` ```
from team_bot.pyinfra import deploy_team_bot from teams_bot.pyinfra import deploy_teams_bot
deploy_team_bot( deploy_teams_bot(
unix_user='root', # an existing UNIX user (doesn't need root or sudo privileges) unix_user='root', # an existing UNIX user (doesn't need root or sudo privileges)
bot_email='helpdesk@example.org', # the email address your team wants to use bot_email='helpdesk@example.org', # the email address your team wants to use
bot_passwd='p4ssw0rd', # the password to the email account bot_passwd='p4ssw0rd', # the password to the email account
@ -79,7 +79,7 @@ login to the user with ssh
and run: and run:
``` ```
export $(cat ~/.env | xargs) && ~/.local/lib/team-bot.venv/bin/team-bot init export $(cat ~/.env | xargs) && ~/.local/lib/teams-bot.venv/bin/teams-bot init
``` ```
Then, Then,
@ -88,11 +88,11 @@ and keep it running in the background,
run: run:
``` ```
systemctl --user enable --now team-bot systemctl --user enable --now teams-bot
``` ```
You can view the log output You can view the log output
with `journalctl --user -fu team-bot` with `journalctl --user -fu teams-bot`
to confirm that it works. to confirm that it works.
## Development Environment ## Development Environment
@ -103,6 +103,7 @@ run:
``` ```
python3 -m venv venv python3 -m venv venv
. venv/bin/activate . venv/bin/activate
pip install -e .[dev] pip install pytest tox black pytest-xdist pytest-timeout
pip install -e .
tox tox
``` ```

24
setup.cfg
View file

@ -1,17 +1,17 @@
[metadata] [metadata]
name = team_bot name = teams-bot
version = 1.1.0 version = 0.0.1
author = missytake author = missytake
author_email = missytake@systemli.org author_email = missytake@systemli.org
description = This bot connects your team to the outside and makes it addressable. description = This bot connects your team to the outside and makes it addressable.
long_description = file: README.md long_description = file: README.md
long_description_content_type = text/markdown long_description_content_type = text/markdown
url = https://github.com/deltachat-bot/team-bot url = https://git.0x90.space/missytake/teams-bot
project_urls = project_urls =
Bug Tracker = https://github.com/deltachat-bot/team-bot/issues Bug Tracker = https://git.0x90.space/missytake/teams-bot/issues
classifiers = classifiers =
Programming Language :: Python :: 3 Programming Language :: Python :: 3
License :: OSI Approved :: MIT License License :: OSI Approved :: ISC License (ISCL)
Operating System :: OS Independent Operating System :: OS Independent
[options] [options]
@ -22,24 +22,16 @@ python_requires = >=3.8
install_requires = install_requires =
click click
pyinfra pyinfra
pickleDB==0.9 pickleDB
qrcode qrcode
deltachat>=1.142.7 deltachat
[options.extras_require]
dev =
pytest
tox
black
pytest-xdist
pytest-timeout
[options.packages.find] [options.packages.find]
where = src where = src
[options.entry_points] [options.entry_points]
console_scripts = console_scripts =
team-bot = team_bot.cli:main teams-bot = teams_bot.cli:main
[tox:tox] [tox:tox]
envlist = lint, py310 envlist = lint, py310

6
src/team_bot/pyinfra_assets/setup-venv.sh
View file

@ -1,6 +0,0 @@
#!/usr/bin/env bash
python3 -m venv ~/.local/lib/team-bot.venv
source ~/.local/lib/team-bot.venv/bin/activate
pip install -U pip wheel

0
src/team_bot/__init__.py → src/teams_bot/__init__.py
View file

57
src/team_bot/bot.py → src/teams_bot/bot.py
View file

@ -11,7 +11,6 @@ from .commands import (
crew_help, crew_help,
set_display_name, set_display_name,
set_avatar, set_avatar,
generate_invite,
start_chat, start_chat,
outside_help, outside_help,
set_outside_help, set_outside_help,
@ -41,7 +40,6 @@ class SetupPlugin:
class RelayPlugin: class RelayPlugin:
def __init__(self, account: deltachat.Account, kvstore: pickledb.PickleDB): def __init__(self, account: deltachat.Account, kvstore: pickledb.PickleDB):
self.account = account self.account = account
self.account.set_config("bcc_self", "1")
self.kvstore = kvstore self.kvstore = kvstore
self.crew = account.get_chat_by_id(kvstore.get("crew_id")) self.crew = account.get_chat_by_id(kvstore.get("crew_id"))
if not kvstore.get("relays"): if not kvstore.get("relays"):
@ -66,14 +64,15 @@ class RelayPlugin:
relay_group = self.get_relay_group(message.chat.id) relay_group = self.get_relay_group(message.chat.id)
relay_group.send_text(f"Sending Message failed:\n\n{error}") relay_group.send_text(f"Sending Message failed:\n\n{error}")
@account_hookimpl
def ac_member_removed(self, chat, contact, actor, message):
if chat == self.crew:
self.offboard(contact)
@account_hookimpl @account_hookimpl
def ac_incoming_message(self, message: deltachat.Message): def ac_incoming_message(self, message: deltachat.Message):
"""This method is called on every incoming message and decides what to do with it.""" """This method is called on every incoming message and decides what to do with it."""
logging.info(
"New message from %s in chat %s: %s",
message.get_sender_contact().addr,
message.chat.get_name(),
message.text,
)
if message.is_system_message(): if message.is_system_message():
if message.chat.id == self.crew.id: if message.chat.id == self.crew.id:
@ -114,9 +113,6 @@ class RelayPlugin:
if arguments[0] == "/set_avatar": if arguments[0] == "/set_avatar":
result = set_avatar(self.account, message, self.crew) result = set_avatar(self.account, message, self.crew)
self.reply(message.chat, result, quote=message) self.reply(message.chat, result, quote=message)
if arguments[0] == "/generate-invite":
text = generate_invite(self.account)
self.reply(message.chat, text, quote=message)
if arguments[0] == "/start_chat": if arguments[0] == "/start_chat":
outside_chat, result = start_chat( outside_chat, result = start_chat(
self.account, self.account,
@ -156,7 +152,6 @@ class RelayPlugin:
else: else:
logging.debug("Ignoring message, just the crew chatting") logging.debug("Ignoring message, just the crew chatting")
else: else:
self.mark_last_messages_read(message.chat)
logging.debug("Ignoring message, just the crew chatting") logging.debug("Ignoring message, just the crew chatting")
else: else:
@ -199,11 +194,7 @@ class RelayPlugin:
) )
return return
""":TODO don't forward if message is the explanation message""" """:TODO don't forward if message is the explanation message"""
try:
outside_chat.send_msg(message) outside_chat.send_msg(message)
except Exception as e:
self.reply(message.chat, "Sending message failed.", quote=message)
raise e
def forward_to_relay_group(self, message: deltachat.Message, started_by_crew=False): def forward_to_relay_group(self, message: deltachat.Message, started_by_crew=False):
"""forward a request to a relay group; create one if it doesn't exist yet.""" """forward a request to a relay group; create one if it doesn't exist yet."""
@ -240,10 +231,21 @@ class RelayPlugin:
def is_relay_group(self, chat: deltachat.Chat) -> bool: def is_relay_group(self, chat: deltachat.Chat) -> bool:
"""Check whether a chat is a relay group.""" """Check whether a chat is a relay group."""
for mapping in self.kvstore.get("relays"): if not chat.get_name().startswith(
if mapping[1] == chat.id: "[%s] " % (self.account.get_config("addr").split("@")[0],)
):
return False # all relay groups' names begin with a [tag] with the localpart of the teamsbot's address
if (
chat.get_messages()[0].get_sender_contact()
!= self.account.get_self_contact()
):
return False # all relay groups were started by the teamsbot
if chat.is_protected():
return False # relay groups don't need to be protected, so they are not
for crew_member in self.crew.get_contacts():
if crew_member not in chat.get_contacts():
return False # all crew members have to be in any relay group
return True return True
return False
def get_outside_chat(self, relay_group_id: int) -> deltachat.Chat: def get_outside_chat(self, relay_group_id: int) -> deltachat.Chat:
"""Get the corresponding outside chat for the ID of a relay group. """Get the corresponding outside chat for the ID of a relay group.
@ -268,22 +270,3 @@ class RelayPlugin:
if mapping[0] == outside_id: if mapping[0] == outside_id:
return self.account.get_chat_by_id(mapping[1]) return self.account.get_chat_by_id(mapping[1])
return None return None
def offboard(self, ex_admin: deltachat.Contact) -> None:
"""Remove a former crew member from all relay groups they are part of.
:param ex_admin: a contact which just got removed from the crew.
"""
for mapping in self.kvstore.get("relays"):
relay_group = self.account.get_chat_by_id(mapping[1])
if ex_admin in relay_group.get_contacts():
relay_group.remove_contact(ex_admin)
def mark_last_messages_read(self, relay_group: deltachat.Chat) -> None:
"""Mark the last incoming messages as read for a corresponding relay group.
:param relay_group: the relay group in which the messages which should marked read were forwarded.
"""
outside_chat = self.get_outside_chat(relay_group.id)
for msg in outside_chat.get_messages():
msg.mark_seen()

18
src/team_bot/cli.py → src/teams_bot/cli.py
View file

@ -29,11 +29,11 @@ def set_log_level(verbose: int, db: str):
cls=click.Group, context_settings={"help_option_names": ["-h", "--help"]} cls=click.Group, context_settings={"help_option_names": ["-h", "--help"]}
) )
@click.pass_context @click.pass_context
def team_bot(ctx): def teams_bot(ctx):
"""This bot connects your team to the outside and makes it addressable.""" """This bot connects your team to the outside and makes it addressable."""
@team_bot.command() @teams_bot.command()
@click.option("--email", type=str, default=None, help="the email account for the bot") @click.option("--email", type=str, default=None, help="the email account for the bot")
@click.option( @click.option(
"--password", type=str, default=None, help="the password of the email account" "--password", type=str, default=None, help="the password of the email account"
@ -41,7 +41,7 @@ def team_bot(ctx):
@click.option( @click.option(
"--dbdir", "--dbdir",
type=str, type=str,
default="team_bot_data", default="teams_bot_data",
help="path to the bot's database", help="path to the bot's database",
envvar="TEAMS_DBDIR", envvar="TEAMS_DBDIR",
) )
@ -59,9 +59,9 @@ def init(ctx, email: str, password: str, dbdir: str, verbose: int):
set_log_level(verbose, delta_db) set_log_level(verbose, delta_db)
ac = deltachat.Account(delta_db) ac = deltachat.Account(delta_db)
ac.run_account(addr=email, password=password, show_ffi=verbose)
ac.set_config("mvbox_move", "1") ac.set_config("mvbox_move", "1")
ac.set_config("sentbox_watch", "0") ac.set_config("sentbox_watch", "0")
ac.run_account(addr=email, password=password, show_ffi=verbose)
crew_id_old = kvstore.get("crew_id") crew_id_old = kvstore.get("crew_id")
@ -123,11 +123,11 @@ def init(ctx, email: str, password: str, dbdir: str, verbose: int):
logging.info("Successfully changed crew ID to the new group.") logging.info("Successfully changed crew ID to the new group.")
@team_bot.command() @teams_bot.command()
@click.option( @click.option(
"--dbdir", "--dbdir",
type=str, type=str,
default="team_bot_data", default="teams_bot_data",
help="path to the bot's database", help="path to the bot's database",
envvar="TEAMS_DBDIR", envvar="TEAMS_DBDIR",
) )
@ -157,11 +157,11 @@ def run(ctx, dbdir: str, verbose: int):
ac.wait_shutdown() ac.wait_shutdown()
@team_bot.command() @teams_bot.command()
@click.option( @click.option(
"--dbdir", "--dbdir",
type=str, type=str,
default="team_bot_data", default="teams_bot_data",
help="path to the bot's database", help="path to the bot's database",
envvar="TEAMS_DBDIR", envvar="TEAMS_DBDIR",
) )
@ -191,7 +191,7 @@ def verify_crypto(ctx, dbdir: str, verbose: int):
def main(): def main():
team_bot(auto_envvar_prefix="TEAMS") teams_bot(auto_envvar_prefix="TEAMS")
if __name__ == "__main__": if __name__ == "__main__":

9
src/team_bot/commands.py → src/teams_bot/commands.py
View file

@ -15,7 +15,6 @@ def crew_help() -> str:
Start a chat:\t/start_chat alice@example.org,bob@example.org Chat_Title Hello friends! Start a chat:\t/start_chat alice@example.org,bob@example.org Chat_Title Hello friends!
Change the bot's name:\t/set_name Name Change the bot's name:\t/set_name Name
Change the bot's avatar:\t/set_avatar <attach image> Change the bot's avatar:\t/set_avatar <attach image>
Generate invite link:\t\t/generate-invite
Show this help text:\t\t/help Show this help text:\t\t/help
Change the help message for outsiders:\t/set_outside_help Hello outsider Change the help message for outsiders:\t/set_outside_help Hello outsider
""" """
@ -64,14 +63,6 @@ def set_avatar(
return "Avatar changed to this image." return "Avatar changed to this image."
def generate_invite(account: deltachat.Account) -> str:
"""Return a https://i.delta.chat invite link for chatting with the bot.
:return: the invite link, e.g.: https://i.delta.chat
"""
return account.get_setup_contact_qr()
def start_chat( def start_chat(
ac: deltachat.Account, ac: deltachat.Account,
command: deltachat.Message, command: deltachat.Message,

48
src/team_bot/pyinfra.py → src/teams_bot/pyinfra.py
View file

@ -6,38 +6,47 @@ from pyinfra import host
from pyinfra.facts.systemd import SystemdStatus from pyinfra.facts.systemd import SystemdStatus
def deploy_team_bot(unix_user: str, bot_email: str, bot_passwd: str, dbdir: str = None): def deploy_teams_bot(
unix_user: str, bot_email: str, bot_passwd: str, dbdir: str = None
):
"""Deploy TeamsBot to a UNIX user, with specified credentials """Deploy TeamsBot to a UNIX user, with specified credentials
:param unix_user: the existing UNIX user of the bot :param unix_user: the existing UNIX user of the bot
:param bot_email: the email address for the bot account :param bot_email: the email address for the bot account
:param bot_passwd: the password for the bot's email account :param bot_passwd: the password for the bot's email account
:param dbdir: the directory where the bot's data will be stored. default: ~/.config/team-bot/email@example.org :param dbdir: the directory where the bot's data will be stored. default: ~/.config/teams-bot/email@example.org
""" """
clone_repo = git.repo( clone_repo = git.repo(
name="Pull the team-bot repository", name="Pull the teams-bot repository",
src="https://github.com/deltachat-bot/team-bot", src="https://git.0x90.space/missytake/teams-bot",
dest=f"/home/{unix_user}/team-bot", dest=f"/home/{unix_user}/teams-bot",
rebase=True, rebase=True,
_su_user=unix_user, _su_user=unix_user,
_use_su_login=True, _use_su_login=True,
) )
if clone_repo.changed: if clone_repo.changed:
server.script(
name="Setup virtual environment for teams-bot",
src=importlib.resources.files(__package__)
/ "pyinfra_assets"
/ "setup-venv.sh",
_su_user=unix_user,
_use_su_login=True,
)
server.shell( server.shell(
name="Compile team-bot", name="Compile teams-bot",
commands=[ commands=[
"python3 -m venv ~/.local/lib/team-bot.venv", f". .local/lib/teams-bot.venv/bin/activate && cd /home/{unix_user}/teams-bot && pip install ."
". ~/.local/lib/team-bot.venv/bin/activate && pip install -U pip wheel",
f". .local/lib/team-bot.venv/bin/activate && cd /home/{unix_user}/team-bot && pip install ."
], ],
_su_user=unix_user, _su_user=unix_user,
_use_su_login=True, _use_su_login=True,
) )
if not dbdir: if not dbdir:
dbdir = f"/home/{unix_user}/.config/team_bot/{bot_email}/" dbdir = f"/home/{unix_user}/.config/teams_bot/{bot_email}/"
secrets = [ secrets = [
f"TEAMS_DBDIR={dbdir}", f"TEAMS_DBDIR={dbdir}",
f"TEAMS_INIT_EMAIL={bot_email}", f"TEAMS_INIT_EMAIL={bot_email}",
@ -61,18 +70,23 @@ def deploy_team_bot(unix_user: str, bot_email: str, bot_passwd: str, dbdir: str
) )
files.template( files.template(
name="upload team-bot systemd unit", name="upload teams-bot systemd unit",
src=importlib.resources.files(__package__) src=importlib.resources.files(__package__)
/ "pyinfra_assets" / "pyinfra_assets"
/ "team-bot.service.j2", / "teams-bot.service.j2",
dest=f"/home/{unix_user}/.config/systemd/user/team-bot.service", dest=f"/home/{unix_user}/.config/systemd/user/teams-bot.service",
user=unix_user, user=unix_user,
unix_user=unix_user, unix_user=unix_user,
bot_email=bot_email, bot_email=bot_email,
) )
server.shell(
name=f"enable {unix_user}'s systemd units to auto-start at boot",
commands=[f"loginctl enable-linger {unix_user}"],
)
systemd.daemon_reload( systemd.daemon_reload(
name=f"{unix_user}: load team-bot systemd service", name=f"{unix_user}: load teams-bot systemd service",
user_name=unix_user, user_name=unix_user,
user_mode=True, user_mode=True,
_su_user=unix_user, _su_user=unix_user,
@ -92,10 +106,10 @@ def deploy_team_bot(unix_user: str, bot_email: str, bot_passwd: str, dbdir: str
_use_su_login=True, _use_su_login=True,
) )
try: try:
if services["team-bot.service"]: if services["teams-bot.service"]:
systemd.service( systemd.service(
name=f"{unix_user}: restart team-bot systemd service", name=f"{unix_user}: restart teams-bot systemd service",
service="team-bot.service", service="teams-bot.service",
running=True, running=True,
restarted=True, restarted=True,
user_mode=True, user_mode=True,

6
src/teams_bot/pyinfra_assets/setup-venv.sh Normal file
View file

@ -0,0 +1,6 @@
#!/usr/bin/env bash
python3 -m venv ~/.local/lib/teams-bot.venv
source ~/.local/lib/teams-bot.venv/bin/activate
pip install -U pip wheel

4
src/team_bot/pyinfra_assets/team-bot.service.j2 → src/teams_bot/pyinfra_assets/teams-bot.service.j2
View file

@ -1,8 +1,8 @@
[Unit] [Unit]
Description=run deltachat team-bot: {{ bot_email }} Description=run deltachat teams-bot: {{ bot_email }}
[Service] [Service]
ExecStart=/home/{{ unix_user }}/.local/lib/team-bot.venv/bin/team-bot run -v ExecStart=/home/{{ unix_user }}/.local/lib/teams-bot.venv/bin/teams-bot run -v
EnvironmentFile=/home/{{ unix_user }}/.env EnvironmentFile=/home/{{ unix_user }}/.env
Restart=on-failure Restart=on-failure
RestartSec=5s RestartSec=5s

41
tests/conftest.py
View file

@ -7,7 +7,7 @@ import deltachat
import pytest import pytest
from _pytest.pytester import LineMatcher from _pytest.pytester import LineMatcher
from team_bot.bot import RelayPlugin from teams_bot.bot import RelayPlugin
class ClickRunner: class ClickRunner:
@ -65,9 +65,9 @@ def _perform_match(output, fnl):
@pytest.fixture @pytest.fixture
def cmd(): def cmd():
"""invoke a command line subcommand.""" """invoke a command line subcommand."""
from team_bot.cli import team_bot from teams_bot.cli import teams_bot
return ClickRunner(team_bot) return ClickRunner(teams_bot)
@pytest.fixture @pytest.fixture
@ -80,7 +80,7 @@ def tmp_file_path(request, tmpdir):
@pytest.fixture @pytest.fixture
def relaycrew(crew) -> deltachat.Chat: def relaycrew(crew):
crew.bot.relayplugin = RelayPlugin(crew.bot, crew.kvstore) crew.bot.relayplugin = RelayPlugin(crew.bot, crew.kvstore)
crew.bot.add_account_plugin(crew.bot.relayplugin) crew.bot.add_account_plugin(crew.bot.relayplugin)
assert not crew.bot.relayplugin.is_relay_group(crew) assert not crew.bot.relayplugin.is_relay_group(crew)
@ -88,24 +88,28 @@ def relaycrew(crew) -> deltachat.Chat:
@pytest.fixture @pytest.fixture
def crew(team_bot, team_user, tmpdir) -> deltachat.Chat: def crew(teams_bot, teams_user, tmpdir):
from team_bot.bot import SetupPlugin from teams_bot.bot import SetupPlugin
crew = team_bot.create_group_chat( crew = teams_bot.create_group_chat(
f"Team: {team_bot.get_config('addr')}", verified=True f"Team: {teams_bot.get_config('addr')}", verified=True
) )
setupplugin = SetupPlugin(crew.id) setupplugin = SetupPlugin(crew.id)
team_bot.add_account_plugin(setupplugin) teams_bot.add_account_plugin(setupplugin)
qr = crew.get_join_qr() qr = crew.get_join_qr()
team_user.qr_join_chat(qr) teams_user.qr_join_chat(qr)
setupplugin.member_added.wait(timeout=30) setupplugin.member_added.wait(timeout=30)
crew.user = team_user crew.user = teams_user
crew.bot = team_bot crew.bot = teams_bot
crew.bot.setupplugin = setupplugin crew.bot.setupplugin = setupplugin
# wait until old user is properly added to crew # wait until old user is properly added to crew
team_user._evtracker.wait_securejoin_joiner_progress(1000) last_message = teams_user.wait_next_incoming_message().text
team_user._evtracker.wait_next_incoming_message() # member added message while (
f"Member Me ({teams_user.get_config('addr')}) added by bot" not in last_message
):
print("User received message:", last_message)
last_message = teams_user.wait_next_incoming_message().text
crew.kvstore = pickledb.load(tmpdir + "pickle.db", True) crew.kvstore = pickledb.load(tmpdir + "pickle.db", True)
crew.kvstore.set("crew_id", crew.id) crew.kvstore.set("crew_id", crew.id)
@ -113,7 +117,7 @@ def crew(team_bot, team_user, tmpdir) -> deltachat.Chat:
@pytest.fixture @pytest.fixture
def team_bot(tmpdir) -> deltachat.Account: def teams_bot(tmpdir):
ac = account(tmpdir + "/bot.sqlite", show_ffi=True) ac = account(tmpdir + "/bot.sqlite", show_ffi=True)
yield ac yield ac
ac.shutdown() ac.shutdown()
@ -121,7 +125,7 @@ def team_bot(tmpdir) -> deltachat.Account:
@pytest.fixture @pytest.fixture
def team_user(tmpdir) -> deltachat.Account: def teams_user(tmpdir):
ac = account(tmpdir + "/user.sqlite") ac = account(tmpdir + "/user.sqlite")
yield ac yield ac
ac.shutdown() ac.shutdown()
@ -129,20 +133,19 @@ def team_user(tmpdir) -> deltachat.Account:
@pytest.fixture @pytest.fixture
def outsider(tmpdir) -> deltachat.Account: def outsider(tmpdir):
ac = account(tmpdir + "/outsider.sqlite") ac = account(tmpdir + "/outsider.sqlite")
yield ac yield ac
ac.shutdown() ac.shutdown()
ac.wait_shutdown() ac.wait_shutdown()
def account(db_path, show_ffi=False) -> deltachat.Account: def account(db_path, show_ffi=False):
token = os.environ.get( token = os.environ.get(
"DCC_NEW_TMP_EMAIL", "https://nine.testrun.org/cgi-bin/newemail.py" "DCC_NEW_TMP_EMAIL", "https://nine.testrun.org/cgi-bin/newemail.py"
) )
print(token) print(token)
ac = deltachat.Account(str(db_path)) ac = deltachat.Account(str(db_path))
ac._evtracker = ac.add_account_plugin(deltachat.events.FFIEventTracker(ac))
credentials = requests.post(token).json() credentials = requests.post(token).json()
email = credentials["email"] email = credentials["email"]
password = credentials["password"] password = credentials["password"]

178
tests/test_bot.py
View file

@ -6,90 +6,46 @@ import pytest
from deltachat.capi import lib as dclib from deltachat.capi import lib as dclib
TIMEOUT = 40 TIMEOUT = 20
def get_user_crew(crewuser: deltachat.Account, id=11) -> deltachat.Chat: @pytest.mark.timeout(60)
"""Get the Team chat from the team member's point of view. def test_not_relay_groups(relaycrew, outsider):
:param crewuser: the account object of the team member
:return: the chat object of the team chat
"""
for chat in crewuser.get_chats():
print(chat.id, chat.get_name())
user_crew = crewuser.get_chat_by_id(id)
assert user_crew.get_name().startswith("Team")
return user_crew
@pytest.mark.timeout(TIMEOUT)
def test_not_relay_groups(relaycrew, outsider, lp):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
def find_msg(ac, text): # bot <-> outsider 1:1 chat
for chat in ac.get_chats():
for msg in chat.get_messages():
if msg.text == text:
return msg
text = "outsider -> bot 1:1 chat"
lp.sec(text)
outsider_botcontact = outsider.create_contact(bot.get_config("addr")) outsider_botcontact = outsider.create_contact(bot.get_config("addr"))
outsider_outside_chat = outsider.create_chat(outsider_botcontact) outsider_outside_chat = outsider.create_chat(outsider_botcontact)
outsider_outside_chat.send_text(text) outsider_outside_chat.send_text("test 1:1 message to bot")
lp.sec("receiving message from outsider in 1:1 chat")
bot_message_from_outsider = bot._evtracker.wait_next_incoming_message() bot_message_from_outsider = bot.wait_next_incoming_message()
bot_outside_chat = bot_message_from_outsider.chat bot_outside_chat = bot_message_from_outsider.chat
assert bot_message_from_outsider.text == text
assert not bot.relayplugin.is_relay_group(bot_outside_chat) assert not bot.relayplugin.is_relay_group(bot_outside_chat)
lp.sec("leave relay group with user") # bot <-> outsider group chat
relayed_msg = find_msg(user, text)
if not relayed_msg:
relayed_msg = user._evtracker.wait_next_incoming_message()
relayed_msg.chat.remove_contact(user.get_config("addr"))
leave_msg = bot._evtracker.wait_next_incoming_message()
assert bot.relayplugin.is_relay_group(leave_msg.chat)
text = "outsider -> bot group chat"
lp.sec(text)
outsider_bot_group = outsider.create_group_chat( outsider_bot_group = outsider.create_group_chat(
"test with outsider", contacts=[outsider_botcontact] "test with outsider", contacts=[outsider_botcontact]
) )
outsider_bot_group.send_text(text) outsider_bot_group.send_text("test message to outsider group")
lp.sec("receiving message from outsider in group chat") bot_message_from_outsider = bot.wait_next_incoming_message()
bot_message_from_outsider = bot._evtracker.wait_next_incoming_message()
assert bot_message_from_outsider.text == text
assert not bot.relayplugin.is_relay_group(bot_message_from_outsider.chat) assert not bot.relayplugin.is_relay_group(bot_message_from_outsider.chat)
text = "user -> bot 1:1 chat" # bot <-> user 1:1 chat
lp.sec(text)
user_botcontact = user.create_contact(bot.get_config("addr")) user_botcontact = user.create_contact(bot.get_config("addr"))
user_to_bot = user.create_chat(user_botcontact) user_to_bot = user.create_chat(user_botcontact)
user_to_bot.send_text(text) user_to_bot.send_text("test message to bot")
lp.sec("receiving message from user in 1:1 chat") bot_message_from_user = bot.wait_next_incoming_message()
# somehow the message doesn't trigger DC_EVENT_INCOMING_MSG
# bot._evtracker.wait_next_incoming_message()
bot_message_from_user = find_msg(bot, text)
while not bot_message_from_user:
bot_message_from_user = find_msg(bot, text)
time.sleep(1)
assert bot_message_from_user.text == text
assert not bot.relayplugin.is_relay_group(bot_message_from_user.chat) assert not bot.relayplugin.is_relay_group(bot_message_from_user.chat)
text = "user -> bot group chat" # bot <-> user group chat
lp.sec(text)
user_group = user.create_group_chat("test with user", contacts=[user_botcontact]) user_group = user.create_group_chat("test with user", contacts=[user_botcontact])
user_group.send_text(text) user_group.send_text("testing message to user group")
lp.sec("receiving message from user in group chat") bot_message_from_user = bot.wait_next_incoming_message()
bot_message_from_user = bot._evtracker.wait_next_incoming_message()
assert bot_message_from_user.text == text
assert not bot.relayplugin.is_relay_group(bot_message_from_user.chat) assert not bot.relayplugin.is_relay_group(bot_message_from_user.chat)
@pytest.mark.timeout(TIMEOUT) @pytest.mark.timeout(60)
def test_relay_group_forwarding(relaycrew, outsider): def test_relay_group_forwarding(relaycrew, outsider):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
@ -100,18 +56,18 @@ def test_relay_group_forwarding(relaycrew, outsider):
outsider_outside_chat.send_text("test 1:1 message to bot") outsider_outside_chat.send_text("test 1:1 message to bot")
# get outside chat # get outside chat
message_from_outsider = bot._evtracker.wait_next_incoming_message() message_from_outsider = bot.wait_next_incoming_message()
bot_outside_chat = message_from_outsider.chat bot_outside_chat = message_from_outsider.chat
assert not bot.relayplugin.is_relay_group(bot_outside_chat) assert not bot.relayplugin.is_relay_group(bot_outside_chat)
assert message_from_outsider.is_in_fresh()
# get relay group # get relay group
user_forwarded_message_from_outsider = user._evtracker.wait_next_incoming_message() user.wait_next_incoming_message() # group added message
user_forwarded_message_from_outsider = user.wait_next_incoming_message()
user_relay_group = user_forwarded_message_from_outsider.create_chat() user_relay_group = user_forwarded_message_from_outsider.create_chat()
user_relay_group.send_text( user_relay_group.send_text(
"Chatter in relay group" "Chatter in relay group"
) # send normal reply, not forwarded ) # send normal reply, not forwarded
bot_chatter_in_relay_group = bot._evtracker.wait_next_incoming_message() bot_chatter_in_relay_group = bot.wait_next_incoming_message()
bot_relay_group = bot_chatter_in_relay_group.chat bot_relay_group = bot_chatter_in_relay_group.chat
# check if relay group has relay group properties # check if relay group has relay group properties
@ -133,10 +89,9 @@ def test_relay_group_forwarding(relaycrew, outsider):
user._dc_context, user_relay_group.id, user_direct_reply._dc_msg user._dc_context, user_relay_group.id, user_direct_reply._dc_msg
) )
assert sent_id == user_direct_reply.id assert sent_id == user_direct_reply.id
assert message_from_outsider.is_in_seen()
# check that direct reply was forwarded to outsider # check that direct reply was forwarded to outsider
outsider_direct_reply = outsider._evtracker.wait_next_incoming_message() outsider_direct_reply = outsider.wait_next_incoming_message()
assert outsider_direct_reply.text == "This should be forwarded to the outsider" assert outsider_direct_reply.text == "This should be forwarded to the outsider"
assert outsider_direct_reply.chat == outsider_outside_chat assert outsider_direct_reply.chat == outsider_outside_chat
assert outsider_direct_reply.get_sender_contact() == outsider_botcontact assert outsider_direct_reply.get_sender_contact() == outsider_botcontact
@ -150,7 +105,7 @@ def test_relay_group_forwarding(relaycrew, outsider):
outsider_outside_chat.send_text("Second message by outsider") outsider_outside_chat.send_text("Second message by outsider")
# check that outsider's reply ends up in the same chat # check that outsider's reply ends up in the same chat
user_second_message_from_outsider = user._evtracker.wait_next_incoming_message() user_second_message_from_outsider = user.wait_next_incoming_message()
assert user_second_message_from_outsider.chat == user_relay_group assert user_second_message_from_outsider.chat == user_relay_group
# check that relay group explanation is not forwarded to outsider # check that relay group explanation is not forwarded to outsider
@ -159,44 +114,6 @@ def test_relay_group_forwarding(relaycrew, outsider):
assert "This is the relay group for" not in msg.text assert "This is the relay group for" not in msg.text
@pytest.mark.timeout(TIMEOUT)
def test_offboarding(team_bot, relaycrew, outsider, team_user):
# outsider sends message, creates relay group
outsider_botcontact = outsider.create_contact(team_bot.get_config("addr"))
outsider_outside_chat = outsider.create_chat(outsider_botcontact)
outsider_outside_chat.send_text("test 1:1 message to bot")
# get relay group
user_relay_group = team_user._evtracker.wait_next_incoming_message().chat
bot_relay_group = team_bot.get_chats()[-1]
# outsider gets added to crew
qr = relaycrew.get_join_qr()
outsider.qr_join_chat(qr)
outsider._evtracker.wait_securejoin_joiner_progress(1000)
# user kicks outsider from crew
user_crew = get_user_crew(team_user)
user_crew.remove_contact(team_user.create_contact(outsider))
team_bot._evtracker.wait_next_incoming_message()
# user leaves crew
user_crew.remove_contact(team_user)
# make sure they are also offboarded from relay group
team_bot._evtracker.wait_next_incoming_message()
team_user._evtracker.wait_next_incoming_message()
team_user._evtracker.wait_next_incoming_message()
team_user._evtracker.wait_next_incoming_message()
for contact in bot_relay_group.get_contacts():
assert team_user.get_config("addr") != contact.addr
# make sure there is no message in relay group that outsider was kicked
for msg in user_relay_group.get_messages():
print(msg.text)
assert outsider.get_config("addr") + " removed by " not in msg.text
@pytest.mark.timeout(TIMEOUT)
def test_default_outside_help(relaycrew, outsider): def test_default_outside_help(relaycrew, outsider):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
@ -207,7 +124,7 @@ def test_default_outside_help(relaycrew, outsider):
outsider_outside_chat.send_text("/help") outsider_outside_chat.send_text("/help")
# get response # get response
outside_help_message = outsider._evtracker.wait_next_incoming_message() outside_help_message = outsider.wait_next_incoming_message()
assert "I forward messages to the " in outside_help_message.text assert "I forward messages to the " in outside_help_message.text
# assert no relay group was created # assert no relay group was created
@ -215,7 +132,6 @@ def test_default_outside_help(relaycrew, outsider):
assert len(user.get_chats()) == 1 assert len(user.get_chats()) == 1
@pytest.mark.timeout(TIMEOUT)
def test_empty_outside_help(relaycrew, outsider): def test_empty_outside_help(relaycrew, outsider):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
@ -227,7 +143,7 @@ def test_empty_outside_help(relaycrew, outsider):
assert user_crew.get_name().startswith("Team") assert user_crew.get_name().startswith("Team")
user_crew.send_text("/set_outside_help") user_crew.send_text("/set_outside_help")
# ensure /set_outside_help arrives before sending /help # ensure /set_outside_help arrives before sending /help
bot._evtracker.wait_next_incoming_message() bot.wait_next_incoming_message()
# create outside chat # create outside chat
outsider_botcontact = outsider.create_contact(bot.get_config("addr")) outsider_botcontact = outsider.create_contact(bot.get_config("addr"))
@ -235,13 +151,12 @@ def test_empty_outside_help(relaycrew, outsider):
outsider_outside_chat.send_text("/help") outsider_outside_chat.send_text("/help")
# get forwarded /help message # get forwarded /help message
user._evtracker.wait_next_incoming_message() # "Removed help message for outsiders" user.wait_next_incoming_message() # group added message
user._evtracker.wait_next_incoming_message() # explanation message user.wait_next_incoming_message() # explanation message
user_forwarded_message_from_outsider = user._evtracker.wait_next_incoming_message() user_forwarded_message_from_outsider = user.wait_next_incoming_message()
assert user_forwarded_message_from_outsider.text == "/help" assert user_forwarded_message_from_outsider.text == "/help"
@pytest.mark.timeout(TIMEOUT)
def test_changed_outside_help(relaycrew, outsider): def test_changed_outside_help(relaycrew, outsider):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
@ -254,7 +169,7 @@ def test_changed_outside_help(relaycrew, outsider):
outside_help_text = "Hi friend :) send me messages to chat with the team" outside_help_text = "Hi friend :) send me messages to chat with the team"
user_crew.send_text("/set_outside_help " + outside_help_text) user_crew.send_text("/set_outside_help " + outside_help_text)
# ensure /set_outside_help arrives before sending /help # ensure /set_outside_help arrives before sending /help
bot._evtracker.wait_next_incoming_message() bot.wait_next_incoming_message()
# create outside chat # create outside chat
outsider_botcontact = outsider.create_contact(bot.get_config("addr")) outsider_botcontact = outsider.create_contact(bot.get_config("addr"))
@ -262,7 +177,7 @@ def test_changed_outside_help(relaycrew, outsider):
outsider_outside_chat.send_text("/help") outsider_outside_chat.send_text("/help")
# get response # get response
outside_help_message = outsider._evtracker.wait_next_incoming_message() outside_help_message = outsider.wait_next_incoming_message()
assert outside_help_message.text == outside_help_text assert outside_help_message.text == outside_help_text
# assert no relay group was created # assert no relay group was created
@ -270,7 +185,6 @@ def test_changed_outside_help(relaycrew, outsider):
assert len(user.get_chats()) == 1 assert len(user.get_chats()) == 1
@pytest.mark.timeout(TIMEOUT)
def test_change_avatar(relaycrew): def test_change_avatar(relaycrew):
bot = relaycrew.bot bot = relaycrew.bot
user = relaycrew.user user = relaycrew.user
@ -288,23 +202,25 @@ def test_change_avatar(relaycrew):
pytest.skip(f"example image not available: {example_png_path}") pytest.skip(f"example image not available: {example_png_path}")
# set avatar to example image # set avatar to example image
user_crew = get_user_crew(user) for chat in user.get_chats():
print(chat.id, chat.get_name())
user_crew = user.get_chat_by_id(11)
assert user_crew.get_name().startswith("Team")
msg = deltachat.Message.new_empty(user, "image") msg = deltachat.Message.new_empty(user, "image")
msg.set_text("/set_avatar") msg.set_text("/set_avatar")
msg.set_file(example_png_path) msg.set_file(example_png_path)
sent_id = dclib.dc_send_msg(user._dc_context, user_crew.id, msg._dc_msg) sent_id = dclib.dc_send_msg(user._dc_context, user_crew.id, msg._dc_msg)
assert sent_id == msg.id assert sent_id == msg.id
group_avatar_changed_msg = user._evtracker.wait_next_incoming_message() group_avatar_changed_msg = user.wait_next_incoming_message()
assert "Group image changed" in group_avatar_changed_msg.text assert "Group image changed" in group_avatar_changed_msg.text
assert user_crew.get_profile_image() assert user_crew.get_profile_image()
confirmation_msg = user._evtracker.wait_next_incoming_message() confirmation_msg = user.wait_next_incoming_message()
assert confirmation_msg.text == "Avatar changed to this image." assert confirmation_msg.text == "Avatar changed to this image."
assert botcontact.get_profile_image() assert botcontact.get_profile_image()
@pytest.mark.timeout(TIMEOUT * 2)
def test_forward_sending_errors_to_relay_group(relaycrew): def test_forward_sending_errors_to_relay_group(relaycrew):
usercrew = relaycrew.user.get_chats()[-1] usercrew = relaycrew.user.get_chats()[-1]
usercrew.send_text("/start_chat alice@example.org This_Message_will_fail test") usercrew.send_text("/start_chat alice@example.org This_Message_will_fail test")
@ -321,9 +237,7 @@ def test_forward_sending_errors_to_relay_group(relaycrew):
while len(relaycrew.user.get_chats()) < 2 and int(time.time()) < begin + TIMEOUT: while len(relaycrew.user.get_chats()) < 2 and int(time.time()) < begin + TIMEOUT:
time.sleep(0.1) time.sleep(0.1)
for chat in relaycrew.user.get_chats(): relay_group = relaycrew.user.get_chats()[-2]
if "This Message will fail" in chat.get_name():
relay_group = chat
while len(relay_group.get_messages()) < 3 and int(time.time()) < begin + TIMEOUT: while len(relay_group.get_messages()) < 3 and int(time.time()) < begin + TIMEOUT:
print(relay_group.get_messages()[-1].text) print(relay_group.get_messages()[-1].text)
@ -336,21 +250,3 @@ def test_forward_sending_errors_to_relay_group(relaycrew):
"Invalid unencrypted mail to <alice@example.org>" "Invalid unencrypted mail to <alice@example.org>"
in relay_group.get_messages()[-1].text in relay_group.get_messages()[-1].text
) )
@pytest.mark.timeout(TIMEOUT)
def test_public_invite(relaycrew, outsider):
crew = get_user_crew(relaycrew.user)
crew.send_text("/generate-invite")
result = relaycrew.user._evtracker.wait_next_incoming_message()
# assert result.filename
# assert result.text.startswith("https://i.delta.chat")
# qr = result.filename
# invite = "OPENPGP4FPR:" + result.text[22::]
chat = outsider.qr_setup_contact(result.text)
outsider._evtracker.wait_securejoin_joiner_progress(1000)
while not chat.is_protected():
print(chat.get_messages()[-1].text)
time.sleep(1)

2
tests/test_cli.py
View file

@ -2,7 +2,7 @@ def test_help(cmd):
cmd.run_ok( cmd.run_ok(
[], [],
""" """
Usage: team-bot [OPTIONS] COMMAND [ARGS]... Usage: teams-bot [OPTIONS] COMMAND [ARGS]...
* -h, --help Show this message and exit. * -h, --help Show this message and exit.
* init Scan a QR code to create a crew and join it * init Scan a QR code to create a crew and join it
* run Run the bot, so it relays messages from and to the outside * run Run the bot, so it relays messages from and to the outside