test(instagram): update platform tests and add discovery fixtures

surfsense_backend/tests/unit/platforms/instagram/test_discovery.py

@@ -0,0 +1,78 @@
+"""Offline tests for Google-backed Instagram discovery.
+
+Discovery is profile-only (hashtag/place feeds are login-walled). A valid handle
+resolves directly; any other query falls back to the ``google_search`` platform
+(``site:instagram.com``), classifying organic results with ``resolve_url`` and
+keeping only profile hits. These tests inject a fake ``scrape_serps`` so there is
+no network: they pin the classification, de-dup, and ``limit`` cap.
+"""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.instagram import scraper
+
+
+def _fake_serps(*organic_urls: str):
+    async def _scrape_serps(input_model, *, limit=None):
+        assert input_model.site == "instagram.com"
+        return [{"organicResults": [{"url": u} for u in organic_urls]}]
+
+    return _scrape_serps
+
+
+async def test_google_discovery_keeps_only_profiles(monkeypatch):
+    # A non-handle query goes to Google; only profile URLs survive (hashtag /
+    # post / non-instagram results are dropped since discovery is profile-only).
+    monkeypatch.setattr(
+        scraper,
+        "scrape_serps",
+        _fake_serps(
+            "https://www.instagram.com/natgeo/",
+            "https://www.instagram.com/explore/tags/travel/",
+            "https://www.instagram.com/p/ABC123/",
+            "https://example.com/not-instagram",
+        ),
+    )
+    targets = await scraper._discover(
+        "nat geo photos", search_type="profile", limit=10
+    )
+    assert [(t.kind, t.value) for t in targets] == [("profile", "natgeo")]
+
+
+async def test_google_discovery_dedupes(monkeypatch):
+    monkeypatch.setattr(
+        scraper,
+        "scrape_serps",
+        _fake_serps(
+            "https://www.instagram.com/natgeo/",
+            "https://www.instagram.com/natgeo/",
+        ),
+    )
+    targets = await scraper._discover(
+        "nat geo photos", search_type="profile", limit=10
+    )
+    assert len(targets) == 1
+
+
+async def test_google_discovery_respects_limit(monkeypatch):
+    monkeypatch.setattr(
+        scraper,
+        "scrape_serps",
+        _fake_serps(
+            "https://www.instagram.com/a_a/",
+            "https://www.instagram.com/b_b/",
+            "https://www.instagram.com/c_c/",
+        ),
+    )
+    targets = await scraper._discover("some brand name", search_type="profile", limit=2)
+    assert [t.value for t in targets] == ["a_a", "b_b"]
+
+
+async def test_discover_profile_handle_fast_path_skips_google(monkeypatch):
+    # A valid handle resolves directly without touching Google.
+    async def _boom(input_model, *, limit=None):
+        raise AssertionError("Google should not be called for a valid handle")
+
+    monkeypatch.setattr(scraper, "scrape_serps", _boom)
+    targets = await scraper._discover("messi", search_type="user", limit=10)
+    assert [(t.kind, t.value) for t in targets] == [("profile", "messi")]

surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py

@@ -172,7 +172,7 @@ async def test_login_redirect_fails_fast_without_rotating():
     try:
         raised = False
         try:
-            await fetch_json("api/v1/tags/web_info/", {"tag_name": "travel"})
+            await fetch_json("api/v1/users/web_profile_info/", {"username": "natgeo"})
         except InstagramAccessBlockedError:
             raised = True
     finally:
@@ -185,7 +185,7 @@ async def test_404_returns_none_without_rotating():
     holder = _FakeHolder([_FakeSession(404), _FakeSession(200)])
     token = _current_session.set(holder)
     try:
-        result = await fetch_json("api/v1/tags/web_info/")
+        result = await fetch_json("api/v1/users/web_profile_info/")
     finally:
         _current_session.reset(token)
     assert result is None
@@ -399,12 +399,22 @@ async def test_discover_profile_is_anonymous_handle_lookup():
     assert [(t.kind, t.value) for t in targets] == [("profile", "messi")]
 
 
