From 9268d7a4ba4c762f0470dcccd52bf39953b0746c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 3 Jun 2026 00:47:25 +0000 Subject: [PATCH 1/2] test: add full integration test suite for live API verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 39 tests covering the complete tool surface against a real Webex bot token. All tests are skipped automatically during normal CI runs and opt-in via WEBEX_INTEGRATION_TESTS=1 + WEBEX_ACCESS_TOKEN to prevent unit-test mock interference. Coverage: - People: get_webex_me, list_webex_people - Rooms: full CRUD lifecycle + post-delete GET verification - Messages: text, markdown, space alias, list, delete - Adaptive cards: build_webex_adaptive_card, send_webex_adaptive_card - Memberships: list, response shape, bot-is-moderator assertion - Space aliases: full create→get→list→update→delete lifecycle - Teams: list (bots cannot create teams) - Error handling: structured E-code responses for invalid IDs and missing required parameters across all tool categories Each class manages its own tearDownClass cleanup so resources are deleted even when individual tests fail. https://claude.ai/code/session_01Bykkm6kE75wpwuiuL9MHny --- tests/test_integration.py | 578 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 578 insertions(+) create mode 100644 tests/test_integration.py diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..c30f474 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,578 @@ +""" +Integration tests for webex-bot-mcp. + +These tests make real Webex API calls and verify that each tool works +end-to-end. They are automatically SKIPPED when WEBEX_ACCESS_TOKEN is +not set, so they are safe to include in CI without credentials. + +Run integration tests with an explicit opt-in flag so they are always +skipped during normal `unittest discover` runs: + + # Integration tests only: + WEBEX_ACCESS_TOKEN= WEBEX_INTEGRATION_TESTS=1 \\ + python -m unittest tests/test_integration.py -v + + # With pytest: + WEBEX_ACCESS_TOKEN= WEBEX_INTEGRATION_TESTS=1 \\ + python -m pytest tests/test_integration.py -v + +Each test class creates its own Webex resources in setUpClass and +deletes them in tearDownClass, even when individual tests fail. +""" + +import os +import sys +import time +import types +import unittest +from unittest.mock import MagicMock + +# --------------------------------------------------------------------------- +# Remove any mock modules injected by unit test modules running in the same +# process, so integration tests always use the real SDK and dotenv. +# --------------------------------------------------------------------------- +for _k in list(sys.modules.keys()): + if _k == "webexpythonsdk" or _k.startswith("webexpythonsdk."): + _mod = sys.modules[_k] + if isinstance(getattr(_mod, "WebexAPI", None), MagicMock): + del sys.modules[_k] + # The unit-test stub for dotenv only defines load_dotenv; remove it so the + # real python-dotenv (which also has dotenv_values) is imported instead. + if (_k == "dotenv" or _k.startswith("dotenv.")) and not hasattr( + sys.modules[_k], "dotenv_values" + ): + del sys.modules[_k] + # Evict cached tool modules so they re-import against the real SDK + if _k.startswith("webex_bot_mcp"): + del sys.modules[_k] + +# --------------------------------------------------------------------------- +# Patch FastMCP's get_http_headers so that calling tools directly (outside +# an HTTP request context) falls through to the WEBEX_ACCESS_TOKEN env var. +# --------------------------------------------------------------------------- +try: + import fastmcp.server.dependencies as _fmcp_deps + _orig_get_http_headers = _fmcp_deps.get_http_headers + _fmcp_deps.get_http_headers = lambda: {} + _FMCP_PATCHED = True +except Exception: + _FMCP_PATCHED = False + _orig_get_http_headers = None + + +def tearDownModule(): + """Restore FastMCP's original get_http_headers.""" + if _FMCP_PATCHED and _orig_get_http_headers is not None: + import fastmcp.server.dependencies as _deps + _deps.get_http_headers = _orig_get_http_headers + + +# --------------------------------------------------------------------------- +# Now import the real tool modules +# --------------------------------------------------------------------------- +from webex_bot_mcp.tools.rooms import ( # noqa: E402 + list_webex_rooms, create_webex_room, get_webex_room, + update_webex_room, delete_webex_room, + list_webex_spaces, create_webex_space, get_webex_space, + update_webex_space, delete_webex_space, +) +from webex_bot_mcp.tools.messages import ( # noqa: E402 + send_webex_message, list_webex_messages, delete_webex_message, + send_webex_message_with_mentions, update_webex_message, + send_webex_adaptive_card, build_webex_adaptive_card, + send_webex_space_message, list_webex_space_messages, +) +from webex_bot_mcp.tools.memberships import ( # noqa: E402 + list_webex_memberships, add_webex_membership, + update_webex_membership, delete_webex_membership, + list_webex_space_memberships, +) +from webex_bot_mcp.tools.people import ( # noqa: E402 + get_webex_me, list_webex_people, +) +from webex_bot_mcp.tools.teams import ( # noqa: E402 + list_webex_teams, get_webex_team, + list_webex_team_memberships, +) +from webex_bot_mcp.tools.common import _cached_webex_api # noqa: E402 + +# Clear the LRU token cache so tests start with a fresh client +_cached_webex_api.cache_clear() + +# --------------------------------------------------------------------------- +# Skip all tests unless explicitly opted in. +# +# Using WEBEX_INTEGRATION_TESTS=1 prevents unit tests (which inject a fake +# WEBEX_ACCESS_TOKEN=test-token via os.environ.setdefault) from accidentally +# triggering live API calls during a normal test-discovery run. +# --------------------------------------------------------------------------- +TOKEN = os.environ.get("WEBEX_ACCESS_TOKEN", "") +_RUN = os.environ.get("WEBEX_INTEGRATION_TESTS", "").lower() in ("1", "true", "yes") +_SKIP = not (TOKEN and _RUN) +_SKIP_REASON = ( + "Set WEBEX_ACCESS_TOKEN and WEBEX_INTEGRATION_TESTS=1 to run integration tests" +) + +# Unique prefix so test resources can be identified in cleanup if needed +_PREFIX = f"[mcp-it-{int(time.time())}]" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _ok(result: dict, label: str = ""): + """Assert a tool call succeeded, showing the error message on failure.""" + assert result.get("success"), ( + f"{label + ': ' if label else ''}tool call failed — " + f"error_code={result.get('error_code')} message={result.get('message')}" + ) + return result["data"] + + +# --------------------------------------------------------------------------- +# People +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestPeopleTools(unittest.TestCase): + """Verify the bot can retrieve its own identity and search for people.""" + + def test_get_webex_me(self): + r = get_webex_me() + data = _ok(r, "get_webex_me") + self.assertIn("id", data) + self.assertIn("displayName", data) + self.assertIn(data.get("type"), ("bot", "person")) + + def test_get_me_response_shape(self): + r = get_webex_me() + self.assertIn("timestamp", r) + self.assertIn("server_version", r) + + def test_list_people_by_own_email(self): + me = _ok(get_webex_me(), "get_webex_me for email lookup") + email = me.get("emails", [None])[0] + if not email: + self.skipTest("Bot has no email address") + r = list_webex_people(email=email) + data = _ok(r, "list_webex_people") + self.assertIn("people", data) + ids = [p["id"] for p in data["people"]] + self.assertIn(me["id"], ids) + + +# --------------------------------------------------------------------------- +# Rooms — full CRUD lifecycle +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestRoomsLifecycle(unittest.TestCase): + """Create → get → list → update → delete a Webex room.""" + + room_id: str = "" + room_title = f"{_PREFIX} rooms-lifecycle" + + @classmethod + def tearDownClass(cls): + if cls.room_id: + try: + delete_webex_room(cls.room_id) + except Exception: + pass + + def test_01_create_room(self): + r = create_webex_room(title=self.room_title) + data = _ok(r, "create_webex_room") + self.assertEqual(data["title"], self.room_title) + self.assertIn("id", data) + TestRoomsLifecycle.room_id = data["id"] + + def test_02_get_room(self): + self.assertTrue(self.room_id, "room not created in test_01") + r = get_webex_room(self.room_id) + data = _ok(r, "get_webex_room") + self.assertEqual(data["id"], self.room_id) + self.assertEqual(data["title"], self.room_title) + + def test_03_list_rooms_includes_created(self): + self.assertTrue(self.room_id, "room not created in test_01") + r = list_webex_rooms(max_results=100) + data = _ok(r, "list_webex_rooms") + self.assertIn("rooms", data) + ids = [room["id"] for room in data["rooms"]] + self.assertIn(self.room_id, ids) + + def test_04_update_room_title(self): + self.assertTrue(self.room_id, "room not created in test_01") + new_title = f"{self.room_title} updated" + r = update_webex_room(self.room_id, title=new_title) + data = _ok(r, "update_webex_room") + self.assertEqual(data["title"], new_title) + # Verify the update persisted via a fresh GET + r2 = get_webex_room(self.room_id) + data2 = _ok(r2, "get_webex_room after update") + self.assertEqual(data2["title"], new_title) + + def test_05_delete_room(self): + self.assertTrue(self.room_id, "room not created in test_01") + r = delete_webex_room(self.room_id) + _ok(r, "delete_webex_room") + # Verify deletion — subsequent GET should return an error + r2 = get_webex_room(self.room_id) + self.assertFalse(r2["success"]) + self.assertIn(r2["error_code"], ("E404", "E403", "E600")) + TestRoomsLifecycle.room_id = "" # prevent tearDownClass from double-deleting + + +# --------------------------------------------------------------------------- +# Messages — full lifecycle inside a dedicated room +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestMessagesLifecycle(unittest.TestCase): + """Send, list, and delete messages in an integration-test room.""" + + room_id: str = "" + message_ids: list = [] + + @classmethod + def setUpClass(cls): + cls.message_ids = [] + r = create_webex_room(title=f"{_PREFIX} messages-lifecycle") + if r["success"]: + cls.room_id = r["data"]["id"] + + @classmethod + def tearDownClass(cls): + for mid in list(cls.message_ids): + try: + delete_webex_message(mid) + except Exception: + pass + if cls.room_id: + try: + delete_webex_room(cls.room_id) + except Exception: + pass + + def test_01_send_text_message(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + r = send_webex_message(room_id=self.room_id, text="integration-test plain text") + data = _ok(r, "send_webex_message text") + self.assertEqual(data["roomId"], self.room_id) + self.assertIn("id", data) + TestMessagesLifecycle.message_ids.append(data["id"]) + + def test_02_send_markdown_message(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + r = send_webex_message( + room_id=self.room_id, + markdown="**integration-test** _markdown_", + text="integration-test markdown fallback", + ) + data = _ok(r, "send_webex_message markdown") + self.assertEqual(data["roomId"], self.room_id) + TestMessagesLifecycle.message_ids.append(data["id"]) + + def test_03_list_messages_contains_sent(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + self.assertTrue(self.message_ids, "no messages sent in earlier tests") + r = list_webex_messages(room_id=self.room_id, max_results=20) + data = _ok(r, "list_webex_messages") + self.assertIn("messages", data) + listed_ids = {m["id"] for m in data["messages"]} + for mid in self.message_ids: + self.assertIn(mid, listed_ids, f"message {mid} not found in listing") + + def test_04_list_messages_response_shape(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + r = list_webex_messages(room_id=self.room_id, max_results=5) + data = _ok(r, "list_webex_messages shape") + for msg in data["messages"]: + self.assertIn("id", msg) + self.assertIn("roomId", msg) + self.assertIn("created", msg) + + def test_05_delete_one_message(self): + self.assertTrue(self.message_ids, "no messages to delete") + mid = TestMessagesLifecycle.message_ids.pop() + r = delete_webex_message(mid) + _ok(r, "delete_webex_message") + # Verify deletion — the message should not appear in the listing + r2 = list_webex_messages(room_id=self.room_id, max_results=50) + if r2["success"]: + listed_ids = {m["id"] for m in r2["data"]["messages"]} + self.assertNotIn(mid, listed_ids) + + def test_06_send_space_message_alias(self): + """send_webex_space_message delegates to send_webex_message.""" + self.assertTrue(self.room_id, "room not created in setUpClass") + r = send_webex_space_message(room_id=self.room_id, text="space-alias message") + data = _ok(r, "send_webex_space_message") + TestMessagesLifecycle.message_ids.append(data["id"]) + + def test_07_list_space_messages_alias(self): + """list_webex_space_messages delegates to list_webex_messages.""" + self.assertTrue(self.room_id, "room not created in setUpClass") + r = list_webex_space_messages(room_id=self.room_id, max_results=5) + data = _ok(r, "list_webex_space_messages") + self.assertIn("messages", data) + + +# --------------------------------------------------------------------------- +# Adaptive cards +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestAdaptiveCards(unittest.TestCase): + """Verify card building utility and card send tool.""" + + room_id: str = "" + + @classmethod + def setUpClass(cls): + r = create_webex_room(title=f"{_PREFIX} adaptive-cards") + if r["success"]: + cls.room_id = r["data"]["id"] + + @classmethod + def tearDownClass(cls): + if cls.room_id: + try: + delete_webex_room(cls.room_id) + except Exception: + pass + + def test_build_adaptive_card_basic(self): + """build_webex_adaptive_card constructs a card without an API call.""" + r = build_webex_adaptive_card( + title="Test Card", + text="Hello from integration tests", + ) + self.assertTrue(r["success"], r.get("message")) + card = r["data"] + self.assertIn("type", card) + self.assertEqual(card["type"], "AdaptiveCard") + + def test_send_adaptive_card(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + card = { + "type": "AdaptiveCard", + "version": "1.2", + "body": [{"type": "TextBlock", "text": "Integration test card"}], + } + r = send_webex_adaptive_card(room_id=self.room_id, card=card) + _ok(r, "send_webex_adaptive_card") + + +# --------------------------------------------------------------------------- +# Memberships +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestMembershipsTools(unittest.TestCase): + """Verify membership listing in a test room the bot created.""" + + room_id: str = "" + + @classmethod + def setUpClass(cls): + r = create_webex_room(title=f"{_PREFIX} memberships") + if r["success"]: + cls.room_id = r["data"]["id"] + + @classmethod + def tearDownClass(cls): + if cls.room_id: + try: + delete_webex_room(cls.room_id) + except Exception: + pass + + def test_list_memberships_not_empty(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + r = list_webex_memberships(room_id=self.room_id) + data = _ok(r, "list_webex_memberships") + self.assertIn("memberships", data) + # Bot that created the room must be a member + self.assertGreater(len(data["memberships"]), 0) + + def test_list_memberships_response_shape(self): + self.assertTrue(self.room_id, "room not created in setUpClass") + r = list_webex_memberships(room_id=self.room_id) + data = _ok(r, "list_webex_memberships shape") + for m in data["memberships"]: + self.assertIn("id", m) + self.assertIn("roomId", m) + self.assertEqual(m["roomId"], self.room_id) + self.assertIn("personId", m) + + def test_bot_is_moderator_of_own_room(self): + """A bot is the moderator of a room it created.""" + self.assertTrue(self.room_id, "room not created in setUpClass") + me = _ok(get_webex_me(), "get_webex_me") + r = list_webex_memberships(room_id=self.room_id) + data = _ok(r, "list_webex_memberships for moderator check") + bot_memberships = [ + m for m in data["memberships"] if m.get("personId") == me["id"] + ] + self.assertEqual(len(bot_memberships), 1, "bot should appear exactly once") + self.assertTrue(bot_memberships[0].get("isModerator")) + + def test_list_space_memberships_alias(self): + """list_webex_space_memberships delegates to list_webex_memberships.""" + self.assertTrue(self.room_id, "room not created in setUpClass") + r = list_webex_space_memberships(room_id=self.room_id) + data = _ok(r, "list_webex_space_memberships") + self.assertIn("memberships", data) + + +# --------------------------------------------------------------------------- +# Space aliases — full CRUD lifecycle mirrors rooms +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestSpaceAliases(unittest.TestCase): + """Verify space-alias tools delegate correctly to room tools.""" + + space_id: str = "" + space_title = f"{_PREFIX} space-aliases" + + @classmethod + def tearDownClass(cls): + if cls.space_id: + try: + delete_webex_space(cls.space_id) + except Exception: + pass + + def test_01_create_space(self): + r = create_webex_space(title=self.space_title) + data = _ok(r, "create_webex_space") + self.assertEqual(data["title"], self.space_title) + TestSpaceAliases.space_id = data["id"] + + def test_02_get_space(self): + self.assertTrue(self.space_id, "space not created in test_01") + r = get_webex_space(self.space_id) + data = _ok(r, "get_webex_space") + self.assertEqual(data["id"], self.space_id) + + def test_03_list_spaces_includes_created(self): + self.assertTrue(self.space_id, "space not created in test_01") + r = list_webex_spaces(max_results=100) + data = _ok(r, "list_webex_spaces") + self.assertIn("spaces", data) + ids = [s["id"] for s in data["spaces"]] + self.assertIn(self.space_id, ids) + + def test_04_update_space_title(self): + self.assertTrue(self.space_id, "space not created in test_01") + new_title = f"{self.space_title} updated" + r = update_webex_space(self.space_id, title=new_title) + data = _ok(r, "update_webex_space") + self.assertEqual(data["title"], new_title) + + def test_05_delete_space(self): + self.assertTrue(self.space_id, "space not created in test_01") + r = delete_webex_space(self.space_id) + _ok(r, "delete_webex_space") + r2 = get_webex_space(self.space_id) + self.assertFalse(r2["success"]) + TestSpaceAliases.space_id = "" + + +# --------------------------------------------------------------------------- +# Teams — listing only (bots cannot create teams via POST /teams) +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestTeamsTools(unittest.TestCase): + """Verify team-related tools that don't require team creation.""" + + def test_list_teams_returns_list(self): + r = list_webex_teams() + self.assertTrue(r["success"], r.get("message")) + self.assertIn("teams", r["data"]) + self.assertIsInstance(r["data"]["teams"], list) + + def test_list_teams_response_shape(self): + r = list_webex_teams() + self.assertTrue(r["success"]) + self.assertIn("timestamp", r) + self.assertIn("server_version", r) + + def test_get_team_invalid_id_returns_error(self): + r = get_webex_team("NOT-A-REAL-TEAM-ID") + self.assertFalse(r["success"]) + self.assertIn("error_code", r) + self.assertIn("message", r) + + def test_list_team_memberships_invalid_team(self): + r = list_webex_team_memberships("NOT-A-REAL-TEAM-ID") + self.assertFalse(r["success"]) + self.assertIn("error_code", r) + + +# --------------------------------------------------------------------------- +# Error handling — verify structured error responses for bad inputs +# --------------------------------------------------------------------------- + +@unittest.skipIf(_SKIP, _SKIP_REASON) +class TestErrorHandling(unittest.TestCase): + """Verify all tools return structured errors (not bare exceptions).""" + + def _assert_error_shape(self, result: dict, label: str): + self.assertFalse(result.get("success"), f"{label}: expected failure but got success") + self.assertIn("error_code", result, f"{label}: missing error_code") + self.assertIn("message", result, f"{label}: missing message") + self.assertIn("timestamp", result, f"{label}: missing timestamp") + self.assertIn("server_version", result, f"{label}: missing server_version") + + def test_get_room_invalid_id(self): + r = get_webex_room("INVALID-ROOM-ID-XXXX") + self._assert_error_shape(r, "get_webex_room") + + def test_delete_room_invalid_id(self): + r = delete_webex_room("INVALID-ROOM-ID-XXXX") + self._assert_error_shape(r, "delete_webex_room") + + def test_update_room_invalid_id(self): + r = update_webex_room("INVALID-ROOM-ID-XXXX", title="new title") + self._assert_error_shape(r, "update_webex_room") + + def test_delete_message_invalid_id(self): + r = delete_webex_message("INVALID-MSG-ID-XXXX") + self._assert_error_shape(r, "delete_webex_message") + + def test_list_messages_invalid_room(self): + r = list_webex_messages(room_id="INVALID-ROOM-ID-XXXX") + self._assert_error_shape(r, "list_webex_messages") + + def test_send_message_missing_destination(self): + """Sending a message with no room or person target must return E001/E002.""" + r = send_webex_message(text="orphan message with no destination") + self._assert_error_shape(r, "send_webex_message no destination") + self.assertIn(r["error_code"], ("E001", "E002", "E003")) + + def test_list_memberships_invalid_room(self): + r = list_webex_memberships(room_id="INVALID-ROOM-ID-XXXX") + self._assert_error_shape(r, "list_webex_memberships") + + def test_get_space_invalid_id(self): + r = get_webex_space("INVALID-SPACE-ID-XXXX") + self._assert_error_shape(r, "get_webex_space") + + def test_error_codes_are_in_expected_range(self): + """All error codes from API failures must match the documented E-code scheme.""" + r = get_webex_room("INVALID-ROOM-ID-XXXX") + code = r.get("error_code", "") + self.assertRegex( + code, r"^E\d{3}$", + f"error_code '{code}' doesn't match E-code format" + ) + + +if __name__ == "__main__": + unittest.main() From e4476263a44ce1cdde5d8b13078d3499d6e2381c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 3 Jun 2026 01:03:49 +0000 Subject: [PATCH 2/2] test: fix integration tests against live Webex API All 39 tests now pass against the real bot token. Fixes from live run: - Correct response shapes: get_webex_me returns {'user': {...}}, room tools return {'room': {...}}, space aliases return {'space': {...}} / {'spaces': [...]} - Space alias tools take space_id param (not room_id) and rename keys - build_webex_adaptive_card returns raw {'card_body', 'card_actions'} dict (not a success/error wrapper); corrected signature (body_text, not text) and test assertions - send_webex_adaptive_card takes card_body/card_actions kwargs; spread build output directly with **card - list_webex_messages returns E403 for bots in group spaces (Webex enforces this for all bot tokens); test accepts either success or structured error - Bot is not automatically a moderator of rooms it creates; weaken assertion to membership presence only - FactSet is nested inside Container in card_body; traverse items https://claude.ai/code/session_01Bykkm6kE75wpwuiuL9MHny --- tests/test_integration.py | 253 +++++++++++++++++++++++--------------- 1 file changed, 155 insertions(+), 98 deletions(-) diff --git a/tests/test_integration.py b/tests/test_integration.py index c30f474..df27d8c 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2,8 +2,8 @@ Integration tests for webex-bot-mcp. These tests make real Webex API calls and verify that each tool works -end-to-end. They are automatically SKIPPED when WEBEX_ACCESS_TOKEN is -not set, so they are safe to include in CI without credentials. +end-to-end. They are automatically SKIPPED when the opt-in flag is not +set, so they are safe to include in CI without credentials. Run integration tests with an explicit opt-in flag so they are always skipped during normal `unittest discover` runs: @@ -18,12 +18,18 @@ Each test class creates its own Webex resources in setUpClass and deletes them in tearDownClass, even when individual tests fail. + +Known Webex bot limitations reflected in these tests: +- Bots cannot list messages in group spaces (GET /messages returns 403) + unless they were @mentioned. Tests that call list_webex_messages accept + either a success response or an E403 error as correct behaviour. +- Bots cannot create teams (POST /teams returns 401). Team tests only + cover read-only operations. """ import os import sys import time -import types import unittest from unittest.mock import MagicMock @@ -84,7 +90,6 @@ def tearDownModule(): ) from webex_bot_mcp.tools.memberships import ( # noqa: E402 list_webex_memberships, add_webex_membership, - update_webex_membership, delete_webex_membership, list_webex_space_memberships, ) from webex_bot_mcp.tools.people import ( # noqa: E402 @@ -113,7 +118,7 @@ def tearDownModule(): "Set WEBEX_ACCESS_TOKEN and WEBEX_INTEGRATION_TESTS=1 to run integration tests" ) -# Unique prefix so test resources can be identified in cleanup if needed +# Unique prefix so test resources can be identified even if cleanup fails _PREFIX = f"[mcp-it-{int(time.time())}]" @@ -121,7 +126,7 @@ def tearDownModule(): # Helpers # --------------------------------------------------------------------------- -def _ok(result: dict, label: str = ""): +def _ok(result: dict, label: str = "") -> dict: """Assert a tool call succeeded, showing the error message on failure.""" assert result.get("success"), ( f"{label + ': ' if label else ''}tool call failed — " @@ -136,14 +141,17 @@ def _ok(result: dict, label: str = ""): @unittest.skipIf(_SKIP, _SKIP_REASON) class TestPeopleTools(unittest.TestCase): - """Verify the bot can retrieve its own identity and search for people.""" + """Verify the bot can retrieve its own identity.""" def test_get_webex_me(self): r = get_webex_me() data = _ok(r, "get_webex_me") - self.assertIn("id", data) - self.assertIn("displayName", data) - self.assertIn(data.get("type"), ("bot", "person")) + # Response shape: {'user': {'id': ..., 'displayName': ..., 'type': ...}} + self.assertIn("user", data) + user = data["user"] + self.assertIn("id", user) + self.assertIn("displayName", user) + self.assertIn(user.get("type"), ("bot", "person")) def test_get_me_response_shape(self): r = get_webex_me() @@ -151,15 +159,16 @@ def test_get_me_response_shape(self): self.assertIn("server_version", r) def test_list_people_by_own_email(self): - me = _ok(get_webex_me(), "get_webex_me for email lookup") - email = me.get("emails", [None])[0] - if not email: + r = get_webex_me() + data = _ok(r, "get_webex_me for email lookup") + emails = data["user"].get("emails", []) + if not emails: self.skipTest("Bot has no email address") - r = list_webex_people(email=email) - data = _ok(r, "list_webex_people") - self.assertIn("people", data) - ids = [p["id"] for p in data["people"]] - self.assertIn(me["id"], ids) + r2 = list_webex_people(email=emails[0]) + data2 = _ok(r2, "list_webex_people") + self.assertIn("people", data2) + ids = [p["id"] for p in data2["people"]] + self.assertIn(data["user"]["id"], ids) # --------------------------------------------------------------------------- @@ -170,6 +179,7 @@ def test_list_people_by_own_email(self): class TestRoomsLifecycle(unittest.TestCase): """Create → get → list → update → delete a Webex room.""" + # Class-level state threaded through ordered tests room_id: str = "" room_title = f"{_PREFIX} rooms-lifecycle" @@ -184,21 +194,27 @@ def tearDownClass(cls): def test_01_create_room(self): r = create_webex_room(title=self.room_title) data = _ok(r, "create_webex_room") - self.assertEqual(data["title"], self.room_title) - self.assertIn("id", data) - TestRoomsLifecycle.room_id = data["id"] + # Response shape: {'room': {'id': ..., 'title': ...}} + self.assertIn("room", data) + room = data["room"] + self.assertEqual(room["title"], self.room_title) + self.assertIn("id", room) + TestRoomsLifecycle.room_id = room["id"] def test_02_get_room(self): self.assertTrue(self.room_id, "room not created in test_01") r = get_webex_room(self.room_id) data = _ok(r, "get_webex_room") - self.assertEqual(data["id"], self.room_id) - self.assertEqual(data["title"], self.room_title) + # Response shape: {'room': {...}} + room = data["room"] + self.assertEqual(room["id"], self.room_id) + self.assertEqual(room["title"], self.room_title) def test_03_list_rooms_includes_created(self): self.assertTrue(self.room_id, "room not created in test_01") r = list_webex_rooms(max_results=100) data = _ok(r, "list_webex_rooms") + # Response shape: {'rooms': [...]} self.assertIn("rooms", data) ids = [room["id"] for room in data["rooms"]] self.assertIn(self.room_id, ids) @@ -208,30 +224,36 @@ def test_04_update_room_title(self): new_title = f"{self.room_title} updated" r = update_webex_room(self.room_id, title=new_title) data = _ok(r, "update_webex_room") - self.assertEqual(data["title"], new_title) - # Verify the update persisted via a fresh GET + # Response shape: {'room': {...}} + self.assertEqual(data["room"]["title"], new_title) + # Verify the update persisted r2 = get_webex_room(self.room_id) - data2 = _ok(r2, "get_webex_room after update") - self.assertEqual(data2["title"], new_title) + self.assertEqual(_ok(r2, "get after update")["room"]["title"], new_title) def test_05_delete_room(self): self.assertTrue(self.room_id, "room not created in test_01") r = delete_webex_room(self.room_id) - _ok(r, "delete_webex_room") + data = _ok(r, "delete_webex_room") + # Response shape: {'deleted': True, 'room_id': ...} + self.assertTrue(data.get("deleted")) # Verify deletion — subsequent GET should return an error r2 = get_webex_room(self.room_id) self.assertFalse(r2["success"]) self.assertIn(r2["error_code"], ("E404", "E403", "E600")) - TestRoomsLifecycle.room_id = "" # prevent tearDownClass from double-deleting + TestRoomsLifecycle.room_id = "" # prevent tearDownClass double-delete # --------------------------------------------------------------------------- -# Messages — full lifecycle inside a dedicated room +# Messages — lifecycle inside a dedicated room # --------------------------------------------------------------------------- @unittest.skipIf(_SKIP, _SKIP_REASON) class TestMessagesLifecycle(unittest.TestCase): - """Send, list, and delete messages in an integration-test room.""" + """Send and delete messages in an integration-test room. + + Note: Webex bots cannot list messages in group spaces unless @mentioned + (API returns 403). list_webex_messages tests accept either success or E403. + """ room_id: str = "" message_ids: list = [] @@ -241,7 +263,7 @@ def setUpClass(cls): cls.message_ids = [] r = create_webex_room(title=f"{_PREFIX} messages-lifecycle") if r["success"]: - cls.room_id = r["data"]["id"] + cls.room_id = r["data"]["room"]["id"] @classmethod def tearDownClass(cls): @@ -260,8 +282,9 @@ def test_01_send_text_message(self): self.assertTrue(self.room_id, "room not created in setUpClass") r = send_webex_message(room_id=self.room_id, text="integration-test plain text") data = _ok(r, "send_webex_message text") - self.assertEqual(data["roomId"], self.room_id) + # Response shape: flat {'id': ..., 'roomId': ..., 'text': ..., ...} self.assertIn("id", data) + self.assertEqual(data["roomId"], self.room_id) TestMessagesLifecycle.message_ids.append(data["id"]) def test_02_send_markdown_message(self): @@ -272,52 +295,57 @@ def test_02_send_markdown_message(self): text="integration-test markdown fallback", ) data = _ok(r, "send_webex_message markdown") - self.assertEqual(data["roomId"], self.room_id) + self.assertIn("id", data) TestMessagesLifecycle.message_ids.append(data["id"]) - def test_03_list_messages_contains_sent(self): - self.assertTrue(self.room_id, "room not created in setUpClass") - self.assertTrue(self.message_ids, "no messages sent in earlier tests") - r = list_webex_messages(room_id=self.room_id, max_results=20) - data = _ok(r, "list_webex_messages") - self.assertIn("messages", data) - listed_ids = {m["id"] for m in data["messages"]} - for mid in self.message_ids: - self.assertIn(mid, listed_ids, f"message {mid} not found in listing") + def test_03_list_messages_structured_response(self): + """list_webex_messages returns a valid structured response. - def test_04_list_messages_response_shape(self): + Bots receive E403 in group spaces unless @mentioned — both outcomes + are verified to ensure the tool returns a correct response envelope. + """ self.assertTrue(self.room_id, "room not created in setUpClass") - r = list_webex_messages(room_id=self.room_id, max_results=5) - data = _ok(r, "list_webex_messages shape") - for msg in data["messages"]: - self.assertIn("id", msg) - self.assertIn("roomId", msg) - self.assertIn("created", msg) - - def test_05_delete_one_message(self): + r = list_webex_messages(room_id=self.room_id, max_results=20) + # Must be a valid structured response either way + self.assertIn("success", r) + self.assertIn("timestamp", r) + self.assertIn("server_version", r) + if r["success"]: + self.assertIn("messages", r["data"]) + self.assertIsInstance(r["data"]["messages"], list) + else: + # E403 is expected for bots in group spaces (Webex limitation) + self.assertIn("error_code", r) + self.assertIn("message", r) + + def test_04_delete_message(self): self.assertTrue(self.message_ids, "no messages to delete") mid = TestMessagesLifecycle.message_ids.pop() r = delete_webex_message(mid) - _ok(r, "delete_webex_message") - # Verify deletion — the message should not appear in the listing - r2 = list_webex_messages(room_id=self.room_id, max_results=50) - if r2["success"]: - listed_ids = {m["id"] for m in r2["data"]["messages"]} - self.assertNotIn(mid, listed_ids) - - def test_06_send_space_message_alias(self): + data = _ok(r, "delete_webex_message") + # Response shape: {'deleted': True, 'message_id': ...} + self.assertTrue(data.get("deleted")) + + def test_05_send_space_message_alias(self): """send_webex_space_message delegates to send_webex_message.""" self.assertTrue(self.room_id, "room not created in setUpClass") - r = send_webex_space_message(room_id=self.room_id, text="space-alias message") + # Space alias uses space_id parameter + r = send_webex_space_message(space_id=self.room_id, text="space-alias message") data = _ok(r, "send_webex_space_message") + self.assertIn("id", data) TestMessagesLifecycle.message_ids.append(data["id"]) - def test_07_list_space_messages_alias(self): + def test_06_list_space_messages_alias(self): """list_webex_space_messages delegates to list_webex_messages.""" self.assertTrue(self.room_id, "room not created in setUpClass") - r = list_webex_space_messages(room_id=self.room_id, max_results=5) - data = _ok(r, "list_webex_space_messages") - self.assertIn("messages", data) + # Space alias uses space_id parameter; same E403 caveat as test_03 + r = list_webex_space_messages(space_id=self.room_id, max_results=5) + self.assertIn("success", r) + self.assertIn("timestamp", r) + if r["success"]: + self.assertIn("messages", r["data"]) + else: + self.assertIn("error_code", r) # --------------------------------------------------------------------------- @@ -334,7 +362,7 @@ class TestAdaptiveCards(unittest.TestCase): def setUpClass(cls): r = create_webex_room(title=f"{_PREFIX} adaptive-cards") if r["success"]: - cls.room_id = r["data"]["id"] + cls.room_id = r["data"]["room"]["id"] @classmethod def tearDownClass(cls): @@ -345,25 +373,48 @@ def tearDownClass(cls): pass def test_build_adaptive_card_basic(self): - """build_webex_adaptive_card constructs a card without an API call.""" - r = build_webex_adaptive_card( + """build_webex_adaptive_card constructs a card dict without an API call.""" + # Returns raw {'card_body': [...], 'card_actions': [...]} — not a success wrapper + result = build_webex_adaptive_card( title="Test Card", - text="Hello from integration tests", + body_text="Hello from integration tests", ) - self.assertTrue(r["success"], r.get("message")) - card = r["data"] - self.assertIn("type", card) - self.assertEqual(card["type"], "AdaptiveCard") + self.assertIn("card_body", result) + self.assertIn("card_actions", result) + self.assertIsInstance(result["card_body"], list) + self.assertGreater(len(result["card_body"]), 0) + + def test_build_adaptive_card_with_facts(self): + result = build_webex_adaptive_card( + title="Deployment", + subtitle="v1.0.0", + facts=[{"title": "Region", "value": "us-east-1"}], + ) + self.assertIn("card_body", result) + # card_body is a list of Container elements; FactSet lives inside items + all_types = [] + for elem in result["card_body"]: + all_types.append(elem.get("type")) + for inner in elem.get("items", []): + all_types.append(inner.get("type")) + self.assertIn("FactSet", all_types) def test_send_adaptive_card(self): self.assertTrue(self.room_id, "room not created in setUpClass") - card = { - "type": "AdaptiveCard", - "version": "1.2", - "body": [{"type": "TextBlock", "text": "Integration test card"}], - } - r = send_webex_adaptive_card(room_id=self.room_id, card=card) - _ok(r, "send_webex_adaptive_card") + card = build_webex_adaptive_card( + title="Integration Test Card", + body_text="Sent by the MCP integration test suite.", + ) + # Spread card_body / card_actions directly into send_webex_adaptive_card + r = send_webex_adaptive_card( + room_id=self.room_id, + fallback_text="Integration Test Card", + **card, + ) + data = _ok(r, "send_webex_adaptive_card") + # Same flat message shape as send_webex_message + self.assertIn("id", data) + self.assertEqual(data["roomId"], self.room_id) # --------------------------------------------------------------------------- @@ -380,7 +431,7 @@ class TestMembershipsTools(unittest.TestCase): def setUpClass(cls): r = create_webex_room(title=f"{_PREFIX} memberships") if r["success"]: - cls.room_id = r["data"]["id"] + cls.room_id = r["data"]["room"]["id"] @classmethod def tearDownClass(cls): @@ -394,8 +445,8 @@ def test_list_memberships_not_empty(self): self.assertTrue(self.room_id, "room not created in setUpClass") r = list_webex_memberships(room_id=self.room_id) data = _ok(r, "list_webex_memberships") + # Response shape: {'memberships': [...]} self.assertIn("memberships", data) - # Bot that created the room must be a member self.assertGreater(len(data["memberships"]), 0) def test_list_memberships_response_shape(self): @@ -407,29 +458,29 @@ def test_list_memberships_response_shape(self): self.assertIn("roomId", m) self.assertEqual(m["roomId"], self.room_id) self.assertIn("personId", m) + self.assertIn("isModerator", m) - def test_bot_is_moderator_of_own_room(self): - """A bot is the moderator of a room it created.""" + def test_bot_is_member_of_own_room(self): + """A bot must be a member of a room it created.""" self.assertTrue(self.room_id, "room not created in setUpClass") - me = _ok(get_webex_me(), "get_webex_me") + me_data = _ok(get_webex_me(), "get_webex_me") + bot_id = me_data["user"]["id"] r = list_webex_memberships(room_id=self.room_id) - data = _ok(r, "list_webex_memberships for moderator check") - bot_memberships = [ - m for m in data["memberships"] if m.get("personId") == me["id"] - ] - self.assertEqual(len(bot_memberships), 1, "bot should appear exactly once") - self.assertTrue(bot_memberships[0].get("isModerator")) + data = _ok(r, "list_webex_memberships for member check") + bot_entries = [m for m in data["memberships"] if m.get("personId") == bot_id] + self.assertEqual(len(bot_entries), 1, "bot should appear exactly once as a member") def test_list_space_memberships_alias(self): """list_webex_space_memberships delegates to list_webex_memberships.""" self.assertTrue(self.room_id, "room not created in setUpClass") - r = list_webex_space_memberships(room_id=self.room_id) + # Space alias uses space_id parameter + r = list_webex_space_memberships(space_id=self.room_id) data = _ok(r, "list_webex_space_memberships") self.assertIn("memberships", data) # --------------------------------------------------------------------------- -# Space aliases — full CRUD lifecycle mirrors rooms +# Space aliases — full CRUD lifecycle # --------------------------------------------------------------------------- @unittest.skipIf(_SKIP, _SKIP_REASON) @@ -450,19 +501,23 @@ def tearDownClass(cls): def test_01_create_space(self): r = create_webex_space(title=self.space_title) data = _ok(r, "create_webex_space") - self.assertEqual(data["title"], self.space_title) - TestSpaceAliases.space_id = data["id"] + # Space alias transforms 'room' → 'space' in the response + self.assertIn("space", data) + self.assertEqual(data["space"]["title"], self.space_title) + TestSpaceAliases.space_id = data["space"]["id"] def test_02_get_space(self): self.assertTrue(self.space_id, "space not created in test_01") r = get_webex_space(self.space_id) data = _ok(r, "get_webex_space") - self.assertEqual(data["id"], self.space_id) + # Response shape: {'space': {...}} + self.assertEqual(data["space"]["id"], self.space_id) def test_03_list_spaces_includes_created(self): self.assertTrue(self.space_id, "space not created in test_01") r = list_webex_spaces(max_results=100) data = _ok(r, "list_webex_spaces") + # Space alias transforms 'rooms' → 'spaces' self.assertIn("spaces", data) ids = [s["id"] for s in data["spaces"]] self.assertIn(self.space_id, ids) @@ -472,12 +527,14 @@ def test_04_update_space_title(self): new_title = f"{self.space_title} updated" r = update_webex_space(self.space_id, title=new_title) data = _ok(r, "update_webex_space") - self.assertEqual(data["title"], new_title) + # Response shape: {'space': {...}} + self.assertEqual(data["space"]["title"], new_title) def test_05_delete_space(self): self.assertTrue(self.space_id, "space not created in test_01") r = delete_webex_space(self.space_id) - _ok(r, "delete_webex_space") + data = _ok(r, "delete_webex_space") + self.assertTrue(data.get("deleted")) r2 = get_webex_space(self.space_id) self.assertFalse(r2["success"]) TestSpaceAliases.space_id = ""