"""Tests for the `opensre messaging` CLI commands.""" from __future__ import annotations import json from pathlib import Path import pytest from click.testing import CliRunner from integrations.messaging_security import verify_pairing_code from surfaces.cli.commands.messaging import messaging @pytest.fixture() def _isolated_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: """Redirect the integration store to a temp directory.""" store_path = tmp_path / "integrations.json" store_path.write_text(json.dumps({"version": 2, "integrations": []})) monkeypatch.setattr("integrations.store.STORE_PATH", store_path) return store_path class TestMessagingPairCommand: def test_pair_generates_code(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["pair", "--platform", "telegram"]) assert result.exit_code == 0 assert "Pairing code generated" in result.output def test_pair_stores_hash_in_integration(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["pair", "--platform", "telegram"]) assert result.exit_code == 0 # Read back the store and verify the hash was persisted data = json.loads(_isolated_store.read_text()) integrations = data["integrations"] assert len(integrations) == 1 record = integrations[0] assert record["service"] == "telegram" creds = record["instances"][0]["credentials"] policy_data = creds["identity_policy"] assert policy_data["pairing_secret_hash"] is not None assert policy_data["inbound_enabled"] is True assert policy_data["require_dm_pairing"] is True def test_pair_code_is_verifiable(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["pair", "--platform", "discord"]) assert result.exit_code == 0 # Extract the code from output (it's between bold markers) lines = result.output.strip().split("\n") code_line = [line for line in lines if len(line.strip()) == 6 and line.strip().isalnum()] assert len(code_line) == 1 code = code_line[0].strip() # Verify the stored hash matches data = json.loads(_isolated_store.read_text()) creds = data["integrations"][0]["instances"][0]["credentials"] stored_hash = creds["identity_policy"]["pairing_secret_hash"] assert verify_pairing_code(code, stored_hash) def test_pair_requires_platform(self) -> None: runner = CliRunner() result = runner.invoke(messaging, ["pair"]) assert result.exit_code != 0 assert "Missing option" in result.output or "required" in result.output.lower() class TestMessagingAllowCommand: def test_allow_adds_user(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["allow", "--platform", "telegram", "--user-id", "12345"]) assert result.exit_code == 0 assert "Added user 12345" in result.output data = json.loads(_isolated_store.read_text()) creds = data["integrations"][0]["instances"][0]["credentials"] assert "12345" in creds["identity_policy"]["allowed_user_ids"] def test_allow_duplicate_user_warns(self, _isolated_store: Path) -> None: runner = CliRunner() runner.invoke(messaging, ["allow", "--platform", "slack", "--user-id", "U001"]) result = runner.invoke(messaging, ["allow", "--platform", "slack", "--user-id", "U001"]) assert result.exit_code == 0 assert "already in the allowed list" in result.output def test_allow_rejects_at_handle(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke( messaging, ["allow", "--platform", "telegram", "--user-id", "@opensre_yauhen_bot"] ) assert result.exit_code != 0 assert "@username" in result.output # Nothing should have been persisted for a rejected id. data = json.loads(_isolated_store.read_text()) assert data["integrations"] == [] def test_allow_rejects_non_numeric_telegram_id(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["allow", "--platform", "telegram", "--user-id", "user1"]) assert result.exit_code != 0 assert "numeric" in result.output def test_allow_accepts_numeric_telegram_id(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke( messaging, ["allow", "--platform", "telegram", "--user-id", " 6514715683 "] ) assert result.exit_code == 0 assert "Added user 6514715683" in result.output # normalized (stripped) def test_allow_accepts_slack_native_id(self, _isolated_store: Path) -> None: # Non-Telegram platforms keep their native id shape (Slack U..., etc.); # only the @handle guard applies. runner = CliRunner() result = runner.invoke(messaging, ["allow", "--platform", "slack", "--user-id", "U12345"]) assert result.exit_code == 0 assert "Added user U12345" in result.output class TestMessagingRevokeCommand: def test_revoke_removes_user(self, _isolated_store: Path) -> None: runner = CliRunner() # First add a user runner.invoke(messaging, ["allow", "--platform", "telegram", "--user-id", "99"]) # Then revoke result = runner.invoke(messaging, ["revoke", "--platform", "telegram", "--user-id", "99"]) assert result.exit_code == 0 assert "Removed user 99" in result.output data = json.loads(_isolated_store.read_text()) creds = data["integrations"][0]["instances"][0]["credentials"] assert "99" not in creds["identity_policy"]["allowed_user_ids"] def test_revoke_nonexistent_user_warns(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["revoke", "--platform", "discord", "--user-id", "ghost"]) assert result.exit_code == 0 assert "not in the allowed list" in result.output class TestMessagingStatusCommand: def test_status_no_integration(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["status", "--platform", "telegram"]) assert result.exit_code == 0 assert "No telegram integration configured" in result.output def test_status_with_configured_integration(self, _isolated_store: Path) -> None: runner = CliRunner() # Set up a user first (Telegram ids are numeric) runner.invoke(messaging, ["allow", "--platform", "telegram", "--user-id", "6514715683"]) result = runner.invoke(messaging, ["status", "--platform", "telegram"]) assert result.exit_code == 0 assert "Inbound enabled" in result.output assert "6514715683" in result.output class TestMessagingGroupNoSubcommand: def test_bare_group_shows_help_and_exits_zero(self) -> None: # A bare group is a help request, not an error — must exit 0 (not Click's 2) # so the interactive shell doesn't surface a scary "non-zero code" line. runner = CliRunner() result = runner.invoke(messaging, []) assert result.exit_code == 0 assert "Commands:" in result.output for sub in ("allow", "pair", "revoke", "status"): assert sub in result.output def test_pair_message_points_to_platform_not_shell(self, _isolated_store: Path) -> None: runner = CliRunner() result = runner.invoke(messaging, ["pair", "--platform", "telegram"]) assert result.exit_code == 0 assert "not this shell" in result.output assert "/pair" in result.output assert "gateway" in result.output.lower()