test(instagram): add platform-layer unit tests

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

@@ -0,0 +1,97 @@
+"""Offline budget tests: per-target caps, cross-target de-dup, and the limit guard.
+
+No network. ``fetch_json`` is stubbed with a synthetic profile payload and the
+fan-out proxy holders are replaced with no-ops, so the orchestrator's paging and
+de-dup policy is exercised deterministically.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+import pytest
+
+from app.proprietary.platforms.instagram import scraper
+from app.proprietary.platforms.instagram.schemas import InstagramScrapeInput
+
+
+class _NoopHolder:
+    async def close(self) -> None:
+        return None
+
+
+@pytest.fixture
+def _stub_proxy(monkeypatch):
+    async def _open():
+        return _NoopHolder()
+
+    @asynccontextmanager
+    async def _bind(_holder):
+        yield _holder
+
+    monkeypatch.setattr(scraper, "open_proxy_holder", _open)
+    monkeypatch.setattr(scraper, "bind_proxy_holder", _bind)
+
+
+def _profile_payload(n: int) -> dict:
+    return {
+        "data": {
+            "user": {
+                "id": "9",
+                "username": "acct",
+                "edge_owner_to_timeline_media": {
+                    "count": n,
+                    "edges": [
+                        {"node": {"id": str(i), "shortcode": f"S{i}"}}
+                        for i in range(n)
+                    ],
+                },
+            }
+        }
+    }
+
+
+async def test_per_target_cap_limits_media(_stub_proxy, monkeypatch):
+    async def _fetch(path, params=None):
+        return _profile_payload(50)
+
+    monkeypatch.setattr(scraper, "fetch_json", _fetch)
+    model = InstagramScrapeInput(
+        resultsType="posts",
+        directUrls=["https://www.instagram.com/acct/"],
+        resultsLimit=5,
+    )
+    items = [i async for i in scraper.iter_instagram(model)]
+    assert len(items) == 5
+
+
+async def test_cross_target_dedup_by_id(_stub_proxy, monkeypatch):
+    async def _fetch(path, params=None):
+        return _profile_payload(3)  # both targets return ids 0,1,2
+
+    monkeypatch.setattr(scraper, "fetch_json", _fetch)
+    model = InstagramScrapeInput(
+        resultsType="posts",
+        directUrls=[
+            "https://www.instagram.com/one/",
+            "https://www.instagram.com/two/",
+        ],
+        resultsLimit=10,
+    )
+    items = [i async for i in scraper.iter_instagram(model)]
+    ids = sorted(i["id"] for i in items)
+    assert ids == ["0", "1", "2"]
+
+
+async def test_scrape_instagram_honors_limit(_stub_proxy, monkeypatch):
+    async def _fetch(path, params=None):
+        return _profile_payload(50)
+
+    monkeypatch.setattr(scraper, "fetch_json", _fetch)
+    model = InstagramScrapeInput(
+        resultsType="posts",
+        directUrls=["https://www.instagram.com/acct/"],
+        resultsLimit=100,
+    )
+    items = await scraper.scrape_instagram(model, limit=7)
+    assert len(items) == 7

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

@@ -0,0 +1,308 @@
+"""Offline resilience tests for the Instagram fetch seam and fan-out worker pool.
+
+No network. Fake sessions drive the ``csrftoken`` warm-up + rotate-on-block +
+backoff paths deterministically (in live runs the first IP warms and returns
+200s, so these branches rarely fire). Mirrors the reddit sibling's
+``test_fetch_resilience.py`` shape, adapted to Instagram's cookie warm-up.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from app.proprietary.platforms.instagram import fetch, scraper
+from app.proprietary.platforms.instagram.fetch import (
+    InstagramAccessBlockedError,
+    _current_session,
+    fetch_json,
+)
+
+_PAYLOAD = {"data": {"user": {"username": "natgeo"}}}
+
+
+class _FakePage:
+    def __init__(self, status: int, *, cookies: dict | None = None, payload=None):
+        self.status = status
+        self.cookies = cookies or {}
+        self._payload = payload if payload is not None else _PAYLOAD
+
+    def json(self):
+        return self._payload
+
+    @property
+    def body(self) -> str:
+        return json.dumps(self._payload)
+
+
+class _FakeSession:
+    """One 'IP': the warm-up GET mints csrftoken per flag; endpoint GETs return ``status``."""
+
+    def __init__(self, status: int = 200, *, csrftoken: bool = True, payload=None) -> None:
+        self.status = status
+        self.csrftoken = csrftoken
+        self.payload = payload
+        self.json_calls = 0
+        self.warm_calls = 0
+
+    async def get(self, url, headers=None, cookies=None):
+        if url.rstrip("/") == "https://www.instagram.com":
+            self.warm_calls += 1
+            ck = {"csrftoken": "x", "mid": "y"} if self.csrftoken else {}
+            return _FakePage(200, cookies=ck)
+        self.json_calls += 1
+        return _FakePage(self.status, payload=self.payload)
+
+
+class _FakeHolder:
+    """Holder whose ``rotate()`` advances to the next fake session (a new IP)."""
+
+    def __init__(self, sessions: list[_FakeSession]) -> None:
+        self._sessions = sessions
+        self.session = sessions[0]
+        self.rotations = 0
+        self.warmed = False
+
+    async def rotate(self):
+        self.rotations += 1
+        self.session = self._sessions[min(self.rotations, len(self._sessions) - 1)]
+        self.warmed = False  # cookies bind to the IP: re-warm on the fresh one
+        return self.session
+
+    async def pace(self) -> None:
+        return None
+
+    async def close(self) -> None:
+        return None
+
+
+def _no_sleep(monkeypatch) -> None:
+    async def _noop(_seconds):
+        return None
+
+    monkeypatch.setattr(fetch.asyncio, "sleep", _noop)
+
+
+async def test_warms_then_returns_json():
+    holder = _FakeHolder([_FakeSession(200, csrftoken=True)])
+    token = _current_session.set(holder)
+    try:
+        result = await fetch_json("api/v1/users/web_profile_info/", {"username": "natgeo"})
+    finally:
+        _current_session.reset(token)
+    assert result == _PAYLOAD
+    assert holder.rotations == 0
+    assert holder.session.warm_calls == 1  # warmed exactly once
+
+
+async def test_rotates_when_warm_fails_then_succeeds():
+    holder = _FakeHolder(
+        [_FakeSession(200, csrftoken=False), _FakeSession(200, csrftoken=True)]
+    )
+    token = _current_session.set(holder)
+    try:
+        result = await fetch_json("api/v1/users/web_profile_info/")
+    finally:
+        _current_session.reset(token)
+    assert result == _PAYLOAD
+    assert holder.rotations == 1
+
+
+async def test_raises_when_no_ip_can_warm():
+    holder = _FakeHolder(
+        [_FakeSession(200, csrftoken=False) for _ in range(fetch._MAX_ROTATIONS + 1)]
+    )
+    token = _current_session.set(holder)
+    try:
+        raised = False
+        try:
+            await fetch_json("api/v1/users/web_profile_info/")
+        except InstagramAccessBlockedError:
+            raised = True
+    finally:
+        _current_session.reset(token)
+    assert raised
+    assert holder.rotations == fetch._MAX_ROTATIONS
+
+
+async def test_rotates_and_rewarms_on_403():
+    holder = _FakeHolder([_FakeSession(403), _FakeSession(200)])
+    token = _current_session.set(holder)
+    try:
+        result = await fetch_json("api/v1/users/web_profile_info/")
+    finally:
+        _current_session.reset(token)
+    assert result == _PAYLOAD
+    assert holder.rotations == 1
+    assert holder.session.warm_calls == 1  # re-warmed on the fresh IP
+
+
+async def test_rotates_on_401_login_wall():
+    holder = _FakeHolder([_FakeSession(401), _FakeSession(200)])
+    token = _current_session.set(holder)
+    try:
+        result = await fetch_json("api/v1/users/web_profile_info/")
+    finally:
+        _current_session.reset(token)
+    assert result == _PAYLOAD
+    assert holder.rotations == 1
+
+
+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/")
+    finally:
+        _current_session.reset(token)
+    assert result is None
+    assert holder.rotations == 0
+
+
+async def test_429_backs_off_without_rotating(monkeypatch):
+    _no_sleep(monkeypatch)
+    session = _FakeSession(429)
+
+    async def _get(url, headers=None, cookies=None):
+        if url.rstrip("/") == "https://www.instagram.com":
+            session.warm_calls += 1
+            return _FakePage(200, cookies={"csrftoken": "x"})
+        session.json_calls += 1
+        return _FakePage(429 if session.json_calls == 1 else 200)
+
+    session.get = _get  # type: ignore[method-assign]
+    holder = _FakeHolder([session])
+    token = _current_session.set(holder)
+    try:
+        result = await fetch_json("api/v1/users/web_profile_info/")
+    finally:
+        _current_session.reset(token)
+    assert result == _PAYLOAD
+    assert holder.rotations == 0
+
+
+async def test_persistent_403_raises_blocked(monkeypatch):
+    _no_sleep(monkeypatch)
+    holder = _FakeHolder([_FakeSession(403) for _ in range(fetch._MAX_ROTATIONS + 1)])
+    token = _current_session.set(holder)
+    try:
+        raised = False
+        try:
+            await fetch_json("api/v1/users/web_profile_info/")
+        except InstagramAccessBlockedError:
+            raised = True
+    finally:
+        _current_session.reset(token)
+    assert raised
+    assert holder.rotations == fetch._MAX_ROTATIONS
+
+
+class _TrackingHolder:
+    """Fake fan-out session holder that records whether it was closed."""
+
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+async def test_fan_out_closes_all_sessions_on_early_stop(monkeypatch):
+    holders: list[_TrackingHolder] = []
+
+    async def _fake_open():
+        h = _TrackingHolder()
+        holders.append(h)
+        return h
+
+    @asynccontextmanager
+    async def _fake_bind(_holder):
+        yield _holder
+
+    monkeypatch.setattr(scraper, "open_proxy_holder", _fake_open)
+    monkeypatch.setattr(scraper, "bind_proxy_holder", _fake_bind)
+
+    async def _job(n: int) -> AsyncIterator[dict]:
+        for i in range(5):
+            yield {"job": n, "i": i}
+
+    jobs = [_job(n) for n in range(20)]
+    gen = scraper.fan_out(jobs, concurrency=4)
+    collected = []
+    async for item in gen:
+        collected.append(item)
+        if len(collected) >= 3:
+            break
+    await gen.aclose()
+
+    assert len(collected) >= 3
+    assert holders, "workers should have opened sessions"
+    assert all(h.closed for h in holders), "a worker leaked its proxy session"
+
+
+async def test_fan_out_empty_jobs_is_noop():
+    out = [x async for x in scraper.fan_out([])]
+    assert out == []
+
+
+def _profile_payload(username: str, n: int) -> dict:
+    # IDs namespaced per target so cross-target de-dup doesn't collapse them.
+    return {
+        "data": {
+            "user": {
+                "id": f"u_{username}",
+                "username": username,
+                "edge_owner_to_timeline_media": {
+                    "count": n,
+                    "edges": [{"node": {"id": f"{username}:{i}"}} for i in range(n)],
+                },
+            }
+        }
+    }
+
+
+async def test_scrape_instagram_closes_sessions_when_limit_stops_inflight_workers(
+    monkeypatch,
+):
+    """Hitting ``limit`` must tear down the whole fan-out chain deterministically.
+
+    Regression: closing the outer ``iter_instagram`` generator does NOT
+    synchronously close the inner ``fan_out`` it loops over — CPython defers that
+    to async-gen GC — so without an explicit ``aclosing`` the collector's early
+    ``break`` leaked every warm proxy session that was still mid-fetch. The
+    ``fan_out``-direct test misses this because instant jobs self-drain before
+    cancellation ever runs; here each fetch yields to the loop so workers are
+    genuinely in-flight when the limit trips.
+    """
+    holders: list[_TrackingHolder] = []
+
+    async def _fake_open():
+        h = _TrackingHolder()
+        holders.append(h)
+        return h
+
+    @asynccontextmanager
+    async def _fake_bind(_holder):
+        yield _holder
+
+    async def _fetch(path, params=None):
+        await asyncio.sleep(0)  # yield control: keep sibling workers in-flight
+        username = (params or {}).get("username", "acct")
+        return _profile_payload(username, 5)
+
+    monkeypatch.setattr(scraper, "open_proxy_holder", _fake_open)
+    monkeypatch.setattr(scraper, "bind_proxy_holder", _fake_bind)
+    monkeypatch.setattr(scraper, "fetch_json", _fetch)
+
+    model = scraper.InstagramScrapeInput(
+        resultsType="posts",
+        directUrls=[f"https://www.instagram.com/acct{i}/" for i in range(50)],
+        resultsLimit=5,
+    )
+    items = await scraper.scrape_instagram(model, limit=3)
+
+    assert len(items) == 3
+    assert holders, "workers should have opened sessions"
+    assert all(h.closed for h in holders), "early stop leaked a proxy session"

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

