"""Tests for LinkedIn GraphQL mutations and queries (User Story 1). Tests cover: - initiateLinkedInOAuth mutation returns authorization URL - availableLinkedInOrganizations query returns orgs from API - connectLinkedInOrganizations mutation creates pages and sources - linkedInPages query returns workspace pages - disconnectLinkedInPage mutation archives page """ from unittest.mock import MagicMock, patch import pytest from django.core.cache import cache from accounts.tests.factories import UserFactory, WorkspaceFactory, WorkspaceMembershipFactory from integrations.models import LinkedInPage, LinkedInSyncStatus from integrations.tests.factories import LinkedInPageFactory from sources.models import Source @pytest.fixture def user(): """Create a test user.""" return UserFactory() @pytest.fixture def workspace(): """Create a test workspace.""" return WorkspaceFactory() @pytest.fixture def admin_membership(user, workspace): """Create admin membership for user in workspace.""" return WorkspaceMembershipFactory(user=user, workspace=workspace, role="admin") @pytest.fixture def context(user): """Create a mock GraphQL context with authenticated user. The GraphQL Info object has info.context.user structure, so we need to set up the mock to match. """ info = MagicMock() info.context.user = user return info @pytest.fixture def cached_tokens(workspace): """Set up cached OAuth tokens for the workspace.""" cache_key = f"linkedin_oauth_{workspace.pk}" token_data = { "access_token": "cached_access_token_123", "refresh_token": "cached_refresh_token_456", "expires_in": 5184000, "refresh_token_expires_in": 31536000, } cache.set(cache_key, token_data, timeout=3600) return token_data @pytest.mark.django_db class TestInitiateLinkedInOAuthMutation: """Tests for initiateLinkedInOAuth mutation.""" def test_initiate_returns_authorization_url(self, context, workspace, admin_membership): """Test that mutation returns a valid authorization URL.""" from strawberry import ID from integrations.graphql import IntegrationsMutation mutation = IntegrationsMutation() result = mutation.initiate_linkedin_oauth(context, workspace_id=ID(str(workspace.pk))) assert result.oauth_url is not None assert "linkedin.com/oauth" in result.oauth_url assert "response_type=code" in result.oauth_url assert "state=" in result.oauth_url assert result.state is not None def test_initiate_includes_correct_scope(self, context, workspace, admin_membership): """Test that authorization URL includes correct scope.""" from strawberry import ID from integrations.graphql import IntegrationsMutation mutation = IntegrationsMutation() result = mutation.initiate_linkedin_oauth(context, workspace_id=ID(str(workspace.pk))) # Should include rw_organization_admin scope assert "rw_organization_admin" in result.oauth_url def test_initiate_requires_authentication(self, workspace, admin_membership): """Test that mutation requires authentication.""" from strawberry import ID from integrations.graphql import IntegrationsMutation context = MagicMock() context.context.user = None mutation = IntegrationsMutation() with pytest.raises(Exception) as exc_info: mutation.initiate_linkedin_oauth(context, workspace_id=ID(str(workspace.pk))) assert "authentication" in str(exc_info.value).lower() def test_initiate_requires_admin_role(self, context, workspace): """Test that mutation requires admin role.""" from strawberry import ID from integrations.graphql import IntegrationsMutation # User is not a member of the workspace mutation = IntegrationsMutation() with pytest.raises(Exception) as exc_info: mutation.initiate_linkedin_oauth(context, workspace_id=ID(str(workspace.pk))) assert "permission" in str(exc_info.value).lower() or "denied" in str(exc_info.value).lower() @pytest.mark.django_db class TestAvailableLinkedInOrganizationsQuery: """Tests for availableLinkedInOrganizations query.""" def setup_method(self): """Clear cache before each test.""" cache.clear() @patch("integrations.services.linkedin_client.LinkedInClient") def test_returns_organizations_from_api( self, mock_client_class, context, workspace, admin_membership, cached_tokens ): """Test that query returns organizations from LinkedIn API.""" from strawberry import ID from integrations.graphql import IntegrationsQuery from integrations.services.linkedin_client import LinkedInOrganization # Setup mock client mock_client = MagicMock() mock_client_class.return_value = mock_client mock_orgs = [ LinkedInOrganization( organization_urn="urn:li:organization:12345", organization_id="12345", name="Acme Corp", vanity_name="acme-corp", logo_url="https://logo.url/acme.png", role="ADMINISTRATOR", state="APPROVED", ), LinkedInOrganization( organization_urn="urn:li:organization:67890", organization_id="67890", name="TechStartup", vanity_name="techstartup", logo_url="https://logo.url/tech.png", role="ADMINISTRATOR", state="APPROVED", ), ] mock_client.get_organizations.return_value = mock_orgs query = IntegrationsQuery() result = query.available_linkedin_organizations(context, workspace_id=ID(str(workspace.pk))) assert len(result) == 2 assert result[0].organization_urn == "urn:li:organization:12345" assert result[0].name == "Acme Corp" assert result[1].organization_urn == "urn:li:organization:67890" @patch("integrations.services.linkedin_client.LinkedInClient") def test_marks_connected_organizations( self, mock_client_class, context, workspace, admin_membership, cached_tokens ): """Test that already connected orgs are marked as connected.""" from strawberry import ID from integrations.graphql import IntegrationsQuery from integrations.services.linkedin_client import LinkedInOrganization # Create an existing connected page LinkedInPageFactory( workspace=workspace, organization_urn="urn:li:organization:12345", ) mock_client = MagicMock() mock_client_class.return_value = mock_client mock_orgs = [ LinkedInOrganization( organization_urn="urn:li:organization:12345", organization_id="12345", name="Acme Corp", vanity_name="acme-corp", logo_url=None, role="ADMINISTRATOR", state="APPROVED", ), ] mock_client.get_organizations.return_value = mock_orgs query = IntegrationsQuery() result = query.available_linkedin_organizations(context, workspace_id=ID(str(workspace.pk))) assert len(result) == 1 assert result[0].is_connected is True def test_requires_cached_tokens(self, context, workspace, admin_membership): """Test that query fails without cached tokens.""" from strawberry import ID from integrations.graphql import IntegrationsQuery query = IntegrationsQuery() with pytest.raises(Exception) as exc_info: query.available_linkedin_organizations(context, workspace_id=ID(str(workspace.pk))) assert "authorize" in str(exc_info.value).lower() or "token" in str(exc_info.value).lower() @pytest.mark.django_db class TestConnectLinkedInOrganizationsMutation: """Tests for connectLinkedInOrganizations mutation.""" def setup_method(self): """Clear cache before each test.""" cache.clear() @patch("integrations.graphql.django_rq") @patch("integrations.services.linkedin_client.LinkedInClient") def test_creates_linkedin_page_and_source( self, mock_client_class, mock_django_rq, context, workspace, admin_membership, cached_tokens ): """Test that mutation creates LinkedInPage and Source records.""" from strawberry import ID from integrations.graphql import IntegrationsMutation from integrations.services.linkedin_client import LinkedInOrganization mock_client = MagicMock() mock_client_class.return_value = mock_client mock_queue = MagicMock() mock_django_rq.get_queue.return_value = mock_queue mock_org = LinkedInOrganization( organization_urn="urn:li:organization:12345", organization_id="12345", name="Acme Corp", vanity_name="acme-corp", logo_url="https://logo.url/acme.png", role="ADMINISTRATOR", state="APPROVED", ) mock_client.get_organizations.return_value = [mock_org] mutation = IntegrationsMutation() result = mutation.connect_linkedin_organizations( context, workspace_id=ID(str(workspace.pk)), organization_urns=["urn:li:organization:12345"], ) assert result.success is True assert len(result.pages) == 1 # Verify LinkedInPage was created page = LinkedInPage.objects.get(organization_urn="urn:li:organization:12345") assert page.workspace == workspace assert page.name == "Acme Corp" assert page.access_token == "cached_access_token_123" # Verify Source was created source = Source.objects.get(linkedin_page=page) assert source.workspace == workspace # Verify backfill was queued mock_queue.enqueue.assert_called_once() call_kwargs = mock_queue.enqueue.call_args.kwargs assert call_kwargs["page_id"] == page.pk assert call_kwargs["full_sync"] is True assert source.kind == "linkedin_page" @patch("integrations.graphql.django_rq") @patch("integrations.services.linkedin_client.LinkedInClient") def test_connects_multiple_organizations( self, mock_client_class, mock_django_rq, context, workspace, admin_membership, cached_tokens ): """Test that mutation can connect multiple organizations.""" from strawberry import ID from integrations.graphql import IntegrationsMutation from integrations.services.linkedin_client import LinkedInOrganization mock_client = MagicMock() mock_client_class.return_value = mock_client mock_queue = MagicMock() mock_django_rq.get_queue.return_value = mock_queue mock_orgs = [ LinkedInOrganization( organization_urn="urn:li:organization:111", organization_id="111", name="Org One", vanity_name="org-one", logo_url=None, role="ADMINISTRATOR", state="APPROVED", ), LinkedInOrganization( organization_urn="urn:li:organization:222", organization_id="222", name="Org Two", vanity_name="org-two", logo_url=None, role="ADMINISTRATOR", state="APPROVED", ), ] mock_client.get_organizations.return_value = mock_orgs mutation = IntegrationsMutation() result = mutation.connect_linkedin_organizations( context, workspace_id=ID(str(workspace.pk)), organization_urns=["urn:li:organization:111", "urn:li:organization:222"], ) assert result.success is True # Verify backfill was queued for both pages assert mock_queue.enqueue.call_count == 2 assert len(result.pages) == 2 assert LinkedInPage.objects.filter(workspace=workspace).count() == 2 @patch("integrations.services.linkedin_client.LinkedInClient") def test_rejects_unauthorized_organization( self, mock_client_class, context, workspace, admin_membership, cached_tokens ): """Test that mutation rejects organization user doesn't admin.""" from strawberry import ID from integrations.graphql import IntegrationsMutation mock_client = MagicMock() mock_client_class.return_value = mock_client mock_client.get_organizations.return_value = [] # User has no orgs mutation = IntegrationsMutation() result = mutation.connect_linkedin_organizations( context, workspace_id=ID(str(workspace.pk)), organization_urns=["urn:li:organization:unauthorized"], ) assert result.success is False assert "not authorized" in result.error.lower() or "not found" in result.error.lower() @patch("integrations.services.linkedin_client.LinkedInClient") def test_skips_already_connected_organization( self, mock_client_class, context, workspace, admin_membership, cached_tokens ): """Test that mutation skips already connected organizations.""" from strawberry import ID from integrations.graphql import IntegrationsMutation from integrations.services.linkedin_client import LinkedInOrganization # Create existing page LinkedInPageFactory( workspace=workspace, organization_urn="urn:li:organization:12345", ) mock_client = MagicMock() mock_client_class.return_value = mock_client mock_org = LinkedInOrganization( organization_urn="urn:li:organization:12345", organization_id="12345", name="Acme Corp", vanity_name="acme-corp", logo_url=None, role="ADMINISTRATOR", state="APPROVED", ) mock_client.get_organizations.return_value = [mock_org] mutation = IntegrationsMutation() mutation.connect_linkedin_organizations( context, workspace_id=ID(str(workspace.pk)), organization_urns=["urn:li:organization:12345"], ) # Should succeed but not create duplicate assert ( LinkedInPage.objects.filter( workspace=workspace, organization_urn="urn:li:organization:12345", ).count() == 1 ) @pytest.mark.django_db class TestLinkedInPagesQuery: """Tests for linkedInPages query.""" def test_returns_workspace_pages(self, context, workspace, admin_membership): """Test that query returns pages for the workspace.""" from strawberry import ID from integrations.graphql import IntegrationsQuery # Create test pages LinkedInPageFactory(workspace=workspace, name="Page One") LinkedInPageFactory(workspace=workspace, name="Page Two") # Create page in different workspace other_workspace = WorkspaceFactory() LinkedInPageFactory(workspace=other_workspace, name="Other Page") query = IntegrationsQuery() result = query.linkedin_pages(context, workspace_id=ID(str(workspace.pk))) assert len(result) == 2 page_names = {p.name for p in result} assert "Page One" in page_names assert "Page Two" in page_names assert "Other Page" not in page_names def test_excludes_archived_pages(self, context, workspace, admin_membership): """Test that query excludes archived pages.""" from strawberry import ID from integrations.graphql import IntegrationsQuery active_page = LinkedInPageFactory(workspace=workspace, is_archived=False) LinkedInPageFactory( workspace=workspace, is_archived=True, organization_urn="urn:li:organization:99999", ) query = IntegrationsQuery() result = query.linkedin_pages(context, workspace_id=ID(str(workspace.pk))) assert len(result) == 1 assert result[0].id == str(active_page.pk) @pytest.mark.django_db class TestDisconnectLinkedInPageMutation: """Tests for disconnectLinkedInPage mutation.""" def test_archives_page(self, context, workspace, admin_membership): """Test that mutation archives the page.""" from strawberry import ID from integrations.graphql import IntegrationsMutation page = LinkedInPageFactory(workspace=workspace, is_archived=False) mutation = IntegrationsMutation() result = mutation.disconnect_linkedin_page(context, page_id=ID(str(page.pk))) assert result.success is True page.refresh_from_db() assert page.is_archived is True assert page.sync_status == LinkedInSyncStatus.PAUSED def test_sets_sync_status_paused(self, context, workspace, admin_membership): """Test that mutation sets sync status to paused.""" from strawberry import ID from integrations.graphql import IntegrationsMutation page = LinkedInPageFactory( workspace=workspace, sync_status=LinkedInSyncStatus.ACTIVE, ) mutation = IntegrationsMutation() mutation.disconnect_linkedin_page(context, page_id=ID(str(page.pk))) page.refresh_from_db() assert page.sync_status == LinkedInSyncStatus.PAUSED def test_requires_workspace_admin(self, context, workspace): """Test that mutation requires workspace admin role.""" from strawberry import ID from integrations.graphql import IntegrationsMutation page = LinkedInPageFactory(workspace=workspace) # User is not a member mutation = IntegrationsMutation() with pytest.raises(Exception) as exc_info: mutation.disconnect_linkedin_page(context, page_id=ID(str(page.pk))) assert "permission" in str(exc_info.value).lower() or "denied" in str(exc_info.value).lower() def test_handles_nonexistent_page(self, context, workspace, admin_membership): """Test that mutation handles nonexistent page.""" from strawberry import ID from integrations.graphql import IntegrationsMutation mutation = IntegrationsMutation() result = mutation.disconnect_linkedin_page( context, page_id=ID("999999"), # Non-existent integer ID ) assert result.success is False assert "not found" in result.error.lower()