| 1 | """Pipeline gating, flag handling, and artifact wiring for the Meta Ads source.""" |
| 2 | |
| 3 | import sys |
| 4 | from unittest.mock import patch |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | import last30days |
| 9 | from lib import env, pipeline |
| 10 | |
| 11 | |
| 12 | class TestMetaAdsSourceGating: |
| 13 | """Dual-gated on the Amazon precedent: key present AND the run asks.""" |
| 14 | |
| 15 | def _config(self, include=""): |
| 16 | return {"SCRAPECREATORS_API_KEY": "fake-key", "INCLUDE_SOURCES": include} |
| 17 | |
| 18 | def test_unavailable_without_a_key_even_when_requested(self): |
| 19 | available = pipeline.available_sources( |
| 20 | {"INCLUDE_SOURCES": "meta_ads"}, ["meta_ads"] |
| 21 | ) |
| 22 | assert "meta_ads" not in available |
| 23 | |
| 24 | def test_unavailable_with_a_key_when_not_requested(self): |
| 25 | # Holding a ScrapeCreators key for TikTok must not start spending |
| 26 | # credits resolving advertisers. |
| 27 | assert "meta_ads" not in pipeline.available_sources(self._config(), None) |
| 28 | |
| 29 | def test_available_via_per_run_request(self): |
| 30 | assert "meta_ads" in pipeline.available_sources(self._config(), ["meta_ads"]) |
| 31 | |
| 32 | def test_available_via_durable_include_sources(self): |
| 33 | assert "meta_ads" in pipeline.available_sources(self._config("meta_ads"), None) |
| 34 | |
| 35 | def test_exclude_sources_wins_over_activation(self): |
| 36 | config = self._config("meta_ads") |
| 37 | config["EXCLUDE_SOURCES"] = "meta_ads" |
| 38 | assert "meta_ads" not in pipeline.available_sources(config, ["meta_ads"]) |
| 39 | |
| 40 | def test_never_inferred_from_a_brand_shaped_topic(self): |
| 41 | # Topic shape must not activate it: keyword ad search on the wrong |
| 42 | # topic resolves the wrong company at full credit cost. |
| 43 | assert "meta_ads" not in pipeline.available_sources(self._config(), []) |
| 44 | |
| 45 | def test_capped_at_one_fetch_per_run(self): |
| 46 | assert pipeline.MAX_SOURCE_FETCHES["meta_ads"] == 1 |
| 47 | |
| 48 | def test_exempt_from_thin_source_retry(self): |
| 49 | # A brand that genuinely ran two creatives is a complete result; a |
| 50 | # retry would re-resolve the page and re-spend the discovery credit. |
| 51 | assert "meta_ads" in pipeline.THIN_RETRY_EXEMPT |
| 52 | |
| 53 | |
| 54 | class TestSearchFlag: |
| 55 | def test_canonical_token_is_accepted(self): |
| 56 | assert "meta_ads" in last30days.parse_search_flag("reddit,x,meta_ads") |
| 57 | |
| 58 | @pytest.mark.parametrize("alias", ["meta", "meta-ads"]) |
| 59 | def test_aliases_resolve(self, alias): |
| 60 | assert "meta_ads" in last30days.parse_search_flag(f"reddit,{alias}") |
| 61 | |
| 62 | def test_source_is_in_the_known_source_registry(self): |
| 63 | # parse_search_flag rejects any token absent from this list, so a |
| 64 | # missing entry silently breaks the per-run activation path. |
| 65 | assert "meta_ads" in pipeline.MOCK_AVAILABLE_SOURCES |
| 66 | |
| 67 | |
| 68 | class TestPageOverrideParsing: |
| 69 | def test_bare_numeric_page_id(self): |
| 70 | assert last30days.parse_meta_ads_page("123456789012345") == "123456789012345" |
| 71 | |
| 72 | def test_ad_library_url(self): |
| 73 | url = ( |
| 74 | "https://www.facebook.com/ads/library/" |
| 75 | "?active_status=all&view_all_page_id=123456789012345" |
| 76 | ) |
| 77 | assert last30days.parse_meta_ads_page(url) == "123456789012345" |
| 78 | |
| 79 | def test_vanity_url_is_rejected(self): |
| 80 | # A vanity handle is not a page id: one live check resolved a |
| 81 | # brand-looking handle to a private person's profile. |
| 82 | assert last30days.parse_meta_ads_page("https://facebook.com/somebrand") == "" |
| 83 | |
| 84 | def test_blank_is_rejected(self): |
| 85 | assert last30days.parse_meta_ads_page(" ") == "" |
| 86 | |
| 87 | def test_short_numeric_string_is_rejected(self): |
| 88 | assert last30days.parse_meta_ads_page("42") == "" |
| 89 | |
| 90 | |
| 91 | class TestEnvContract: |
| 92 | def test_country_key_is_resolvable(self): |
| 93 | config = env.get_config() |
| 94 | assert "LAST30DAYS_META_ADS_COUNTRY" in config |
| 95 | |
| 96 | def test_country_defaults_to_us(self): |
| 97 | assert env.get_config().get("LAST30DAYS_META_ADS_COUNTRY") == "US" |
| 98 | |
| 99 | def test_no_durable_page_override_key_exists(self): |
| 100 | # A page id is per-topic state, and env keys ride through the |
| 101 | # competitor runner's config copy. |
| 102 | assert "LAST30DAYS_META_ADS_PAGE" not in env.get_config() |
| 103 | |
| 104 | |
| 105 | class _FakeBundle: |
| 106 | def __init__(self, artifacts): |
| 107 | self.artifacts = artifacts |
| 108 | |
| 109 | |
| 110 | class TestArtifactLift: |
| 111 | """Stream artifacts only reach the report as anonymous grounding entries.""" |
| 112 | |
| 113 | def test_footer_inputs_are_promoted_to_named_artifacts(self): |
| 114 | bundle = _FakeBundle( |
| 115 | { |
| 116 | "grounding": [ |
| 117 | {"x_receipts": ["unrelated"]}, |
| 118 | { |
| 119 | "meta_ads_page": {"id": "1", "name": "Brightpan"}, |
| 120 | "meta_ads_tally": {"launched_in_window": 7}, |
| 121 | }, |
| 122 | ] |
| 123 | } |
| 124 | ) |
| 125 | pipeline._lift_stream_artifacts(bundle) |
| 126 | assert bundle.artifacts["meta_ads_page"]["name"] == "Brightpan" |
| 127 | assert bundle.artifacts["meta_ads_tally"]["launched_in_window"] == 7 |
| 128 | |
| 129 | def test_lift_survives_a_zero_item_run(self): |
| 130 | # The advertiser must still be nameable when no creative landed. |
| 131 | bundle = _FakeBundle( |
| 132 | {"grounding": [{"meta_ads_page": {"id": "1", "name": "Brightpan"}}]} |
| 133 | ) |
| 134 | pipeline._lift_stream_artifacts(bundle) |
| 135 | assert bundle.artifacts["meta_ads_page"]["name"] == "Brightpan" |
| 136 | |
| 137 | def test_lift_ignores_non_dict_entries(self): |
| 138 | bundle = _FakeBundle({"grounding": ["not a dict", None]}) |
| 139 | pipeline._lift_stream_artifacts(bundle) |
| 140 | assert "meta_ads_page" not in bundle.artifacts |
| 141 | |
| 142 | def test_lift_is_a_no_op_without_grounding_artifacts(self): |
| 143 | bundle = _FakeBundle({}) |
| 144 | pipeline._lift_stream_artifacts(bundle) |
| 145 | assert bundle.artifacts == {} |
| 146 | |
| 147 | |
| 148 | class TestRetrievalBranch: |
| 149 | """The dispatch branch itself: brand resolution, config threading, outcome.""" |
| 150 | |
| 151 | def _run(self, result, config=None, raw_topic="Brightpan", search_query="kettle reviews"): |
| 152 | from lib import schema |
| 153 | |
| 154 | subquery = schema.SubQuery( |
| 155 | label="primary", |
| 156 | search_query=search_query, |
| 157 | ranking_query="q", |
| 158 | sources=["meta_ads"], |
| 159 | ) |
| 160 | cfg = {"SCRAPECREATORS_API_KEY": "fake-key"} |
| 161 | cfg.update(config or {}) |
| 162 | captured = {} |
| 163 | |
| 164 | def fake_search(topic, from_date, to_date, **kwargs): |
| 165 | captured["topic"] = topic |
| 166 | captured["from_date"] = from_date |
| 167 | captured["to_date"] = to_date |
| 168 | captured.update(kwargs) |
| 169 | return result |
| 170 | |
| 171 | with patch.object(pipeline.meta_ads, "search_meta_ads", side_effect=fake_search): |
| 172 | items, artifact = pipeline._retrieve_stream_impl( |
| 173 | topic="Brightpan", |
| 174 | subquery=subquery, |
| 175 | source="meta_ads", |
| 176 | config=cfg, |
| 177 | depth="default", |
| 178 | date_range=("2026-08-15", "2026-09-14"), |
| 179 | runtime=schema.ProviderRuntime( |
| 180 | reasoning_provider="none", |
| 181 | planner_model="none", |
| 182 | rerank_model="none", |
| 183 | ), |
| 184 | mock=False, |
| 185 | raw_topic=raw_topic, |
| 186 | ) |
| 187 | return items, artifact or {}, captured |
| 188 | |
| 189 | def _result(self, **over): |
| 190 | base = { |
| 191 | "ads": [{"id": "1"}], |
| 192 | "page": {"id": "300", "name": "Brightpan"}, |
| 193 | "tally": {"launched_in_window": 1, "resolution": "resolved"}, |
| 194 | } |
| 195 | base.update(over) |
| 196 | return base |
| 197 | |
| 198 | def test_advertiser_resolves_from_the_research_topic_not_the_subquery(self): |
| 199 | # A subquery like "kettle reviews" would resolve a different company |
| 200 | # than the brand the run is actually about. |
| 201 | _items, _artifact, captured = self._run(self._result()) |
| 202 | assert captured["topic"] == "Brightpan" |
| 203 | |
| 204 | def test_config_is_threaded_into_the_lane(self): |
| 205 | _items, _artifact, captured = self._run( |
| 206 | self._result(), |
| 207 | config={ |
| 208 | "LAST30DAYS_META_ADS_COUNTRY": "GB", |
| 209 | "_meta_ads_page": "123456789012345", |
| 210 | }, |
| 211 | ) |
| 212 | assert captured["token"] == "fake-key" |
| 213 | assert captured["country"] == "GB" |
| 214 | assert captured["page_override"] == "123456789012345" |
| 215 | |
| 216 | def test_country_defaults_when_unset(self): |
| 217 | _items, _artifact, captured = self._run(self._result()) |
| 218 | assert captured["country"] == pipeline.meta_ads.DEFAULT_COUNTRY |
| 219 | |
| 220 | def test_footer_inputs_ride_on_the_stream_artifact(self): |
| 221 | _items, artifact, _captured = self._run(self._result()) |
| 222 | assert artifact["meta_ads_page"]["name"] == "Brightpan" |
| 223 | assert artifact["meta_ads_tally"]["launched_in_window"] == 1 |
| 224 | |
| 225 | def test_partial_result_forces_a_partial_outcome(self): |
| 226 | from lib import schema |
| 227 | |
| 228 | _items, artifact, _captured = self._run( |
| 229 | self._result(partial=True, error="lane budget of 120.0s exceeded") |
| 230 | ) |
| 231 | assert artifact["_source_outcome"]["state"] == schema.PARTIAL |
| 232 | assert "budget" in artifact["_source_outcome"]["detail"] |
| 233 | |
| 234 | def test_items_come_back_as_the_stream(self): |
| 235 | items, _artifact, _captured = self._run(self._result()) |
| 236 | assert items == [{"id": "1"}] |
| 237 | |
| 238 | def test_zero_item_run_still_carries_the_advertiser(self): |
| 239 | _items, artifact, _captured = self._run( |
| 240 | self._result(ads=[], tally={"launched_in_window": 0, "resolution": "resolved"}) |
| 241 | ) |
| 242 | assert artifact["meta_ads_page"]["name"] == "Brightpan" |
| 243 | |
| 244 | |
| 245 | class TestCompetitorIsolation: |
| 246 | def test_page_override_is_dropped_for_competitor_sub_runs(self): |
| 247 | # Left in place, one brand's advertiser page would be fetched for |
| 248 | # every peer in a comparison and rendered as that peer's ads. |
| 249 | import inspect |
| 250 | |
| 251 | import last30days as engine |
| 252 | |
| 253 | source = inspect.getsource(engine) |
| 254 | assert 'entity_config.pop("_meta_ads_page", None)' in source |
| 255 |