First pass processing tests run OK

app/helpers.py

@@ -15,7 +15,6 @@ def index_ojects_by_parameter(object_list: List, param: Any):
     """
     Create a dictionary from passed list where each list entry is keyed
     by passed param
-    n
     """
 
     results = {}

app/urma.py

@@ -94,7 +94,7 @@ class MastodonAPI:
                 # Have we reached minimum id?
                 if datum.id < since:
                     break
-                # Is this our post that we boosted?
+                # Is this a post that we boosted?
                 if datum.account.id == self.me.id and datum.reblog:
                     # Add in new data
                     results.append(datum)
@@ -154,13 +154,14 @@ def main() -> None:
         update_followed_hashtags(session, mastapi)
 
         favourited = mastapi.get_favourited(since)
-        process_favourited_posts(session, favourited)
+        process_favourited_posts(session, favourited, mastapi.me.id)
 
-        boosted = mastapi.get_boosted(since)
-        process_boosted_posts(session, boosted)
+        boosted_by_me = mastapi.get_boosted(since)
+        process_posts_boosted_by_me(session, boosted_by_me, mastapi.me.id)
 
         bookmarked = mastapi.get_bookmarked(since)
-        process_bookmarked_posts(session, mastapi, bookmarked)
+        process_bookmarked_posts(session, bookmarked, mastapi.me.id)
+        # TODO: mastapi.unbookmark(int(post.id))
 
 
 def get_since_id(session: Session) -> int:
@@ -186,53 +187,44 @@ def get_since_id(session: Session) -> int:
         return max(max_days_ago_id, newest_db_id)
 
 
-def process_bookmarked_posts(session, mastapi, posts) -> None:
+def process_bookmarked_posts(session: Session,
+                             posts: List[Posts], me_id: int) -> None:
     """
     Process bookmarked posts
     """
 
     for post in posts:
-        record = _process_post(session, post)
+        record = _process_post(session, post, me_id)
         # Posts that are favourited and bookmarked are genuine bookmark
         # posts: ignore.
         if record.favourited:
             continue
         record.bookmarked = True
         return
-    # TODO: mastapi.unbookmark(int(post.id))
 
 
-def process_boosted_posts(session, posts) -> None:
-    """
-    Process boosted posts
-    """
-
-    for post in posts:
-        record = _process_post(session, post)
-        record.boosted = True
-
-
-def process_favourited_posts(session, posts) -> None:
+def process_favourited_posts(session: Session,
+                             posts: List[Posts], me_id: int) -> None:
     """
     Process favourited posts
     """
 
     for post in posts:
-        record = _process_post(session, post)
+        if post.favourited:
+            record = _process_post(session, post, me_id)
             record.favourited = True
+        else:
+            log.debug(
+                f"process_favourited_posts({post.id=}) not favourited"
+            )
 
 
-def _process_post(session: Session, post) -> Posts:
+def _process_post(session: Session, post: Posts, me_id) -> Posts:
     """
     Add passsed post to database
     """
 
     log.debug(f"{post.id=} processing")
-    rec = Posts.get_or_create(session, str(post.id))
-    if rec.account_id is not None:
-        # We already have this post
-        log.debug(f"{post.id=} already in db")
-        return rec
 
     # Create account record if needed
     log.debug(f"{post.id=} processing {post.account.id=}")
@@ -244,6 +236,24 @@ def _process_post(session: Session, post) -> Posts:
         account_rec.display_name = post.account.display_name
         account_rec.bot = post.account.bot
         account_rec.url = post.account.url
+
+    if post.reblog:
+        # We're only interesting the boosted post, not this onej
+        log.debug(f"{post.id=} {post.reblog.id=}")
+        boosted_record = _process_post(session, post.reblog, me_id)
+        # Record who bosed the post unless it was us
+        if post.account.id == me_id:
+            boosted_record.boosted_by_id = None
+        else:
+            boosted_record.boosted_by_id = account_rec.id
+        return boosted_record
+
+    rec = Posts.get_or_create(session, str(post.id))
+    if rec.account_id is not None:
+        # We already have this post
+        log.debug(f"{post.id=} already in db")
+        return rec
+    else:
         rec.account_id = account_rec.id
 
     # Create hashtag records as needed
@@ -255,14 +265,31 @@ def _process_post(session: Session, post) -> Posts:
     rec.created_at = post.created_at
     rec.uri = post.uri
 
