Set up integration tests against a local instance

requirements-dev.txt

@@ -1,10 +1,10 @@
 keyring
+psycopg2
 pytest
 pytest-cov
 pyxdg
+pyyaml
 sphinx
 sphinx-autobuild
-stdeb
 twine
 wheel
-pyyaml

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