LinkedIn Content Creation MCP Server

""" Tests for LinkedInManager and Draft. """ import json import tempfile from pathlib import Path import pytest from chuk_mcp_linkedin.manager import Draft, LinkedInManager class TestDraft: """Test Draft class""" def test_draft_creation(self): """Test creating a draft""" draft = Draft( draft_id="draft_1", name="My Post", post_type="text", content={"text": "Hello world"}, theme="thought_leader", ) assert draft.draft_id == "draft_1" assert draft.name == "My Post" assert draft.post_type == "text" assert draft.theme == "thought_leader" def test_draft_update_content(self): """Test updating draft content""" import time draft = Draft( draft_id="draft_1", name="My Post", post_type="text", content={"text": "Hello"} ) original_updated = draft.updated_at # Small delay to ensure timestamp changes on all platforms (including Windows) time.sleep(0.01) draft.update_content({"text": "Hello world"}) assert draft.content["text"] == "Hello world" assert draft.updated_at != original_updated def test_draft_to_dict(self): """Test converting draft to dictionary""" draft = Draft( draft_id="draft_1", name="My Post", post_type="text", content={"text": "Hello"}, theme="thought_leader", ) draft_dict = draft.to_dict() assert draft_dict["draft_id"] == "draft_1" assert draft_dict["name"] == "My Post" assert draft_dict["theme"] == "thought_leader" assert "created_at" in draft_dict assert "updated_at" in draft_dict def test_draft_from_dict(self): """Test creating draft from dictionary""" data = { "draft_id": "draft_1", "name": "My Post", "post_type": "text", "content": {"text": "Hello"}, "theme": "thought_leader", } draft = Draft.from_dict(data) assert draft.draft_id == "draft_1" assert draft.name == "My Post" assert draft.theme == "thought_leader" def test_draft_name_setter(self): """Test setting draft name updates timestamp""" import time draft = Draft(draft_id="draft_1", name="Original", post_type="text", content={}) original_updated = draft.updated_at time.sleep(0.01) draft.name = "Updated Name" assert draft.name == "Updated Name" assert draft.updated_at != original_updated def test_draft_created_at_setter(self): """Test setting created_at timestamp""" draft = Draft(draft_id="draft_1", name="Test", post_type="text", content={}) custom_timestamp = "2024-01-01T12:00:00" draft.created_at = custom_timestamp assert draft.created_at == custom_timestamp def test_draft_updated_at_setter(self): """Test setting updated_at timestamp""" draft = Draft(draft_id="draft_1", name="Test", post_type="text", content={}) custom_timestamp = "2024-01-01T13:00:00" draft.updated_at = custom_timestamp assert draft.updated_at == custom_timestamp def test_draft_metadata_property(self): """Test accessing draft metadata""" draft = Draft(draft_id="draft_1", name="Test", post_type="text", content={}) # Metadata should be accessible assert isinstance(draft.metadata, dict) def test_draft_preview_token_property(self): """Test accessing preview token""" draft = Draft(draft_id="draft_1", name="Test", post_type="text", content={}) # Preview token should be accessible and non-empty assert draft.preview_token assert isinstance(draft.preview_token, str) class TestLinkedInManager: """Test LinkedInManager class""" @pytest.fixture def temp_storage(self): """Create temporary storage for tests""" with tempfile.TemporaryDirectory() as tmpdir: yield tmpdir @pytest.fixture def manager(self, temp_storage): """Create manager with temporary storage""" return LinkedInManager(storage_path=temp_storage) def test_manager_initialization(self, temp_storage): """Test manager initializes correctly""" manager = LinkedInManager(storage_path=temp_storage) assert manager.storage_path == Path(temp_storage) assert len(manager.drafts) == 0 assert manager.current_draft_id is None def test_create_draft(self, manager): """Test creating a draft""" draft = manager.create_draft( name="My Post", post_type="text", content={"text": "Hello world"}, theme="thought_leader", ) assert draft.name == "My Post" assert draft.post_type == "text" assert draft.theme == "thought_leader" assert draft.draft_id in manager.drafts assert manager.current_draft_id == draft.draft_id def test_create_multiple_drafts(self, manager): """Test creating multiple drafts""" manager.create_draft("Post 1", "text") draft2 = manager.create_draft("Post 2", "poll") assert len(manager.drafts) == 2 assert manager.current_draft_id == draft2.draft_id def test_get_draft(self, manager): """Test getting a draft by ID""" draft = manager.create_draft("My Post", "text") retrieved = manager.get_draft(draft.draft_id) assert retrieved is not None assert retrieved.draft_id == draft.draft_id def test_get_draft_not_found(self, manager): """Test getting non-existent draft""" draft = manager.get_draft("nonexistent") assert draft is None def test_get_current_draft(self, manager): """Test getting current draft""" draft = manager.create_draft("My Post", "text") current = manager.get_current_draft() assert current is not None assert current.draft_id == draft.draft_id def test_get_current_draft_none(self, manager): """Test getting current draft when none exists""" current = manager.get_current_draft() assert current is None def test_list_drafts(self, manager): """Test listing all drafts""" manager.create_draft("Post 1", "text") manager.create_draft("Post 2", "poll") drafts = manager.list_drafts() assert len(drafts) == 2 assert drafts[0]["name"] == "Post 1" assert drafts[1]["name"] == "Post 2" assert drafts[1]["is_current"] is True def test_switch_draft(self, manager): """Test switching to a different draft""" draft1 = manager.create_draft("Post 1", "text") manager.create_draft("Post 2", "poll") # Switch back to draft1 success = manager.switch_draft(draft1.draft_id) assert success is True assert manager.current_draft_id == draft1.draft_id def test_switch_draft_invalid(self, manager): """Test switching to invalid draft""" success = manager.switch_draft("nonexistent") assert success is False def test_update_draft(self, manager): """Test updating a draft""" draft = manager.create_draft("My Post", "text", content={"text": "Hello"}) success = manager.update_draft( draft.draft_id, content={"text": "Hello world"}, theme="personal_brand" ) assert success is True updated_draft = manager.get_draft(draft.draft_id) assert updated_draft.content["text"] == "Hello world" assert updated_draft.theme == "personal_brand" def test_update_draft_not_found(self, manager): """Test updating non-existent draft""" success = manager.update_draft("nonexistent", content={"text": "Hello"}) assert success is False def test_delete_draft(self, manager): """Test deleting a draft""" draft = manager.create_draft("My Post", "text") success = manager.delete_draft(draft.draft_id) assert success is True assert draft.draft_id not in manager.drafts def test_delete_draft_updates_current(self, manager): """Test deleting current draft updates current_draft_id""" draft1 = manager.create_draft("Post 1", "text") draft2 = manager.create_draft("Post 2", "poll") # Delete current draft manager.delete_draft(draft2.draft_id) assert manager.current_draft_id == draft1.draft_id def test_delete_draft_not_found(self, manager): """Test deleting non-existent draft""" success = manager.delete_draft("nonexistent") assert success is False def test_clear_all_drafts(self, manager): """Test clearing all drafts""" manager.create_draft("Post 1", "text") manager.create_draft("Post 2", "poll") manager.create_draft("Post 3", "document") count = manager.clear_all_drafts() assert count == 3 assert len(manager.drafts) == 0 assert manager.current_draft_id is None def test_export_draft(self, manager): """Test exporting draft as JSON""" draft = manager.create_draft( "My Post", "text", content={"text": "Hello"}, theme="thought_leader" ) json_str = manager.export_draft(draft.draft_id) assert json_str is not None # Parse and verify data = json.loads(json_str) assert data["name"] == "My Post" assert data["theme"] == "thought_leader" def test_export_draft_not_found(self, manager): """Test exporting non-existent draft""" json_str = manager.export_draft("nonexistent") assert json_str is None def test_import_draft(self, manager): """Test importing draft from JSON""" draft_json = json.dumps( { "draft_id": "imported_1", "name": "Imported Post", "post_type": "text", "content": {"text": "Imported content"}, "theme": "storyteller", } ) draft = manager.import_draft(draft_json) assert draft is not None assert draft.name == "Imported Post" assert draft.theme == "storyteller" assert draft.draft_id in manager.drafts def test_import_draft_duplicate_id(self, manager): """Test importing draft with duplicate ID gets new ID""" draft1 = manager.create_draft("Post 1", "text") draft_json = json.dumps( { "draft_id": draft1.draft_id, # Same ID "name": "Duplicate", "post_type": "text", "content": {"text": "Content"}, } ) draft2 = manager.import_draft(draft_json) assert draft2.draft_id != draft1.draft_id def test_import_draft_invalid_json(self, manager): """Test importing invalid JSON""" draft = manager.import_draft("invalid json{") assert draft is None def test_get_draft_preview(self, manager): """Test getting draft preview""" draft = manager.create_draft("My Post", "text", content={"commentary": "x" * 300}) preview = manager.get_draft_preview(draft.draft_id, chars=210) assert len(preview) <= 213 # 210 + "..." def test_get_draft_preview_not_found(self, manager): """Test getting preview of non-existent draft""" preview = manager.get_draft_preview("nonexistent") assert preview is None def test_get_draft_stats(self, manager): """Test getting draft statistics""" draft = manager.create_draft( "My Post", "text", content={ "commentary": "Hello world this is my post", "hook": "Attention grabber", "cta": "What do you think?", "hashtags": ["Marketing", "LinkedIn"], }, ) stats = manager.get_draft_stats(draft.draft_id) assert stats is not None assert stats["word_count"] > 0 assert stats["char_count"] > 0 assert stats["has_hook"] is True assert stats["has_cta"] is True assert stats["hashtag_count"] == 2 def test_get_draft_stats_not_found(self, manager): """Test getting stats of non-existent draft""" stats = manager.get_draft_stats("nonexistent") assert stats is None def test_get_info(self, manager): """Test getting manager info""" manager.create_draft("Post 1", "text") manager.create_draft("Post 2", "poll") info = manager.get_info() assert info["total_drafts"] == 2 assert info["current_draft_id"] is not None assert "text" in info["draft_types"] assert "poll" in info["draft_types"] def test_persistence(self, temp_storage): """Test drafts persist across manager instances""" # Create drafts with first manager manager1 = LinkedInManager(storage_path=temp_storage) draft1 = manager1.create_draft("Post 1", "text") draft2 = manager1.create_draft("Post 2", "poll") # Create new manager instance with same storage manager2 = LinkedInManager(storage_path=temp_storage) # Drafts should be loaded assert len(manager2.drafts) == 2 assert draft1.draft_id in manager2.drafts assert draft2.draft_id in manager2.drafts def test_save_and_load_preserves_data(self, temp_storage): """Test save and load preserves all draft data""" manager1 = LinkedInManager(storage_path=temp_storage) draft = manager1.create_draft( "My Post", "text", content={"text": "Hello", "hook": "Stat"}, theme="thought_leader", variant_config={"style": "insight"}, ) # Load in new manager manager2 = LinkedInManager(storage_path=temp_storage) loaded_draft = manager2.get_draft(draft.draft_id) assert loaded_draft.name == "My Post" assert loaded_draft.post_type == "text" assert loaded_draft.theme == "thought_leader" assert loaded_draft.content["text"] == "Hello" assert loaded_draft.variant_config["style"] == "insight" def test_update_draft_with_variant_config(self, manager): """Test updating draft with variant config""" draft = manager.create_draft("My Post", "text", variant_config={"style": "insight"}) # Update variant config manager.update_draft(draft.draft_id, variant_config={"tone": "professional"}) updated_draft = manager.get_draft(draft.draft_id) assert updated_draft.variant_config["style"] == "insight" assert updated_draft.variant_config["tone"] == "professional" def test_get_draft_preview_with_text_fallback(self, manager): """Test getting draft preview falls back to text field""" draft = manager.create_draft("My Post", "text", content={"text": "x" * 300}) preview = manager.get_draft_preview(draft.draft_id, chars=210) assert len(preview) <= 213 # 210 + "..." def test_get_draft_preview_short_content(self, manager): """Test getting draft preview with short content""" draft = manager.create_draft("My Post", "text", content={"commentary": "Short"}) preview = manager.get_draft_preview(draft.draft_id, chars=210) assert preview == "Short" def test_load_drafts_with_corrupted_file(self, temp_storage): """Test loading drafts skips corrupted files""" manager1 = LinkedInManager(storage_path=temp_storage) manager1.create_draft("Valid Post", "text") # Create a corrupted JSON file corrupted_file = Path(temp_storage) / "corrupted.json" with open(corrupted_file, "w") as f: f.write("invalid json{") # Load in new manager - should skip corrupted file manager2 = LinkedInManager(storage_path=temp_storage) assert len(manager2.drafts) == 1 # Only the valid draft loaded def test_user_id_generation(self, temp_storage): """Test that manager has a user ID""" manager = LinkedInManager(storage_path=temp_storage) # Default user_id is None (used for backward compatibility) assert manager.user_id is None or isinstance(manager.user_id, str) def test_custom_user_id(self, temp_storage): """Test manager with custom user ID""" manager = LinkedInManager(storage_path=temp_storage, user_id="user123") assert manager.user_id == "user123" def test_get_draft_by_preview_token(self, manager): """Test getting draft by preview token""" draft1 = manager.create_draft("Post 1", "text") draft2 = manager.create_draft("Post 2", "text") # Get draft by its preview token found_draft = manager.get_draft_by_preview_token(draft1.preview_token) assert found_draft is not None assert found_draft.draft_id == draft1.draft_id assert found_draft.name == "Post 1" # Get second draft by its token found_draft2 = manager.get_draft_by_preview_token(draft2.preview_token) assert found_draft2 is not None assert found_draft2.draft_id == draft2.draft_id def test_get_draft_by_preview_token_not_found(self, manager): """Test getting draft with invalid preview token""" manager.create_draft("Post 1", "text") # Non-existent token found_draft = manager.get_draft_by_preview_token("invalid-token-123") assert found_draft is None def test_user_isolation(self, temp_storage): """Test that different users have isolated drafts""" # User 1 creates drafts manager1 = LinkedInManager(storage_path=temp_storage, user_id="user1") draft1 = manager1.create_draft("User 1 Draft", "text") assert draft1 is not None # User 2 should have independent storage manager2 = LinkedInManager(storage_path=temp_storage, user_id="user2") # User 2 shouldn't see user 1's drafts (when user_id is used for isolation) # Note: With current filesystem implementation, drafts are shared unless # using artifacts with user_id scoping assert len(manager2.drafts) >= 0 # May see shared drafts in filesystem mode def test_draft_creation_with_user_id(self, temp_storage): """Test creating drafts with user ID""" manager = LinkedInManager(storage_path=temp_storage, user_id="testuser") draft = manager.create_draft("Test", "text") assert draft is not None assert draft.draft_id in manager.drafts def test_list_drafts_for_user(self, temp_storage): """Test listing drafts for a specific user""" manager = LinkedInManager(storage_path=temp_storage, user_id="user1") # Create drafts manager.create_draft("Draft 1", "text") manager.create_draft("Draft 2", "text") # List drafts drafts = manager.list_drafts() assert len(drafts) == 2 def test_list_drafts_includes_metadata(self, manager): """Test draft list includes metadata""" manager.create_draft("Test", "text") drafts = manager.list_drafts() assert len(drafts) == 1 draft_info = drafts[0] # Should include basic draft metadata assert "draft_id" in draft_info assert "name" in draft_info assert "post_type" in draft_info assert "is_current" in draft_info class TestLinkedInManagerWithArtifacts: """Test LinkedInManager with artifact storage""" @pytest.mark.asyncio async def test_artifact_storage_initialization(self): """Test artifact storage initialization""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: assert manager.use_artifacts is True assert manager.artifact_provider == "memory" @pytest.mark.asyncio async def test_store_draft_as_artifact(self): """Test storing draft as artifact""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft( name="Test Draft", post_type="text", content={"commentary": "Test"} ) artifact_id = await manager.store_draft_as_artifact(draft.draft_id) assert artifact_id is not None @pytest.mark.asyncio async def test_store_nonexistent_draft_as_artifact(self): """Test storing non-existent draft as artifact""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: artifact_id = await manager.store_draft_as_artifact("nonexistent") assert artifact_id is None @pytest.mark.asyncio async def test_retrieve_draft_from_artifact(self): """Test retrieving draft from artifact""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: # Create and store draft original = manager.create_draft( name="Test Draft", post_type="text", content={"commentary": "Test content"} ) artifact_id = await manager.store_draft_as_artifact(original.draft_id) # Retrieve from artifact retrieved = await manager.retrieve_draft_from_artifact(artifact_id) assert retrieved is not None assert retrieved.name == "Test Draft" assert retrieved.content["commentary"] == "Test content" @pytest.mark.asyncio async def test_retrieve_nonexistent_artifact(self): """Test retrieving non-existent artifact""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: retrieved = await manager.retrieve_draft_from_artifact("nonexistent") assert retrieved is None @pytest.mark.asyncio async def test_artifact_storage_disabled(self): """Test that artifact storage is disabled by default""" manager = LinkedInManager(use_artifacts=False) assert manager.use_artifacts is False @pytest.mark.asyncio async def test_artifact_provider_filesystem(self): """Test artifact provider with filesystem""" async with LinkedInManager(use_artifacts=True, artifact_provider="filesystem") as manager: assert manager.artifact_provider == "filesystem" draft = manager.create_draft("FS Test", "text") artifact_id = await manager.store_draft_as_artifact(draft.draft_id) assert artifact_id is not None @pytest.mark.asyncio async def test_store_draft_artifact_store_none(self): """Test storing draft when artifact store is None""" manager = LinkedInManager(use_artifacts=True, artifact_provider="memory") draft = manager.create_draft("Test", "text") # Access private attribute to set it to None (edge case testing) manager._artifact_store = None manager._artifact_initialized = True artifact_id = await manager.store_draft_as_artifact(draft.draft_id) assert artifact_id is None @pytest.mark.asyncio async def test_retrieve_draft_when_disabled(self): """Test retrieving draft from artifact when artifacts are disabled""" manager = LinkedInManager(use_artifacts=False) result = await manager.retrieve_draft_from_artifact("any-id") assert result is None @pytest.mark.asyncio async def test_retrieve_draft_artifact_store_none(self): """Test retrieving draft when artifact store is None""" manager = LinkedInManager(use_artifacts=True, artifact_provider="memory") # Set store to None after initialization (edge case) manager._artifact_store = None manager._artifact_initialized = True result = await manager.retrieve_draft_from_artifact("any-id") assert result is None @pytest.mark.asyncio async def test_retrieve_draft_artifact_not_found(self): """Test retrieving draft when artifact data is None""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: # Try to retrieve with ID that doesn't exist result = await manager.retrieve_draft_from_artifact("missing-id") assert result is None @pytest.mark.asyncio async def test_generate_html_preview_draft_not_found(self): """Test HTML preview generation when draft doesn't exist""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: result = await manager.generate_html_preview_async("nonexistent-draft") assert result is None @pytest.mark.asyncio async def test_generate_html_preview_artifact_store_none(self): """Test HTML preview generation when artifact store is None""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft("Test", "text") # Set store to None (edge case) manager._artifact_store = None manager._artifact_initialized = True result = await manager.generate_html_preview_async(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_read_preview_html_draft_not_found(self): """Test reading preview HTML when draft doesn't exist""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: result = await manager.read_preview_html_async("nonexistent-draft") assert result is None @pytest.mark.asyncio async def test_read_preview_html_artifact_store_none(self): """Test reading preview HTML when artifact store is None""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft("Test", "text") # Set store to None (edge case) manager._artifact_store = None manager._artifact_initialized = True result = await manager.read_preview_html_async(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_read_preview_html_artifact_metadata_no_match(self): """Test reading preview HTML when metadata doesn't match""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft("Test", "text") # Generate preview first await manager.generate_html_preview_async(draft.draft_id) # Now try to read with a different draft_id result = await manager.read_preview_html_async("different-draft-id") # Should generate new preview since no matching artifact found assert result is None @pytest.mark.asyncio async def test_read_preview_html_exception_handling(self): """Test reading preview HTML handles exceptions gracefully""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft("Test", "text") # Mock the artifact store to raise an exception from unittest.mock import AsyncMock manager._artifact_store.list_by_session = AsyncMock(side_effect=Exception("Test error")) result = await manager.read_preview_html_async(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_generate_preview_url_draft_not_found(self): """Test generating preview URL when draft doesn't exist""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: result = await manager.generate_preview_url("nonexistent-draft") assert result is None @pytest.mark.asyncio async def test_generate_preview_url_s3(self): """Test generating preview URL with S3 storage""" async with LinkedInManager(use_artifacts=True, artifact_provider="s3") as manager: from unittest.mock import AsyncMock draft = manager.create_draft("Test", "text") # Mock the get_shareable_url method if manager._artifact_store: manager._artifact_store.get_shareable_url = AsyncMock( return_value="https://s3.amazonaws.com/signed-url" ) result = await manager.generate_preview_url(draft.draft_id) assert result == "https://s3.amazonaws.com/signed-url" @pytest.mark.asyncio async def test_generate_preview_url_s3_generation_failed(self): """Test generating preview URL with S3 when preview generation fails""" async with LinkedInManager(use_artifacts=True, artifact_provider="s3") as manager: draft = manager.create_draft("Test", "text") # Mock generate_html_preview_async to return None from unittest.mock import AsyncMock manager.generate_html_preview_async = AsyncMock(return_value=None) result = await manager.generate_preview_url(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_generate_preview_url_s3_store_none(self): """Test generating preview URL with S3 when store is None""" async with LinkedInManager(use_artifacts=True, artifact_provider="s3") as manager: draft = manager.create_draft("Test", "text") # Set store to None manager._artifact_store = None manager._artifact_initialized = True result = await manager.generate_preview_url(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_generate_preview_url_s3_no_signed_url(self): """Test generating preview URL with S3 when signed URL returns None""" async with LinkedInManager(use_artifacts=True, artifact_provider="s3") as manager: from unittest.mock import AsyncMock draft = manager.create_draft("Test", "text") # Mock get_shareable_url to return None if manager._artifact_store: manager._artifact_store.get_shareable_url = AsyncMock(return_value=None) result = await manager.generate_preview_url(draft.draft_id) assert result is None @pytest.mark.asyncio async def test_generate_preview_url_memory_provider(self): """Test generating preview URL with memory provider""" async with LinkedInManager(use_artifacts=True, artifact_provider="memory") as manager: draft = manager.create_draft("Test", "text") result = await manager.generate_preview_url( draft.draft_id, base_url="http://localhost:8000" ) assert result is not None assert "http://localhost:8000/preview/" in result assert draft.preview_token in result @pytest.mark.asyncio async def test_preview_cross_user_isolation(self): """Test that preview HTML respects user isolation - no cross-user leakage""" import tempfile with tempfile.TemporaryDirectory() as tmpdir: # User 1: Create draft and generate preview async with LinkedInManager( storage_path=tmpdir, user_id="user1", use_artifacts=True, artifact_provider="memory", ) as manager1: draft1 = manager1.create_draft( "User 1 Draft", "text", content={"commentary": "User 1 content"} ) # Generate preview for user 1 draft artifact_id1 = await manager1.generate_html_preview_async(draft1.draft_id) assert artifact_id1 is not None # User 1 can read their own preview html1 = await manager1.read_preview_html_async(draft1.draft_id) assert html1 is not None assert "User 1 content" in html1 # User 2: Each user has their own artifact store (memory provider is per-manager) async with LinkedInManager( storage_path=tmpdir, user_id="user2", use_artifacts=True, artifact_provider="memory", ) as manager2: # User 2 can see draft1 exists (filesystem-based draft storage is shared in this test) _ = manager2.get_draft(draft1.draft_id) # In production with artifacts, user2 wouldn't see user1's drafts # But in this test with shared filesystem, they might # User 2 creates their own draft draft2 = manager2.create_draft( "User 2 Draft", "text", content={"commentary": "User 2 content"} ) # User 2 can generate/read their own preview artifact_id2 = await manager2.generate_html_preview_async(draft2.draft_id) assert artifact_id2 is not None html2 = await manager2.read_preview_html_async(draft2.draft_id) assert html2 is not None assert "User 2 content" in html2 # SECURITY VERIFICATION: Verify user1 and user2 artifacts are isolated # Re-open user1 and verify they can only access their own data async with LinkedInManager( storage_path=tmpdir, user_id="user1", use_artifacts=True, artifact_provider="memory", ) as manager1_reopen: draft1_reloaded = manager1_reopen.get_draft(draft1.draft_id) assert draft1_reloaded is not None # User 1 needs to regenerate preview (memory artifact store was cleared) html1_new = await manager1_reopen.read_preview_html_async(draft1.draft_id) assert html1_new is not None assert "User 1 content" in html1_new