-    if post.reblog:
-        log.debug(f"{post.id=} {post.reblog.id=}")
-        boosted_rec = _process_post(session, post.reblog)
-        rec.boosted_by_id = boosted_rec.account_id
-
     return rec
 
 
+def process_posts_boosted_by_me(session: Session,
+                                posts: List[Posts], me_id: int) -> None:
+    """
+    Process boosted posts
+    """
+
+    for post in posts:
+        if post.account.id != me_id:
+            log.debug(
+                f"{post.id=} in process_posts_boosted_by_me but not "
+                f"boosted by me"
+            )
+        if post.reblog:
+            record = _process_post(session, post, me_id)
+            record.boosted = True
+            record.boosted_by = None
+        else:
+            log.debug(
+                f"{post.id=} in process_boosted_posts but not boosted"
+            )
+
+
 def update_followed_accounts(session: Session, mastapi: MastodonAPI) -> None:
     """
     Retrieve list of followed accounts and update accounts

tests/test_processing_posts.py

@@ -0,0 +1,119 @@
+import os
+import pickle
+
+from dateutil.tz import tzutc
+
+from sqlalchemy import select
+
+from models import (
+    Hashtags,
+    Posts,
+)
+from urma import (
+    process_bookmarked_posts,
+    process_posts_boosted_by_me,
+    process_favourited_posts,
+)
+
+BOOSTED_POST = "tests/boosted.pickle"
+BOOSTING_POST = "tests/boosting.pickle"
+FAVOURITED_POST = "tests/favourited.pickle"
+ME_ID = 109568725613662482
+
+
+def test_process_favourited_no_fave(session):
+    """Test post not marked as favourite"""
+
+    with open(FAVOURITED_POST, "rb") as inp:
+        posts = pickle.load(inp)
+
+    # Sanity check
+    assert len(posts) == 1
+
+    # Set not favourited
+    posts[0]['favourited'] = False
+
+    process_favourited_posts(session, posts, ME_ID)
+
+    all_posts = session.execute(select(Posts)).scalars().all()
+    assert len(all_posts) == 0
+
+
+def test_process_favourited_with_fave(session):
+    """Test post marked as favourite"""
+
+    with open(FAVOURITED_POST, "rb") as inp:
+        posts = pickle.load(inp)
+
+    # Sanity check
+    assert len(posts) == 1
+
+    process_favourited_posts(session, posts, ME_ID)
+
+    all_posts = session.execute(select(Posts)).scalars().all()
+    assert len(all_posts) == 1
+
+    original_post = posts[0]
+    retrieved_post = all_posts[0]
+
+    assert original_post.id == int(retrieved_post.post_id)
+    assert original_post.account.id == int(retrieved_post.account.account_id)
+    assert original_post.created_at == retrieved_post.created_at
+    assert retrieved_post.favourited is True
+
+
+def test_process_post_hashtags(session):
+    """Test hashtags correctly parsed"""
+
+    with open(FAVOURITED_POST, "rb") as inp:
+        posts = pickle.load(inp)
+
+    # Sanity check
+    assert len(posts) == 1
+
+    process_favourited_posts(session, posts, ME_ID)
+
+    all_tags = Hashtags.get_all(session)
+
+    expected = ['fdroid', 'apps', 'android', 'foss', 'free', 'AndroidAppRain']
+    for hashtag in all_tags:
+        assert hashtag.name in expected
+
+
+def test_process_boosts_by_me_no_boost(session):
+    """Test processing boosted posts with no post that's been boosted"""
+
+    with open(BOOSTING_POST, "rb") as inp:
+        post = pickle.load(inp)
+
+    post['reblog'] = None
+
+    process_posts_boosted_by_me(session, [post], ME_ID)
+
+    all_posts = session.execute(select(Posts)).scalars().all()
+    assert len(all_posts) == 0
+
+
+def test_process_boosts_by_me(session):
+    """Test processing posts boosted by me"""
+
+    with open(BOOSTING_POST, "rb") as inp:
+        boosting = pickle.load(inp)
+    with open(BOOSTED_POST, "rb") as inp:
+        boosted = pickle.load(inp)
+
+    process_posts_boosted_by_me(session, [boosting], ME_ID)
+
+    all_posts = session.execute(select(Posts)).scalars().all()
+    assert len(all_posts) == 1
+
+    stored_post = all_posts[0]
+
+    assert boosted.account.id == int(stored_post.account.account_id)
+    assert boosted.id == int(stored_post.post_id)
+
+
+def test_process_favourite_containing_boosted_post(session):
+    """Test processing a favourited post that is a boost"""
+
+    pass