diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9c8ebd5..01d3bfd 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -4,12 +4,17 @@ on: pull_request: branches: - main + types: [opened, synchronize, reopened] + push: + branches-ignore: + - main jobs: test: - name: Run unit tests + name: Run unit tests (Python ${{ matrix.python-version }}) runs-on: ubuntu-latest strategy: + fail-fast: false matrix: python-version: ["3.12", "3.13"] @@ -30,4 +35,12 @@ jobs: run: uv sync - name: Run tests - run: uv run python -m unittest discover -s tests -v + run: uv run python -m unittest discover -s tests -v 2>&1 | tee test-output.txt; exit ${PIPESTATUS[0]} + + - name: Summarize results + if: always() + run: | + echo "## Test Results (Python ${{ matrix.python-version }})" >> $GITHUB_STEP_SUMMARY + echo '```' >> $GITHUB_STEP_SUMMARY + tail -20 test-output.txt >> $GITHUB_STEP_SUMMARY + echo '```' >> $GITHUB_STEP_SUMMARY diff --git a/tests/test_memberships.py b/tests/test_memberships.py new file mode 100644 index 0000000..09fa8c0 --- /dev/null +++ b/tests/test_memberships.py @@ -0,0 +1,397 @@ +""" +Unit tests for Webex Membership management tools. + +webexpythonsdk is mocked at the sys.modules level so tests can run without +real Webex credentials or the SDK being installed. +""" +import os +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +# ── Provide fake env var and SDK before any tools module is imported ────────── +os.environ.setdefault("WEBEX_ACCESS_TOKEN", "test-token") + +_sdk_mod = types.ModuleType("webexpythonsdk") +_sdk_mod.WebexAPI = MagicMock(return_value=MagicMock()) +sys.modules.setdefault("webexpythonsdk", _sdk_mod) + +# ── Also stub python-dotenv if missing ─────────────────────────────────────── +if "dotenv" not in sys.modules: + _dotenv_mod = types.ModuleType("dotenv") + _dotenv_mod.load_dotenv = lambda *a, **kw: None + sys.modules["dotenv"] = _dotenv_mod + +import webex_bot_mcp.tools.memberships as _memberships_mod # noqa: E402 +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, + add_webex_space_membership, +) + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _fake_membership(**overrides): + m = MagicMock() + m.id = "MEMBERSHIPID" + m.roomId = "ROOMID" + m.personId = "PERSONID" + m.personEmail = "user@example.com" + m.personDisplayName = "Test User" + m.personOrgId = "ORGID" + m.isModerator = False + m.isMonitor = False + m.created = "2025-01-01T00:00:00Z" + m.roomType = None # optional + for k, v in overrides.items(): + setattr(m, k, v) + return m + + +# ── list_webex_memberships ──────────────────────────────────────────────────── + +class TestListWebexMemberships(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + return mock_api, list_webex_memberships(**kwargs) + + def test_returns_memberships_list(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership(), _fake_membership(id="MID2")] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertTrue(r['success']) + self.assertEqual(len(r['data']['memberships']), 2) + + def test_empty_list(self): + mock_api, r = self._call() + self.assertTrue(r['success']) + self.assertEqual(r['data']['memberships'], []) + + def test_metadata_count_matches(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership()] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertEqual(r['metadata']['count'], 1) + + def test_room_id_filter_passed(self): + mock_api, _ = self._call(room_id="ROOMID") + mock_api.memberships.list.assert_called_once_with(roomId="ROOMID") + + def test_person_id_filter_passed(self): + mock_api, _ = self._call(person_id="PID") + mock_api.memberships.list.assert_called_once_with(personId="PID") + + def test_person_email_filter_passed(self): + mock_api, _ = self._call(person_email="user@example.com") + mock_api.memberships.list.assert_called_once_with(personEmail="user@example.com") + + def test_max_results_passed(self): + mock_api, _ = self._call(max_results=25) + mock_api.memberships.list.assert_called_once_with(max=25) + + def test_no_filters_calls_list_with_no_args(self): + mock_api, _ = self._call() + mock_api.memberships.list.assert_called_once_with() + + def test_membership_dict_has_required_fields(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership()] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + m = r['data']['memberships'][0] + for field in ('id', 'roomId', 'personId', 'personEmail', + 'personDisplayName', 'isModerator', 'isMonitor', 'created'): + self.assertIn(field, m) + + def test_optional_room_type_included_when_set(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership(roomType="group")] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertEqual(r['data']['memberships'][0]['roomType'], "group") + + def test_room_type_absent_when_none(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership()] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertNotIn('roomType', r['data']['memberships'][0]) + + def test_unauthorized_error(self): + mock_api = MagicMock() + mock_api.memberships.list.side_effect = Exception("unauthorized") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_not_found_error(self): + mock_api = MagicMock() + mock_api.memberships.list.side_effect = Exception("not found") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_rate_limit_error_is_temporary(self): + mock_api = MagicMock() + mock_api.memberships.list.side_effect = Exception("rate limit exceeded") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_memberships() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E503') + self.assertTrue(r.get('temporary')) + + +# ── add_webex_membership ────────────────────────────────────────────────────── + +class TestAddWebexMembership(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.memberships.create.return_value = _fake_membership() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + return mock_api, add_webex_membership(**kwargs) + + def test_missing_person_id_and_email_returns_error(self): + _, r = self._call(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_success_with_person_id(self): + _, r = self._call(room_id="ROOMID", person_id="PID") + self.assertTrue(r['success']) + + def test_success_with_person_email(self): + _, r = self._call(room_id="ROOMID", person_email="user@example.com") + self.assertTrue(r['success']) + + def test_person_id_passed_when_both_given(self): + mock_api, _ = self._call(room_id="ROOMID", person_id="PID", person_email="u@e.com") + call_kwargs = mock_api.memberships.create.call_args[1] + self.assertIn('personId', call_kwargs) + self.assertNotIn('personEmail', call_kwargs) + + def test_email_used_when_no_person_id(self): + mock_api, _ = self._call(room_id="ROOMID", person_email="user@example.com") + call_kwargs = mock_api.memberships.create.call_args[1] + self.assertIn('personEmail', call_kwargs) + self.assertNotIn('personId', call_kwargs) + + def test_is_moderator_passed(self): + mock_api, _ = self._call(room_id="ROOMID", person_id="PID", is_moderator=True) + call_kwargs = mock_api.memberships.create.call_args[1] + self.assertTrue(call_kwargs['isModerator']) + + def test_room_id_always_passed(self): + mock_api, _ = self._call(room_id="ROOMID", person_id="PID") + call_kwargs = mock_api.memberships.create.call_args[1] + self.assertEqual(call_kwargs['roomId'], "ROOMID") + + def test_returns_membership_in_data(self): + _, r = self._call(room_id="ROOMID", person_id="PID") + self.assertIn('membership', r['data']) + + def test_operation_in_metadata(self): + _, r = self._call(room_id="ROOMID", person_id="PID") + self.assertEqual(r['metadata']['operation'], 'add_membership') + + def test_forbidden_returns_error(self): + mock_api = MagicMock() + mock_api.memberships.create.side_effect = Exception("forbidden") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = add_webex_membership(room_id="ROOMID", person_id="PID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E403') + + def test_has_timestamp(self): + _, r = self._call(room_id="ROOMID", person_id="PID") + self.assertIn('timestamp', r) + + +# ── update_webex_membership ─────────────────────────────────────────────────── + +class TestUpdateWebexMembership(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.memberships.update.return_value = _fake_membership() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + return mock_api, update_webex_membership(**kwargs) + + def test_no_fields_returns_error(self): + _, r = self._call(membership_id="MID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_success_with_is_moderator(self): + _, r = self._call(membership_id="MID", is_moderator=True) + self.assertTrue(r['success']) + + def test_success_with_is_monitor(self): + _, r = self._call(membership_id="MID", is_monitor=False) + self.assertTrue(r['success']) + + def test_is_moderator_passed_to_sdk(self): + mock_api, _ = self._call(membership_id="MID", is_moderator=True) + call_kwargs = mock_api.memberships.update.call_args[1] + self.assertTrue(call_kwargs['isModerator']) + + def test_is_monitor_passed_to_sdk(self): + mock_api, _ = self._call(membership_id="MID", is_monitor=True) + call_kwargs = mock_api.memberships.update.call_args[1] + self.assertTrue(call_kwargs['isMonitor']) + + def test_membership_id_passed_to_sdk(self): + mock_api, _ = self._call(membership_id="MID123", is_moderator=False) + call_kwargs = mock_api.memberships.update.call_args[1] + self.assertEqual(call_kwargs['membershipId'], "MID123") + + def test_operation_in_metadata(self): + _, r = self._call(membership_id="MID", is_moderator=True) + self.assertEqual(r['metadata']['operation'], 'update_membership') + + def test_returns_membership_in_data(self): + _, r = self._call(membership_id="MID", is_moderator=False) + self.assertIn('membership', r['data']) + + def test_not_found_returns_error(self): + mock_api = MagicMock() + mock_api.memberships.update.side_effect = Exception("not found") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = update_webex_membership(membership_id="MID", is_moderator=True) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_forbidden_returns_error(self): + mock_api = MagicMock() + mock_api.memberships.update.side_effect = Exception("forbidden") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = update_webex_membership(membership_id="MID", is_moderator=True) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E403') + + +# ── delete_webex_membership ─────────────────────────────────────────────────── + +class TestDeleteWebexMembership(unittest.TestCase): + def test_success_returns_deleted_true(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertTrue(r['success']) + self.assertTrue(r['data']['deleted']) + + def test_membership_id_in_response(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertEqual(r['data']['membership_id'], "MID") + + def test_membership_id_passed_to_sdk(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + delete_webex_membership(membership_id="DELID") + mock_api.memberships.delete.assert_called_once_with(membershipId="DELID") + + def test_empty_membership_id_returns_error(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_not_found_returns_error(self): + mock_api = MagicMock() + mock_api.memberships.delete.side_effect = Exception("not found") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_unauthorized_returns_error(self): + mock_api = MagicMock() + mock_api.memberships.delete.side_effect = Exception("unauthorized") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_operation_in_metadata(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertEqual(r['metadata']['operation'], 'delete_membership') + + def test_has_timestamp(self): + mock_api = MagicMock() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_membership(membership_id="MID") + self.assertIn('timestamp', r) + + +# ── list_webex_space_memberships (alias) ────────────────────────────────────── + +class TestListWebexSpaceMemberships(unittest.TestCase): + def test_delegates_to_list_memberships(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [_fake_membership()] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_space_memberships(space_id="SID") + mock_api.memberships.list.assert_called_once_with(roomId="SID") + self.assertTrue(r['success']) + + def test_all_filters_forwarded(self): + mock_api = MagicMock() + mock_api.memberships.list.return_value = [] + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + list_webex_space_memberships( + space_id="SID", person_id="PID", person_email="u@e.com", max_results=10 + ) + mock_api.memberships.list.assert_called_once_with( + roomId="SID", personId="PID", personEmail="u@e.com", max=10 + ) + + def test_error_passthrough(self): + mock_api = MagicMock() + mock_api.memberships.list.side_effect = Exception("forbidden") + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_space_memberships(space_id="SID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E403') + + +# ── add_webex_space_membership (alias) ──────────────────────────────────────── + +class TestAddWebexSpaceMembership(unittest.TestCase): + def test_delegates_to_add_membership(self): + mock_api = MagicMock() + mock_api.memberships.create.return_value = _fake_membership() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + r = add_webex_space_membership(space_id="SID", person_id="PID") + mock_api.memberships.create.assert_called_once_with(roomId="SID", personId="PID") + self.assertTrue(r['success']) + + def test_validation_still_enforced(self): + r = add_webex_space_membership(space_id="SID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_moderator_flag_forwarded(self): + mock_api = MagicMock() + mock_api.memberships.create.return_value = _fake_membership() + with patch.object(_memberships_mod, 'get_webex_api', return_value=mock_api): + add_webex_space_membership(space_id="SID", person_email="u@e.com", is_moderator=True) + call_kwargs = mock_api.memberships.create.call_args[1] + self.assertTrue(call_kwargs['isModerator']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_people.py b/tests/test_people.py new file mode 100644 index 0000000..b075812 --- /dev/null +++ b/tests/test_people.py @@ -0,0 +1,265 @@ +""" +Unit tests for Webex People management tools. + +webexpythonsdk is mocked at the sys.modules level so tests can run without +real Webex credentials or the SDK being installed. +""" +import os +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +# ── Provide fake env var and SDK before any tools module is imported ────────── +os.environ.setdefault("WEBEX_ACCESS_TOKEN", "test-token") + +_sdk_mod = types.ModuleType("webexpythonsdk") +_sdk_mod.WebexAPI = MagicMock(return_value=MagicMock()) +sys.modules.setdefault("webexpythonsdk", _sdk_mod) + +# ── Also stub python-dotenv if missing ─────────────────────────────────────── +if "dotenv" not in sys.modules: + _dotenv_mod = types.ModuleType("dotenv") + _dotenv_mod.load_dotenv = lambda *a, **kw: None + sys.modules["dotenv"] = _dotenv_mod + +import webex_bot_mcp.tools.people as _people_mod # noqa: E402 +from webex_bot_mcp.tools.people import ( # noqa: E402 + get_webex_me, + list_webex_people, +) + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _fake_person(**overrides): + p = MagicMock() + p.id = "PERSONID" + p.emails = ["user@example.com"] + p.displayName = "Test User" + p.nickName = "Test" + p.firstName = "Test" + p.lastName = "User" + p.avatar = "https://avatar.example.com/avatar.jpg" + p.orgId = "ORGID" + p.created = "2025-01-01T00:00:00Z" + p.status = "active" + p.type = "person" + # Optional attributes — must be falsy so they are not included by default + p.userName = None + p.lastModified = None + p.roles = None + p.licenses = None + p.phoneNumbers = None + p.extension = None + p.locationId = None + p.addresses = None + p.timezone = None + for k, v in overrides.items(): + setattr(p, k, v) + return p + + +# ── get_webex_me ────────────────────────────────────────────────────────────── + +class TestGetWebexMe(unittest.TestCase): + def test_success_returns_user(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertTrue(r['success']) + self.assertIn('user', r['data']) + + def test_no_parameters_passed_to_sdk(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + get_webex_me() + mock_api.people.me.assert_called_once_with() + + def test_user_dict_has_required_fields(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + user = r['data']['user'] + for field in ('id', 'emails', 'displayName', 'nickName', 'firstName', + 'lastName', 'orgId', 'created', 'status', 'type'): + self.assertIn(field, user) + + def test_optional_fields_absent_when_none(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + user = r['data']['user'] + for field in ('userName', 'roles', 'licenses', 'phoneNumbers', 'timezone'): + self.assertNotIn(field, user) + + def test_optional_timezone_included_when_set(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person(timezone="America/New_York") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertEqual(r['data']['user']['timezone'], "America/New_York") + + def test_optional_roles_included_when_set(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person(roles=["admin"]) + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertEqual(r['data']['user']['roles'], ["admin"]) + + def test_has_timestamp(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertIn('timestamp', r) + + def test_has_server_version(self): + mock_api = MagicMock() + mock_api.people.me.return_value = _fake_person() + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertIn('server_version', r) + + def test_unauthorized_returns_error(self): + mock_api = MagicMock() + mock_api.people.me.side_effect = Exception("unauthorized") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_network_error_is_temporary(self): + mock_api = MagicMock() + mock_api.people.me.side_effect = Exception("network connection refused") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E601') + self.assertTrue(r.get('temporary')) + + def test_forbidden_returns_error(self): + mock_api = MagicMock() + mock_api.people.me.side_effect = Exception("forbidden") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_me() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E403') + + +# ── list_webex_people ───────────────────────────────────────────────────────── + +class TestListWebexPeople(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.people.list.return_value = [] + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + return mock_api, list_webex_people(**kwargs) + + def test_returns_people_list(self): + mock_api = MagicMock() + mock_api.people.list.return_value = [_fake_person(), _fake_person(id="PID2")] + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + self.assertTrue(r['success']) + self.assertEqual(len(r['data']['people']), 2) + + def test_empty_list(self): + mock_api, r = self._call() + self.assertTrue(r['success']) + self.assertEqual(r['data']['people'], []) + + def test_metadata_count_matches(self): + mock_api = MagicMock() + mock_api.people.list.return_value = [_fake_person(), _fake_person()] + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + self.assertEqual(r['metadata']['count'], 2) + + def test_email_filter_passed(self): + mock_api, _ = self._call(email="user@example.com") + mock_api.people.list.assert_called_once_with(email="user@example.com") + + def test_display_name_filter_passed(self): + mock_api, _ = self._call(display_name="Test User") + mock_api.people.list.assert_called_once_with(displayName="Test User") + + def test_person_id_filter_passed(self): + mock_api, _ = self._call(person_id="PID") + mock_api.people.list.assert_called_once_with(id="PID") + + def test_org_id_filter_passed(self): + mock_api, _ = self._call(org_id="OID") + mock_api.people.list.assert_called_once_with(orgId="OID") + + def test_calling_data_filter_passed(self): + mock_api, _ = self._call(calling_data=True) + mock_api.people.list.assert_called_once_with(callingData=True) + + def test_location_id_filter_passed(self): + mock_api, _ = self._call(location_id="LID") + mock_api.people.list.assert_called_once_with(locationId="LID") + + def test_max_results_passed(self): + mock_api, _ = self._call(max_results=50) + mock_api.people.list.assert_called_once_with(max=50) + + def test_no_filters_calls_list_with_no_args(self): + mock_api, _ = self._call() + mock_api.people.list.assert_called_once_with() + + def test_multiple_filters_combined(self): + mock_api, _ = self._call(email="u@e.com", org_id="OID", max_results=10) + mock_api.people.list.assert_called_once_with(email="u@e.com", orgId="OID", max=10) + + def test_person_dict_has_required_fields(self): + mock_api = MagicMock() + mock_api.people.list.return_value = [_fake_person()] + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + person = r['data']['people'][0] + for field in ('id', 'emails', 'displayName', 'nickName', + 'firstName', 'lastName', 'orgId', 'created', 'status', 'type'): + self.assertIn(field, person) + + def test_filters_applied_in_metadata(self): + mock_api, r = self._call(email="u@e.com") + self.assertIn('filters_applied', r['metadata']) + self.assertEqual(r['metadata']['filters_applied']['email'], 'u@e.com') + + def test_unauthorized_error(self): + mock_api = MagicMock() + mock_api.people.list.side_effect = Exception("unauthorized") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_not_found_error(self): + mock_api = MagicMock() + mock_api.people.list.side_effect = Exception("not found") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_rate_limit_error_is_temporary(self): + mock_api = MagicMock() + mock_api.people.list.side_effect = Exception("rate limit exceeded") + with patch.object(_people_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_people() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E503') + self.assertTrue(r.get('temporary')) + + def test_has_timestamp(self): + mock_api, r = self._call() + self.assertIn('timestamp', r) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_rooms.py b/tests/test_rooms.py new file mode 100644 index 0000000..fda5564 --- /dev/null +++ b/tests/test_rooms.py @@ -0,0 +1,533 @@ +""" +Unit tests for Webex Room/Space management tools. + +webexpythonsdk is mocked at the sys.modules level so tests can run without +real Webex credentials or the SDK being installed. +""" +import os +import sys +import types +import unittest +from unittest.mock import MagicMock, patch + +# ── Provide fake env var and SDK before any tools module is imported ────────── +os.environ.setdefault("WEBEX_ACCESS_TOKEN", "test-token") + +_sdk_mod = types.ModuleType("webexpythonsdk") +_sdk_mod.WebexAPI = MagicMock(return_value=MagicMock()) +sys.modules.setdefault("webexpythonsdk", _sdk_mod) + +# ── Also stub python-dotenv if missing ─────────────────────────────────────── +if "dotenv" not in sys.modules: + _dotenv_mod = types.ModuleType("dotenv") + _dotenv_mod.load_dotenv = lambda *a, **kw: None + sys.modules["dotenv"] = _dotenv_mod + +import webex_bot_mcp.tools.rooms as _rooms_mod # noqa: E402 +from webex_bot_mcp.tools.rooms import ( # noqa: E402 + list_webex_rooms, + create_webex_room, + update_webex_room, + get_webex_room, + delete_webex_room, + list_webex_spaces, + create_webex_space, + update_webex_space, + get_webex_space, + delete_webex_space, +) + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _fake_room(**overrides): + r = MagicMock() + r.id = "ROOMID" + r.title = "Test Room" + r.type = "group" + r.isLocked = False + r.lastActivity = "2025-01-01T00:00:00Z" + r.created = "2025-01-01T00:00:00Z" + r.creatorId = "CREATORID" + # Optional attributes must be None so getattr(..., None) returns None + r.teamId = None + r.sipAddress = None + r.description = None + r.isPublic = None + r.isAnnouncementOnly = None + r.ownerId = None + r.classificationId = None + for k, v in overrides.items(): + setattr(r, k, v) + return r + + +# ── list_webex_rooms ────────────────────────────────────────────────────────── + +class TestListWebexRooms(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + return mock_api, list_webex_rooms(**kwargs) + + def test_returns_rooms_list(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room(), _fake_room(id="ROOMID2", title="Room 2")] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertTrue(r['success']) + self.assertEqual(len(r['data']['rooms']), 2) + + def test_empty_list(self): + mock_api, r = self._call() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertTrue(r['success']) + self.assertEqual(r['data']['rooms'], []) + + def test_metadata_count_matches(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room(), _fake_room()] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertEqual(r['metadata']['count'], 2) + + def test_team_id_filter_passed(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_rooms(team_id="TEAMID") + mock_api.rooms.list.assert_called_once_with(teamId="TEAMID") + + def test_room_type_filter_passed(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_rooms(room_type="direct") + mock_api.rooms.list.assert_called_once_with(type="direct") + + def test_sort_by_filter_passed(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_rooms(sort_by="lastactivity") + mock_api.rooms.list.assert_called_once_with(sortBy="lastactivity") + + def test_max_results_passed(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_rooms(max_results=50) + mock_api.rooms.list.assert_called_once_with(max=50) + + def test_no_filters_calls_list_with_no_args(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_rooms() + mock_api.rooms.list.assert_called_once_with() + + def test_room_dict_has_required_fields(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room()] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + room = r['data']['rooms'][0] + for field in ('id', 'title', 'type', 'isLocked', 'lastActivity', 'created', 'creatorId'): + self.assertIn(field, room) + + def test_optional_team_id_included_when_set(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room(teamId="TEAMID123")] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertEqual(r['data']['rooms'][0]['teamId'], "TEAMID123") + + def test_optional_fields_absent_when_none(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room()] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + room = r['data']['rooms'][0] + self.assertNotIn('teamId', room) + self.assertNotIn('description', room) + + def test_unauthorized_error(self): + mock_api = MagicMock() + mock_api.rooms.list.side_effect = Exception("unauthorized") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_not_found_error(self): + mock_api = MagicMock() + mock_api.rooms.list.side_effect = Exception("not found") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_rate_limit_error_is_temporary(self): + mock_api = MagicMock() + mock_api.rooms.list.side_effect = Exception("rate limit exceeded") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms() + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E503') + self.assertTrue(r.get('temporary')) + + def test_filters_applied_in_metadata(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_rooms(room_type="group") + self.assertIn('filters_applied', r['metadata']) + self.assertEqual(r['metadata']['filters_applied']['type'], 'group') + + +# ── create_webex_room ───────────────────────────────────────────────────────── + +class TestCreateWebexRoom(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.rooms.create.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + return mock_api, create_webex_room(**kwargs) + + def test_success_returns_room(self): + _, r = self._call(title="New Room") + self.assertTrue(r['success']) + self.assertIn('room', r['data']) + + def test_title_passed_to_sdk(self): + mock_api, _ = self._call(title="My Room") + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertEqual(call_kwargs['title'], "My Room") + + def test_is_locked_and_is_moderated_conflict_returns_error(self): + _, r = self._call(title="R", is_locked=True, is_moderated=True) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_is_moderated_used_when_is_locked_absent(self): + mock_api, r = self._call(title="R", is_moderated=True) + self.assertTrue(r['success']) + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertTrue(call_kwargs['isLocked']) + + def test_is_locked_used_when_is_moderated_absent(self): + mock_api, r = self._call(title="R", is_locked=False) + self.assertTrue(r['success']) + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertFalse(call_kwargs['isLocked']) + + def test_team_id_passed(self): + mock_api, _ = self._call(title="R", team_id="TID") + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertEqual(call_kwargs['teamId'], "TID") + + def test_description_passed(self): + mock_api, _ = self._call(title="R", description="A room") + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertEqual(call_kwargs['description'], "A room") + + def test_is_public_passed(self): + mock_api, _ = self._call(title="R", is_public=True) + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertTrue(call_kwargs['isPublic']) + + def test_is_announcement_only_passed(self): + mock_api, _ = self._call(title="R", is_announcement_only=True) + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertTrue(call_kwargs['isAnnouncementOnly']) + + def test_classification_id_passed(self): + mock_api, _ = self._call(title="R", classification_id="CID") + call_kwargs = mock_api.rooms.create.call_args[1] + self.assertEqual(call_kwargs['classificationId'], "CID") + + def test_operation_in_metadata(self): + _, r = self._call(title="R") + self.assertEqual(r['metadata']['operation'], 'create_room') + + def test_sdk_exception_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.create.side_effect = Exception("forbidden") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = create_webex_room(title="R") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E403') + + def test_has_timestamp(self): + _, r = self._call(title="R") + self.assertIn('timestamp', r) + + +# ── update_webex_room ───────────────────────────────────────────────────────── + +class TestUpdateWebexRoom(unittest.TestCase): + def _call(self, **kwargs): + mock_api = MagicMock() + mock_api.rooms.update.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + return mock_api, update_webex_room(**kwargs) + + def test_success_returns_room(self): + _, r = self._call(room_id="RID", title="New Title") + self.assertTrue(r['success']) + self.assertIn('room', r['data']) + + def test_no_fields_returns_error(self): + _, r = self._call(room_id="RID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_is_locked_and_is_moderated_conflict_returns_error(self): + _, r = self._call(room_id="RID", is_locked=True, is_moderated=False) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_title_passed_to_sdk(self): + mock_api, _ = self._call(room_id="RID", title="Updated") + call_kwargs = mock_api.rooms.update.call_args[1] + self.assertEqual(call_kwargs['title'], "Updated") + + def test_room_id_passed_to_sdk(self): + mock_api, _ = self._call(room_id="RID123", title="T") + call_kwargs = mock_api.rooms.update.call_args[1] + self.assertEqual(call_kwargs['roomId'], "RID123") + + def test_description_passed(self): + mock_api, _ = self._call(room_id="RID", description="New desc") + call_kwargs = mock_api.rooms.update.call_args[1] + self.assertEqual(call_kwargs['description'], "New desc") + + def test_is_public_passed(self): + mock_api, _ = self._call(room_id="RID", is_public=False) + call_kwargs = mock_api.rooms.update.call_args[1] + self.assertFalse(call_kwargs['isPublic']) + + def test_operation_in_metadata(self): + _, r = self._call(room_id="RID", title="T") + self.assertEqual(r['metadata']['operation'], 'update_room') + + def test_not_found_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.update.side_effect = Exception("not found") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = update_webex_room(room_id="RID", title="T") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_is_moderated_maps_to_is_locked(self): + mock_api, _ = self._call(room_id="RID", is_moderated=True) + call_kwargs = mock_api.rooms.update.call_args[1] + self.assertTrue(call_kwargs['isLocked']) + + +# ── get_webex_room ──────────────────────────────────────────────────────────── + +class TestGetWebexRoom(unittest.TestCase): + def test_success_returns_room(self): + mock_api = MagicMock() + mock_api.rooms.get.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_room(room_id="ROOMID") + self.assertTrue(r['success']) + self.assertIn('room', r['data']) + + def test_room_id_passed_to_sdk(self): + mock_api = MagicMock() + mock_api.rooms.get.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + get_webex_room(room_id="RID123") + mock_api.rooms.get.assert_called_once_with(roomId="RID123") + + def test_metadata_has_room_id(self): + mock_api = MagicMock() + mock_api.rooms.get.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_room(room_id="ROOMID") + self.assertEqual(r['metadata']['room_id'], "ROOMID") + + def test_not_found_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.get.side_effect = Exception("404 not found") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_room(room_id="MISSING") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_unauthorized_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.get.side_effect = Exception("unauthorized") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_room(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + +# ── delete_webex_room ───────────────────────────────────────────────────────── + +class TestDeleteWebexRoom(unittest.TestCase): + def test_success_returns_deleted_true(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="ROOMID") + self.assertTrue(r['success']) + self.assertTrue(r['data']['deleted']) + + def test_room_id_in_response(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="ROOMID") + self.assertEqual(r['data']['room_id'], "ROOMID") + + def test_room_id_passed_to_sdk(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + delete_webex_room(room_id="DELID") + mock_api.rooms.delete.assert_called_once_with(roomId="DELID") + + def test_empty_room_id_returns_error(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + def test_not_found_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.delete.side_effect = Exception("not found") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E404') + + def test_unauthorized_returns_error(self): + mock_api = MagicMock() + mock_api.rooms.delete.side_effect = Exception("unauthorized") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="ROOMID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E401') + + def test_operation_in_metadata(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_room(room_id="ROOMID") + self.assertEqual(r['metadata']['operation'], 'delete_room') + + +# ── list_webex_spaces (alias) ───────────────────────────────────────────────── + +class TestListWebexSpaces(unittest.TestCase): + def test_returns_spaces_key_not_rooms(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [_fake_room()] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_spaces() + self.assertTrue(r['success']) + self.assertIn('spaces', r['data']) + self.assertNotIn('rooms', r['data']) + + def test_delegates_space_type_as_room_type(self): + mock_api = MagicMock() + mock_api.rooms.list.return_value = [] + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + list_webex_spaces(space_type="group") + mock_api.rooms.list.assert_called_once_with(type="group") + + def test_error_passthrough(self): + mock_api = MagicMock() + mock_api.rooms.list.side_effect = Exception("unauthorized") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = list_webex_spaces() + self.assertFalse(r['success']) + + +# ── create_webex_space (alias) ──────────────────────────────────────────────── + +class TestCreateWebexSpace(unittest.TestCase): + def test_returns_space_key_not_room(self): + mock_api = MagicMock() + mock_api.rooms.create.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = create_webex_space(title="My Space") + self.assertTrue(r['success']) + self.assertIn('space', r['data']) + self.assertNotIn('room', r['data']) + + def test_error_passthrough(self): + mock_api = MagicMock() + mock_api.rooms.create.side_effect = Exception("forbidden") + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = create_webex_space(title="S") + self.assertFalse(r['success']) + + def test_conflict_validation_still_works(self): + r = create_webex_space(title="S", is_locked=True, is_moderated=True) + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + +# ── update_webex_space (alias) ──────────────────────────────────────────────── + +class TestUpdateWebexSpace(unittest.TestCase): + def test_returns_space_key_not_room(self): + mock_api = MagicMock() + mock_api.rooms.update.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = update_webex_space(space_id="SID", title="New Title") + self.assertTrue(r['success']) + self.assertIn('space', r['data']) + self.assertNotIn('room', r['data']) + + def test_no_fields_error_propagated(self): + r = update_webex_space(space_id="SID") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + +# ── get_webex_space (alias) ─────────────────────────────────────────────────── + +class TestGetWebexSpace(unittest.TestCase): + def test_returns_space_key_not_room(self): + mock_api = MagicMock() + mock_api.rooms.get.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = get_webex_space(space_id="SID") + self.assertTrue(r['success']) + self.assertIn('space', r['data']) + self.assertNotIn('room', r['data']) + + def test_space_id_passed_to_sdk(self): + mock_api = MagicMock() + mock_api.rooms.get.return_value = _fake_room() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + get_webex_space(space_id="SID123") + mock_api.rooms.get.assert_called_once_with(roomId="SID123") + + +# ── delete_webex_space (alias) ──────────────────────────────────────────────── + +class TestDeleteWebexSpace(unittest.TestCase): + def test_delegates_to_delete_room(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_space(space_id="SID") + self.assertTrue(r['success']) + mock_api.rooms.delete.assert_called_once_with(roomId="SID") + + def test_empty_space_id_returns_error(self): + mock_api = MagicMock() + with patch.object(_rooms_mod, 'get_webex_api', return_value=mock_api): + r = delete_webex_space(space_id="") + self.assertFalse(r['success']) + self.assertEqual(r['error_code'], 'E001') + + +if __name__ == '__main__': + unittest.main()