-async def test_discover_hashtag_search_blocks_anonymously():
-    # hashtag/place keyword discovery has no anonymous endpoint at all, so it must
-    # fail loud (clear message) rather than return a misleading empty success.
-    raised = False
-    try:
-        await scraper._discover("travel", search_type="hashtag", limit=10)
-    except InstagramAccessBlockedError:
-        raised = True
-    assert raised
+async def test_discover_nonhandle_routes_through_google(monkeypatch):
+    # A non-handle profile query goes through Google (site:instagram.com) and
+    # classifies the organic results into profile targets (the only kind now).
+    async def _fake_scrape_serps(input_model, *, limit=None):
+        assert input_model.site == "instagram.com"
+        return [
+            {
+                "organicResults": [
+                    {"url": "https://www.instagram.com/natgeo/"},
+                    {"url": "https://www.instagram.com/p/Cabc/"},  # wrong kind
+                ]
+            }
+        ]
+
+    monkeypatch.setattr(scraper, "scrape_serps", _fake_scrape_serps)
+    targets = await scraper._discover(
+        "national geographic", search_type="profile", limit=10
+    )
+    assert [(t.kind, t.value) for t in targets] == [("profile", "natgeo")]

surfsense_backend/tests/unit/platforms/instagram/test_parsers.py

@@ -14,10 +14,8 @@ from pathlib import Path
 import pytest
 
 from app.proprietary.platforms.instagram.parsers import (
-    parse_comment,
-    parse_hashtag,
     parse_media,
-    parse_place,
+    parse_post,
     parse_profile,
 )
 
@@ -62,23 +60,6 @@ def test_parse_media_marks_video_type():
     assert item["videoViewCount"] == 99
 
 
