Migrate timeline commands

Makefile

@@ -15,7 +15,7 @@ test:
 coverage:
 	coverage erase
 	coverage run
-	coverage html --omit toot/tui/*
+	coverage html --omit "toot/tui/*"
 	coverage report
 
 clean :

tests/integration/test_timelines.py

@@ -0,0 +1,142 @@
+import pytest
+
+from time import sleep
+from uuid import uuid4
+
+from toot import api, cli
+from toot.entities import from_dict, Status
+from tests.integration.conftest import TOOT_TEST_BASE_URL, register_account
+
+
+# TODO: If fixture is not overriden here, tests fail, not sure why, figure it out
+@pytest.fixture(scope="module")
+def user(app):
+    return register_account(app)
+
+
+@pytest.fixture(scope="module")
+def other_user(app):
+    return register_account(app)
+
+
+@pytest.fixture(scope="module")
+def friend_user(app, user):
+    friend = register_account(app)
+    friend_account = api.find_account(app, user, friend.username)
+    api.follow(app, user, friend_account["id"])
+    return friend
+
+
+@pytest.fixture(scope="module")
+def friend_list(app, user, friend_user):
+    friend_account = api.find_account(app, user, friend_user.username)
+    list = api.create_list(app, user, str(uuid4()))
+    api.add_accounts_to_list(app, user, list["id"], account_ids=[friend_account["id"]])
+    return list
+
+
+# TODO: break up big test and set up module level fixtures
+def test_timelines(app, user, other_user, friend_user, friend_list, run):
+    status1 = _post_status(app, user, "#foo")
+    status2 = _post_status(app, other_user, "#bar")
+    status3 = _post_status(app, friend_user, "#foo #bar")
+
+    # Give mastodon time to process things :/
+    # Tests fail if this is removed, required delay depends on server speed
+    sleep(1)
+
+    # Home timeline
+    result = run(cli.timeline)
+    assert result.exit_code == 0
+    print()
+    print(result.stdout)
+    print()
+
+    assert status1.id in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    # Public timeline
+    result = run(cli.timeline, "--public")
+    assert result.exit_code == 0
+    assert status1.id in result.stdout
+    assert status2.id in result.stdout
+    assert status3.id in result.stdout
+
+    # Anon public timeline
+    result = run(cli.timeline, "--instance", TOOT_TEST_BASE_URL, "--public")
+    assert result.exit_code == 0
+    assert status1.id in result.stdout
+    assert status2.id in result.stdout
+    assert status3.id in result.stdout
+
+    # Tag timeline
+    result = run(cli.timeline, "--tag", "foo")
+    assert result.exit_code == 0
+    assert status1.id in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    result = run(cli.timeline, "--tag", "bar")
+    assert result.exit_code == 0
+    assert status1.id not in result.stdout
+    assert status2.id in result.stdout
+    assert status3.id in result.stdout
+
+    # Anon tag timeline
+    result = run(cli.timeline, "--instance", TOOT_TEST_BASE_URL, "--tag", "foo")
+    assert result.exit_code == 0
+    assert status1.id in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    # List timeline (by list name)
+    result = run(cli.timeline, "--list", friend_list["title"])
+    assert result.exit_code == 0
+    assert status1.id not in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    # List timeline (by list ID)
+    result = run(cli.timeline, "--list", friend_list["id"])
+    assert result.exit_code == 0
+    assert status1.id not in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    # Account timeline
+    result = run(cli.timeline, "--account", friend_user.username)
+    assert result.exit_code == 0
+    assert status1.id not in result.stdout
+    assert status2.id not in result.stdout
+    assert status3.id in result.stdout
+
+    result = run(cli.timeline, "--account", other_user.username)
+    assert result.exit_code == 0
+    assert status1.id not in result.stdout
+    assert status2.id in result.stdout
+    assert status3.id not in result.stdout
+
+
+def test_timeline_cant_combine_timelines(run):
+    result = run(cli.timeline, "--tag", "foo", "--account", "bar")
+    assert result.exit_code == 1
+    assert result.stderr.strip() == "Error: Only one of --public, --tag, --account, or --list can be used at one time."
+
+
+def test_timeline_local_needs_public_or_tag(run):
+    result = run(cli.timeline, "--local")
+    assert result.exit_code == 1
+    assert result.stderr.strip() == "Error: The --local option is only valid alongside --public or --tag."
+
+
+def test_timeline_instance_needs_public_or_tag(run):
+    result = run(cli.timeline, "--instance", TOOT_TEST_BASE_URL)
+    assert result.exit_code == 1
+    assert result.stderr.strip() == "Error: The --instance option is only valid alongside --public or --tag."
+
+
+def _post_status(app, user, text=None) -> Status:
+    text = text or str(uuid4())
+    response = api.post_status(app, user, text)
+    return from_dict(Status, response.json())

toot/api.py

@@ -8,7 +8,7 @@
 from urllib.parse import urlparse, urlencode, quote
 
 from toot import App, User, http, CLIENT_NAME, CLIENT_WEBSITE
-from toot.exceptions import AuthenticationError, ConsoleError
+from toot.exceptions import ConsoleError
 from toot.utils import drop_empty_values, str_bool, str_bool_nullable
 
 
@@ -300,6 +300,35 @@ def reblogged_by(app, user, status_id) -> Response:
     return http.get(app, user, url)
 
 
+def get_timeline_generator(
+    app: Optional[App],
+    user: Optional[User],
+    base_url: Optional[str] = None,
+    account: Optional[str] = None,
+    list_id: Optional[str] = None,
+    tag: Optional[str] = None,
+    local: bool = False,
+    public: bool = False,
+    limit=20,  # TODO
+):
+    if public:
+        if base_url:
+            return anon_public_timeline_generator(base_url, local=local, limit=limit)
+        else:
+            return public_timeline_generator(app, user, local=local, limit=limit)
+    elif tag:
+        if base_url:
+            return anon_tag_timeline_generator(base_url, tag, limit=limit)
+        else:
+            return tag_timeline_generator(app, user, tag, local=local, limit=limit)
+    elif account:
+        return account_timeline_generator(app, user, account, limit=limit)
+    elif list_id:
+        return timeline_list_generator(app, user, list_id, limit=limit)
+    else:
+        return home_timeline_generator(app, user, limit=limit)
+
+
 def _get_next_path(headers):
     """Given timeline response headers, returns the path to the next batch"""
     links = headers.get('Link', '')
@@ -309,6 +338,14 @@ def _get_next_path(headers):
         return "?".join([parsed.path, parsed.query])
 
 
+def _get_next_url(headers) -> Optional[str]:
+    """Given timeline response headers, returns the url to the next batch"""
+    links = headers.get('Link', '')
+    match = re.match('<([^>]+)>; rel="next"', links)
+    if match:
+        return match.group(1)
+
+
 def _timeline_generator(app, user, path, params=None):
     while path:
         response = http.get(app, user, path, params)
@@ -369,7 +406,7 @@ def conversation_timeline_generator(app, user, limit=20):
     return _conversation_timeline_generator(app, user, path, params)
 
 
-def account_timeline_generator(app: App, user: User, account_name: str, replies=False, reblogs=False, limit=20):
+def account_timeline_generator(app, user, account_name: str, replies=False, reblogs=False, limit=20):
     account = find_account(app, user, account_name)
     path = f"/api/v1/accounts/{account['id']}/statuses"
     params = {"limit": limit, "exclude_replies": not replies, "exclude_reblogs": not reblogs}
@@ -381,24 +418,23 @@ def timeline_list_generator(app, user, list_id, limit=20):
     return _timeline_generator(app, user, path, {'limit': limit})
 
 
-def _anon_timeline_generator(instance, path, params=None):
-    while path:
-        url = f"https://{instance}{path}"
+def _anon_timeline_generator(url, params=None):
+    while url:
         response = http.anon_get(url, params)
         yield response.json()
-        path = _get_next_path(response.headers)
+        url = _get_next_url(response.headers)
 
 
-def anon_public_timeline_generator(instance, local=False, limit=20):
-    path = '/api/v1/timelines/public'
-    params = {'local': str_bool(local), 'limit': limit}
-    return _anon_timeline_generator(instance, path, params)
+def anon_public_timeline_generator(base_url, local=False, limit=20):
+    query = urlencode({"local": str_bool(local), "limit": limit})
+    url = f"{base_url}/api/v1/timelines/public?{query}"
+    return _anon_timeline_generator(url)
 
 
-def anon_tag_timeline_generator(instance, hashtag, local=False, limit=20):
-    path = f"/api/v1/timelines/tag/{quote(hashtag)}"
-    params = {'local': str_bool(local), 'limit': limit}
-    return _anon_timeline_generator(instance, path, params)
+def anon_tag_timeline_generator(base_url, hashtag, local=False, limit=20):
+    query = urlencode({"local": str_bool(local), "limit": limit})
+    url = f"{base_url}/api/v1/timelines/tag/{quote(hashtag)}?{query}"
+    return _anon_timeline_generator(url)
 
 
 def get_media(app: App, user: User, id: str):
@@ -570,7 +606,7 @@ def get_list_accounts(app, user, list_id):
     return _get_response_list(app, user, path)
 
 
-def create_list(app, user, title, replies_policy):
+def create_list(app, user, title, replies_policy="none"):
     url = "/api/v1/lists"
     json = {'title': title}
     if replies_policy:

toot/cli/__init__.py

@@ -1,9 +1,11 @@
-from toot.cli.base import cli, Context  # noqa
+# flake8: noqa
+from toot.cli.base import cli, Context
 
-from toot.cli.auth import *
 from toot.cli.accounts import *
+from toot.cli.auth import *
 from toot.cli.lists import *
 from toot.cli.post import *
 from toot.cli.read import *
 from toot.cli.statuses import *
 from toot.cli.tags import *
+from toot.cli.timelines import *

toot/cli/base.py

@@ -40,7 +40,7 @@ def get_default_visibility() -> str:
 
 # Data object to add to Click context
 class Context(NamedTuple):
-    app: Optional[App] = None
+    app: Optional[App]
     user: Optional[User] = None
     color: bool = False
     debug: bool = False