""" Tests for the Integrations module """ from __future__ import annotations import pytest from stack0.integrations import ( AsyncIntegrations, Communication, CRM, Integrations, Productivity, Storage, ) from stack0.integrations.types import ( CompleteOAuthRequest, CreateCompanyInput, CreateContactInput, CreateDealInput, CreateDocumentInput, CreateFolderInput, CreateTableRowInput, InitiateOAuthRequest, ListConnectionsRequest, ListLogsRequest, ListOptions, PassthroughRequest, ReconnectConnectionRequest, SendMessageInput, UpdateConnectionRequest, UpdateContactInput, UploadFileInput, ) from tests.conftest import MockAsyncHTTPClient, MockHTTPClient class TestCRM: """Tests for CRM operations""" def test_list_contacts(self) -> None: """Test listing CRM contacts""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/contacts?connectionId=conn_123": { "items": [ { "id": "contact_1", "firstName": "John", "lastName": "Doe", "email": "[email protected]", }, ], "cursor": None, } }) crm = CRM(mock_http) # type: ignore result = crm.list_contacts("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["email"] == "[email protected]" def test_list_contacts_with_options(self) -> None: """Test listing contacts with pagination options""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/contacts?connectionId=conn_123&limit=10&sortBy=createdAt&sortOrder=desc": { "items": [], "cursor": None, } }) crm = CRM(mock_http) # type: ignore result = crm.list_contacts("conn_123", ListOptions( limit=10, sort_by="createdAt", sort_order="desc", )) assert "items" in result def test_get_contact(self) -> None: """Test getting a contact by ID""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/contacts/contact_123?connectionId=conn_123": { "id": "contact_123", "firstName": "John", "lastName": "Doe", "email": "[email protected]", "phone": "+1234567890", } }) crm = CRM(mock_http) # type: ignore contact = crm.get_contact("conn_123", "contact_123") assert contact.id == "contact_123" assert contact.first_name == "John" assert contact.email == "[email protected]" def test_create_contact(self) -> None: """Test creating a contact""" mock_http = MockHTTPClient(responses={ "POST:/integrations/crm/contacts": { "id": "contact_new", "firstName": "Jane", "lastName": "Smith", "email": "[email protected]", } }) crm = CRM(mock_http) # type: ignore contact = crm.create_contact("conn_123", CreateContactInput( first_name="Jane", last_name="Smith", email="[email protected]", )) assert contact.id == "contact_new" assert contact.first_name == "Jane" def test_update_contact(self) -> None: """Test updating a contact""" mock_http = MockHTTPClient(responses={ "PATCH:/integrations/crm/contacts/contact_123": { "id": "contact_123", "firstName": "John", "lastName": "Updated", "email": "[email protected]", } }) crm = CRM(mock_http) # type: ignore contact = crm.update_contact("conn_123", "contact_123", UpdateContactInput( last_name="Updated", email="[email protected]", )) assert contact.last_name == "Updated" def test_delete_contact(self) -> None: """Test deleting a contact""" mock_http = MockHTTPClient(responses={ "DELETE:/integrations/crm/contacts/contact_123?connectionId=conn_123": { "success": True, } }) crm = CRM(mock_http) # type: ignore result = crm.delete_contact("conn_123", "contact_123") assert result["success"] is True def test_list_companies(self) -> None: """Test listing companies""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/companies?connectionId=conn_123": { "items": [ { "id": "company_1", "name": "Acme Corp", "domain": "acme.com", }, ], "cursor": None, } }) crm = CRM(mock_http) # type: ignore result = crm.list_companies("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["name"] == "Acme Corp" def test_get_company(self) -> None: """Test getting a company""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/companies/company_123?connectionId=conn_123": { "id": "company_123", "name": "Acme Corp", "domain": "acme.com", "industry": "Technology", } }) crm = CRM(mock_http) # type: ignore company = crm.get_company("conn_123", "company_123") assert company.id == "company_123" assert company.name == "Acme Corp" def test_create_company(self) -> None: """Test creating a company""" mock_http = MockHTTPClient(responses={ "POST:/integrations/crm/companies": { "id": "company_new", "name": "New Corp", "domain": "newcorp.com", } }) crm = CRM(mock_http) # type: ignore company = crm.create_company("conn_123", CreateCompanyInput( name="New Corp", domain="newcorp.com", )) assert company.id == "company_new" def test_list_deals(self) -> None: """Test listing deals""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/deals?connectionId=conn_123": { "items": [ { "id": "deal_1", "name": "Big Deal", "amount": 10000.0, "stage": "negotiation", }, ], "cursor": None, } }) crm = CRM(mock_http) # type: ignore result = crm.list_deals("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["amount"] == 10000.0 def test_get_deal(self) -> None: """Test getting a deal""" mock_http = MockHTTPClient(responses={ "GET:/integrations/crm/deals/deal_123?connectionId=conn_123": { "id": "deal_123", "name": "Big Deal", "amount": 10000.0, "stage": "negotiation", } }) crm = CRM(mock_http) # type: ignore deal = crm.get_deal("conn_123", "deal_123") assert deal.id == "deal_123" assert deal.amount == 10000.0 def test_create_deal(self) -> None: """Test creating a deal""" mock_http = MockHTTPClient(responses={ "POST:/integrations/crm/deals": { "id": "deal_new", "name": "New Deal", "amount": 5000.0, "stage": "qualification", } }) crm = CRM(mock_http) # type: ignore deal = crm.create_deal("conn_123", CreateDealInput( name="New Deal", amount=5000.0, )) assert deal.id == "deal_new" class TestStorage: """Tests for Storage operations""" def test_list_files(self) -> None: """Test listing files""" mock_http = MockHTTPClient(responses={ "GET:/integrations/storage/files?connectionId=conn_123": { "items": [ { "id": "file_1", "name": "document.pdf", "mimeType": "application/pdf", "size": 1024, }, ], "cursor": None, } }) storage = Storage(mock_http) # type: ignore result = storage.list_files("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["name"] == "document.pdf" def test_list_files_in_folder(self) -> None: """Test listing files in a specific folder""" mock_http = MockHTTPClient(responses={ "GET:/integrations/storage/files?connectionId=conn_123&folderId=folder_123": { "items": [], "cursor": None, } }) storage = Storage(mock_http) # type: ignore result = storage.list_files("conn_123", folder_id="folder_123") assert "items" in result def test_get_file(self) -> None: """Test getting a file""" mock_http = MockHTTPClient(responses={ "GET:/integrations/storage/files/file_123?connectionId=conn_123": { "id": "file_123", "name": "document.pdf", "mimeType": "application/pdf", "size": 1024, "createdAt": "2024-01-15T10:00:00Z", } }) storage = Storage(mock_http) # type: ignore file = storage.get_file("conn_123", "file_123") assert file.id == "file_123" assert file.name == "document.pdf" def test_upload_file(self) -> None: """Test uploading a file""" mock_http = MockHTTPClient(responses={ "POST:/integrations/storage/files": { "id": "file_new", "name": "uploaded.txt", "mimeType": "text/plain", "size": 100, } }) storage = Storage(mock_http) # type: ignore file = storage.upload_file("conn_123", UploadFileInput( name="uploaded.txt", mime_type="text/plain", data=b"Hello, World!", )) assert file.id == "file_new" assert file.name == "uploaded.txt" def test_delete_file(self) -> None: """Test deleting a file""" mock_http = MockHTTPClient(responses={ "DELETE:/integrations/storage/files/file_123?connectionId=conn_123": { "success": True, } }) storage = Storage(mock_http) # type: ignore result = storage.delete_file("conn_123", "file_123") assert result["success"] is True def test_list_folders(self) -> None: """Test listing folders""" mock_http = MockHTTPClient(responses={ "GET:/integrations/storage/folders?connectionId=conn_123": { "items": [ { "id": "folder_1", "name": "Documents", }, ], "cursor": None, } }) storage = Storage(mock_http) # type: ignore result = storage.list_folders("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["name"] == "Documents" def test_get_folder(self) -> None: """Test getting a folder""" mock_http = MockHTTPClient(responses={ "GET:/integrations/storage/folders/folder_123?connectionId=conn_123": { "id": "folder_123", "name": "Documents", "parentId": None, } }) storage = Storage(mock_http) # type: ignore folder = storage.get_folder("conn_123", "folder_123") assert folder.id == "folder_123" assert folder.name == "Documents" def test_create_folder(self) -> None: """Test creating a folder""" mock_http = MockHTTPClient(responses={ "POST:/integrations/storage/folders": { "id": "folder_new", "name": "New Folder", } }) storage = Storage(mock_http) # type: ignore folder = storage.create_folder("conn_123", CreateFolderInput( name="New Folder", )) assert folder.id == "folder_new" class TestCommunication: """Tests for Communication operations""" def test_list_channels(self) -> None: """Test listing channels""" mock_http = MockHTTPClient(responses={ "GET:/integrations/communication/channels?connectionId=conn_123": { "items": [ { "id": "channel_1", "name": "general", "type": "public", }, ], "cursor": None, } }) communication = Communication(mock_http) # type: ignore result = communication.list_channels("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["name"] == "general" def test_get_channel(self) -> None: """Test getting a channel""" mock_http = MockHTTPClient(responses={ "GET:/integrations/communication/channels/channel_123?connectionId=conn_123": { "id": "channel_123", "name": "general", "type": "public", "memberCount": 50, } }) communication = Communication(mock_http) # type: ignore channel = communication.get_channel("conn_123", "channel_123") assert channel.id == "channel_123" assert channel.name == "general" def test_list_messages(self) -> None: """Test listing messages""" mock_http = MockHTTPClient(responses={ "GET:/integrations/communication/messages?connectionId=conn_123&channelId=channel_123": { "items": [ { "id": "msg_1", "content": "Hello!", "channelId": "channel_123", }, ], "cursor": None, } }) communication = Communication(mock_http) # type: ignore result = communication.list_messages("conn_123", "channel_123") assert len(result["items"]) == 1 assert result["items"][0]["content"] == "Hello!" def test_send_message(self) -> None: """Test sending a message""" mock_http = MockHTTPClient(responses={ "POST:/integrations/communication/messages": { "id": "msg_new", "content": "Hello, World!", "channelId": "channel_123", } }) communication = Communication(mock_http) # type: ignore message = communication.send_message("conn_123", SendMessageInput( channel_id="channel_123", content="Hello, World!", )) assert message.id == "msg_new" assert message.content == "Hello, World!" def test_list_users(self) -> None: """Test listing users""" mock_http = MockHTTPClient(responses={ "GET:/integrations/communication/users?connectionId=conn_123": { "items": [ { "id": "user_1", "name": "John Doe", "email": "[email protected]", }, ], "cursor": None, } }) communication = Communication(mock_http) # type: ignore result = communication.list_users("conn_123") assert len(result["items"]) == 1 class TestProductivity: """Tests for Productivity operations""" def test_list_documents(self) -> None: """Test listing documents""" mock_http = MockHTTPClient(responses={ "GET:/integrations/productivity/documents?connectionId=conn_123": { "items": [ { "id": "doc_1", "title": "Project Plan", "type": "document", }, ], "cursor": None, } }) productivity = Productivity(mock_http) # type: ignore result = productivity.list_documents("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["title"] == "Project Plan" def test_get_document(self) -> None: """Test getting a document""" mock_http = MockHTTPClient(responses={ "GET:/integrations/productivity/documents/doc_123?connectionId=conn_123": { "id": "doc_123", "title": "Project Plan", "content": "# Project Plan\n\nThis is the plan.", "type": "document", } }) productivity = Productivity(mock_http) # type: ignore document = productivity.get_document("conn_123", "doc_123") assert document.id == "doc_123" assert document.title == "Project Plan" def test_create_document(self) -> None: """Test creating a document""" mock_http = MockHTTPClient(responses={ "POST:/integrations/productivity/documents": { "id": "doc_new", "title": "New Document", "content": "# New Document", "type": "document", } }) productivity = Productivity(mock_http) # type: ignore document = productivity.create_document("conn_123", CreateDocumentInput( title="New Document", content="# New Document", )) assert document.id == "doc_new" def test_list_tables(self) -> None: """Test listing tables""" mock_http = MockHTTPClient(responses={ "GET:/integrations/productivity/tables?connectionId=conn_123": { "items": [ { "id": "table_1", "name": "Tasks", }, ], "cursor": None, } }) productivity = Productivity(mock_http) # type: ignore result = productivity.list_tables("conn_123") assert len(result["items"]) == 1 assert result["items"][0]["name"] == "Tasks" def test_get_table(self) -> None: """Test getting a table""" mock_http = MockHTTPClient(responses={ "GET:/integrations/productivity/tables/table_123?connectionId=conn_123": { "id": "table_123", "name": "Tasks", "columns": [ {"id": "col_1", "name": "Name", "type": "text"}, ], } }) productivity = Productivity(mock_http) # type: ignore table = productivity.get_table("conn_123", "table_123") assert table.id == "table_123" assert table.name == "Tasks" def test_list_table_rows(self) -> None: """Test listing table rows""" mock_http = MockHTTPClient(responses={ "GET:/integrations/productivity/tables/table_123/rows?connectionId=conn_123": { "items": [ { "id": "row_1", "values": {"Name": "Task 1"}, }, ], "cursor": None, } }) productivity = Productivity(mock_http) # type: ignore result = productivity.list_table_rows("conn_123", "table_123") assert len(result["items"]) == 1 def test_create_table_row(self) -> None: """Test creating a table row""" mock_http = MockHTTPClient(responses={ "POST:/integrations/productivity/tables/table_123/rows": { "id": "row_new", "values": {"Name": "New Task"}, } }) productivity = Productivity(mock_http) # type: ignore row = productivity.create_table_row("conn_123", "table_123", CreateTableRowInput( values={"Name": "New Task"}, )) assert row.id == "row_new" class TestIntegrations: """Tests for main Integrations client""" def test_list_connectors(self) -> None: """Test listing connectors""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connectors": [ { "slug": "hubspot", "name": "HubSpot", "category": "crm", "authType": "oauth2", }, { "slug": "google-drive", "name": "Google Drive", "category": "storage", "authType": "oauth2", }, ] }) integrations = Integrations(mock_http) # type: ignore connectors = integrations.list_connectors() assert len(connectors) == 2 assert connectors[0].slug == "hubspot" def test_list_connectors_by_category(self) -> None: """Test listing connectors filtered by category""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connectors?category=crm": [ { "slug": "hubspot", "name": "HubSpot", "category": "crm", "authType": "oauth2", }, ] }) integrations = Integrations(mock_http) # type: ignore connectors = integrations.list_connectors(category="crm") assert len(connectors) == 1 assert connectors[0].category == "crm" def test_get_connector(self) -> None: """Test getting a connector""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connectors/hubspot": { "slug": "hubspot", "name": "HubSpot", "category": "crm", "authType": "oauth2", "scopes": ["contacts", "deals"], } }) integrations = Integrations(mock_http) # type: ignore connector = integrations.get_connector("hubspot") assert connector.slug == "hubspot" assert connector.auth_type == "oauth2" def test_list_connections(self) -> None: """Test listing connections""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connections": { "items": [ { "id": "conn_123", "connectorSlug": "hubspot", "status": "active", }, ], "total": 1, } }) integrations = Integrations(mock_http) # type: ignore response = integrations.list_connections() assert len(response.items) == 1 assert response.items[0].connector_slug == "hubspot" def test_list_connections_with_filters(self) -> None: """Test listing connections with filters""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connections?status=active&connectorSlug=hubspot": { "items": [], "total": 0, } }) integrations = Integrations(mock_http) # type: ignore response = integrations.list_connections(ListConnectionsRequest( status="active", connector_slug="hubspot", )) assert response.total == 0 def test_get_connection(self) -> None: """Test getting a connection""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connections/conn_123": { "id": "conn_123", "connectorSlug": "hubspot", "status": "active", "connectorName": "HubSpot", "metadata": {}, } }) integrations = Integrations(mock_http) # type: ignore connection = integrations.get_connection("conn_123") assert connection.id == "conn_123" assert connection.status == "active" def test_initiate_oauth(self) -> None: """Test initiating OAuth flow""" mock_http = MockHTTPClient(responses={ "POST:/integrations/connections/oauth/initiate": { "authorizationUrl": "https://oauth.hubspot.com/authorize?...", "state": "state_123", } }) integrations = Integrations(mock_http) # type: ignore response = integrations.initiate_oauth(InitiateOAuthRequest( connector_slug="hubspot", redirect_url="https://myapp.com/callback", )) assert response.authorization_url.startswith("https://oauth.hubspot.com") assert response.state == "state_123" def test_complete_oauth(self) -> None: """Test completing OAuth flow""" mock_http = MockHTTPClient(responses={ "POST:/integrations/connections/oauth/callback": { "connectionId": "conn_new", "status": "active", } }) integrations = Integrations(mock_http) # type: ignore response = integrations.complete_oauth(CompleteOAuthRequest( code="auth_code_123", state="state_123", )) assert response.connection_id == "conn_new" assert response.status == "active" def test_update_connection(self) -> None: """Test updating a connection""" mock_http = MockHTTPClient(responses={ "PATCH:/integrations/connections/conn_123": { "success": True, } }) integrations = Integrations(mock_http) # type: ignore response = integrations.update_connection(UpdateConnectionRequest( connection_id="conn_123", name="My HubSpot Connection", )) assert response.success is True def test_delete_connection(self) -> None: """Test deleting a connection""" mock_http = MockHTTPClient(responses={ "DELETE:/integrations/connections/conn_123": { "success": True, } }) integrations = Integrations(mock_http) # type: ignore result = integrations.delete_connection("conn_123") assert result["success"] is True def test_reconnect_connection(self) -> None: """Test reconnecting a connection""" mock_http = MockHTTPClient(responses={ "POST:/integrations/connections/conn_123/reconnect": { "authorizationUrl": "https://oauth.hubspot.com/authorize?...", } }) integrations = Integrations(mock_http) # type: ignore response = integrations.reconnect_connection(ReconnectConnectionRequest( connection_id="conn_123", redirect_url="https://myapp.com/callback", )) assert response.authorization_url.startswith("https://oauth.hubspot.com") def test_get_stats(self) -> None: """Test getting integration stats""" mock_http = MockHTTPClient(responses={ "GET:/integrations/connections/stats": { "totalConnections": 10, "activeConnections": 8, "errorConnections": 2, "byConnector": { "hubspot": 5, "google-drive": 3, }, } }) integrations = Integrations(mock_http) # type: ignore stats = integrations.get_stats() assert stats.total_connections == 10 assert stats.active_connections == 8 def test_list_logs(self) -> None: """Test listing API logs""" mock_http = MockHTTPClient(responses={ "GET:/integrations/logs": { "items": [ { "id": "log_1", "connectionId": "conn_123", "method": "GET", "path": "/contacts", "statusCode": 200, }, ], "cursor": None, } }) integrations = Integrations(mock_http) # type: ignore response = integrations.list_logs() assert len(response.items) == 1 assert response.items[0].status_code == 200 def test_list_logs_with_filters(self) -> None: """Test listing logs with filters""" mock_http = MockHTTPClient(responses={ "GET:/integrations/logs?connectionId=conn_123&statusCode=500": { "items": [], "cursor": None, } }) integrations = Integrations(mock_http) # type: ignore response = integrations.list_logs(ListLogsRequest( connection_id="conn_123", status_code=500, )) assert len(response.items) == 0 def test_passthrough(self) -> None: """Test passthrough request""" mock_http = MockHTTPClient(responses={ "POST:/integrations/passthrough": { "data": {"custom": "response"}, } }) integrations = Integrations(mock_http) # type: ignore result = integrations.passthrough(PassthroughRequest( connection_id="conn_123", method="GET", path="/custom/endpoint", )) assert result["data"]["custom"] == "response" def test_sub_clients_initialized(self) -> None: """Test that sub-clients are properly initialized""" mock_http = MockHTTPClient() integrations = Integrations(mock_http) # type: ignore assert integrations.crm is not None assert integrations.storage is not None assert integrations.communication is not None assert integrations.productivity is not None class TestAsyncIntegrations: """Tests for async Integrations client""" @pytest.mark.asyncio async def test_list_connectors(self) -> None: """Test async list connectors""" mock_http = MockAsyncHTTPClient(responses={ "GET:/integrations/connectors": [ { "slug": "hubspot", "name": "HubSpot", "category": "crm", "authType": "oauth2", }, ] }) integrations = AsyncIntegrations(mock_http) # type: ignore connectors = await integrations.list_connectors() assert len(connectors) == 1 @pytest.mark.asyncio async def test_list_connections(self) -> None: """Test async list connections""" mock_http = MockAsyncHTTPClient(responses={ "GET:/integrations/connections": { "items": [ { "id": "conn_123", "connectorSlug": "hubspot", "status": "active", }, ], "total": 1, } }) integrations = AsyncIntegrations(mock_http) # type: ignore response = await integrations.list_connections() assert len(response.items) == 1 @pytest.mark.asyncio async def test_crm_list_contacts(self) -> None: """Test async CRM list contacts""" mock_http = MockAsyncHTTPClient(responses={ "GET:/integrations/crm/contacts?connectionId=conn_123": { "items": [ { "id": "contact_1", "firstName": "John", "lastName": "Doe", "email": "[email protected]", }, ], "cursor": None, } }) integrations = AsyncIntegrations(mock_http) # type: ignore result = await integrations.crm.list_contacts("conn_123") assert len(result["items"]) == 1 @pytest.mark.asyncio async def test_storage_list_files(self) -> None: """Test async storage list files""" mock_http = MockAsyncHTTPClient(responses={ "GET:/integrations/storage/files?connectionId=conn_123": { "items": [ { "id": "file_1", "name": "document.pdf", "mimeType": "application/pdf", "size": 1024, }, ], "cursor": None, } }) integrations = AsyncIntegrations(mock_http) # type: ignore result = await integrations.storage.list_files("conn_123") assert len(result["items"]) == 1