-def test_parse_comment():
-    node = {
-        "id": "c1",
-        "text": "nice",
-        "created_at": 1_600_000_000,
-        "shortcode": "Cabc",
-        "owner": {"username": "bob", "id": "5"},
-        "edge_liked_by": {"count": 3},
-    }
-    item = parse_comment(node, post_url="https://www.instagram.com/p/Cabc/")
-    assert item["id"] == "c1"
-    assert item["text"] == "nice"
-    assert item["ownerUsername"] == "bob"
-    assert item["likesCount"] == 3
-    assert item["postUrl"] == "https://www.instagram.com/p/Cabc/"
-
-
 def test_parse_profile_flattens_counts_and_latest_posts():
     user = {
         "id": "9",
@@ -100,45 +81,63 @@ def test_parse_profile_flattens_counts_and_latest_posts():
     assert len(item["latestPosts"]) == 1
 
 
-def test_parse_hashtag():
-    data = {
-        "data": {
-            "id": "h1",
-            "name": "crossfit",
-            "edge_hashtag_to_media": {
-                "count": 5,
-                "edges": [{"node": {"id": "m1", "shortcode": "A"}}],
-            },
-            "edge_hashtag_to_top_posts": {
-                "edges": [{"node": {"id": "t1", "shortcode": "B"}}]
-            },
-        }
-    }
-    item = parse_hashtag(data)
-    assert item["detailKind"] == "hashtag"
-    assert item["name"] == "crossfit"
-    assert item["postsCount"] == 5
-    assert len(item["topPosts"]) == 1
-    assert len(item["posts"]) == 1
+_POST_URL = "https://www.instagram.com/p/Cabc/"
 
 
-def test_parse_place():
-    data = {
-        "location": {
-            "id": "7538318",
-            "name": "Copenhagen",
-            "slug": "copenhagen",
-            "edge_location_to_media": {
-                "count": 3,
-                "edges": [{"node": {"id": "m1", "shortcode": "A"}}],
-            },
-        }
-    }
-    item = parse_place(data)
-    assert item["detailKind"] == "place"
-    assert item["name"] == "Copenhagen"
-    assert item["location_id"] == "7538318"
-    assert len(item["posts"]) == 1
+def test_parse_post_prefers_ldjson():
+    html = """
+    <html><head>
+    <script type="application/ld+json">
+    {"@type": "VideoObject", "articleBody": "sunset over #bali with @friend",
+     "uploadDate": "2024-01-02T03:04:05Z",
+     "author": {"@type": "Person", "alternateName": "@natgeo"},
+     "video": {"contentUrl": "https://cdn/v.mp4"},
+     "image": {"url": "https://cdn/i.jpg"},
+     "interactionStatistic": [
+       {"interactionType": "https://schema.org/LikeAction", "userInteractionCount": 4200},
+       {"interactionType": "https://schema.org/CommentAction", "userInteractionCount": 37}
+     ]}
+    </script>
+    </head></html>
+    """
+    item = parse_post(html, url=_POST_URL, shortcode="Cabc")
+    assert item is not None
+    assert item["type"] == "Video"
+    assert item["shortCode"] == "Cabc"
+    assert item["url"] == _POST_URL
+    assert item["ownerUsername"] == "natgeo"
+    assert item["caption"] == "sunset over #bali with @friend"
+    assert item["hashtags"] == ["bali"]
+    assert item["mentions"] == ["friend"]
+    assert item["likesCount"] == 4200
+    assert item["commentsCount"] == 37
+    assert item["videoUrl"] == "https://cdn/v.mp4"
+    assert item["timestamp"] == "2024-01-02T03:04:05Z"
+
+
+def test_parse_post_falls_back_to_og_meta():
+    html = """
+    <html><head>
+    <meta property="og:type" content="video.other" />
+    <meta property="og:image" content="https://cdn/i.jpg" />
+    <meta property="og:description"
+      content="1,234 likes, 56 comments - natgeo on January 2, 2024: &quot;a caption&quot;" />
+    </head></html>
+    """
+    item = parse_post(html, url=_POST_URL, shortcode="Cabc")
+    assert item is not None
+    assert item["likesCount"] == 1234
+    assert item["commentsCount"] == 56
+    assert item["displayUrl"] == "https://cdn/i.jpg"
+    assert item["type"] == "Video"
+
+
+def test_parse_post_returns_none_without_surfaces():
+    # A login interstitial / empty doc carries neither ld+json nor og -> None,
+    # never a silent empty-success item.
+    assert parse_post("<html><body>login</body></html>", url=_POST_URL) is None
+    assert parse_post(None, url=_POST_URL) is None
+    assert parse_post("", url=_POST_URL) is None
 
 
 @pytest.mark.skipif(
@@ -151,3 +150,14 @@ def test_fixture_profile_maps():
     item = parse_profile(user)
     assert item["detailKind"] == "profile"
     assert item["username"]
+
+
+@pytest.mark.skipif(
+    not (_FIXTURES / "post.json").exists(),
+    reason="captured fixture absent (run the single-post probe to dump /p/ HTML)",
+)
+def test_fixture_post_maps():
+    raw = json.loads((_FIXTURES / "post.json").read_text())
+    item = parse_post(raw["html"], url=raw["url"], shortcode=raw.get("shortcode"))
+    assert item is not None, "captured /p/ HTML produced no media item"
+    assert item["url"] == raw["url"]

surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py

@@ -1,9 +1,10 @@
 """Offline skeleton tests: input surface parity + URL classification.
 
 No network. Locks the two invariants the reference-compatible surface promises —
-no auth fields ever, and additive ``extra="allow"`` parity — plus the full
+no auth fields ever, and additive ``extra="allow"`` parity — plus the
 ``url_resolver`` classification/normalization table (``_u/`` and profilecard
-stripping, story→profile, ID-only locations, numeric post-ID flagging).
+stripping, story→profile, numeric post-ID flagging). Hashtag/place URLs are
+login-walled and deliberately resolve to ``None``.
 """
 
 from __future__ import annotations
@@ -32,7 +33,7 @@ def test_input_has_no_auth_fields():
 def test_input_defaults():
     model = InstagramScrapeInput()
     assert model.resultsType == "posts"
-    assert model.searchType == "hashtag"
+    assert model.searchType == "profile"
     assert model.directUrls == []
     assert model.addParentData is False
 
@@ -70,19 +71,14 @@ def test_resolve_post_and_reel():
     assert r.kind == "reel" and r.value == "Cxyz"
 
 
-def test_resolve_hashtag():
-    r = resolve_url("https://www.instagram.com/explore/tags/crossfit/")
-    assert r.kind == "hashtag" and r.value == "crossfit"
-
-
-def test_resolve_place_with_slug_and_id_only():
-    with_slug = resolve_url(
-        "https://www.instagram.com/explore/locations/7538318/copenhagen/"
+def test_resolve_hashtag_and_place_unsupported():
+    # Login-walled surfaces: they must resolve to None so the orchestrator skips
+    # them rather than building a target that can only return a login wall.
+    assert resolve_url("https://www.instagram.com/explore/tags/crossfit/") is None
+    assert (
+        resolve_url("https://www.instagram.com/explore/locations/7538318/copenhagen/")
+        is None
     )
-    assert with_slug.kind == "place" and with_slug.value == "7538318"
-    assert with_slug.slug == "copenhagen"
-    id_only = resolve_url("https://www.instagram.com/explore/locations/7538318/")
-    assert id_only.kind == "place" and id_only.value == "7538318"
 
 
 def test_resolve_strips_u_and_profilecard():