From 9da136070d638ce7df2befa81245bd957d564969 Mon Sep 17 00:00:00 2001 From: Anish Sarkar <104695310+AnishSarkar22@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:02:41 +0530 Subject: [PATCH] test(instagram): add platform-layer unit tests --- .../unit/platforms/instagram/__init__.py | 0 .../unit/platforms/instagram/test_budget.py | 97 ++++++ .../instagram/test_fetch_resilience.py | 308 ++++++++++++++++++ .../unit/platforms/instagram/test_parsers.py | 153 +++++++++ .../unit/platforms/instagram/test_skeleton.py | 106 ++++++ 5 files changed, 664 insertions(+) create mode 100644 surfsense_backend/tests/unit/platforms/instagram/__init__.py create mode 100644 surfsense_backend/tests/unit/platforms/instagram/test_budget.py create mode 100644 surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py create mode 100644 surfsense_backend/tests/unit/platforms/instagram/test_parsers.py create mode 100644 surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py diff --git a/surfsense_backend/tests/unit/platforms/instagram/__init__.py b/surfsense_backend/tests/unit/platforms/instagram/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_budget.py b/surfsense_backend/tests/unit/platforms/instagram/test_budget.py new file mode 100644 index 000000000..ea500be40 --- /dev/null +++ b/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 diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py b/surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py new file mode 100644 index 000000000..eefae2fd9 --- /dev/null +++ b/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" diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_parsers.py b/surfsense_backend/tests/unit/platforms/instagram/test_parsers.py new file mode 100644 index 000000000..6f6c1411e --- /dev/null +++ b/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"] diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py b/surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py new file mode 100644 index 000000000..59017c81c --- /dev/null +++ b/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