"""check_availability/book_meeting/cancel_meeting resolve which provider (Google Calendar or Teams) the tenant has active, then call that provider's module. Every provider function is mocked -- these tests only exercise the tool-level routing, feature/connection gating, and the immediate-booking race-condition guard. """ import asyncio from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch from zoneinfo import ZoneInfo import pytest from app.services.chat.tools import ( execute_book_meeting, execute_cancel_meeting, execute_check_availability, ) def _ctx(): return {"tenant_id": "org_test", "thread_id": "thread_1", "user_id": None} def _settings(feature_on=True, provider="google_calendar"): return {"features": {"book_meeting": feature_on}, "providers": {"book_meeting": provider}} def _run(coro): return asyncio.run(coro) def _future_request_args(hours_ahead=3): when = datetime.now(timezone.utc) + timedelta(hours=hours_ahead) hour_12 = when.hour % 12 or 12 meridiem = "AM" if when.hour < 12 else "PM" return {"date": when.date().isoformat(), "hour_12": hour_12, "minute": when.minute, "meridiem": meridiem}, when.replace(second=0, microsecond=0) @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_is_refused_when_the_feature_is_off(mock_settings): mock_settings.return_value = _settings(feature_on=False) args, _ = _future_request_args() result = _run(execute_check_availability(args, _ctx())) assert result["status"] == "error" @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_is_refused_with_no_provider_selected(mock_settings): mock_settings.return_value = _settings(provider=None) args, _ = _future_request_args() result = _run(execute_check_availability(args, _ctx())) assert result["status"] == "error" @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_confirms_the_exact_requested_time_when_free( mock_settings, mock_check, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} args, when = _future_request_args() mock_check.return_value = [{"start": when, "end": when + timedelta(minutes=30)}] result = _run(execute_check_availability(args, _ctx())) mock_check.assert_called_once() assert result["status"] == "ok" assert result["available"] is True assert result["requested"]["start"] == when.isoformat() @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_honors_an_explicit_visitor_timezone( mock_settings, mock_check, mock_conn): # The business's calendar is Sydney, but the visitor asked in IST -- the # exact-time check must happen against 9 AM India time, not 9 AM Sydney. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "Australia/Sydney"} when = datetime.now(timezone.utc).astimezone(ZoneInfo("Asia/Kolkata")) + timedelta(hours=3) when = when.replace(second=0, microsecond=0) hour_12 = when.hour % 12 or 12 meridiem = "AM" if when.hour < 12 else "PM" mock_check.return_value = [{"start": when, "end": when + timedelta(minutes=30)}] result = _run(execute_check_availability( {"date": when.date().isoformat(), "hour_12": hour_12, "minute": when.minute, "meridiem": meridiem, "visitor_timezone": "Asia/Kolkata"}, _ctx())) assert result["status"] == "ok" assert result["available"] is True # The result's own timezone must reflect the zone actually used (the # visitor's IST), not the business's Sydney zone -- otherwise the model # relays a time labelled with the wrong zone. assert result["timezone"] == "Asia/Kolkata" called_start = mock_check.call_args[0][1] assert called_start == when assert called_start.utcoffset() == timedelta(hours=5, minutes=30) @patch("app.services.chat.tools.get_teams_connection") @patch("app.services.chat.tools.teams_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_routes_to_teams(mock_settings, mock_check, mock_conn): mock_settings.return_value = _settings(provider="teams") mock_conn.return_value = {"access_token": "tok", "user_email": "a@b.com", "timezone": "UTC"} mock_check.return_value = [] args, _ = _future_request_args() result = _run(execute_check_availability(args, _ctx())) mock_check.assert_called() assert result["status"] == "ok" assert result["available"] is False @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_rejects_malformed_request_fields(mock_settings, mock_check, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} result = _run(execute_check_availability( {"date": "not-a-date", "hour_12": 9, "minute": 0, "meridiem": "AM"}, _ctx())) assert result["status"] == "error" mock_check.assert_not_called() @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_checks_a_narrow_window_around_the_exact_time( mock_settings, mock_check, mock_conn): # It must check the specific requested slot, not dump a whole day/week. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} args, when = _future_request_args() mock_check.return_value = [] # not free -> falls through to the nearby search _run(execute_check_availability(args, _ctx())) first_call_start = mock_check.call_args_list[0][0][1] first_call_end = mock_check.call_args_list[0][0][2] assert first_call_start == when assert (first_call_end - first_call_start) <= timedelta(minutes=31) @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_offers_real_nearby_slots_when_the_exact_time_is_taken( mock_settings, mock_check, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} args, when = _future_request_args() nearby = when + timedelta(hours=1, minutes=30) far = when + timedelta(hours=3, minutes=30) # outside the nearby search window def fake_check(connection, start, end): # The narrow exact-time check (first call) never matches -- the slot # is taken. The wider nearby search (second call) returns real # alternatives, including one outside the search window that must # not appear in the result. if (end - start) <= timedelta(minutes=31): return [] return [{"start": nearby, "end": nearby + timedelta(minutes=30)}, {"start": far, "end": far + timedelta(minutes=30)}] mock_check.side_effect = fake_check result = _run(execute_check_availability(args, _ctx())) assert result["status"] == "ok" assert result["available"] is False starts = [s["start"] for s in result["nearby_slots"]] assert nearby.isoformat() in starts @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.get_tool_settings") def test_check_availability_treats_a_past_request_as_not_free(mock_settings, mock_check, mock_conn): # A past requested time skips the exact-time check entirely (there is no # point calling the provider for a slot that already elapsed) and falls # straight through to the nearby search. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} past = datetime.now(timezone.utc) - timedelta(hours=2) hour_12 = past.hour % 12 or 12 meridiem = "AM" if past.hour < 12 else "PM" mock_check.return_value = [] result = _run(execute_check_availability( {"date": past.date().isoformat(), "hour_12": hour_12, "minute": past.minute, "meridiem": meridiem}, _ctx())) assert result["status"] == "ok" assert result["available"] is False mock_check.assert_called_once() # only the nearby search, no exact-time check @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.gcal_create_event") @patch("app.services.chat.tools.get_tool_settings") def test_book_meeting_books_a_still_free_slot(mock_settings, mock_create, mock_check, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} start = datetime(2026, 9, 1, 9, tzinfo=timezone.utc) mock_check.return_value = [{"start": start, "end": datetime(2026, 9, 1, 9, 30, tzinfo=timezone.utc)}] mock_create.return_value = {"event_id": "evt123", "meeting_link": "https://meet.google.com/x"} result = _run(execute_book_meeting( {"date": "2026-09-01", "hour_12": 9, "minute": 0, "meridiem": "AM", "visitor_email": "jo@example.com", "visitor_name": "Jo"}, _ctx())) mock_create.assert_called_once() assert result["status"] == "ok" assert result["meeting_link"] == "https://meet.google.com/x" @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.gcal_create_event") @patch("app.services.chat.tools.get_tool_settings") def test_book_meeting_honors_an_explicit_visitor_timezone( mock_settings, mock_create, mock_check, mock_conn): # The business's own calendar is in Sydney, but the visitor explicitly # asked for "9 AM IST" -- the booked instant must be 9 AM India time, # not 9 AM Sydney time (a ~4.5 hour difference). mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "Australia/Sydney"} expected_start = datetime(2026, 9, 1, 9, tzinfo=ZoneInfo("Asia/Kolkata")) mock_check.return_value = [{"start": expected_start, "end": expected_start + timedelta(minutes=30)}] mock_create.return_value = {"event_id": "evt123", "meeting_link": "https://meet.google.com/x"} result = _run(execute_book_meeting( {"date": "2026-09-01", "hour_12": 9, "minute": 0, "meridiem": "AM", "visitor_timezone": "Asia/Kolkata", "visitor_email": "jo@example.com", "visitor_name": "Jo"}, _ctx())) assert result["status"] == "ok" # confirmed_start/timezone must reflect IST, not the business's Sydney # zone -- otherwise the confirmation message states the wrong zone even # though the actual booked instant is correct. assert result["timezone"] == "Asia/Kolkata" assert result["confirmed_start"] == expected_start.isoformat() called_start = mock_check.call_args[0][1] assert called_start == expected_start assert called_start.utcoffset() == timedelta(hours=5, minutes=30) @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.gcal_create_event") @patch("app.services.chat.tools.get_tool_settings") def test_book_meeting_ignores_an_unrecognized_visitor_timezone( mock_settings, mock_create, mock_check, mock_conn): # A bad/unparseable visitor_timezone must not fail the call -- it should # just fall back to the business's own timezone. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} start = datetime(2026, 9, 1, 9, tzinfo=timezone.utc) mock_check.return_value = [{"start": start, "end": start + timedelta(minutes=30)}] mock_create.return_value = {"event_id": "evt123", "meeting_link": "https://meet.google.com/x"} result = _run(execute_book_meeting( {"date": "2026-09-01", "hour_12": 9, "minute": 0, "meridiem": "AM", "visitor_timezone": "Not/AZone", "visitor_email": "jo@example.com", "visitor_name": "Jo"}, _ctx())) assert result["status"] == "ok" called_start = mock_check.call_args[0][1] assert called_start == start @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.gcal_create_event") @patch("app.services.chat.tools.get_tool_settings") def test_book_meeting_refuses_a_slot_that_is_no_longer_free( mock_settings, mock_create, mock_check, mock_conn): # The re-check right before booking finds no matching open slot -- e.g. # another visitor took it between the suggestion and this pick. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} mock_check.return_value = [] # nothing free any more result = _run(execute_book_meeting( {"date": "2026-09-01", "hour_12": 9, "minute": 0, "meridiem": "AM", "visitor_email": "jo@example.com", "visitor_name": "Jo"}, _ctx())) mock_create.assert_not_called() assert result["status"] == "error" @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_check_availability") @patch("app.services.chat.tools.gcal_create_event") @patch("app.services.chat.tools.get_tool_settings") def test_book_meeting_rejects_a_start_time_already_in_the_past( mock_settings, mock_create, mock_check, mock_conn): # Correctness backstop independent of check_availability's own filtering # -- the model could send any date/hour/minute regardless of what # check_availability actually returned. mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} past_start = datetime.now(timezone.utc) - timedelta(hours=1) hour_12 = past_start.hour % 12 or 12 meridiem = "AM" if past_start.hour < 12 else "PM" result = _run(execute_book_meeting( {"date": past_start.date().isoformat(), "hour_12": hour_12, "minute": past_start.minute, "meridiem": meridiem, "visitor_email": "jo@example.com", "visitor_name": "Jo"}, _ctx())) assert result == {"status": "error", "detail": "That time has already passed."} mock_check.assert_not_called() mock_create.assert_not_called() @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_cancel_event") @patch("app.services.chat.tools.get_tool_settings") def test_cancel_meeting_routes_to_google_calendar(mock_settings, mock_cancel, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} mock_cancel.return_value = True result = _run(execute_cancel_meeting({"visitor_email": "jo@example.com"}, _ctx())) mock_cancel.assert_called_once() assert result["status"] == "ok" @patch("app.services.chat.tools.get_google_calendar_connection") @patch("app.services.chat.tools.gcal_cancel_event") @patch("app.services.chat.tools.get_tool_settings") def test_cancel_meeting_reports_an_ambiguous_match_as_an_error( mock_settings, mock_cancel, mock_conn): mock_settings.return_value = _settings(provider="google_calendar") mock_conn.return_value = {"access_token": "tok", "calendar_email": "a@b.com", "timezone": "UTC"} mock_cancel.return_value = False # zero or multiple matches result = _run(execute_cancel_meeting({"visitor_email": "jo@example.com"}, _ctx())) assert result["status"] == "error" def test_book_meeting_tool_names_are_registered(): from app.services.chat.tools import TOOL_EXECUTORS, TOOL_SCHEMAS names = {t["function"]["name"] for t in TOOL_SCHEMAS} assert {"check_availability", "book_meeting", "cancel_meeting"} <= names assert {"check_availability", "book_meeting", "cancel_meeting"} <= set(TOOL_EXECUTORS)