test(instagram): add capability unit tests

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"

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

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"]
+        )