diff --git a/requirements-dev.txt b/requirements-dev.txt index 8e1d32f..e559c3b 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,10 +1,10 @@ keyring +psycopg2 pytest pytest-cov pyxdg +pyyaml sphinx sphinx-autobuild -stdeb twine wheel -pyyaml diff --git a/tests/assets/test1.png b/tests/assets/test1.png new file mode 100644 index 0000000..3ecb770 Binary files /dev/null and b/tests/assets/test1.png differ diff --git a/tests/assets/test2.png b/tests/assets/test2.png new file mode 100644 index 0000000..90800c7 Binary files /dev/null and b/tests/assets/test2.png differ diff --git a/tests/assets/test3.png b/tests/assets/test3.png new file mode 100644 index 0000000..f0ffc78 Binary files /dev/null and b/tests/assets/test3.png differ diff --git a/tests/assets/test4.png b/tests/assets/test4.png new file mode 100644 index 0000000..af99441 Binary files /dev/null and b/tests/assets/test4.png differ diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..8c5c5cc --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,274 @@ +""" +This module contains integration tests meant to run against a test Mastodon instance. + +You can set up a test instance locally by following this guide: +https://docs.joinmastodon.org/dev/setup/ + +To enable integration tests, export the following environment variables to match +your test server and database: + +``` +export TOOT_TEST_HOSTNAME="localhost:3000" +export TOOT_TEST_DATABASE_DSN="mastodon_development" +``` +""" + +import os +import psycopg2 +import pytest +import re +import uuid + +from os import path +from toot import CLIENT_NAME, CLIENT_WEBSITE, api, App, User +from toot.console import run_command +from toot.exceptions import NotFoundError +from toot.utils import get_text + +# Host name of a test instance to run integration tests against +# DO NOT USE PUBLIC INSTANCES!!! +HOSTNAME = os.getenv("TOOT_TEST_HOSTNAME") + +# Mastodon database name, used to confirm user registration without having to click the link +DATABASE_DSN = os.getenv("TOOT_TEST_DATABASE_DSN") + + +if not HOSTNAME or not DATABASE_DSN: + pytest.skip("Skipping integration tests", allow_module_level=True) + +# ------------------------------------------------------------------------------ +# Fixtures +# ------------------------------------------------------------------------------ + + +def create_app(): + response = api.create_app(HOSTNAME, scheme="http") + return App(HOSTNAME, f"http://{HOSTNAME}", response["client_id"], response["client_secret"]) + + +def register_account(app: App): + username = str(uuid.uuid4())[-10:] + email = f"{username}@example.com" + + response = api.register_account(app, username, email, "password", "en") + confirm_user(email) + return User(app.instance, username, response["access_token"]) + + +def confirm_user(email): + conn = psycopg2.connect(DATABASE_DSN) + cursor = conn.cursor() + cursor.execute("UPDATE users SET confirmed_at = now() WHERE email = %s;", (email,)) + conn.commit() + + +@pytest.fixture(scope="session") +def app(): + return create_app() + + +@pytest.fixture(scope="session") +def user(app): + return register_account(app) + + +# ------------------------------------------------------------------------------ +# Tests +# ------------------------------------------------------------------------------ + +def test_get_instance(app): + response = api.get_instance(HOSTNAME, scheme="http") + assert response["title"] == "Mastodon" + assert response["uri"] == app.instance + + +def test_post(app, user, capsys): + text = "i wish i was a #lumberjack" + run_command(app, user, "post", [text]) + status_id = _posted_status_id(capsys) + + status = api.fetch_status(app, user, status_id) + assert text == get_text(status["content"]) + assert status["account"]["acct"] == user.username + assert status["application"]["name"] == CLIENT_NAME + assert status["application"]["website"] == CLIENT_WEBSITE + assert status["visibility"] == "public" + assert status["sensitive"] is False + assert status["spoiler_text"] == "" + + +def test_post_visibility(app, user, capsys): + for visibility in ["public", "unlisted", "private", "direct"]: + run_command(app, user, "post", ["foo", "--visibility", visibility]) + status_id = _posted_status_id(capsys) + status = api.fetch_status(app, user, status_id) + assert status["visibility"] == visibility + + +def test_media_attachments(app, user, capsys): + assets_dir = path.realpath(path.join(path.dirname(__file__), "assets")) + + path1 = path.join(assets_dir, "test1.png") + path2 = path.join(assets_dir, "test2.png") + path3 = path.join(assets_dir, "test3.png") + path4 = path.join(assets_dir, "test4.png") + + run_command(app, user, "post", [ + "--media", path1, + "--media", path2, + "--media", path3, + "--media", path4, + "--description", "Test 1", + "--description", "Test 2", + "--description", "Test 3", + "--description", "Test 4", + "some text" + ]) + + status_id = _posted_status_id(capsys) + status = api.fetch_status(app, user, status_id) + + [a1, a2, a3, a4] = status["media_attachments"] + + assert a1["meta"]["original"]["size"] == "50x50" + assert a2["meta"]["original"]["size"] == "50x60" + assert a3["meta"]["original"]["size"] == "50x70" + assert a4["meta"]["original"]["size"] == "50x80" + + assert a1["description"] == "Test 1" + assert a2["description"] == "Test 2" + assert a3["description"] == "Test 3" + assert a4["description"] == "Test 4" + + +def test_delete_status(app, user): + status = api.post_status(app, user, "foo") + + response = api.delete_status(app, user, status["id"]).json() + assert response["id"] == status["id"] + + with pytest.raises(NotFoundError): + api.fetch_status(app, user, response["id"]) + + +def test_favourite(app, user, capsys): + status = api.post_status(app, user, "foo") + assert not status["favourited"] + + run_command(app, user, "favourite", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status favourited" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert status["favourited"] + + run_command(app, user, "unfavourite", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status unfavourited" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert not status["favourited"] + + +def test_reblog(app, user, capsys): + status = api.post_status(app, user, "foo") + assert not status["reblogged"] + + run_command(app, user, "reblog", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status reblogged" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert status["reblogged"] + + run_command(app, user, "reblogged_by", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == f"@{user.username}" + + run_command(app, user, "unreblog", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status unreblogged" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert not status["reblogged"] + + +def test_pin(app, user, capsys): + status = api.post_status(app, user, "foo") + assert not status["pinned"] + + run_command(app, user, "pin", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status pinned" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert status["pinned"] + + run_command(app, user, "unpin", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status unpinned" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert not status["pinned"] + + +def test_bookmark(app, user, capsys): + status = api.post_status(app, user, "foo") + assert not status["bookmarked"] + + run_command(app, user, "bookmark", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status bookmarked" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert status["bookmarked"] + + run_command(app, user, "unbookmark", [status["id"]]) + + out, err = capsys.readouterr() + assert strip_ansi(out) == "✓ Status unbookmarked" + assert err == "" + + status = api.fetch_status(app, user, status["id"]) + assert not status["bookmarked"] + + +# ------------------------------------------------------------------------------ +# Utils +# ------------------------------------------------------------------------------ + +strip_ansi_pattern = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") + + +def strip_ansi(string): + return strip_ansi_pattern.sub("", string).strip() + + +def _posted_status_id(capsys): + out, err = capsys.readouterr() + out = strip_ansi(out) + assert err == "" + + pattern = re.compile(r"Toot posted: http://([^/]+)/@([^/]+)/(.+)") + match = re.search(pattern, out) + assert match + + host, _, status_id = match.groups() + assert host == HOSTNAME + + return status_id