{{ theme_description }}
-{% endif %} diff --git a/docs/advanced.rst b/docs/advanced.md similarity index 52% rename from docs/advanced.rst rename to docs/advanced.md index 88810cf..6b4c5f2 100644 --- a/docs/advanced.rst +++ b/docs/advanced.md @@ -1,40 +1,39 @@ -============== Advanced usage ============== Disabling HTTPS --------------- -You may pass the ``--disable-https`` flag to use unencrypted HTTP instead of +You may pass the `--disable-https` flag to use unencrypted HTTP instead of HTTPS for a given instance. This is inherently insecure and should be used only when connecting to local development instances. -.. code-block:: sh - - toot login --disable-https --instance localhost:8080 +```sh +toot login --disable-https --instance localhost:8080 +``` Using proxies ------------- -You can configure proxies by setting the ``HTTPS_PROXY`` or ``HTTP_PROXY`` +You can configure proxies by setting the `HTTPS_PROXY` or `HTTP_PROXY` environment variables. This will cause all http(s) requests to be proxied through the specified server. For example: -.. code-block:: sh - - export HTTPS_PROXY="http://1.2.3.4:5678" - toot login --instance mastodon.social +```sh +export HTTPS_PROXY="http://1.2.3.4:5678" +toot login --instance mastodon.social +``` **NB:** This feature is provided by -`requestsThe computer can't tell you the emotional story. It can give you the exact mathematical design, but what's missing is the eyebrows.
", - 'reblog': None, - 'in_reply_to_id': None, - 'media_attachments': [], - }]) - - console.run_command(app, user, 'timeline', ['--once']) - - mock_get.assert_called_once_with(app, user, '/api/v1/timelines/home?limit=10', None) - - out, err = capsys.readouterr() - lines = out.split("\n") - - assert "Frank Zappa 🎸" in lines[1] - assert "@fz" in lines[1] - assert "2017-04-12 15:53 UTC" in lines[1] - - assert ( - "The computer can't tell you the emotional story. It can give you the " - "exact mathematical design, but\nwhat's missing is the eyebrows." in out) - - assert "111111111111111111" in lines[-3] - - assert err == "" - - -@mock.patch('toot.http.get') -def test_timeline_with_re(mock_get, monkeypatch, capsys): - mock_get.return_value = MockResponse([{ - 'id': '111111111111111111', - 'created_at': '2017-04-12T15:53:18.174Z', - 'account': { - 'display_name': 'Frank Zappa', - 'acct': 'fz' - }, - 'reblog': { - 'account': { - 'display_name': 'Johnny Cash', - 'acct': 'jc' - }, - 'content': "The computer can't tell you the emotional story. It can give you the exact mathematical design, but what's missing is the eyebrows.
", - 'media_attachments': [], - }, - 'in_reply_to_id': '111111111111111110', - 'media_attachments': [], - }]) - - console.run_command(app, user, 'timeline', ['--once']) - - mock_get.assert_called_once_with(app, user, '/api/v1/timelines/home?limit=10', None) - - out, err = capsys.readouterr() - lines = out.split("\n") - - assert "Frank Zappa" in lines[1] - assert "@fz" in lines[1] - assert "2017-04-12 15:53 UTC" in lines[1] - - assert ( - "The computer can't tell you the emotional story. It can give you the " - "exact mathematical design, but\nwhat's missing is the eyebrows." in out) - - assert "111111111111111111" in lines[-3] - assert "↻ Reblogged @jc" in lines[-3] - - assert err == "" - - -@mock.patch('toot.http.get') -def test_thread(mock_get, monkeypatch, capsys): - mock_get.side_effect = [ - MockResponse({ - 'id': '111111111111111111', - 'account': { - 'display_name': 'Frank Zappa', - 'acct': 'fz' - }, - 'created_at': '2017-04-12T15:53:18.174Z', - 'content': "my response in the middle", - 'reblog': None, - 'in_reply_to_id': '111111111111111110', - 'media_attachments': [], - }), - MockResponse({ - 'ancestors': [{ - 'id': '111111111111111110', - 'account': { - 'display_name': 'Frank Zappa', - 'acct': 'fz' - }, - 'created_at': '2017-04-12T15:53:18.174Z', - 'content': "original content", - 'media_attachments': [], - 'reblog': None, - 'in_reply_to_id': None}], - 'descendants': [{ - 'id': '111111111111111112', - 'account': { - 'display_name': 'Frank Zappa', - 'acct': 'fz' - }, - 'created_at': '2017-04-12T15:53:18.174Z', - 'content': "response message", - 'media_attachments': [], - 'reblog': None, - 'in_reply_to_id': '111111111111111111'}], - }), - ] - - console.run_command(app, user, 'thread', ['111111111111111111']) - - calls = [ - mock.call(app, user, '/api/v1/statuses/111111111111111111'), - mock.call(app, user, '/api/v1/statuses/111111111111111111/context'), - ] - mock_get.assert_has_calls(calls, any_order=False) - - out, err = capsys.readouterr() - - assert not err - - # Display order - assert out.index('original content') < out.index('my response in the middle') - assert out.index('my response in the middle') < out.index('response message') - - assert "original content" in out - assert "my response in the middle" in out - assert "response message" in out - assert "Frank Zappa" in out - assert "@fz" in out - assert "111111111111111111" in out - assert "In reply to" in out - -@mock.patch('toot.http.get') -def test_reblogged_by(mock_get, monkeypatch, capsys): - mock_get.return_value = MockResponse([{ - 'display_name': 'Terry Bozzio', - 'acct': 'bozzio@drummers.social', - }, { - 'display_name': 'Dweezil', - 'acct': 'dweezil@zappafamily.social', - }]) - - console.run_command(app, user, 'reblogged_by', ['111111111111111111']) - - calls = [ - mock.call(app, user, '/api/v1/statuses/111111111111111111/reblogged_by'), - ] - mock_get.assert_has_calls(calls, any_order=False) - - out, err = capsys.readouterr() - - # Display order - expected = "\n".join([ - "Terry Bozzio", - " @bozzio@drummers.social", - "Dweezil", - " @dweezil@zappafamily.social", - "", - ]) - assert out == expected - - -@mock.patch('toot.http.post') -def test_upload(mock_post, capsys): - mock_post.return_value = MockResponse({ - 'id': 123, - 'url': 'https://bigfish.software/123/456', - 'preview_url': 'https://bigfish.software/789/012', - 'url': 'https://bigfish.software/345/678', - 'type': 'image', - }) - - console.run_command(app, user, 'upload', [__file__]) - - mock_post.call_count == 1 - - args, kwargs = http.post.call_args - assert args == (app, user, '/api/v1/media') - assert isinstance(kwargs['files']['file'], io.BufferedReader) - - out, err = capsys.readouterr() - assert "Uploading media" in out - assert __file__ in out - - -@mock.patch('toot.http.get') -def test_search(mock_get, capsys): - mock_get.return_value = MockResponse({ - 'hashtags': [ - { - 'history': [], - 'name': 'foo', - 'url': 'https://mastodon.social/tags/foo' - }, - { - 'history': [], - 'name': 'bar', - 'url': 'https://mastodon.social/tags/bar' - }, - { - 'history': [], - 'name': 'baz', - 'url': 'https://mastodon.social/tags/baz' - }, - ], - 'accounts': [{ - 'acct': 'thequeen', - 'display_name': 'Freddy Mercury' - }, { - 'acct': 'thequeen@other.instance', - 'display_name': 'Mercury Freddy' - }], - 'statuses': [], - }) - - console.run_command(app, user, 'search', ['freddy']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', { - 'q': 'freddy', - 'type': None, - 'resolve': False, - }) - - out, err = capsys.readouterr() - assert "Hashtags:\n#foo, #bar, #baz" in out - assert "Accounts:" in out - assert "@thequeen Freddy Mercury" in out - assert "@thequeen@other.instance Mercury Freddy" in out - - -@mock.patch('toot.http.post') -@mock.patch('toot.http.get') -def test_follow(mock_get, mock_post, capsys): - mock_get.return_value = MockResponse({ - "accounts": [ - {"id": 123, "acct": "blixa@other.acc"}, - {"id": 321, "acct": "blixa"}, - ] - }) - mock_post.return_value = MockResponse() - - console.run_command(app, user, 'follow', ['blixa']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', {'q': 'blixa', 'type': 'accounts', 'resolve': True}) - mock_post.assert_called_once_with(app, user, '/api/v1/accounts/321/follow') - - out, err = capsys.readouterr() - assert "You are now following blixa" in out - - -@mock.patch('toot.http.post') -@mock.patch('toot.http.get') -def test_follow_case_insensitive(mock_get, mock_post, capsys): - mock_get.return_value = MockResponse({ - "accounts": [ - {"id": 123, "acct": "blixa@other.acc"}, - {"id": 321, "acct": "blixa"}, - ] - }) - mock_post.return_value = MockResponse() - - console.run_command(app, user, 'follow', ['bLiXa@oThEr.aCc']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', {'q': 'bLiXa@oThEr.aCc', 'type': 'accounts', 'resolve': True}) - mock_post.assert_called_once_with(app, user, '/api/v1/accounts/123/follow') - - out, err = capsys.readouterr() - assert "You are now following bLiXa@oThEr.aCc" in out - - -@mock.patch('toot.http.get') -def test_follow_not_found(mock_get, capsys): - mock_get.return_value = MockResponse({"accounts": []}) - - with pytest.raises(ConsoleError) as ex: - console.run_command(app, user, 'follow', ['blixa']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', {'q': 'blixa', 'type': 'accounts', 'resolve': True}) - assert "Account not found" == str(ex.value) - - -@mock.patch('toot.http.post') -@mock.patch('toot.http.get') -def test_unfollow(mock_get, mock_post, capsys): - mock_get.return_value = MockResponse({ - "accounts": [ - {'id': 123, 'acct': 'blixa@other.acc'}, - {'id': 321, 'acct': 'blixa'}, - ] - }) - - mock_post.return_value = MockResponse() - - console.run_command(app, user, 'unfollow', ['blixa']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', {'q': 'blixa', 'type': 'accounts', 'resolve': True}) - mock_post.assert_called_once_with(app, user, '/api/v1/accounts/321/unfollow') - - out, err = capsys.readouterr() - assert "You are no longer following blixa" in out - - -@mock.patch('toot.http.get') -def test_unfollow_not_found(mock_get, capsys): - mock_get.return_value = MockResponse({"accounts": []}) - - with pytest.raises(ConsoleError) as ex: - console.run_command(app, user, 'unfollow', ['blixa']) - - mock_get.assert_called_once_with(app, user, '/api/v2/search', {'q': 'blixa', 'type': 'accounts', 'resolve': True}) - - assert "Account not found" == str(ex.value) - - -@mock.patch('toot.http.get') -def test_whoami(mock_get, capsys): - mock_get.return_value = MockResponse({ - 'acct': 'ihabunek', - 'avatar': 'https://files.mastodon.social/accounts/avatars/000/046/103/original/6a1304e135cac514.jpg?1491312434', - 'avatar_static': 'https://files.mastodon.social/accounts/avatars/000/046/103/original/6a1304e135cac514.jpg?1491312434', - 'created_at': '2017-04-04T13:23:09.777Z', - 'display_name': 'Ivan Habunek', - 'followers_count': 5, - 'following_count': 9, - 'header': '/headers/original/missing.png', - 'header_static': '/headers/original/missing.png', - 'id': 46103, - 'locked': False, - 'note': 'A developer.', - 'statuses_count': 19, - 'url': 'https://mastodon.social/@ihabunek', - 'username': 'ihabunek' - }) - - console.run_command(app, user, 'whoami', []) - - mock_get.assert_called_once_with(app, user, '/api/v1/accounts/verify_credentials') - - out, err = capsys.readouterr() - out = uncolorize(out) - - assert "@ihabunek Ivan Habunek" in out - assert "A developer." in out - assert "https://mastodon.social/@ihabunek" in out - assert "ID: 46103" in out - assert "Since: 2017-04-04" in out - assert "Followers: 5" in out - assert "Following: 9" in out - assert "Statuses: 19" in out - - -@mock.patch('toot.http.get') -def test_notifications(mock_get, capsys): - mock_get.return_value = MockResponse([{ - 'id': '1', - 'type': 'follow', - 'created_at': '2019-02-16T07:01:20.714Z', - 'account': { - 'display_name': 'Frank Zappa', - 'acct': 'frank@zappa.social', - }, - }, { - 'id': '2', - 'type': 'mention', - 'created_at': '2017-01-12T12:12:12.0Z', - 'account': { - 'display_name': 'Dweezil Zappa', - 'acct': 'dweezil@zappa.social', - }, - 'status': { - 'id': '111111111111111111', - 'account': { - 'display_name': 'Dweezil Zappa', - 'acct': 'dweezil@zappa.social', - }, - 'created_at': '2017-04-12T15:53:18.174Z', - 'content': "We still have fans in 2017 @fan123
", - 'reblog': None, - 'in_reply_to_id': None, - 'media_attachments': [], - }, - }, { - 'id': '3', - 'type': 'reblog', - 'created_at': '1983-11-03T03:03:03.333Z', - 'account': { - 'display_name': 'Terry Bozzio', - 'acct': 'terry@bozzio.social', - }, - 'status': { - 'id': '1234', - 'account': { - 'display_name': 'Zappa Fan', - 'acct': 'fan123@zappa-fans.social' - }, - 'created_at': '1983-11-04T15:53:18.174Z', - 'content': "The Black Page, a masterpiece
", - 'reblog': None, - 'in_reply_to_id': None, - 'media_attachments': [], - }, - }, { - 'id': '4', - 'type': 'favourite', - 'created_at': '1983-12-13T01:02:03.444Z', - 'account': { - 'display_name': 'Zappa Old Fan', - 'acct': 'fan9@zappa-fans.social', - }, - 'status': { - 'id': '1234', - 'account': { - 'display_name': 'Zappa Fan', - 'acct': 'fan123@zappa-fans.social' - }, - 'created_at': '1983-11-04T15:53:18.174Z', - 'content': "The Black Page, a masterpiece
", - 'reblog': None, - 'in_reply_to_id': None, - 'media_attachments': [], - }, - }]) - - console.run_command(app, user, 'notifications', []) - - mock_get.assert_called_once_with(app, user, '/api/v1/notifications', {'exclude_types[]': [], 'limit': 20}) - - out, err = capsys.readouterr() - out = uncolorize(out) - - assert not err - assert out == "\n".join([ - "────────────────────────────────────────────────────────────────────────────────────────────────────", - "Frank Zappa @frank@zappa.social now follows you", - "────────────────────────────────────────────────────────────────────────────────────────────────────", - "Dweezil Zappa @dweezil@zappa.social mentioned you in", - "Dweezil Zappa @dweezil@zappa.social 2017-04-12 15:53 UTC", - "", - "We still have fans in 2017 @fan123", - "", - "ID 111111111111111111 ", - "────────────────────────────────────────────────────────────────────────────────────────────────────", - "Terry Bozzio @terry@bozzio.social reblogged your status", - "Zappa Fan @fan123@zappa-fans.social 1983-11-04 15:53 UTC", - "", - "The Black Page, a masterpiece", - "", - "ID 1234 ", - "────────────────────────────────────────────────────────────────────────────────────────────────────", - "Zappa Old Fan @fan9@zappa-fans.social favourited your status", - "Zappa Fan @fan123@zappa-fans.social 1983-11-04 15:53 UTC", - "", - "The Black Page, a masterpiece", - "", - "ID 1234 ", - "────────────────────────────────────────────────────────────────────────────────────────────────────", - "", - ]) - - -@mock.patch('toot.http.get') -def test_notifications_empty(mock_get, capsys): - mock_get.return_value = MockResponse([]) - - console.run_command(app, user, 'notifications', []) - - mock_get.assert_called_once_with(app, user, '/api/v1/notifications', {'exclude_types[]': [], 'limit': 20}) - - out, err = capsys.readouterr() - out = uncolorize(out) - - assert not err - assert out == "No notification\n" - - -@mock.patch('toot.http.post') -def test_notifications_clear(mock_post, capsys): - console.run_command(app, user, 'notifications', ['--clear']) - out, err = capsys.readouterr() - out = uncolorize(out) - - mock_post.assert_called_once_with(app, user, '/api/v1/notifications/clear') - assert not err - assert out == 'Cleared notifications\n' - - -def u(user_id, access_token="abc"): - username, instance = user_id.split("@") - return { - "instance": instance, - "username": username, - "access_token": access_token, - } - - -@mock.patch('toot.config.save_config') -@mock.patch('toot.config.load_config') -def test_logout(mock_load, mock_save, capsys): - mock_load.return_value = { - "users": { - "king@gizzard.social": u("king@gizzard.social"), - "lizard@wizard.social": u("lizard@wizard.social"), - }, - "active_user": "king@gizzard.social", - } - - console.run_command(app, user, "logout", ["king@gizzard.social"]) - - mock_save.assert_called_once_with({ - 'users': { - 'lizard@wizard.social': u("lizard@wizard.social") - }, - 'active_user': None - }) - - out, err = capsys.readouterr() - assert "✓ User king@gizzard.social logged out" in out - - -@mock.patch('toot.config.save_config') -@mock.patch('toot.config.load_config') -def test_activate(mock_load, mock_save, capsys): - mock_load.return_value = { - "users": { - "king@gizzard.social": u("king@gizzard.social"), - "lizard@wizard.social": u("lizard@wizard.social"), - }, - "active_user": "king@gizzard.social", - } - - console.run_command(app, user, "activate", ["lizard@wizard.social"]) - - mock_save.assert_called_once_with({ - 'users': { - "king@gizzard.social": u("king@gizzard.social"), - 'lizard@wizard.social': u("lizard@wizard.social") - }, - 'active_user': "lizard@wizard.social" - }) - - out, err = capsys.readouterr() - assert "✓ User lizard@wizard.social active" in out diff --git a/tests/test_integration.py b/tests/test_integration.py deleted file mode 100644 index 49a7a78..0000000 --- a/tests/test_integration.py +++ /dev/null @@ -1,518 +0,0 @@ -""" -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="dbname=mastodon_development" -``` -""" - -import os -import psycopg2 -import pytest -import re -import time -import uuid - -from datetime import datetime, timedelta, timezone -from os import path -from toot import CLIENT_NAME, CLIENT_WEBSITE, api, App, User -from toot.console import run_command -from toot.exceptions import ConsoleError, NotFoundError -from toot.utils import get_text -from unittest import mock - -# 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) - - -@pytest.fixture(scope="session") -def friend(app): - return register_account(app) - - -@pytest.fixture -def run(app, user, capsys): - def _run(command, *params, as_user=None): - run_command(app, as_user or user, command, params or []) - out, err = capsys.readouterr() - assert err == "" - return strip_ansi(out) - return _run - - -@pytest.fixture -def run_anon(capsys): - def _run(command, *params): - run_command(None, None, command, params or []) - out, err = capsys.readouterr() - assert err == "" - return strip_ansi(out) - return _run - -# ------------------------------------------------------------------------------ -# Tests -# ------------------------------------------------------------------------------ - - -def test_instance(app, run): - out = run("instance", "--disable-https") - assert "Mastodon" in out - assert app.instance in out - assert "running Mastodon" in out - - -def test_instance_anon(app, run_anon): - out = run_anon("instance", "--disable-https", HOSTNAME) - assert "Mastodon" in out - assert app.instance in out - assert "running Mastodon" in out - - # Need to specify the instance name when running anon - with pytest.raises(ConsoleError) as exc: - run_anon("instance") - assert str(exc.value) == "Please specify instance name." - - -def test_post(app, user, run): - text = "i wish i was a #lumberjack" - out = run("post", text) - status_id = _posted_status_id(out) - - status = api.fetch_status(app, user, status_id) - assert text == get_text(status["content"]) - assert status["visibility"] == "public" - assert status["sensitive"] is False - assert status["spoiler_text"] == "" - - # Pleroma doesn't return the application - if status["application"]: - assert status["application"]["name"] == CLIENT_NAME - assert status["application"]["website"] == CLIENT_WEBSITE - - -def test_post_visibility(app, user, run): - for visibility in ["public", "unlisted", "private", "direct"]: - out = run("post", "foo", "--visibility", visibility) - status_id = _posted_status_id(out) - status = api.fetch_status(app, user, status_id) - assert status["visibility"] == visibility - - -def test_post_scheduled_at(app, user, run): - text = str(uuid.uuid4()) - scheduled_at = datetime.now(timezone.utc).replace(microsecond=0) + timedelta(minutes=10) - - out = run("post", text, "--scheduled-at", scheduled_at.isoformat()) - assert "Toot scheduled for" in out - - statuses = api.scheduled_statuses(app, user) - [status] = [s for s in statuses if s["params"]["text"] == text] - assert datetime.strptime(status["scheduled_at"], "%Y-%m-%dT%H:%M:%S.%f%z") == scheduled_at - - -def test_post_scheduled_in(app, user, run): - text = str(uuid.uuid4()) - - variants = [ - ("1 day", timedelta(days=1)), - ("1 day 6 hours", timedelta(days=1, hours=6)), - ("1 day 6 hours 13 minutes", timedelta(days=1, hours=6, minutes=13)), - ("1 day 6 hours 13 minutes 51 second", timedelta(days=1, hours=6, minutes=13, seconds=51)), - ("2d", timedelta(days=2)), - ("2d6h", timedelta(days=2, hours=6)), - ("2d6h13m", timedelta(days=2, hours=6, minutes=13)), - ("2d6h13m51s", timedelta(days=2, hours=6, minutes=13, seconds=51)), - ] - - datetimes = [] - for scheduled_in, delta in variants: - out = run("post", text, "--scheduled-in", scheduled_in) - dttm = datetime.utcnow() + delta - assert out.startswith(f"Toot scheduled for: {str(dttm)[:16]}") - datetimes.append(dttm) - - scheduled = api.scheduled_statuses(app, user) - scheduled = [s for s in scheduled if s["params"]["text"] == text] - scheduled = sorted(scheduled, key=lambda s: s["scheduled_at"]) - assert len(scheduled) == 8 - - for expected, status in zip(datetimes, scheduled): - actual = datetime.strptime(status["scheduled_at"], "%Y-%m-%dT%H:%M:%S.%fZ") - delta = expected - actual - assert delta.total_seconds() < 5 - - -def test_post_language(app, user, run): - out = run("post", "test", "--language", "hr") - status_id = _posted_status_id(out) - status = api.fetch_status(app, user, status_id) - assert status["language"] == "hr" - - out = run("post", "test", "--language", "zh") - status_id = _posted_status_id(out) - status = api.fetch_status(app, user, status_id) - assert status["language"] == "zh" - - -def test_media_attachments(app, user, run): - 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") - - out = run( - "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(out) - status = api.fetch_status(app, user, status_id) - - [a1, a2, a3, a4] = status["media_attachments"] - - # Pleroma doesn't send metadata - if "meta" in a1: - 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" - - -@mock.patch("toot.utils.multiline_input") -@mock.patch("sys.stdin.read") -def test_media_attachment_without_text(mock_read, mock_ml, app, user, run): - # No status from stdin or readline - mock_read.return_value = "" - mock_ml.return_value = "" - - assets_dir = path.realpath(path.join(path.dirname(__file__), "assets")) - media_path = path.join(assets_dir, "test1.png") - - out = run("post", "--media", media_path) - status_id = _posted_status_id(out) - - status = api.fetch_status(app, user, status_id) - assert status["content"] == "" - - [attachment] = status["media_attachments"] - assert not attachment["description"] - - # Pleroma doesn't send metadata - if "meta" in attachment: - assert attachment["meta"]["original"]["size"] == "50x50" - - -def test_delete_status(app, user, run): - status = api.post_status(app, user, "foo") - - out = run("delete", status["id"]) - assert out == "✓ Status deleted" - - with pytest.raises(NotFoundError): - api.fetch_status(app, user, status["id"]) - - -def test_reply_thread(app, user, friend, run): - status = api.post_status(app, friend, "This is the status") - - out = run("post", "--reply-to", status["id"], "This is the reply") - status_id = _posted_status_id(out) - reply = api.fetch_status(app, user, status_id) - - assert reply["in_reply_to_id"] == status["id"] - - out = run("thread", status["id"]) - [s1, s2] = [s.strip() for s in re.split(r"─+", out) if s.strip()] - - assert "This is the status" in s1 - assert "This is the reply" in s2 - assert friend.username in s1 - assert user.username in s2 - assert status["id"] in s1 - assert reply["id"] in s2 - - -def test_favourite(app, user, run): - status = api.post_status(app, user, "foo") - assert not status["favourited"] - - out = run("favourite", status["id"]) - assert out == "✓ Status favourited" - - status = api.fetch_status(app, user, status["id"]) - assert status["favourited"] - - out = run("unfavourite", status["id"]) - assert out == "✓ Status unfavourited" - - # A short delay is required before the server returns new data - time.sleep(0.1) - - status = api.fetch_status(app, user, status["id"]) - assert not status["favourited"] - - -def test_reblog(app, user, run): - status = api.post_status(app, user, "foo") - assert not status["reblogged"] - - out = run("reblog", status["id"]) - assert out == "✓ Status reblogged" - - status = api.fetch_status(app, user, status["id"]) - assert status["reblogged"] - - out = run("reblogged_by", status["id"]) - assert out == f"@{user.username}" - - out = run("unreblog", status["id"]) - assert out == "✓ Status unreblogged" - - status = api.fetch_status(app, user, status["id"]) - assert not status["reblogged"] - - -def test_pin(app, user, run): - status = api.post_status(app, user, "foo") - assert not status["pinned"] - - out = run("pin", status["id"]) - assert out == "✓ Status pinned" - - status = api.fetch_status(app, user, status["id"]) - assert status["pinned"] - - out = run("unpin", status["id"]) - assert out == "✓ Status unpinned" - - status = api.fetch_status(app, user, status["id"]) - assert not status["pinned"] - - -def test_bookmark(app, user, run): - status = api.post_status(app, user, "foo") - assert not status["bookmarked"] - - out = run("bookmark", status["id"]) - assert out == "✓ Status bookmarked" - - status = api.fetch_status(app, user, status["id"]) - assert status["bookmarked"] - - out = run("unbookmark", status["id"]) - assert out == "✓ Status unbookmarked" - - status = api.fetch_status(app, user, status["id"]) - assert not status["bookmarked"] - - -def test_whoami(user, run): - out = run("whoami") - # TODO: test other fields once updating account is supported - assert f"@{user.username}" in out - assert f"http://{HOSTNAME}/@{user.username}" in out - - -def test_whois(app, friend, run): - variants = [ - friend.username, - f"@{friend.username}", - f"{friend.username}@{app.instance}", - f"@{friend.username}@{app.instance}", - ] - - for username in variants: - out = run("whois", username) - assert f"@{friend.username}" in out - assert f"http://{HOSTNAME}/@{friend.username}" in out - - -def test_search_account(friend, run): - out = run("search", friend.username) - assert out == f"Accounts:\n* @{friend.username}" - - -def test_search_hashtag(app, user, run): - api.post_status(app, user, "#hashtag_x") - api.post_status(app, user, "#hashtag_y") - api.post_status(app, user, "#hashtag_z") - - out = run("search", "#hashtag") - assert out == "Hashtags:\n#hashtag_x, #hashtag_y, #hashtag_z" - - -def test_follow(friend, run): - out = run("follow", friend.username) - assert out == f"✓ You are now following {friend.username}" - - out = run("unfollow", friend.username) - assert out == f"✓ You are no longer following {friend.username}" - - -def test_follow_case_insensitive(friend, run): - username = friend.username.upper() - - out = run("follow", username) - assert out == f"✓ You are now following {username}" - - out = run("unfollow", username) - assert out == f"✓ You are no longer following {username}" - - -# TODO: improve testing stderr, catching exceptions is not optimal -def test_follow_not_found(run): - with pytest.raises(ConsoleError) as ex_info: - run("follow", "banana") - assert str(ex_info.value) == "Account not found" - - -def test_mute(app, user, friend, run): - out = run("mute", friend.username) - assert out == f"✓ You have muted {friend.username}" - - [muted_account] = api.get_muted_accounts(app, user) - assert muted_account["acct"] == friend.username - - out = run("unmute", friend.username) - assert out == f"✓ {friend.username} is no longer muted" - - assert api.get_muted_accounts(app, user) == [] - - -def test_block(app, user, friend, run): - out = run("block", friend.username) - assert out == f"✓ You are now blocking {friend.username}" - - [blockd_account] = api.get_blocked_accounts(app, user) - assert blockd_account["acct"] == friend.username - - out = run("unblock", friend.username) - assert out == f"✓ {friend.username} is no longer blocked" - - assert api.get_blocked_accounts(app, user) == [] - - -def test_following_followers(user, friend, run): - out = run("following", user.username) - assert out == "" - - run("follow", friend.username) - - out = run("following", user.username) - assert out == f"* @{friend.username}" - - out = run("followers", friend.username) - assert out == f"* @{user.username}" - - -def test_tags(run): - out = run("tags_followed") - assert out == "You're not following any hashtags." - - out = run("tags_follow", "foo") - assert out == "✓ You are now following #foo" - - out = run("tags_followed") - assert out == "* #foo\thttp://localhost:3000/tags/foo" - - out = run("tags_follow", "bar") - assert out == "✓ You are now following #bar" - - out = run("tags_followed") - assert out == "\n".join([ - "* #bar\thttp://localhost:3000/tags/bar", - "* #foo\thttp://localhost:3000/tags/foo", - ]) - - out = run("tags_unfollow", "foo") - assert out == "✓ You are no longer following #foo" - - out = run("tags_followed") - assert out == "* #bar\thttp://localhost:3000/tags/bar" - - -# ------------------------------------------------------------------------------ -# 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(out): - 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 diff --git a/tests/test_output.py b/tests/test_output.py deleted file mode 100644 index cc31e5c..0000000 --- a/tests/test_output.py +++ /dev/null @@ -1,26 +0,0 @@ -from toot.output import colorize, strip_tags, STYLES - -reset = STYLES["reset"] -red = STYLES["red"] -green = STYLES["green"] -bold = STYLES["bold"] - - -def test_colorize(): - assert colorize("foo") == "foo" - assert colorize("foo
+foo bar baz
+ """.strip() + + [foo, divider, bar] = html_to_widgets(html) + + assert isinstance(foo, Pile) + assert isinstance(divider, Divider) + assert isinstance(bar, Pile) + + [(foo_embed, _)] = foo.contents + assert foo_embed.embedded == [] + assert foo_embed.attrib == [] + assert foo_embed.text == "foo" + + [(bar_embed, _)] = bar.contents + assert bar_embed.embedded == [] + assert bar_embed.attrib == [(None, 4), ("b", 3), (None, 1), ("i", 3)] + assert bar_embed.text == "foo bar baz" diff --git a/tests/utils.py b/tests/utils.py index cdae09c..817bdb9 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -2,6 +2,9 @@ Helpers for testing. """ +import time +from typing import Any, Callable + class MockResponse: def __init__(self, response_data={}, ok=True, is_redirect=False): @@ -19,3 +22,23 @@ class MockResponse: def retval(val): return lambda *args, **kwargs: val + + +def run_with_retries(fn: Callable[..., Any]): + """ + Run the the given function repeatedly until it finishes without raising an + AssertionError. Sleep a bit between attempts. If the function doesn't + succeed in the given number of tries raises the AssertionError. Used for + tests which should eventually succeed. + """ + + # Wait upto 6 seconds with incrementally longer sleeps + delays = [0.1, 0.2, 0.3, 0.4, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5] + + for delay in delays: + try: + return fn() + except AssertionError: + time.sleep(delay) + + fn() diff --git a/toot/__init__.py b/toot/__init__.py index 459cafc..2a3f4ab 100644 --- a/toot/__init__.py +++ b/toot/__init__.py @@ -1,11 +1,45 @@ -from collections import namedtuple +import os +import sys -__version__ = '0.33.1' +from os.path import join, expanduser +from typing import NamedTuple -App = namedtuple('App', ['instance', 'base_url', 'client_id', 'client_secret']) -User = namedtuple('User', ['instance', 'username', 'access_token']) +__version__ = '0.41.1' -DEFAULT_INSTANCE = 'mastodon.social' + +class App(NamedTuple): + instance: str + base_url: str + client_id: str + client_secret: str + + +class User(NamedTuple): + instance: str + username: str + access_token: str + + +DEFAULT_INSTANCE = 'https://mastodon.social' CLIENT_NAME = 'toot - a Mastodon CLI client' CLIENT_WEBSITE = 'https://github.com/ihabunek/toot' + +TOOT_CONFIG_DIR_NAME = "toot" + + +def get_config_dir(): + """Returns the path to toot config directory""" + + # On Windows, store the config in roaming appdata + if sys.platform == "win32" and "APPDATA" in os.environ: + return join(os.getenv("APPDATA"), TOOT_CONFIG_DIR_NAME) + + # Respect XDG_CONFIG_HOME env variable if set + # https://specifications.freedesktop.org/basedir-spec/basedir-spec-latest.html + if "XDG_CONFIG_HOME" in os.environ: + config_home = expanduser(os.environ["XDG_CONFIG_HOME"]) + return join(config_home, TOOT_CONFIG_DIR_NAME) + + # Default to ~/.config/toot/ + return join(expanduser("~"), ".config", TOOT_CONFIG_DIR_NAME) diff --git a/toot/__main__.py b/toot/__main__.py new file mode 100644 index 0000000..fa3e807 --- /dev/null +++ b/toot/__main__.py @@ -0,0 +1,3 @@ +from toot.cli import cli + +cli() diff --git a/toot/api.py b/toot/api.py index df008d5..470eabb 100644 --- a/toot/api.py +++ b/toot/api.py @@ -1,28 +1,56 @@ +import mimetypes import re import uuid +from os import path +from requests import Response +from typing import BinaryIO, List, Optional from urllib.parse import urlparse, urlencode, quote -from toot import http, CLIENT_NAME, CLIENT_WEBSITE -from toot.exceptions import AuthenticationError, ApiError -from toot.utils import str_bool +from toot import App, User, http, CLIENT_NAME, CLIENT_WEBSITE +from toot.exceptions import AuthenticationError, ApiError, ConsoleError +from toot.utils import drop_empty_values, str_bool, str_bool_nullable + SCOPES = 'read write follow' -def _account_action(app, user, account, action): +def find_account(app, user, account_name): + if not account_name: + raise ConsoleError("Empty account name given") + + normalized_name = account_name.lstrip("@").lower() + + # Strip @{html}
", recovery_attempt=True) + else: + continue + else: + name = e.name + # if our HTML starts with a tag, but not a block tag + # the HTML is out of spec. Attempt a fix by wrapping the + # HTML with + if (first_tag and not recovery_attempt and name not in BLOCK_TAGS): + return html_to_widgets(f"{html}
", recovery_attempt=True) + + markup = render(name, e) + first_tag = False + + if not isinstance(markup, urwid.Widget): + # plaintext, so create a padded text widget + txt = text_to_widget("", markup) + markup = urwid.Padding( + txt, + align="left", + width=("relative", 100), + min_width=None, + ) + widgets.append(markup) + # separate top level widgets with a blank line + widgets.append(urwid.Divider(" ")) + return widgets[:-1] # but suppress the last blank line + + +def url_to_widget(url: str): + widget = len(url), urwid.Filler(Hyperlink(url, "link", url)) + return TextEmbed(widget) + + +def inline_tag_to_text(tag) -> Tuple: + """Convert html tag to plain text with tag as attributes recursively""" + markups = process_inline_tag_children(tag) + if not markups: + return (tag.name, "") + return (tag.name, markups) + + +def process_inline_tag_children(tag) -> List: + """Recursively retrieve all children + and convert to a list of markup text""" + markups = [] + for child in tag.children: + if isinstance(child, Tag): + markup = render(child.name, child) + markups.append(markup) + else: + markups.append(child) + return markups + + +URL_PATTERN = re.compile(r"(^.+)\x03(.+$)") + + +def text_to_widget(attr, markup) -> urwid.Widget: + markup_list = [] + for run in markup: + if isinstance(run, tuple): + txt, attr_list = decompose_tagmarkup(run) + # find anchor titles with an ETX separator followed by href + match = URL_PATTERN.match(txt) + if match: + label, url = match.groups() + anchor_attr = get_best_anchor_attr(attr_list) + markup_list.append(( + len(label), + urwid.Filler(Hyperlink(url, anchor_attr, label)), + )) + else: + markup_list.append(run) + else: + markup_list.append(run) + + return TextEmbed(markup_list) + + +def process_block_tag_children(tag) -> List[urwid.Widget]: + """Recursively retrieve all children + and convert to a list of widgets + any inline tags containing text will be + converted to Text widgets""" + + pre_widget_markups = [] + post_widget_markups = [] + child_widgets = [] + found_nested_widget = False + + for child in tag.children: + if isinstance(child, Tag): + # child is a nested tag; process using custom method + # or default to inline_tag_to_text + result = render(child.name, child) + if isinstance(result, urwid.Widget): + found_nested_widget = True + child_widgets.append(result) + else: + if not found_nested_widget: + pre_widget_markups.append(result) + else: + post_widget_markups.append(result) + else: + # child is text; append to the appropriate markup list + if not found_nested_widget: + pre_widget_markups.append(child) + else: + post_widget_markups.append(child) + + widget_list = [] + if len(pre_widget_markups): + widget_list.append(text_to_widget(tag.name, pre_widget_markups)) + + if len(child_widgets): + widget_list += child_widgets + + if len(post_widget_markups): + widget_list.append(text_to_widget(tag.name, post_widget_markups)) + + return widget_list + + +def get_urwid_attr_name(tag) -> str: + """Get the class name and translate to a + name suitable for use as an urwid + text attribute name""" + + if "class" in tag.attrs: + clss = tag.attrs["class"] + if len(clss) > 0: + style_name = "class_" + "_".join(clss) + # return the class name, only if we + # find it as a defined palette name + if style_name in STYLE_NAMES: + return style_name + + # fallback to returning the tag name + return tag.name + + +def basic_block_tag_handler(tag) -> urwid.Widget: + """default for block tags that need no special treatment""" + return urwid.Pile(process_block_tag_children(tag)) + + +def get_best_anchor_attr(attrib_list) -> str: + if not attrib_list: + return "" + flat_al = list(flatten(attrib_list)) + + for a in flat_al[0]: + # ref: https://docs.joinmastodon.org/spec/activitypub/ + # these are the class names (translated to attrib names) + # that we can support for display + + try: + if a[0] in ["class_hashtag", "class_mention_hashtag", "class_mention"]: + return a[0] + except KeyError: + continue + + return "a" + + +def render(attr: str, content: str): + if attr in ["a"]: + return render_anchor(content) + + if attr in ["blockquote"]: + return render_blockquote(content) + + if attr in ["br"]: + return render_br(content) + + if attr in ["em"]: + return render_em(content) + + if attr in ["ol"]: + return render_ol(content) + + if attr in ["pre"]: + return render_pre(content) + + if attr in ["span"]: + return render_span(content) + + if attr in ["b", "strong"]: + return render_strong(content) + + if attr in ["ul"]: + return render_ul(content) + + # Glitch-soc and Pleroma allow+ if attr in ["p", "div", "li", "h1", "h2", "h3", "h4", "h5", "h6"]: + return basic_block_tag_handler(content) + + # Fall back to inline_tag_to_text handler + return inline_tag_to_text(content) + + +def render_anchor(tag) -> Tuple: + """anchor tag handler""" + + markups = process_inline_tag_children(tag) + if not markups: + return (tag.name, "") + + href = tag.attrs["href"] + title, attrib_list = decompose_tagmarkup(markups) + if not attrib_list: + attrib_list = [tag] + if href: + # urlencode the path and query portions of the URL + href = urlencode_url(href) + # use ASCII ETX (end of record) as a + # delimiter between the title and the HREF + title += f"\x03{href}" + + attr = get_best_anchor_attr(attrib_list) + + if attr == "a": + # didn't find an attribute to use + # in the child markup, so let's + # try the anchor tag's own attributes + + attr = get_urwid_attr_name(tag) + + # hashtag anchors have a class of "mention hashtag" + # or "hashtag" + # we'll return style "class_mention_hashtag" + # or "class_hashtag" + # in that case; see corresponding palette entry + # in constants.py controlling hashtag highlighting + + return (attr, title) + + +def render_blockquote(tag) -> urwid.Widget: + widget_list = process_block_tag_children(tag) + blockquote_widget = urwid.LineBox( + urwid.Padding( + urwid.Pile(widget_list), + align="left", + width=("relative", 100), + min_width=None, + left=1, + right=1, + ), + tlcorner="", + tline="", + lline="│", + trcorner="", + blcorner="", + rline="", + bline="", + brcorner="", + ) + return urwid.Pile([urwid.AttrMap(blockquote_widget, "blockquote")]) + + +def render_br(tag) -> Tuple: + return ("br", "\n") + + +def render_em(tag) -> Tuple: + # to simplify the number of palette entries + # translate EM to I (italic) + markups = process_inline_tag_children(tag) + if not markups: + return ("i", "") + + # special case processing for bold and italic + for parent in tag.parents: + if parent.name == "b" or parent.name == "strong": + return ("bi", markups) + + return ("i", markups) + + +def render_ol(tag) -> urwid.Widget: + """ordered list tag handler""" + + widgets = [] + list_item_num = 1 + increment = -1 if tag.has_attr("reversed") else 1 + + # get ol start= attribute if present + if tag.has_attr("start") and len(tag.attrs["start"]) > 0: + try: + list_item_num = int(tag.attrs["start"]) + except ValueError: + pass + + for li in tag.find_all("li", recursive=False): + markup = render("li", li) + + # li value= attribute will change the item number + # it also overrides any ol start= attribute + + if li.has_attr("value") and len(li.attrs["value"]) > 0: + try: + list_item_num = int(li.attrs["value"]) + except ValueError: + pass + + if not isinstance(markup, urwid.Widget): + txt = text_to_widget("li", [str(list_item_num), ". ", markup]) + # 1. foo, 2. bar, etc. + widgets.append(txt) + else: + txt = text_to_widget("li", [str(list_item_num), ". "]) + columns = urwid.Columns( + [txt, ("weight", 9999, markup)], dividechars=1, min_width=3 + ) + widgets.append(columns) + + list_item_num += increment + + return urwid.Pile(widgets) + + +def render_pre(tag) -> urwid.Widget: + #
tag spec says that text should not wrap,
+ # but horizontal screen space is at a premium
+ # and we have no horizontal scroll bar, so allow
+ # wrapping.
+
+ widget_list = [urwid.Divider(" ")]
+ widget_list += process_block_tag_children(tag)
+
+ pre_widget = urwid.Padding(
+ urwid.Pile(widget_list),
+ align="left",
+ width=("relative", 100),
+ min_width=None,
+ left=1,
+ right=1,
+ )
+ return urwid.Pile([urwid.AttrMap(pre_widget, "pre")])
+
+
+def render_span(tag) -> Tuple:
+ markups = process_inline_tag_children(tag)
+
+ if not markups:
+ return (tag.name, "")
+
+ # span inherits its parent's class definition
+ # unless it has a specific class definition
+ # of its own
+
+ if "class" in tag.attrs:
+ # uncomment the following code to hide all HTML marked
+ # invisible (generally, the http:// prefix of URLs)
+ # could be a user preference, it's only advisable if
+ # the terminal supports OCS 8 hyperlinks (and that's not
+ # automatically detectable)
+
+ # if "invisible" in tag.attrs["class"]:
+ # return (tag.name, "")
+
+ style_name = get_urwid_attr_name(tag)
+
+ if style_name != "span":
+ # unique class name matches an entry in our palette
+ return (style_name, markups)
+
+ if tag.parent:
+ return (get_urwid_attr_name(tag.parent), markups)
+ else:
+ # fallback
+ return ("span", markups)
+
+
+def render_strong(tag) -> Tuple:
+ # to simplify the number of palette entries
+ # translate STRONG to B (bold)
+ markups = process_inline_tag_children(tag)
+ if not markups:
+ return ("b", "")
+
+ # special case processing for bold and italic
+ for parent in tag.parents:
+ if parent.name == "i" or parent.name == "em":
+ return ("bi", markups)
+
+ return ("b", markups)
+
+
+def render_ul(tag) -> urwid.Widget:
+ """unordered list tag handler"""
+
+ widgets = []
+
+ for li in tag.find_all("li", recursive=False):
+ markup = render("li", li)
+
+ if not isinstance(markup, urwid.Widget):
+ txt = text_to_widget("li", ["\N{bullet} ", markup])
+ # * foo, * bar, etc.
+ widgets.append(txt)
+ else:
+ txt = text_to_widget("li", ["\N{bullet} "])
+ columns = urwid.Columns(
+ [txt, ("weight", 9999, markup)], dividechars=1, min_width=3
+ )
+ widgets.append(columns)
+
+ return urwid.Pile(widgets)
+
+
+def flatten(data):
+ if isinstance(data, tuple):
+ for x in data:
+ yield from flatten(x)
+ else:
+ yield data
diff --git a/toot/tui/scroll.py b/toot/tui/scroll.py
index fa2c3bb..e2bf9f0 100644
--- a/toot/tui/scroll.py
+++ b/toot/tui/scroll.py
@@ -1,8 +1,3 @@
-# scroll.py
-#
-# Copied from the stig project by rndusr@github
-# https://github.com/rndusr/stig
-#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
@@ -36,12 +31,12 @@ class Scrollable(urwid.WidgetDecoration):
def selectable(self):
return True
- def __init__(self, widget):
+ def __init__(self, widget, force_forward_keypress = False):
"""Box widget that makes a fixed or flow widget vertically scrollable
TODO: Focusable widgets are handled, including switching focus, but
possibly not intuitively, depending on the arrangement of widgets. When
- switching focus to a widget that is outside of the visible part of the
+ switching focus to a widget that is ouside of the visible part of the
original widget, the canvas scrolls up/down to the focused widget. It
would be better to scroll until the next focusable widget is in sight
first. But for that to work we must somehow obtain a list of focusable
@@ -54,6 +49,7 @@ class Scrollable(urwid.WidgetDecoration):
self._forward_keypress = None
self._old_cursor_coords = None
self._rows_max_cached = 0
+ self.force_forward_keypress = force_forward_keypress
self.__super.__init__(widget)
def render(self, size, focus=False):
@@ -111,6 +107,51 @@ class Scrollable(urwid.WidgetDecoration):
if canv_full.cursor is not None:
# Full canvas contains the cursor, but scrolled out of view
self._forward_keypress = False
+
+ # Reset cursor position on page/up down scrolling
+ try:
+ if hasattr(ow, "automove_cursor_on_scroll") and ow.automove_cursor_on_scroll:
+ pwi = 0
+ ch = 0
+ last_hidden = False
+ first_visible = False
+ for w,o in ow.contents:
+ wcanv = w.render((maxcol,))
+ wh = wcanv.rows()
+ if wh:
+ ch += wh
+
+ if not last_hidden and ch >= self._trim_top:
+ last_hidden = True
+
+ elif last_hidden:
+ if not first_visible:
+ first_visible = True
+
+ if w.selectable():
+ ow.focus_item = pwi
+
+ st = None
+ nf = ow.get_focus()
+ if hasattr(nf, "key_timeout"):
+ st = nf
+ elif hasattr(nf, "original_widget"):
+ no = nf.original_widget
+ if hasattr(no, "original_widget"):
+ st = no.original_widget
+ else:
+ if hasattr(no, "key_timeout"):
+ st = no
+
+ if st and hasattr(st, "key_timeout") and hasattr(st, "keypress") and callable(st.keypress):
+ st.keypress(None, None)
+
+ break
+
+ pwi += 1
+ except Exception as e:
+ pass
+
else:
# Original widget does not have a cursor, but may be selectable
@@ -132,7 +173,7 @@ class Scrollable(urwid.WidgetDecoration):
def keypress(self, size, key):
# Maybe offer key to original widget
- if self._forward_keypress:
+ if self._forward_keypress or self.force_forward_keypress:
ow = self._original_widget
ow_size = self._get_original_widget_size(size)
@@ -216,7 +257,7 @@ class Scrollable(urwid.WidgetDecoration):
# If the cursor was moved by the most recent keypress, adjust trim_top
# so that the new cursor position is within the displayed canvas part.
# But don't do this if the cursor is at the top/bottom edge so we can still scroll out
- if self._old_cursor_coords is not None and self._old_cursor_coords != canv.cursor:
+ if self._old_cursor_coords is not None and self._old_cursor_coords != canv.cursor and canv.cursor != None:
self._old_cursor_coords = None
curscol, cursrow = canv.cursor
if cursrow < self._trim_top:
@@ -227,10 +268,10 @@ class Scrollable(urwid.WidgetDecoration):
def _get_original_widget_size(self, size):
ow = self._original_widget
sizing = ow.sizing()
- if FIXED in sizing:
- return ()
- elif FLOW in sizing:
+ if FLOW in sizing:
return (size[0],)
+ elif FIXED in sizing:
+ return ()
def get_scrollpos(self, size=None, focus=False):
"""Current scrolling position
@@ -416,11 +457,14 @@ class ScrollBar(urwid.WidgetDecoration):
if not handled and hasattr(ow, 'set_scrollpos'):
if button == 4: # scroll wheel up
pos = ow.get_scrollpos(ow_size)
- ow.set_scrollpos(pos - 1)
+ newpos = pos - 1
+ if newpos < 0:
+ newpos = 0
+ ow.set_scrollpos(newpos)
return True
elif button == 5: # scroll wheel down
pos = ow.get_scrollpos(ow_size)
ow.set_scrollpos(pos + 1)
return True
- return False
\ No newline at end of file
+ return False
diff --git a/toot/tui/timeline.py b/toot/tui/timeline.py
index d4b9b74..b9311e7 100644
--- a/toot/tui/timeline.py
+++ b/toot/tui/timeline.py
@@ -1,17 +1,18 @@
import logging
-import sys
import urwid
import webbrowser
-from typing import Optional
+from typing import List, Optional
-from .entities import Status
-from .scroll import Scrollable, ScrollBar
-from .utils import highlight_hashtags, parse_datetime, highlight_keys
-from .widgets import SelectableText, SelectableColumns
-from toot.utils import format_content
+from toot.tui import app
+from toot.tui.richtext import html_to_widgets, url_to_widget
+from toot.utils.datetime import parse_datetime, time_ago
from toot.utils.language import language_name
-from toot.tui.utils import time_ago
+
+from toot.entities import Status
+from toot.tui.scroll import Scrollable, ScrollBar
+from toot.tui.utils import highlight_keys
+from toot.tui.widgets import SelectableText, SelectableColumns, RoundedLineBox
logger = logging.getLogger("toot")
@@ -21,33 +22,25 @@ class Timeline(urwid.Columns):
Displays a list of statuses to the left, and status details on the right.
"""
signals = [
- "close", # Close thread
- "compose", # Compose a new toot
- "delete", # Delete own status
- "favourite", # Favourite status
- "focus", # Focus changed
- "bookmark", # Bookmark status
- "media", # Display media attachments
- "menu", # Show a context menu
- "next", # Fetch more statuses
- "reblog", # Reblog status
- "reply", # Compose a reply to a status
- "source", # Show status source
- "links", # Show status links
- "thread", # Show thread for status
- "translate", # Translate status
- "save", # Save current timeline
- "zoom", # Open status in scrollable popup window
- "clear-screen", # Clear the screen (used internally)
+ "close", # Close thread
+ "focus", # Focus changed
+ "next", # Fetch more statuses
+ "save", # Save current timeline
]
- def __init__(self, name, statuses, can_translate, followed_tags=[], focus=0, is_thread=False):
+ def __init__(
+ self,
+ tui: "app.TUI",
+ name: str,
+ statuses: List[Status],
+ focus: int = 0,
+ is_thread: bool = False
+ ):
+ self.tui = tui
self.name = name
self.is_thread = is_thread
self.statuses = statuses
- self.can_translate = can_translate
self.status_list = self.build_status_list(statuses, focus=focus)
- self.followed_tags = followed_tags
try:
focused_status = statuses[focus]
@@ -59,16 +52,17 @@ class Timeline(urwid.Columns):
super().__init__([
("weight", 40, self.status_list),
- ("weight", 0, urwid.AttrWrap(urwid.SolidFill("│"), "blue_selected")),
+ ("weight", 0, urwid.AttrWrap(urwid.SolidFill("│"), "columns_divider")),
("weight", 60, status_widget),
])
def wrap_status_details(self, status_details: "StatusDetails") -> urwid.Widget:
- """Wrap StatusDetails widget with a scollbar and footer."""
+ """Wrap StatusDetails widget with a scrollbar and footer."""
+ self.status_detail_scrollable = Scrollable(urwid.Padding(status_details, right=1))
return urwid.Padding(
urwid.Frame(
body=ScrollBar(
- Scrollable(urwid.Padding(status_details, right=1)),
+ self.status_detail_scrollable,
thumb_char="\u2588",
trough_char="\u2591",
),
@@ -85,38 +79,45 @@ class Timeline(urwid.Columns):
return urwid.ListBox(walker)
def build_list_item(self, status):
- item = StatusListItem(status)
+ item = StatusListItem(status, self.tui.options.relative_datetimes)
urwid.connect_signal(item, "click", lambda *args:
- self._emit("menu", status))
+ self.tui.show_context_menu(status))
return urwid.AttrMap(item, None, focus_map={
- "blue": "green_selected",
- "green": "green_selected",
- "yellow": "green_selected",
- "cyan": "green_selected",
- "red": "green_selected",
- None: "green_selected",
+ "status_list_account": "status_list_selected",
+ "status_list_timestamp": "status_list_selected",
+ "highlight": "status_list_selected",
+ "dim": "status_list_selected",
+ None: "status_list_selected",
})
def get_option_text(self, status: Optional[Status]) -> Optional[urwid.Text]:
if not status:
return None
+ poll = status.original.data.get("poll")
+ show_media = status.original.data["media_attachments"] and self.tui.options.media_viewer
+
options = [
+ "[A]ccount" if not status.is_mine else "",
"[B]oost",
"[D]elete" if status.is_mine else "",
+ "[E]dit" if status.is_mine else "",
"B[o]okmark",
"[F]avourite",
"[V]iew",
"[T]hread" if not self.is_thread else "",
- "[L]inks",
+ "L[i]nks",
+ "[M]edia" if show_media else "",
"[R]eply",
+ "[P]oll" if poll and not poll["expired"] else "",
"So[u]rce",
"[Z]oom",
- "Tra[n]slate" if self.can_translate else "",
- "[H]elp",
+ "Tra[n]slate" if self.tui.can_translate else "",
+ "Cop[y]",
+ "Help([?])",
]
options = "\n" + " ".join(o for o in options if o)
- options = highlight_keys(options, "white_bold", "cyan")
+ options = highlight_keys(options, "shortcut_highlight", "shortcut")
return urwid.Text(options)
def get_focused_status(self):
@@ -146,7 +147,9 @@ class Timeline(urwid.Columns):
def refresh_status_details(self):
"""Redraws the details of the focused status."""
status = self.get_focused_status()
+ pos = self.status_detail_scrollable.get_scrollpos()
self.draw_status_details(status)
+ self.status_detail_scrollable.set_scrollpos(pos)
def draw_status_details(self, status):
self.status_details = StatusDetails(self, status)
@@ -169,24 +172,35 @@ class Timeline(urwid.Columns):
if index >= count:
self._emit("next")
+ if key in ("a", "A"):
+ account_id = status.original.data["account"]["id"]
+ self.tui.show_account(account_id)
+ return
+
if key in ("b", "B"):
- self._emit("reblog", status)
+ self.tui.async_toggle_reblog(self, status)
return
if key in ("c", "C"):
- self._emit("compose")
+ self.tui.show_compose()
return
if key in ("d", "D"):
- self._emit("delete", status)
+ if status.is_mine:
+ self.tui.show_delete_confirmation(status)
+ return
+
+ if key in ("e", "E"):
+ if status.is_mine:
+ self.tui.async_edit(status)
return
if key in ("f", "F"):
- self._emit("favourite", status)
+ self.tui.async_toggle_favourite(self, status)
return
if key in ("m", "M"):
- self._emit("media", status)
+ self.tui.show_media(status)
return
if key in ("q", "Q"):
@@ -198,7 +212,7 @@ class Timeline(urwid.Columns):
return
if key in ("r", "R"):
- self._emit("reply", status)
+ self.tui.show_compose(status)
return
if key in ("s", "S"):
@@ -207,39 +221,49 @@ class Timeline(urwid.Columns):
return
if key in ("o", "O"):
- self._emit("bookmark", status)
+ self.tui.async_toggle_bookmark(self, status)
return
- if key in ("l", "L"):
- self._emit("links", status)
+ if key in ("i", "I"):
+ self.tui.show_links(status)
return
if key in ("n", "N"):
- if self.can_translate:
- self._emit("translate", status)
+ if self.tui.can_translate:
+ self.tui.async_translate(self, status)
return
if key in ("t", "T"):
- self._emit("thread", status)
+ self.tui.show_thread(status)
return
if key in ("u", "U"):
- self._emit("source", status)
+ self.tui.show_status_source(status)
return
if key in ("v", "V"):
if status.original.url:
webbrowser.open(status.original.url)
# force a screen refresh; necessary with console browsers
- self._emit("clear-screen")
+ self.tui.clear_screen()
return
- if key in ("p", "P"):
+ if key in ("e", "E"):
self._emit("save", status)
return
if key in ("z", "Z"):
- self._emit("zoom", self.status_details)
+ self.tui.show_status_zoom(self.status_details)
+ return
+
+ if key in ("p", "P"):
+ poll = status.original.data.get("poll")
+ if poll and not poll["expired"]:
+ self.tui.show_poll(status)
+ return
+
+ if key in ("y", "Y"):
+ self.tui.copy_status(status)
return
return super().keypress(size, key)
@@ -294,7 +318,8 @@ class Timeline(urwid.Columns):
class StatusDetails(urwid.Pile):
def __init__(self, timeline: Timeline, status: Optional[Status]):
self.status = status
- self.followed_tags = timeline.followed_tags
+ self.followed_accounts = timeline.tui.followed_accounts
+ self.options = timeline.tui.options
reblogged_by = status.author if status and status.reblog else None
widget_list = list(self.content_generator(status.original, reblogged_by)
@@ -304,13 +329,14 @@ class StatusDetails(urwid.Pile):
def content_generator(self, status, reblogged_by):
if reblogged_by:
text = "♺ {} boosted".format(reblogged_by.display_name or reblogged_by.username)
- yield ("pack", urwid.Text(("gray", text)))
- yield ("pack", urwid.AttrMap(urwid.Divider("-"), "gray"))
+ yield ("pack", urwid.Text(("dim", text)))
+ yield ("pack", urwid.AttrMap(urwid.Divider("-"), "dim"))
if status.author.display_name:
- yield ("pack", urwid.Text(("green", status.author.display_name)))
+ yield ("pack", urwid.Text(("bold", status.author.display_name)))
- yield ("pack", urwid.Text(("yellow", status.author.account)))
+ account_color = "highlight" if status.author.account in self.followed_accounts else "account"
+ yield ("pack", urwid.Text((account_color, status.author.account)))
yield ("pack", urwid.Divider())
if status.data["spoiler_text"]:
@@ -318,23 +344,28 @@ class StatusDetails(urwid.Pile):
yield ("pack", urwid.Divider())
# Show content warning
- if status.data["spoiler_text"] and not status.show_sensitive:
+ if status.data["spoiler_text"] and not status.show_sensitive and not self.options.always_show_sensitive:
yield ("pack", urwid.Text(("content_warning", "Marked as sensitive. Press S to view.")))
else:
- content = status.translation if status.show_translation else status.data["content"]
- for line in format_content(content):
- yield ("pack", urwid.Text(highlight_hashtags(line, self.followed_tags)))
+ if status.data["spoiler_text"]:
+ yield ("pack", urwid.Text(("content_warning", "Marked as sensitive.")))
+
+ content = status.original.translation if status.original.show_translation else status.data["content"]
+ widgetlist = html_to_widgets(content)
+
+ for line in widgetlist:
+ yield (line)
media = status.data["media_attachments"]
if media:
for m in media:
- yield ("pack", urwid.AttrMap(urwid.Divider("-"), "gray"))
+ yield ("pack", urwid.AttrMap(urwid.Divider("-"), "dim"))
yield ("pack", urwid.Text([("bold", "Media attachment"), " (", m["type"], ")"]))
if m["description"]:
yield ("pack", urwid.Text(m["description"]))
- yield ("pack", urwid.Text(("link", m["url"])))
+ yield ("pack", url_to_widget(m["url"]))
- poll = status.data.get("poll")
+ poll = status.original.data.get("poll")
if poll:
yield ("pack", urwid.Divider())
yield ("pack", self.build_linebox(self.poll_generator(poll)))
@@ -347,33 +378,35 @@ class StatusDetails(urwid.Pile):
application = status.data.get("application") or {}
application = application.get("name")
- yield ("pack", urwid.AttrWrap(urwid.Divider("-"), "gray"))
+ yield ("pack", urwid.AttrWrap(urwid.Divider("-"), "dim"))
translated_from = (
- language_name(status.translated_from)
- if status.show_translation and status.translated_from
+ language_name(status.original.translated_from)
+ if status.original.show_translation and status.original.translated_from
else None
)
visibility_colors = {
- "public": "gray",
- "unlisted": "white",
- "private": "cyan",
- "direct": "yellow"
+ "public": "visibility_public",
+ "unlisted": "visibility_unlisted",
+ "private": "visibility_private",
+ "direct": "visibility_direct"
}
visibility = status.visibility.title()
- visibility_color = visibility_colors.get(status.visibility, "gray")
+ visibility_color = visibility_colors.get(status.visibility, "dim")
yield ("pack", urwid.Text([
- ("blue", f"{status.created_at.strftime('%Y-%m-%d %H:%M')} "),
- ("red" if status.bookmarked else "gray", "🠷 "),
- ("gray", f"⤶ {status.data['replies_count']} "),
- ("yellow" if status.reblogged else "gray", f"♺ {status.data['reblogs_count']} "),
- ("yellow" if status.favourited else "gray", f"★ {status.data['favourites_count']}"),
+ ("status_detail_timestamp", f"{status.created_at.strftime('%Y-%m-%d %H:%M')} "),
+ ("status_detail_timestamp",
+ f"(edited {status.edited_at.strftime('%Y-%m-%d %H:%M')}) " if status.edited_at else ""),
+ ("status_detail_bookmarked" if status.bookmarked else "dim", "b "),
+ ("dim", f"⤶ {status.data['replies_count']} "),
+ ("highlight" if status.reblogged else "dim", f"♺ {status.data['reblogs_count']} "),
+ ("highlight" if status.favourited else "dim", f"★ {status.data['favourites_count']}"),
(visibility_color, f" · {visibility}"),
- ("yellow", f" · Translated from {translated_from} " if translated_from else ""),
- ("gray", f" · {application}" if application else ""),
+ ("highlight", f" · Translated from {translated_from} " if translated_from else ""),
+ ("dim", f" · {application}" if application else ""),
]))
# Push things to bottom
@@ -382,17 +415,17 @@ class StatusDetails(urwid.Pile):
def build_linebox(self, contents):
contents = urwid.Pile(list(contents))
contents = urwid.Padding(contents, left=1, right=1)
- return urwid.LineBox(contents)
+ return RoundedLineBox(contents)
def card_generator(self, card):
- yield urwid.Text(("green", card["title"].strip()))
+ yield urwid.Text(("card_title", card["title"].strip()))
if card.get("author_name"):
- yield urwid.Text(["by ", ("yellow", card["author_name"].strip())])
+ yield urwid.Text(["by ", ("card_author", card["author_name"].strip())])
yield urwid.Text("")
if card["description"]:
yield urwid.Text(card["description"].strip())
yield urwid.Text("")
- yield urwid.Text(("link", card["url"]))
+ yield url_to_widget(card["url"])
def poll_generator(self, poll):
for idx, option in enumerate(poll["options"]):
@@ -416,36 +449,36 @@ class StatusDetails(urwid.Pile):
expires_at = parse_datetime(poll["expires_at"]).strftime("%Y-%m-%d %H:%M")
status += " · Closes on {}".format(expires_at)
- yield urwid.Text(("gray", status))
+ yield urwid.Text(("dim", status))
class StatusListItem(SelectableColumns):
- def __init__(self, status):
- edited = status.data["edited_at"]
+ def __init__(self, status, relative_datetimes):
+ edited_at = status.original.edited_at
# TODO: hacky implementation to avoid creating conflicts for existing
- # pull reuqests, refactor when merged.
+ # pull requests, refactor when merged.
created_at = (
time_ago(status.created_at).ljust(3, " ")
- if "--relative-datetimes" in sys.argv
+ if relative_datetimes
else status.created_at.strftime("%Y-%m-%d %H:%M")
)
- edited_flag = "*" if edited else " "
- favourited = ("yellow", "★") if status.original.favourited else " "
- reblogged = ("yellow", "♺") if status.original.reblogged else " "
- is_reblog = ("cyan", "♺") if status.reblog else " "
- is_reply = ("cyan", "⤶") if status.original.in_reply_to else " "
+ edited_flag = "*" if edited_at else " "
+ favourited = ("highlight", "★") if status.original.favourited else " "
+ reblogged = ("highlight", "♺") if status.original.reblogged else " "
+ is_reblog = ("dim", "♺") if status.reblog else " "
+ is_reply = ("dim", "⤶ ") if status.original.in_reply_to else " "
return super().__init__([
- ("pack", SelectableText(("blue", created_at), wrap="clip")),
- ("pack", urwid.Text(("blue", edited_flag))),
+ ("pack", SelectableText(("status_list_timestamp", created_at), wrap="clip")),
+ ("pack", urwid.Text(("status_list_timestamp", edited_flag))),
("pack", urwid.Text(" ")),
("pack", urwid.Text(favourited)),
("pack", urwid.Text(" ")),
("pack", urwid.Text(reblogged)),
("pack", urwid.Text(" ")),
- urwid.Text(("green", status.original.account), wrap="clip"),
+ urwid.Text(("status_list_account", status.original.account), wrap="clip"),
("pack", urwid.Text(is_reply)),
("pack", urwid.Text(is_reblog)),
("pack", urwid.Text(" ")),
diff --git a/toot/tui/utils.py b/toot/tui/utils.py
index e2855c4..5b94624 100644
--- a/toot/tui/utils.py
+++ b/toot/tui/utils.py
@@ -1,58 +1,12 @@
-from html.parser import HTMLParser
-import math
-import os
+import base64
import re
-import shutil
-import subprocess
+import urwid
-from datetime import datetime, timezone
+from functools import reduce
+from html.parser import HTMLParser
+from typing import List
HASHTAG_PATTERN = re.compile(r'(? datetime:
- now = datetime.now().astimezone()
- delta = now.timestamp() - value.timestamp()
-
- if (delta < 1):
- return "now"
-
- if (delta < 8 * DAY):
- if (delta < MINUTE):
- return f"{math.floor(delta / SECOND)}".rjust(2, " ") + "s"
- if (delta < HOUR):
- return f"{math.floor(delta / MINUTE)}".rjust(2, " ") + "m"
- if (delta < DAY):
- return f"{math.floor(delta / HOUR)}".rjust(2, " ") + "h"
- return f"{math.floor(delta / DAY)}".rjust(2, " ") + "d"
-
- if (delta < 53 * WEEK): # not exactly correct but good enough as a boundary
- return f"{math.floor(delta / WEEK)}".rjust(2, " ") + "w"
-
- return ">1y"
def highlight_keys(text, high_attr, low_attr=""):
@@ -79,48 +33,19 @@ def highlight_keys(text, high_attr, low_attr=""):
return list(_gen())
-def highlight_hashtags(line, followed_tags, attr="hashtag", followed_attr="followed_hashtag"):
+def highlight_hashtags(line):
hline = []
for p in re.split(HASHTAG_PATTERN, line):
if p.startswith("#"):
- if p[1:].lower() in (t.lower() for t in followed_tags):
- hline.append((followed_attr, p))
- else:
- hline.append((attr, p))
+ hline.append(("hashtag", p))
else:
hline.append(p)
return hline
-def show_media(paths):
- """
- Attempt to open an image viewer to show given media files.
-
- FIXME: This is not very thought out, but works for me.
- Once settings are implemented, add an option for the user to configure their
- prefered media viewer.
- """
- viewer = None
- potential_viewers = [
- "feh",
- "eog",
- "display"
- ]
- for v in potential_viewers:
- viewer = shutil.which(v)
- if viewer:
- break
-
- if not viewer:
- raise Exception("Cannot find an image viewer")
-
- subprocess.run([viewer] + paths)
-
-
class LinkParser(HTMLParser):
-
def reset(self):
super().reset()
self.links = []
@@ -144,3 +69,43 @@ def parse_content_links(content):
parser = LinkParser()
parser.feed(content)
return parser.links[:]
+
+
+def copy_to_clipboard(screen: urwid.raw_display.Screen, text: str):
+ """ copy text to clipboard using OSC 52
+ This escape sequence is documented
+ here https://iterm2.com/documentation-escape-codes.html
+ It has wide support - XTerm, Windows Terminal,
+ Kitty, iTerm2, others. Some terminals may require a setting to be
+ enabled in order to use OSC 52 clipboard functions.
+ """
+
+ text_bytes = text.encode("utf-8")
+ b64_bytes = base64.b64encode(text_bytes)
+ b64_text = b64_bytes.decode("utf-8")
+
+ screen.write(f"\033]52;c;{b64_text}\a")
+ screen.flush()
+
+
+def get_max_toot_chars(instance, default=500):
+ # Mastodon
+ # https://docs.joinmastodon.org/entities/Instance/#max_characters
+ max_toot_chars = deep_get(instance, ["configuration", "statuses", "max_characters"])
+ if isinstance(max_toot_chars, int):
+ return max_toot_chars
+
+ # Pleroma
+ max_toot_chars = instance.get("max_toot_chars")
+ if isinstance(max_toot_chars, int):
+ return max_toot_chars
+
+ return default
+
+
+def deep_get(adict: dict, path: List[str], default=None):
+ return reduce(
+ lambda d, key: d.get(key, default) if isinstance(d, dict) else default,
+ path,
+ adict
+ )
diff --git a/toot/tui/widgets.py b/toot/tui/widgets.py
index a311d52..db7bf9e 100644
--- a/toot/tui/widgets.py
+++ b/toot/tui/widgets.py
@@ -1,4 +1,5 @@
import urwid
+from wcwidth import wcswidth
class Clickable:
@@ -40,9 +41,67 @@ class Button(urwid.AttrWrap):
"""Styled button."""
def __init__(self, *args, **kwargs):
button = urwid.Button(*args, **kwargs)
- padding = urwid.Padding(button, width=len(args[0]) + 4)
+ padding = urwid.Padding(button, width=wcswidth(args[0]) + 4)
return super().__init__(padding, "button", "button_focused")
def set_label(self, *args, **kwargs):
self.original_widget.original_widget.set_label(*args, **kwargs)
- self.original_widget.width = len(args[0]) + 4
+ self.original_widget.width = wcswidth(args[0]) + 4
+
+
+class CheckBox(urwid.AttrWrap):
+ """Styled checkbox."""
+ def __init__(self, *args, **kwargs):
+ self.button = urwid.CheckBox(*args, **kwargs)
+ padding = urwid.Padding(self.button, width=len(args[0]) + 4)
+ return super().__init__(padding, "button", "button_focused")
+
+ def get_state(self):
+ """Return the state of the checkbox."""
+ return self.button._state
+
+
+class RadioButton(urwid.AttrWrap):
+ """Styled radiobutton."""
+ def __init__(self, *args, **kwargs):
+ button = urwid.RadioButton(*args, **kwargs)
+ padding = urwid.Padding(button, width=len(args[1]) + 4)
+ return super().__init__(padding, "button", "button_focused")
+
+
+class ModalBox(urwid.Frame):
+ def __init__(self, message):
+ text = urwid.Text(message)
+ filler = urwid.Filler(text, valign='top', top=1, bottom=1)
+ padding = urwid.Padding(filler, left=1, right=1)
+ return super().__init__(padding)
+
+
+class RoundedLineBox(urwid.LineBox):
+ """LineBox that defaults to rounded corners."""
+ def __init__(self,
+ original_widget,
+ title="",
+ title_align="center",
+ title_attr=None,
+ tlcorner="\u256d",
+ tline="─",
+ lline="│",
+ trcorner="\u256e",
+ blcorner="\u2570",
+ rline="│",
+ bline="─",
+ brcorner="\u256f",
+ ) -> None:
+ return super().__init__(original_widget,
+ title,
+ title_align,
+ title_attr,
+ tlcorner,
+ tline,
+ lline,
+ trcorner,
+ blcorner,
+ rline,
+ bline,
+ brcorner)
diff --git a/toot/typing_compat.py b/toot/typing_compat.py
new file mode 100644
index 0000000..0c6fe5d
--- /dev/null
+++ b/toot/typing_compat.py
@@ -0,0 +1,147 @@
+# Taken from https://github.com/rossmacarthur/typing-compat/
+# TODO: Remove once the minimum python version is increased to 3.8
+#
+# Licensed under the MIT license
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+# flake8: noqa
+
+import collections
+import typing
+
+
+__all__ = ['get_args', 'get_origin']
+__title__ = 'typing-compat'
+__version__ = '0.1.0'
+__url__ = 'https://github.com/rossmacarthur/typing-compat'
+__author__ = 'Ross MacArthur'
+__author_email__ = 'ross@macarthur.io'
+__description__ = 'Python typing compatibility library'
+
+
+try:
+ # Python >=3.8 should have these functions already
+ from typing import get_args as _get_args # novermin
+ from typing import get_origin as _get_origin # novermin
+except ImportError:
+ if hasattr(typing, '_GenericAlias'): # Python 3.7
+
+ def _get_origin(tp):
+ """Copied from the Python 3.8 typing module"""
+ if isinstance(tp, typing._GenericAlias):
+ return tp.__origin__
+ if tp is typing.Generic:
+ return typing.Generic
+ return None
+
+ def _get_args(tp):
+ """Copied from the Python 3.8 typing module"""
+ if isinstance(tp, typing._GenericAlias):
+ res = tp.__args__
+ if (
+ get_origin(tp) is collections.abc.Callable
+ and res[0] is not Ellipsis
+ ):
+ res = (list(res[:-1]), res[-1])
+ return res
+ return ()
+
+ else: # Python <3.7
+
+ def _resolve_via_mro(tp):
+ if hasattr(tp, '__mro__'):
+ for t in tp.__mro__:
+ if t.__module__ in ('builtins', '__builtin__') and t is not object:
+ return t
+ return tp
+
+ def _get_origin(tp):
+ """Emulate the behaviour of Python 3.8 typing module"""
+ if isinstance(tp, typing._ClassVar):
+ return typing.ClassVar
+ elif isinstance(tp, typing._Union):
+ return typing.Union
+ elif isinstance(tp, typing.GenericMeta):
+ if hasattr(tp, '_gorg'):
+ return _resolve_via_mro(tp._gorg)
+ else:
+ while tp.__origin__ is not None:
+ tp = tp.__origin__
+ return _resolve_via_mro(tp)
+ elif hasattr(typing, '_Literal') and isinstance(tp, typing._Literal): # novermin
+ return typing.Literal # novermin
+
+ def _normalize_arg(args):
+ if isinstance(args, tuple) and len(args) > 1:
+ base, rest = args[0], tuple(_normalize_arg(arg) for arg in args[1:])
+ if isinstance(base, typing.CallableMeta):
+ return typing.Callable[list(rest[:-1]), rest[-1]]
+ elif isinstance(base, (typing.GenericMeta, typing._Union)):
+ return base[rest]
+ return args
+
+ def _get_args(tp):
+ """Emulate the behaviour of Python 3.8 typing module"""
+ if isinstance(tp, typing._ClassVar):
+ return (tp.__type__,)
+ elif hasattr(tp, '_subs_tree'):
+ tree = tp._subs_tree()
+ if isinstance(tree, tuple) and len(tree) > 1:
+ if isinstance(tree[0], typing.CallableMeta) and len(tree) == 2:
+ return ([], _normalize_arg(tree[1]))
+ return tuple(_normalize_arg(arg) for arg in tree[1:])
+ return ()
+
+
+def get_origin(tp):
+ """
+ Get the unsubscripted version of a type.
+
+ This supports generic types, Callable, Tuple, Union, Literal, Final and
+ ClassVar. Returns None for unsupported types.
+
+ Examples:
+
+ get_origin(Literal[42]) is Literal
+ get_origin(int) is None
+ get_origin(ClassVar[int]) is ClassVar
+ get_origin(Generic) is Generic
+ get_origin(Generic[T]) is Generic
+ get_origin(Union[T, int]) is Union
+ get_origin(List[Tuple[T, T]][int]) == list
+ """
+ return _get_origin(tp)
+
+
+def get_args(tp):
+ """
+ Get type arguments with all substitutions performed.
+
+ For unions, basic simplifications used by Union constructor are performed.
+
+ Examples:
+
+ get_args(Dict[str, int]) == (str, int)
+ get_args(int) == ()
+ get_args(Union[int, Union[T, int], str][int]) == (int, str)
+ get_args(Union[int, Tuple[T, int]][str]) == (int, Tuple[str, int])
+ get_args(Callable[[], T][int]) == ([], int)
+ """
+ return _get_args(tp)
diff --git a/toot/utils/__init__.py b/toot/utils/__init__.py
index c76e65f..7bc5c77 100644
--- a/toot/utils/__init__.py
+++ b/toot/utils/__init__.py
@@ -7,8 +7,12 @@ import unicodedata
import warnings
from bs4 import BeautifulSoup
+from typing import Any, Dict, List
+
+import click
from toot.exceptions import ConsoleError
+from urllib.parse import urlparse, urlencode, quote, unquote
def str_bool(b):
@@ -16,20 +20,27 @@ def str_bool(b):
return "true" if b else "false"
-def get_text(html):
- """Converts html to text, strips all tags."""
+def str_bool_nullable(b):
+ """Similar to str_bool, but leave None as None"""
+ return None if b is None else str_bool(b)
+
+def parse_html(html: str) -> BeautifulSoup:
# Ignore warnings made by BeautifulSoup, if passed something that looks like
# a file (e.g. a dot which matches current dict), it will warn that the file
# should be opened instead of passing a filename.
with warnings.catch_warnings():
warnings.simplefilter("ignore")
- text = BeautifulSoup(html.replace(''', "'"), "html.parser").get_text()
-
- return unicodedata.normalize('NFKC', text)
+ return BeautifulSoup(html.replace("'", "'"), "html.parser")
-def parse_html(html):
+def get_text(html):
+ """Converts html to text, strips all tags."""
+ text = parse_html(html).get_text()
+ return unicodedata.normalize("NFKC", text)
+
+
+def html_to_paragraphs(html: str) -> List[List[str]]:
"""Attempt to convert html to plain text while keeping line breaks.
Returns a list of paragraphs, each being a list of lines.
"""
@@ -48,7 +59,7 @@ def format_content(content):
Returns a generator yielding lines of content.
"""
- paragraphs = parse_html(content)
+ paragraphs = html_to_paragraphs(content)
first = True
@@ -100,17 +111,56 @@ Everything below it will be ignored.
"""
-def editor_input(editor, initial_text):
+def editor_input(editor: str, initial_text: str) -> str:
"""Lets user input text using an editor."""
+ tmp_path = _tmp_status_path()
initial_text = (initial_text or "") + EDITOR_INPUT_INSTRUCTIONS
- with tempfile.NamedTemporaryFile(suffix='.toot') as f:
- f.write(initial_text.encode())
- f.flush()
+ if not _use_existing_tmp_file(tmp_path):
+ with open(tmp_path, "w") as f:
+ f.write(initial_text)
+ f.flush()
- subprocess.run([editor, f.name])
+ subprocess.run([editor, tmp_path])
- f.seek(0)
- text = f.read().decode()
+ with open(tmp_path) as f:
+ return f.read().split(EDITOR_DIVIDER)[0].strip()
- return text.split(EDITOR_DIVIDER)[0].strip()
+
+def delete_tmp_status_file() -> None:
+ try:
+ os.unlink(_tmp_status_path())
+ except FileNotFoundError:
+ pass
+
+
+def _tmp_status_path() -> str:
+ tmp_dir = tempfile.gettempdir()
+ return f"{tmp_dir}/.status.toot"
+
+
+def _use_existing_tmp_file(tmp_path: str) -> bool:
+ if os.path.exists(tmp_path):
+ click.echo(f"Found draft status at: {tmp_path}")
+
+ choice = click.Choice(["O", "D"], case_sensitive=False)
+ char = click.prompt("Open or Delete?", type=choice, default="O")
+ return char == "O"
+
+ return False
+
+
+def drop_empty_values(data: Dict[Any, Any]) -> Dict[Any, Any]:
+ """Remove keys whose values are null"""
+ return {k: v for k, v in data.items() if v is not None}
+
+
+def urlencode_url(url: str) -> str:
+ parsed_url = urlparse(url)
+
+ # unencode before encoding, to prevent double-urlencoding
+ encoded_path = quote(unquote(parsed_url.path), safe="-._~()'!*:@,;+&=/")
+ encoded_query = urlencode({k: quote(unquote(v), safe="-._~()'!*:@,;?/") for k, v in parsed_url.params})
+ encoded_url = parsed_url._replace(path=encoded_path, params=encoded_query).geturl()
+
+ return encoded_url
diff --git a/toot/utils/datetime.py b/toot/utils/datetime.py
new file mode 100644
index 0000000..2a214a0
--- /dev/null
+++ b/toot/utils/datetime.py
@@ -0,0 +1,45 @@
+import math
+import os
+
+from datetime import datetime, timezone
+
+
+def parse_datetime(value):
+ """Returns an aware datetime in local timezone"""
+ dttm = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%f%z")
+
+ # When running tests return datetime in UTC so that tests don't depend on
+ # the local timezone
+ if "PYTEST_CURRENT_TEST" in os.environ:
+ return dttm.astimezone(timezone.utc)
+
+ return dttm.astimezone()
+
+
+SECOND = 1
+MINUTE = SECOND * 60
+HOUR = MINUTE * 60
+DAY = HOUR * 24
+WEEK = DAY * 7
+
+
+def time_ago(value: datetime) -> str:
+ now = datetime.now().astimezone()
+ delta = now.timestamp() - value.timestamp()
+
+ if delta < 1:
+ return "now"
+
+ if delta < 8 * DAY:
+ if delta < MINUTE:
+ return f"{math.floor(delta / SECOND)}".rjust(2, " ") + "s"
+ if delta < HOUR:
+ return f"{math.floor(delta / MINUTE)}".rjust(2, " ") + "m"
+ if delta < DAY:
+ return f"{math.floor(delta / HOUR)}".rjust(2, " ") + "h"
+ return f"{math.floor(delta / DAY)}".rjust(2, " ") + "d"
+
+ if delta < 53 * WEEK: # not exactly correct but good enough as a boundary
+ return f"{math.floor(delta / WEEK)}".rjust(2, " ") + "w"
+
+ return ">1y"
diff --git a/toot/wcstring.py b/toot/wcstring.py
index cc4bee1..31fe2c4 100644
--- a/toot/wcstring.py
+++ b/toot/wcstring.py
@@ -3,11 +3,12 @@ Utilities for dealing with string containing wide characters.
"""
import re
+from typing import Generator, List
from wcwidth import wcwidth, wcswidth
-def _wc_hard_wrap(line, length):
+def _wc_hard_wrap(line: str, length: int) -> Generator[str, None, None]:
"""
Wrap text to length characters, breaking when target length is reached,
taking into account character width.
@@ -20,7 +21,7 @@ def _wc_hard_wrap(line, length):
char_len = wcwidth(char)
if chars_len + char_len > length:
yield "".join(chars)
- chars = []
+ chars: List[str] = []
chars_len = 0
chars.append(char)
@@ -30,7 +31,7 @@ def _wc_hard_wrap(line, length):
yield "".join(chars)
-def wc_wrap(text, length):
+def wc_wrap(text: str, length: int) -> Generator[str, None, None]:
"""
Wrap text to given length, breaking on whitespace and taking into account
character width.
@@ -38,7 +39,7 @@ def wc_wrap(text, length):
Meant for use on a single line or paragraph. Will destroy spacing between
words and paragraphs and any indentation.
"""
- line_words = []
+ line_words: List[str] = []
line_len = 0
words = re.split(r"\s+", text.strip())
@@ -66,7 +67,7 @@ def wc_wrap(text, length):
yield from _wc_hard_wrap(line, length)
-def trunc(text, length):
+def trunc(text: str, length: int) -> str:
"""
Truncates text to given length, taking into account wide characters.
@@ -98,7 +99,7 @@ def trunc(text, length):
return text[:-n].strip() + '…'
-def pad(text, length):
+def pad(text: str, length: int) -> str:
"""Pads text to given length, taking into account wide characters."""
text_length = wcswidth(text)
@@ -108,7 +109,7 @@ def pad(text, length):
return text
-def fit_text(text, length):
+def fit_text(text: str, length: int) -> str:
"""Makes text fit the given length by padding or truncating it."""
text_length = wcswidth(text)