| 1 | """Contract tests for the Grok Bot host slice of SKILL.md (R4, R12, R16; AE9). |
| 2 | |
| 3 | On a Grok Bot host the model-facing contract must drive X through the |
| 4 | official path only: the X connector lane first, the X API bearer or the xAI |
| 5 | key as backups, keys written only through the engine's ``setup --store-key`` |
| 6 | path, and no browser-session step of any kind. These tests read SKILL.md as |
| 7 | text - the model's runtime contract - the way tests/test_onboarding_contract.py |
| 8 | and tests/test_codex_host_contract.py do, and slice the Grok Bot passages so a |
| 9 | word that is fine elsewhere (the cookie recipes for Linux / Mac mini) cannot |
| 10 | satisfy or fail an assertion here. |
| 11 | """ |
| 12 | |
| 13 | from __future__ import annotations |
| 14 | |
| 15 | import re |
| 16 | import unittest |
| 17 | from pathlib import Path |
| 18 | |
| 19 | ROOT = Path(__file__).resolve().parents[1] |
| 20 | SKILL_MD = ROOT / "skills" / "last30days" / "SKILL.md" |
| 21 | CONFIGURATION = ROOT / "CONFIGURATION.md" |
| 22 | AGENTS_MD = ROOT / "AGENTS.md" |
| 23 | |
| 24 | FLOW_HEADING = "### Grok Bot Prose Flow" |
| 25 | RECIPE_MARKER = "X connector recipe" |
| 26 | RECIPE_END = "**Step 1: Run the research script" |
| 27 | |
| 28 | # R4 vocabulary: none of it may appear in the Grok Bot flow (case-insensitive). |
| 29 | FORBIDDEN = ( |
| 30 | "cookie", |
| 31 | "box-chrome", |
| 32 | "cdp", |
| 33 | "auth_token", |
| 34 | "ct0", |
| 35 | "bird", |
| 36 | "18800", |
| 37 | "xquik", |
| 38 | "grok login", |
| 39 | "from_browser", |
| 40 | "last30days_x_backend", |
| 41 | "askuserquestion", |
| 42 | ) |
| 43 | |
| 44 | LANE_TAGS = ("topic", "from", "mention", "related") |
| 45 | ROW_FIELDS = ( |
| 46 | "id", |
| 47 | "author_handle", |
| 48 | "created_at", |
| 49 | "text", |
| 50 | "likes", |
| 51 | "reposts", |
| 52 | "replies", |
| 53 | "quotes", |
| 54 | ) |
| 55 | KEY_TOKENS = ("X_BEARER_TOKEN", "XAI_API_KEY", "--x-posts", "envelope", ".json") |
| 56 | |
| 57 | HEREDOC_RE = re.compile(r"<<-?\s*([^\s]+)") |
| 58 | |
| 59 | |
| 60 | def _text() -> str: |
| 61 | return SKILL_MD.read_text(encoding="utf-8") |
| 62 | |
| 63 | |
| 64 | def _slice_between(text: str, start_marker: str, end_marker: str) -> str: |
| 65 | start = text.find(start_marker) |
| 66 | assert start != -1, f"missing marker: {start_marker!r}" |
| 67 | end = text.find(end_marker, start + len(start_marker)) |
| 68 | assert end != -1, f"missing end marker after {start_marker!r}: {end_marker!r}" |
| 69 | return text[start:end] |
| 70 | |
| 71 | |
| 72 | def _grok_flow(text: str) -> str: |
| 73 | """The Grok Bot Prose Flow, from its heading to the next ### heading.""" |
| 74 | start = text.find(FLOW_HEADING) |
| 75 | assert start != -1, f"missing {FLOW_HEADING!r}" |
| 76 | match = re.search(r"^### ", text[start + len(FLOW_HEADING):], flags=re.MULTILINE) |
| 77 | assert match is not None, "no ### heading follows the Grok Bot Prose Flow" |
| 78 | return text[start : start + len(FLOW_HEADING) + match.start()] |
| 79 | |
| 80 | |
| 81 | def _recipe(text: str) -> str: |
| 82 | research = text[text.index("## Research Execution") :] |
| 83 | return _slice_between(research, RECIPE_MARKER, RECIPE_END) |
| 84 | |
| 85 | |
| 86 | def _guaranteed_band(text: str) -> str: |
| 87 | return "\n".join(text.splitlines()[:420]) |
| 88 | |
| 89 | |
| 90 | def _extras_passages(text: str) -> dict[str, str]: |
| 91 | step0 = _slice_between(text, "## Step 0: First-Run Setup Wizard", "## CRITICAL: Parse User Intent") |
| 92 | modal = _slice_between(step0, "### Claude Code Modal Flow", "### Non-Modal Prose Flow") |
| 93 | prose = _slice_between(step0, "### Non-Modal Prose Flow", FLOW_HEADING) |
| 94 | manual = step0[step0.index("### Manual Setup Guide") :] |
| 95 | modal_extras = _slice_between(modal, "**Extras-host X login", "**macOS Full Disk Access") |
| 96 | prose_extras = _slice_between(prose, "**Extras hosts", " - On **no**") |
| 97 | manual_extras = _slice_between(manual, "**X on Linux / Mac mini (repair).**", "**Reddit (free") |
| 98 | return {"modal": modal_extras, "prose": prose_extras, "manual": manual_extras} |
| 99 | |
| 100 | |
| 101 | def _forbidden_hits(slice_text: str) -> list[str]: |
| 102 | lowered = slice_text.lower() |
| 103 | return [word for word in FORBIDDEN if word in lowered] |
| 104 | |
| 105 | |
| 106 | class TestGrokBotProseFlow(unittest.TestCase): |
| 107 | def setUp(self): |
| 108 | self.text = _text() |
| 109 | self.flow = _grok_flow(self.text) |
| 110 | |
| 111 | def test_flow_is_the_third_step0_branch(self): |
| 112 | step0 = _slice_between( |
| 113 | self.text, "## Step 0: First-Run Setup Wizard", "## CRITICAL: Parse User Intent" |
| 114 | ) |
| 115 | split = _slice_between(step0, "**Platform split", "### Claude Code Modal Flow") |
| 116 | self.assertIn("Grok Bot Prose Flow", split) |
| 117 | # Cursor stays a Non-Modal host; it is not routed to the Grok Bot flow. |
| 118 | self.assertIn("Cursor", _slice_between(step0, "### Non-Modal Prose Flow", FLOW_HEADING)) |
| 119 | self.assertNotIn("Cursor", self.flow) |
| 120 | |
| 121 | def test_flow_has_no_r4_vocabulary(self): |
| 122 | self.assertEqual([], _forbidden_hits(self.flow)) |
| 123 | |
| 124 | def test_flow_names_the_official_contract(self): |
| 125 | for token in ( |
| 126 | "LAST30DAYS_HOST=grok-bot", |
| 127 | "LAST30DAYS_X_HOST_LANE=1", |
| 128 | "X_BEARER_TOKEN", |
| 129 | "XAI_API_KEY", |
| 130 | "--x-posts", |
| 131 | "search_posts_all", |
| 132 | "generated_at", |
| 133 | "window-unsupported", |
| 134 | "setup --store-key", |
| 135 | "about the last week", |
| 136 | "SETUP_COMPLETE=true", |
| 137 | "X_DECLINED=grok-bot", |
| 138 | ): |
| 139 | self.assertIn(token, self.flow, token) |
| 140 | |
| 141 | def test_flow_names_the_call_counts_and_lanes(self): |
| 142 | self.assertRegex(self.flow, r"10\s*/\s*30\s*/\s*60") |
| 143 | self.assertRegex(self.flow, r"\b8\b.*\b5\b.*\b3\b") |
| 144 | for lane in LANE_TAGS: |
| 145 | self.assertIn(f"`{lane}`", self.flow, lane) |
| 146 | for field in ROW_FIELDS: |
| 147 | self.assertIn(f"`{field}`", self.flow, field) |
| 148 | |
| 149 | def test_flow_names_the_x_for_grok_bot_plugin(self): |
| 150 | """The connector is the marketplace "X for Grok Bot" plugin: the flow |
| 151 | names it and keys the lane on its post-search tools, not on one |
| 152 | tool name alone (search_posts_all stays as the example).""" |
| 153 | self.assertIn('"X for Grok Bot"', self.flow) |
| 154 | self.assertIn("search_posts_all", self.flow) |
| 155 | rule = _text()[: _text().index("## Step 0")] |
| 156 | self.assertIn('"X for Grok Bot"', rule) |
| 157 | |
| 158 | def test_connector_step_precedes_bearer_offer(self): |
| 159 | connector = self.flow.index("search_posts_all") |
| 160 | bearer = self.flow.index("X_BEARER_TOKEN") |
| 161 | self.assertLess(connector, bearer) |
| 162 | |
| 163 | def test_bearer_coverage_caveat_never_implies_parity(self): |
| 164 | self.assertIn( |
| 165 | "recent posts, about the last week, unless your X developer project has full-archive access", |
| 166 | self.flow, |
| 167 | ) |
| 168 | self.assertIn("X developer console", self.flow) |
| 169 | self.assertIn("console.x.ai", self.flow) |
| 170 | |
| 171 | def test_key_persistence_only_through_engine_and_masked(self): |
| 172 | self.assertIn("setup --store-key", self.flow) |
| 173 | self.assertIn("=****", self.flow) |
| 174 | self.assertIn("never echo the value back", self.flow) |
| 175 | for line in self.flow.splitlines(): |
| 176 | if not re.search(r"\b(echo|printf)\b", line): |
| 177 | continue |
| 178 | writes = re.search(r"\b(echo|printf)\b[^\n]*(>>|>|\|)", line) |
| 179 | if writes and any(token in line for token in KEY_TOKENS): |
| 180 | self.fail(f"a shell write of a key or the envelope: {line.strip()!r}") |
| 181 | |
| 182 | def test_every_heredoc_in_flow_uses_single_quoted_delimiter(self): |
| 183 | for delim in HEREDOC_RE.findall(self.flow): |
| 184 | self.assertTrue(delim.startswith("'"), f"unquoted heredoc delimiter {delim!r}") |
| 185 | |
| 186 | def test_flow_declines_write_marker_and_has_no_modals(self): |
| 187 | self.assertIn("X_DECLINED=grok-bot", self.flow) |
| 188 | self.assertNotIn("AskUserQuestion", self.flow) |
| 189 | |
| 190 | |
| 191 | class TestGuaranteedLoadedRule(unittest.TestCase): |
| 192 | def setUp(self): |
| 193 | self.band = _guaranteed_band(_text()) |
| 194 | |
| 195 | def test_rule_lives_in_the_guaranteed_loaded_band(self): |
| 196 | self.assertIn("LAST30DAYS_HOST=grok-bot", self.band) |
| 197 | self.assertIn("LAST30DAYS_X_HOST_LANE=1", self.band) |
| 198 | self.assertIn("never place post text unquoted", self.band) |
| 199 | self.assertIn("CURSOR_AGENT", self.band) |
| 200 | |
| 201 | def test_rule_is_not_keyed_on_cursor_agent_alone(self): |
| 202 | start = self.band.index("LAST30DAYS_HOST=grok-bot") |
| 203 | rule = self.band[max(0, start - 600) : start + 1200] |
| 204 | self.assertIn("CURSOR_AGENT", rule) |
| 205 | self.assertRegex(rule, r"(?i)not .*CURSOR_AGENT.*alone|CURSOR_AGENT.*alone") |
| 206 | |
| 207 | |
| 208 | class TestConnectorRecipe(unittest.TestCase): |
| 209 | def setUp(self): |
| 210 | self.recipe = _recipe(_text()) |
| 211 | |
| 212 | def test_recipe_precedes_the_engine_command(self): |
| 213 | text = _text() |
| 214 | research = text[text.index("## Research Execution") :] |
| 215 | self.assertLess(research.index(RECIPE_MARKER), research.index(RECIPE_END)) |
| 216 | |
| 217 | def test_recipe_names_counts_window_and_status(self): |
| 218 | for token in ( |
| 219 | "search_posts_all", |
| 220 | "-is:retweet", |
| 221 | "window-unsupported", |
| 222 | "partial", |
| 223 | "generated_at", |
| 224 | "last30days-x-posts/1", |
| 225 | "--x-posts", |
| 226 | "x_posts", |
| 227 | "--competitors-plan", |
| 228 | "X via X connector", |
| 229 | "LAW 9", |
| 230 | "stderr", |
| 231 | ): |
| 232 | self.assertIn(token, self.recipe, token) |
| 233 | self.assertRegex(self.recipe, r"10 .*30 .*60") |
| 234 | for lane in LANE_TAGS: |
| 235 | self.assertIn(f'"lane": "{lane}"', self.recipe, lane) |
| 236 | for field in ROW_FIELDS: |
| 237 | self.assertIn(f'"{field}"', self.recipe, field) |
| 238 | |
| 239 | def test_recipe_forbids_raw_tool_output_in_error(self): |
| 240 | self.assertIn("never raw tool output", self.recipe) |
| 241 | for category in ("credits", "not-connected", "unavailable"): |
| 242 | self.assertIn(category, self.recipe) |
| 243 | |
| 244 | def test_recipe_heredocs_are_single_quoted_and_no_echo_writes(self): |
| 245 | for delim in HEREDOC_RE.findall(self.recipe): |
| 246 | self.assertTrue(delim.startswith("'"), f"unquoted heredoc delimiter {delim!r}") |
| 247 | for line in self.recipe.splitlines(): |
| 248 | if re.search(r"\b(echo|printf)\b[^\n]*(>>|>|\|)", line) and any( |
| 249 | token in line for token in KEY_TOKENS |
| 250 | ): |
| 251 | self.fail(f"a shell write of the envelope: {line.strip()!r}") |
| 252 | |
| 253 | def test_recipe_heredoc_sentinel_is_per_run_and_json_is_one_line(self): |
| 254 | """Post text is attacker-controlled: a fixed public sentinel could be |
| 255 | echoed by a post to close the heredoc early. The recipe demands a |
| 256 | per-run nonce in both sentinel lines and single-line JSON.""" |
| 257 | delims = HEREDOC_RE.findall(self.recipe) |
| 258 | self.assertTrue(any("X_POSTS_EOF_{X_POSTS_NONCE}" in d for d in delims), delims) |
| 259 | self.assertNotIn("<<'X_POSTS_EOF'", self.recipe) |
| 260 | self.assertIn("\nX_POSTS_EOF_{X_POSTS_NONCE}\n", self.recipe) |
| 261 | self.assertIn("ONE line", self.recipe) |
| 262 | self.assertIn("random", self.recipe) |
| 263 | |
| 264 | def test_recipe_has_no_r4_vocabulary(self): |
| 265 | self.assertEqual([], _forbidden_hits(self.recipe)) |
| 266 | |
| 267 | |
| 268 | class TestExtrasPassagesRescoped(unittest.TestCase): |
| 269 | def test_extras_passages_no_longer_name_grok_bot(self): |
| 270 | for name, passage in _extras_passages(_text()).items(): |
| 271 | self.assertNotIn("Grok Bot", passage, f"{name} extras passage still names Grok Bot") |
| 272 | self.assertIn("grok-bot", passage, f"{name} extras passage does not exclude the grok-bot host") |
| 273 | |
| 274 | def test_manual_repair_heading_rescoped(self): |
| 275 | text = _text() |
| 276 | self.assertIn("**X on Linux / Mac mini (repair).**", text) |
| 277 | self.assertNotIn("X on Linux / Grok Bot / Mac mini", text) |
| 278 | |
| 279 | |
| 280 | class TestManualSetupGuide(unittest.TestCase): |
| 281 | def setUp(self): |
| 282 | text = _text() |
| 283 | step0 = _slice_between(text, "## Step 0: First-Run Setup Wizard", "## CRITICAL: Parse User Intent") |
| 284 | self.manual = step0[step0.index("### Manual Setup Guide") :] |
| 285 | |
| 286 | def test_bearer_bullet_comes_first_in_x_section(self): |
| 287 | x_section = _slice_between(self.manual, "**X/Twitter (pick one", "**X on Linux / Mac mini (repair).**") |
| 288 | bullets = [line for line in x_section.splitlines() if line.startswith("- ")] |
| 289 | self.assertTrue(bullets, "no X bullets in the Manual Setup Guide") |
| 290 | self.assertIn("X_BEARER_TOKEN", bullets[0]) |
| 291 | self.assertIn("about a week", bullets[0]) |
| 292 | |
| 293 | def test_grok_bot_repair_paragraph_is_official_only(self): |
| 294 | para = _slice_between(self.manual, "**X on a Grok Bot (repair).**", "**X on Linux / Mac mini (repair).**") |
| 295 | self.assertEqual([], _forbidden_hits(para)) |
| 296 | for token in ("connect X", "X_BEARER_TOKEN", "about the last week", "XAI_API_KEY", "top up"): |
| 297 | self.assertIn(token, para, token) |
| 298 | |
| 299 | |
| 300 | class TestSecurityAndFrontmatter(unittest.TestCase): |
| 301 | def test_frontmatter_optional_env_lists_bearer(self): |
| 302 | text = _text() |
| 303 | frontmatter = text[: text.index("---", 3)] |
| 304 | self.assertIn("- X_BEARER_TOKEN", frontmatter) |
| 305 | |
| 306 | def test_security_section_lists_x_api_and_envelope(self): |
| 307 | text = _text() |
| 308 | security = text[text.index("## Security & Permissions") :] |
| 309 | self.assertIn("api.x.com", security) |
| 310 | self.assertIn("--x-posts", security) |
| 311 | self.assertIn("X connector", security) |
| 312 | overview = _slice_between(text, "**Permissions overview:**", "Research ANY topic") |
| 313 | self.assertIn("api.x.com", overview) |
| 314 | |
| 315 | |
| 316 | class TestConfigurationGrokBotSubsection(unittest.TestCase): |
| 317 | def test_configuration_grok_bot_subsection_has_no_r4_vocabulary(self): |
| 318 | text = CONFIGURATION.read_text(encoding="utf-8") |
| 319 | match = re.search(r"^(#{2,4}) [^\n]*Grok Bot[^\n]*$", text, flags=re.MULTILINE) |
| 320 | if match is None: |
| 321 | self.skipTest("CONFIGURATION.md has no Grok Bot subsection yet") |
| 322 | level = len(match.group(1)) |
| 323 | rest = text[match.end() :] |
| 324 | nxt = re.search(rf"^#{{1,{level}}} ", rest, flags=re.MULTILINE) |
| 325 | section = rest if nxt is None else rest[: nxt.start()] |
| 326 | self.assertEqual([], _forbidden_hits(section)) |
| 327 | |
| 328 | |
| 329 | class TestAgentsMd(unittest.TestCase): |
| 330 | def test_agents_md_names_three_branches(self): |
| 331 | text = AGENTS_MD.read_text(encoding="utf-8") |
| 332 | self.assertIn("Step 0 has THREE branches", text) |
| 333 | self.assertIn("Grok Bot Prose Flow", text) |
| 334 | self.assertNotIn("Step 0 has TWO branches", text) |
| 335 | |
| 336 | |
| 337 | if __name__ == "__main__": |
| 338 | unittest.main() |
| 339 |