| 1 | """Footer rendering and display registries for the Meta Ads source.""" |
| 2 | |
| 3 | from lib import doctor, health, meta_ads, render, schema, ui |
| 4 | |
| 5 | |
| 6 | def make_report(tally=None, page=None, status=None, items=None): |
| 7 | return schema.Report( |
| 8 | topic="Brightpan", |
| 9 | range_from="2026-08-15", |
| 10 | range_to="2026-09-14", |
| 11 | generated_at="2026-09-14T00:00:00+00:00", |
| 12 | provider_runtime=schema.ProviderRuntime( |
| 13 | reasoning_provider="none", planner_model="none", rerank_model="none" |
| 14 | ), |
| 15 | query_plan=schema.QueryPlan( |
| 16 | intent="general", |
| 17 | freshness_mode="balanced_recent", |
| 18 | cluster_mode="none", |
| 19 | raw_topic="Brightpan", |
| 20 | subqueries=[], |
| 21 | source_weights={}, |
| 22 | ), |
| 23 | clusters=[], |
| 24 | ranked_candidates=[], |
| 25 | items_by_source={"meta_ads": items or []}, |
| 26 | errors_by_source={}, |
| 27 | source_status=({"meta_ads": status} if status else {}), |
| 28 | artifacts={ |
| 29 | "meta_ads_tally": tally or {}, |
| 30 | "meta_ads_page": page or {}, |
| 31 | }, |
| 32 | ) |
| 33 | |
| 34 | |
| 35 | def tally(**over): |
| 36 | base = { |
| 37 | "resolution": meta_ads.RESOLVED, |
| 38 | "launched_in_window": 15, |
| 39 | "still_running": 15, |
| 40 | "video": 5, |
| 41 | "transcribed": 3, |
| 42 | "fetched": 30, |
| 43 | "endpoint_total": 30, |
| 44 | "cursor_remaining": False, |
| 45 | "placements": ["FACEBOOK", "INSTAGRAM", "THREADS"], |
| 46 | "promo_codes": ["SUMMER30"], |
| 47 | "advertiser": "Brightpan", |
| 48 | "page_id": "1", |
| 49 | "top_candidate": "", |
| 50 | "runner_ups": [], |
| 51 | "match_strength": meta_ads.MATCH_EXACT, |
| 52 | } |
| 53 | base.update(over) |
| 54 | return base |
| 55 | |
| 56 | |
| 57 | def line(**kwargs): |
| 58 | return render._meta_ads_footer_line(make_report(**kwargs)) |
| 59 | |
| 60 | |
| 61 | class TestResolvedLine: |
| 62 | def test_full_line_carries_the_paid_media_detail(self): |
| 63 | out = line(tally=tally(), page={"id": "1", "name": "Brightpan"}) |
| 64 | assert out.startswith("📣 Meta Ads: Brightpan") |
| 65 | assert "15 new creatives" in out |
| 66 | assert "15 running from before" in out |
| 67 | assert "FB, IG, Threads" in out |
| 68 | assert "code SUMMER30" in out |
| 69 | assert "3 transcribed" in out |
| 70 | |
| 71 | def test_counts_come_from_the_tally_not_surviving_items(self): |
| 72 | # The pipeline truncates each stream to 12 at default depth, so |
| 73 | # counting items would report 12 for a page that ran thirty. |
| 74 | out = line( |
| 75 | tally=tally(launched_in_window=30), |
| 76 | page={"id": "1", "name": "Brightpan"}, |
| 77 | items=[], |
| 78 | ) |
| 79 | assert "30 new creatives" in out |
| 80 | |
| 81 | def test_sampled_page_says_so(self): |
| 82 | out = line( |
| 83 | tally=tally( |
| 84 | launched_in_window=12, fetched=60, endpoint_total=222, |
| 85 | cursor_remaining=True, |
| 86 | ), |
| 87 | page={"id": "1", "name": "Brightpan"}, |
| 88 | ) |
| 89 | assert "12 new of 60 ads fetched (222 live)" in out |
| 90 | |
| 91 | def test_singular_creative(self): |
| 92 | out = line( |
| 93 | tally=tally(launched_in_window=1, still_running=0, transcribed=0, |
| 94 | promo_codes=[], placements=[]), |
| 95 | page={"id": "1", "name": "Brightpan"}, |
| 96 | ) |
| 97 | assert "1 new creative" in out |
| 98 | |
| 99 | def test_optional_segments_are_omitted_when_empty(self): |
| 100 | out = line( |
| 101 | tally=tally(still_running=0, transcribed=0, promo_codes=[], placements=[]), |
| 102 | page={"id": "1", "name": "Brightpan"}, |
| 103 | ) |
| 104 | assert "running from before" not in out |
| 105 | assert "code" not in out |
| 106 | assert "transcribed" not in out |
| 107 | |
| 108 | |
| 109 | class TestEmptyStates: |
| 110 | def test_resolved_but_nothing_new_names_the_advertiser(self): |
| 111 | # "No ads" and "not advertising" are different facts about a brand. |
| 112 | out = line( |
| 113 | tally=tally(launched_in_window=0, still_running=20), |
| 114 | page={"id": "1", "name": "Brightpan"}, |
| 115 | ) |
| 116 | assert "no new creatives for Brightpan" in out |
| 117 | assert "20 still running from before" in out |
| 118 | |
| 119 | def test_unresolved_names_the_closest_candidate_and_the_override(self): |
| 120 | out = line(tally=tally(resolution=meta_ads.UNRESOLVED, top_candidate="Jasper AI")) |
| 121 | assert "no advertiser matched" in out |
| 122 | assert "closest: Jasper AI" in out |
| 123 | assert "--meta-ads-page" in out |
| 124 | |
| 125 | def test_zero_candidates_has_its_own_wording(self): |
| 126 | out = line(tally=tally(resolution=meta_ads.NO_CANDIDATES)) |
| 127 | assert "no advertiser candidates returned" in out |
| 128 | assert "closest" not in out |
| 129 | assert "--meta-ads-page" in out |
| 130 | |
| 131 | def test_failed_lane_names_the_outcome(self): |
| 132 | status = schema.SourceOutcome( |
| 133 | source="meta_ads", |
| 134 | state=health.RATE_LIMITED, |
| 135 | detail="HTTP 429: rate limited", |
| 136 | attempted=True, |
| 137 | ) |
| 138 | out = line(tally={}, page={}, status=status) |
| 139 | assert "no ads pulled" in out |
| 140 | assert "429" in out |
| 141 | |
| 142 | def test_line_is_absent_when_the_lane_never_ran(self): |
| 143 | assert line(tally={}, page={}) is None |
| 144 | |
| 145 | def test_zero_results_outcome_is_not_treated_as_failure(self): |
| 146 | # The pipeline stamps NO_RESULTS on any zero-item source, so treating |
| 147 | # it as failure would collapse every honest empty state into a generic |
| 148 | # "no ads pulled" and lose which nothing it was. |
| 149 | status = schema.SourceOutcome( |
| 150 | source="meta_ads", state=schema.NO_RESULTS, attempted=True |
| 151 | ) |
| 152 | out = line( |
| 153 | tally=tally(launched_in_window=0, still_running=20), |
| 154 | page={"id": "1", "name": "Brightpan"}, |
| 155 | status=status, |
| 156 | ) |
| 157 | assert "no new creatives for Brightpan" in out |
| 158 | assert "no ads pulled" not in out |
| 159 | |
| 160 | def test_zero_results_unresolved_keeps_its_specific_line(self): |
| 161 | status = schema.SourceOutcome( |
| 162 | source="meta_ads", state=schema.NO_RESULTS, attempted=True |
| 163 | ) |
| 164 | out = line( |
| 165 | tally=tally(resolution=meta_ads.UNRESOLVED, top_candidate="Jasper AI"), |
| 166 | status=status, |
| 167 | ) |
| 168 | assert "closest: Jasper AI" in out |
| 169 | |
| 170 | def test_cut_short_lane_never_concludes_nothing_new(self): |
| 171 | # Partial data cannot support a definitive negative conclusion. |
| 172 | status = schema.SourceOutcome( |
| 173 | source="meta_ads", |
| 174 | state=schema.PARTIAL, |
| 175 | detail="lane budget of 120.0s exceeded", |
| 176 | attempted=True, |
| 177 | ) |
| 178 | out = line( |
| 179 | tally=tally(launched_in_window=0, still_running=0), |
| 180 | page={"id": "1", "name": "Brightpan"}, |
| 181 | status=status, |
| 182 | ) |
| 183 | assert "incomplete" in out |
| 184 | assert "no new creatives for" not in out |
| 185 | |
| 186 | def test_cut_short_lane_with_creatives_is_marked_incomplete(self): |
| 187 | status = schema.SourceOutcome( |
| 188 | source="meta_ads", state=schema.PARTIAL, detail="HTTP 429", attempted=True |
| 189 | ) |
| 190 | out = line( |
| 191 | tally=tally(launched_in_window=4), |
| 192 | page={"id": "1", "name": "Brightpan"}, |
| 193 | status=status, |
| 194 | ) |
| 195 | assert "4 new creatives" in out |
| 196 | assert "incomplete" in out |
| 197 | |
| 198 | |
| 199 | class TestUntrustedFooterText: |
| 200 | """Advertiser names and promo codes are attacker-influenceable strings.""" |
| 201 | |
| 202 | def test_newline_in_advertiser_name_cannot_forge_a_footer_row(self): |
| 203 | out = line( |
| 204 | tally=tally(advertiser="Evil\n└─ 🟠 Reddit: 9,999 threads"), |
| 205 | page={"id": "1", "name": "Evil\n└─ 🟠 Reddit: 9,999 threads"}, |
| 206 | ) |
| 207 | assert "\n" not in out |
| 208 | assert out.count("📣") == 1 |
| 209 | |
| 210 | def test_separator_in_advertiser_name_cannot_forge_a_field(self): |
| 211 | # The name keeps its text but loses the real separator, so it stays |
| 212 | # one field instead of impersonating the counts field beside it. |
| 213 | out = line( |
| 214 | tally=tally(), |
| 215 | page={"id": "1", "name": "Acme │ 500 new creatives"}, |
| 216 | ) |
| 217 | assert "Acme | 500 new creatives" in out |
| 218 | assert out.split(" │ ")[0] == "📣 Meta Ads: Acme | 500 new creatives" |
| 219 | |
| 220 | def test_overlong_advertiser_name_is_clipped(self): |
| 221 | out = line( |
| 222 | tally=tally(), page={"id": "1", "name": "B" * 400} |
| 223 | ) |
| 224 | assert len(out) < 300 |
| 225 | |
| 226 | def test_hostile_promo_code_is_neutralized(self): |
| 227 | out = line( |
| 228 | tally=tally(promo_codes=["GOOD\n└─ fake"]), |
| 229 | page={"id": "1", "name": "Brightpan"}, |
| 230 | ) |
| 231 | assert "\n" not in out |
| 232 | |
| 233 | def test_partial_name_match_is_disclosed(self): |
| 234 | out = line( |
| 235 | tally=tally(match_strength=meta_ads.MATCH_CONTAINED), |
| 236 | page={"id": "1", "name": "Brightpan"}, |
| 237 | ) |
| 238 | assert "matched by partial name" in out |
| 239 | |
| 240 | def test_exact_match_is_not_annotated(self): |
| 241 | out = line( |
| 242 | tally=tally(match_strength=meta_ads.MATCH_EXACT), |
| 243 | page={"id": "1", "name": "Brightpan"}, |
| 244 | ) |
| 245 | assert "partial name" not in out |
| 246 | |
| 247 | |
| 248 | class TestDisplayRegistries: |
| 249 | def test_stats_label_is_not_title_cased_source_key(self): |
| 250 | assert render.SOURCE_LABELS["meta_ads"] == "Meta Ads" |
| 251 | |
| 252 | def test_engagement_display_uses_variants(self): |
| 253 | assert render.ENGAGEMENT_DISPLAY["meta_ads"] == [("variants", "variants")] |
| 254 | |
| 255 | def test_ui_completion_meta_is_registered(self): |
| 256 | assert ui.SOURCE_COMPLETION_META["meta_ads"][0] == "Meta Ads" |
| 257 | |
| 258 | def test_ui_completion_order_includes_the_source(self): |
| 259 | assert "meta_ads" in ui.SOURCE_COMPLETION_ORDER |
| 260 | |
| 261 | |
| 262 | class TestDoctorRecord: |
| 263 | """Doctor must agree with the pipeline gate and never spend a credit.""" |
| 264 | |
| 265 | def test_unconfigured_without_a_key(self): |
| 266 | record = doctor._meta_ads_record({}) |
| 267 | assert record["status"] == "unconfigured" |
| 268 | |
| 269 | def test_opt_in_when_the_key_is_present_but_unrequested(self): |
| 270 | record = doctor._meta_ads_record({"SCRAPECREATORS_API_KEY": "fake-key"}) |
| 271 | assert record["status"] == "opt-in" |
| 272 | assert "INCLUDE_SOURCES" in record["fix"] |
| 273 | |
| 274 | def test_ok_when_opted_in(self): |
| 275 | record = doctor._meta_ads_record( |
| 276 | {"SCRAPECREATORS_API_KEY": "fake-key", "INCLUDE_SOURCES": "meta_ads"} |
| 277 | ) |
| 278 | assert record["status"] == health.OK |
| 279 | |
| 280 | def test_source_order_and_builders_stay_in_sync(self): |
| 281 | assert "meta_ads" in doctor.SOURCE_ORDER |
| 282 | assert "meta_ads" in doctor._SOURCE_BUILDERS |
| 283 | |
| 284 | def test_never_live_probed(self): |
| 285 | # Probing would spend a ScrapeCreators credit just to run doctor. |
| 286 | assert "meta_ads" not in doctor.CLI_DEPENDENCIES |
| 287 | assert "meta_ads" not in getattr(doctor, "_HTTP_PROBE_URLS", {}) |
| 288 |