From 2e66e714e2cd3818f3224232ef9b2502722d4c3f Mon Sep 17 00:00:00 2001 From: Anish Sarkar <104695310+AnishSarkar22@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:02:48 +0530 Subject: [PATCH] test(instagram): add capability unit tests --- .../unit/capabilities/instagram/__init__.py | 0 .../capabilities/instagram/test_executor.py | 104 ++++++++++++++++++ .../capabilities/instagram/test_registry.py | 35 ++++++ .../capabilities/instagram/test_schemas.py | 80 ++++++++++++++ 4 files changed, 219 insertions(+) create mode 100644 surfsense_backend/tests/unit/capabilities/instagram/__init__.py create mode 100644 surfsense_backend/tests/unit/capabilities/instagram/test_executor.py create mode 100644 surfsense_backend/tests/unit/capabilities/instagram/test_registry.py create mode 100644 surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py diff --git a/surfsense_backend/tests/unit/capabilities/instagram/__init__.py b/surfsense_backend/tests/unit/capabilities/instagram/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py b/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py new file mode 100644 index 000000000..9087b0bcf --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py @@ -0,0 +1,104 @@ +"""Executor tests: lean verb input → ``InstagramScrapeInput`` mapping + wrapping. + +A fake scraper captures the actor input the executor built (no network), so the +snake_case→camelCase mapping and the ``InstagramAccessBlockedError`` → +``ForbiddenError`` translation are asserted deterministically. +""" + +from __future__ import annotations + +import pytest + +from app.capabilities.instagram.comments.executor import build_comments_executor +from app.capabilities.instagram.comments.schemas import CommentsInput, CommentsOutput +from app.capabilities.instagram.details.executor import build_details_executor +from app.capabilities.instagram.details.schemas import DetailsInput, DetailsOutput +from app.capabilities.instagram.scrape.executor import build_scrape_executor +from app.capabilities.instagram.scrape.schemas import ScrapeInput, ScrapeOutput +from app.exceptions import ForbiddenError +from app.proprietary.platforms.instagram import ( + InstagramAccessBlockedError, + InstagramScrapeInput, +) + +pytestmark = pytest.mark.unit + + +class _FakeScraper: + def __init__(self, items: list[dict]) -> None: + self.items = items + self.calls: list[tuple[InstagramScrapeInput, int | None]] = [] + + async def __call__(self, actor_input, *, limit=None): + self.calls.append((actor_input, limit)) + return self.items + + +async def test_scrape_maps_urls_and_wraps_items(): + fake = _FakeScraper([{"id": "1", "shortCode": "abc", "caption": "hi"}]) + execute = build_scrape_executor(fake) + out = await execute(ScrapeInput(urls=["https://www.instagram.com/natgeo/"])) + assert isinstance(out, ScrapeOutput) + assert out.items[0].shortCode == "abc" + actor_input, limit = fake.calls[0] + assert actor_input.resultsType == "posts" + assert actor_input.directUrls == ["https://www.instagram.com/natgeo/"] + assert actor_input.search == "" + assert limit == 10 # default max_items forwarded as the collector limit + + +async def test_scrape_joins_search_queries(): + fake = _FakeScraper([]) + execute = build_scrape_executor(fake) + await execute(ScrapeInput(search_queries=["fit", "gym"], search_type="hashtag")) + actor_input, _ = fake.calls[0] + assert actor_input.search == "fit,gym" + assert actor_input.searchType == "hashtag" + assert actor_input.directUrls == [] + + +async def test_scrape_access_blocked_maps_to_forbidden(): + async def _blocked(actor_input, *, limit=None): + raise InstagramAccessBlockedError("login wall") + + execute = build_scrape_executor(_blocked) + with pytest.raises(ForbiddenError): + await execute(ScrapeInput(urls=["x"])) + + +async def test_comments_maps_flags(): + fake = _FakeScraper([{"id": "c1", "text": "nice"}]) + execute = build_comments_executor(fake) + out = await execute( + CommentsInput( + urls=["https://www.instagram.com/p/Cabc/"], + newest_first=True, + include_replies=True, + max_comments_per_post=25, + ) + ) + assert isinstance(out, CommentsOutput) + assert out.items[0].text == "nice" + actor_input, _ = fake.calls[0] + assert actor_input.resultsType == "comments" + assert actor_input.isNewestComments is True + assert actor_input.includeNestedComments is True + assert actor_input.resultsLimit == 25 + + +async def test_details_maps_and_wraps_discriminated_items(): + fake = _FakeScraper( + [ + { + "detailKind": "profile", + "username": "natgeo", + "url": "https://www.instagram.com/natgeo/", + } + ] + ) + execute = build_details_executor(fake) + out = await execute(DetailsInput(urls=["https://www.instagram.com/natgeo/"])) + assert isinstance(out, DetailsOutput) + assert out.items[0].username == "natgeo" + actor_input, _ = fake.calls[0] + assert actor_input.resultsType == "details" diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py b/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py new file mode 100644 index 000000000..89257253a --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py @@ -0,0 +1,35 @@ +"""The instagram namespace registers its three verbs for the doors/agent to read. + +Unlike the stale reddit assertion (``billing_unit is None``), these assert the +real meters — the Capability definitions are the source of truth. +""" + +from __future__ import annotations + +import pytest + +from app.capabilities import ( + instagram, # noqa: F401 — importing the namespace registers its verbs +) +from app.capabilities.core.store import get_capability +from app.capabilities.core.types import BillingUnit + +pytestmark = pytest.mark.unit + + +def test_instagram_scrape_registered_with_item_meter(): + cap = get_capability("instagram.scrape") + assert cap.name == "instagram.scrape" + assert cap.billing_unit is BillingUnit.INSTAGRAM_ITEM + + +def test_instagram_comments_registered_with_comment_meter(): + cap = get_capability("instagram.comments") + assert cap.name == "instagram.comments" + assert cap.billing_unit is BillingUnit.INSTAGRAM_COMMENT + + +def test_instagram_details_registered_with_item_meter(): + cap = get_capability("instagram.details") + assert cap.name == "instagram.details" + assert cap.billing_unit is BillingUnit.INSTAGRAM_ITEM diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py b/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py new file mode 100644 index 000000000..13efb8a48 --- /dev/null +++ b/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py @@ -0,0 +1,80 @@ +"""``instagram.*`` input guards: source exclusivity and bounded batches.""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from app.capabilities.instagram.comments.schemas import CommentsInput +from app.capabilities.instagram.details.schemas import DetailsInput, DetailsOutput +from app.capabilities.instagram.scrape.schemas import ( + MAX_INSTAGRAM_ITEMS, + MAX_INSTAGRAM_SOURCES, + ScrapeInput, +) + +pytestmark = pytest.mark.unit + + +def test_scrape_rejects_no_source(): + with pytest.raises(ValidationError): + ScrapeInput() + + +def test_scrape_rejects_both_sources(): + with pytest.raises(ValidationError): + ScrapeInput(urls=["https://www.instagram.com/natgeo/"], search_queries=["fit"]) + + +def test_scrape_accepts_urls_only(): + payload = ScrapeInput(urls=["https://www.instagram.com/natgeo/"]) + assert payload.search_queries == [] + assert payload.estimated_units == payload.max_items + + +def test_scrape_bounds(): + with pytest.raises(ValidationError): + ScrapeInput( + urls=["https://www.instagram.com/x/"], + max_items=MAX_INSTAGRAM_ITEMS + 1, + ) + with pytest.raises(ValidationError): + ScrapeInput( + urls=[ + f"https://www.instagram.com/u{i}/" + for i in range(MAX_INSTAGRAM_SOURCES + 1) + ] + ) + + +def test_comments_requires_urls_and_caps_at_50(): + with pytest.raises(ValidationError): + CommentsInput(urls=[]) + with pytest.raises(ValidationError): + CommentsInput( + urls=["https://www.instagram.com/p/Cabc/"], max_comments_per_post=51 + ) + ok = CommentsInput( + urls=["https://www.instagram.com/p/Cabc/"], max_comments_per_post=50 + ) + assert ok.max_comments_per_post == 50 + + +def test_details_discriminated_union_selects_by_detail_kind(): + out = DetailsOutput( + items=[ + {"detailKind": "profile", "username": "natgeo"}, + {"detailKind": "hashtag", "name": "fit"}, + {"detailKind": "place", "name": "Copenhagen"}, + ] + ) + kinds = [type(i).__name__ for i in out.items] + assert kinds == ["InstagramProfile", "InstagramHashtag", "InstagramPlace"] + assert out.billable_units == 3 + + +def test_details_rejects_both_sources(): + with pytest.raises(ValidationError): + DetailsInput( + urls=["https://www.instagram.com/natgeo/"], search_queries=["x"] + )