links/tests/test_bookmarks_api.py

"""Integration tests for the Bookmarks REST API (/api/bookmarks)."""

import pytest

# a minimal full-export object as produced by x-bookmarks-exporter
SAMPLE_EXPORT = {
    "id": "1234567890",
    "url": "https://x.com/user/status/1234567890",
    "author": {
        "screen_name": "testuser",
        "name": "Test User",
        "profile_image_url": "https://example.com/avatar.jpg",
    },
    "text": "Hello world",
    "media": [{"type": "photo", "media_url": "https://example.com/p.jpg",
               "media_large": "https://example.com/p.jpg"}],
    "card": None,
    "quoted": None,
    "retweet": None,
    "favorite_count": 7, "view_count": 21, "bookmark_count": 1,
}


@pytest.mark.django_db
class TestBookmarksAPI:
    BASE = "/api/bookmarks"

    def test_list_empty(self, api_client):
        r = api_client.get(f"{self.BASE}")
        assert r.status_code == 200
        assert r.json()["count"] == 0

    def test_create_then_list_and_detail(self, api_client):
        r = api_client.post(self.BASE, {
            "tweet_id": "111",
            "url": "https://x.com/u/status/111",
            "author_screen_name": "u",
            "text": "hi", "has_photo": True,
        }, format="json")
        assert r.status_code == 201, r.content
        created = r.json()
        assert created["tweet_id"] == "111"

        list_r = api_client.get(self.BASE)
        assert list_r.json()["count"] == 1

        det = api_client.get(f"{self.BASE}/{created['id']}")
        assert det.status_code == 200
        assert det.json()["has_photo"] is True

    def test_filter_has_video(self, api_client):
        api_client.post(self.BASE, {"tweet_id": "v1", "has_video": True}, format="json")
        api_client.post(self.BASE, {"tweet_id": "p1", "has_photo": True}, format="json")
        r = api_client.get(f"{self.BASE}", {"has_video": "true"})
        assert r.json()["count"] == 1
        assert r.json()["results"][0]["tweet_id"] == "v1"

    def test_search_q(self, api_client):
        api_client.post(self.BASE, {"tweet_id": "a1", "text": "python tips"}, format="json")
        api_client.post(self.BASE, {"tweet_id": "a2", "text": "rust tricks"}, format="json")
        r = api_client.get(f"{self.BASE}", {"q": "python"})
        assert r.json()["count"] == 1
        assert r.json()["results"][0]["tweet_id"] == "a1"

    def test_update_summary(self, api_client):
        created = api_client.post(self.BASE, {"tweet_id": "u1"}, format="json").json()
        r = api_client.patch(f"{self.BASE}/{created['id']}", {"summary": "AI summary"},
                             format="json")
        assert r.status_code == 200
        assert r.json()["summary"] == "AI summary"

    def test_delete(self, api_client):
        created = api_client.post(self.BASE, {"tweet_id": "d1"}, format="json").json()
        r = api_client.delete(f"{self.BASE}/{created['id']}")
        assert r.status_code == 204

    def test_bulk_import_upserts(self, api_client):
        r = api_client.post(f"{self.BASE}/import",
                            {"bookmarks": [SAMPLE_EXPORT]},
                            format="json")
        assert r.status_code == 200, r.content
        assert r.json()["created"] == 1
        # second import with changed content → update
        r2 = api_client.post(f"{self.BASE}/import",
                            {"bookmarks": [
                                {**SAMPLE_EXPORT, "text": "updated"}]},
                            format="json")
        assert r2.json()["updated"] == 1
        assert r2.json()["created"] == 0
        from links.models import Bookmark
        assert Bookmark.objects.count() == 1
        assert Bookmark.objects.first().text == "updated"

    def test_bulk_import_unchanged_skips_save(self, api_client):
        """Re-importing identical content should count as 'unchanged', not 'updated'."""
        # first import
        r1 = api_client.post(f"{self.BASE}/import",
                             {"bookmarks": [SAMPLE_EXPORT]},
                             format="json")
        assert r1.json()["created"] == 1
        # second import with identical content
        r2 = api_client.post(f"{self.BASE}/import",
                             {"bookmarks": [SAMPLE_EXPORT]},
                             format="json")
        assert r2.status_code == 200, r2.content
        assert r2.json()["unchanged"] == 1
        assert r2.json()["updated"] == 0
        assert r2.json()["created"] == 0
        from links.models import Bookmark
        bm = Bookmark.objects.first()
        # last_imported_at should be set even when unchanged
        assert bm.last_imported_at is not None
        # content_hash should be populated
        assert bm.content_hash != ''

    def test_bulk_upsert_array_unchanged(self, api_client):
        """Bulk array endpoint should also detect unchanged content."""
        r1 = api_client.post(f"{self.BASE}/bulk",
                             [{"tweet_id": "ub1", "text": "same"}],
                             format="json")
        assert r1.json()["created"] == 1
        # re-import identical
        r2 = api_client.post(f"{self.BASE}/bulk",
                             [{"tweet_id": "ub1", "text": "same"}],
                             format="json")
        assert r2.json()["unchanged"] == 1
        assert r2.json()["updated"] == 0

    def test_bulk_upsert_array(self, api_client):
        r = api_client.post(f"{self.BASE}/bulk",
                            [{"tweet_id": "b1", "text": "one"},
                             {"tweet_id": "b2", "text": "two"}],
                            format="json")
        assert r.status_code == 200
        assert r.json()["created"] == 2
        r2 = api_client.post(f"{self.BASE}/bulk",
                             [{"tweet_id": "b1", "text": "one-edit"}],
                             format="json")
        assert r2.json()["updated"] == 1
        from links.models import Bookmark
        assert Bookmark.objects.get(tweet_id="b1").text == "one-edit"

    def test_stats(self, api_client):
        api_client.post(self.BASE, {"tweet_id": "s1", "has_video": True}, format="json")
        api_client.post(self.BASE, {"tweet_id": "s2", "has_photo": True}, format="json")
        r = api_client.get(f"{self.BASE}/stats")
        assert r.status_code == 200
        s = r.json()
        assert s["total"] == 2
        assert s["has_video"] == 1
        assert s["has_photo"] == 1