@@ -0,0 +1,153 @@
+"""Offline parser tests: raw web JSON -> flat item dicts.
+
+Synthetic nodes cover the GraphQL ``edge_*`` flattening the anonymous web
+payloads use. A fixture-pinned test runs only when a captured fixture is present
+(the live e2e script dumps trimmed, PII-anonymized fixtures), so the suite stays
+green offline.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from app.proprietary.platforms.instagram.parsers import (
+    parse_comment,
+    parse_hashtag,
+    parse_media,
+    parse_place,
+    parse_profile,
+)
+
+_FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def _edge(nodes: list[dict]) -> dict:
+    return {"edges": [{"node": n} for n in nodes]}
+
+
+def test_parse_media_flattens_edges_and_extracts_tags():
+    node = {
+        "id": "1",
+        "shortcode": "Cabc",
+        "__typename": "GraphImage",
+        "taken_at_timestamp": 1_600_000_000,
+        "edge_media_to_caption": _edge([{"text": "love #nasa shot @buzz"}]),
+        "edge_media_to_comment": {"count": 7},
+        "edge_liked_by": {"count": 42},
+        "owner": {"username": "natgeo", "id": "9"},
+    }
+    item = parse_media(node)
+    assert item["shortCode"] == "Cabc"
+    assert item["type"] == "Image"
+    assert item["hashtags"] == ["nasa"]
+    assert item["mentions"] == ["buzz"]
+    assert item["commentsCount"] == 7
+    assert item["likesCount"] == 42
+    assert item["ownerUsername"] == "natgeo"
+    assert item["url"] == "https://www.instagram.com/p/Cabc/"
+
+
+def test_parse_media_passes_through_hidden_like_count():
+    # Instagram reports -1 when the creator hid likes; never coerce it away.
+    item = parse_media({"id": "1", "edge_liked_by": {"count": -1}})
+    assert item["likesCount"] == -1
+
+
+def test_parse_media_marks_video_type():
+    item = parse_media({"id": "1", "is_video": True, "video_view_count": 99})
+    assert item["type"] == "Video"
+    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",
+        "username": "natgeo",
+        "full_name": "Nat Geo",
+        "edge_followed_by": {"count": 1000},
+        "edge_follow": {"count": 50},
+        "edge_owner_to_timeline_media": {
+            "count": 2,
+            "edges": [{"node": {"id": "p1", "shortcode": "A"}}],
+        },
+    }
+    item = parse_profile(user)
+    assert item["detailKind"] == "profile"
+    assert item["username"] == "natgeo"
+    assert item["followersCount"] == 1000
+    assert item["followsCount"] == 50
+    assert item["postsCount"] == 2
+    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
+
+
+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
+
+
+@pytest.mark.skipif(
+    not (_FIXTURES / "profile.json").exists(),
+    reason="captured fixture absent (run scripts/e2e_instagram_scraper.py to dump)",
+)
+def test_fixture_profile_maps():
+    raw = json.loads((_FIXTURES / "profile.json").read_text())
+    user = raw.get("data", {}).get("user", raw)
+    item = parse_profile(user)
+    assert item["detailKind"] == "profile"
+    assert item["username"]

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

@@ -0,0 +1,106 @@
+"""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
+``url_resolver`` classification/normalization table (``_u/`` and profilecard
+stripping, story→profile, ID-only locations, numeric post-ID flagging).
+"""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.instagram.schemas import (
+    InstagramMediaItem,
+    InstagramScrapeInput,
+)
+from app.proprietary.platforms.instagram.url_resolver import resolve_url
+
+
+def test_input_has_no_auth_fields():
+    # Anonymous-only: the input surface must never expose a login/credential seam.
+    forbidden = {
+        "sessionid",
+        "username",
+        "password",
+        "cookies",
+        "authorization",
+        "proxyConfiguration",
+        "loginCredentials",
+    }
+    assert forbidden.isdisjoint(InstagramScrapeInput.model_fields)
+
+
+def test_input_defaults():
+    model = InstagramScrapeInput()
+    assert model.resultsType == "posts"
+    assert model.searchType == "hashtag"
+    assert model.directUrls == []
+    assert model.addParentData is False
+
+
+def test_input_allows_extra_inert_fields():
+    # A reference field the acquisition layer doesn't source is accepted, not rejected.
+    model = InstagramScrapeInput(enhanceUserSearchWithFacebookPage="x")
+    assert model.model_dump().get("enhanceUserSearchWithFacebookPage") == "x"
+
+
+def test_media_item_to_output_keeps_none_keys():
+    out = InstagramMediaItem(id="123", shortCode="abc").to_output()
+    assert out["id"] == "123"
+    assert out["shortCode"] == "abc"
+    # Unsourced fields stay present as None / [] for additive parity.
+    assert out["likesCount"] is None
+    assert out["requestErrorMessages"] == []
+
+
+def test_resolve_profile():
+    r = resolve_url("https://www.instagram.com/natgeo/")
+    assert r.kind == "profile"
+    assert r.value == "natgeo"
+
+
+def test_resolve_bare_profile_id():
+    r = resolve_url("natgeo")
+    assert r.kind == "profile" and r.value == "natgeo"
+
+
+def test_resolve_post_and_reel():
+    r = resolve_url("https://www.instagram.com/p/Cabc123/")
+    assert r.kind == "post" and r.value == "Cabc123" and r.numeric_post_id is False
+    r = resolve_url("https://www.instagram.com/reel/Cxyz/")
+    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/"
+    )
+    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():
+    stripped_u = resolve_url("https://www.instagram.com/_u/natgeo/")
+    assert stripped_u.kind == "profile" and stripped_u.value == "natgeo"
+    card = resolve_url("https://www.instagram.com/natgeo/profilecard/")
+    assert card.kind == "profile" and card.value == "natgeo"
+
+
+def test_resolve_story_reduces_to_profile():
+    r = resolve_url("https://www.instagram.com/stories/natgeo/12345/")
+    assert r.kind == "profile" and r.value == "natgeo"
+
+
+def test_resolve_numeric_post_id_flagged():
+    r = resolve_url("https://www.instagram.com/p/12345/")
+    assert r.kind == "post" and r.numeric_post_id is True
+
+
+def test_resolve_rejects_non_instagram_host():
+    assert resolve_url("https://example.com/natgeo/") is None