feat: Enhance LinkedIn image publishing by supporting bytes data and implementing temporary file handling

LINKEDIN_IMAGE_PUBLISHING_SOLUTION.md

@@ -0,0 +1,250 @@
+# LinkedIn Image Publishing Solution
+
+## Problem Statement
+
+The current implementation stores images as bytes in the database, but LinkedIn's API requires temporary upload URLs for images. This mismatch prevents successful image publishing to LinkedIn.
+
+## Current Implementation Analysis
+
+### Image Storage
+Images are currently stored in the database as bytes using the `ensure_bytes_format` utility function in `backend/utils/image_utils.py`. This function handles:
+- Converting base64 encoded strings to bytes
+- Storing URLs as strings
+- Keeping bytes data as-is
+
+### LinkedIn API Process
+The current `LinkedInService.publish_post` method in `backend/services/linkedin_service.py` follows these steps:
+1. Register upload with LinkedIn API using `registerUploadRequest`
+2. Get temporary upload URL and asset URN
+3. If image_url is a string (URL), download and upload to LinkedIn
+4. Create post with the asset URN
+
+However, when image data is stored as bytes, the current implementation skips the image upload entirely.
+
+## Solution Design
+
+### 1. Converting Stored Image Bytes to LinkedIn-Compatible Format
+
+We need to modify the `publish_post` method to handle bytes data by:
+1. Creating a temporary file from the bytes data
+2. Uploading this file to LinkedIn using the existing register/upload mechanism
+3. Cleaning up the temporary file after upload
+
+### 2. Temporary File Storage Mechanism
+
+We'll implement a temporary file storage solution using Python's `tempfile` module:
+- Create temporary files with secure random names
+- Store files in the system's temporary directory
+- Use appropriate file extensions based on image type
+- Implement automatic cleanup after use
+
+### 3. Integration with Existing LinkedInService
+
+The solution will be integrated into the `LinkedInService.publish_post` method:
+- Add a new code path to handle bytes data
+- Maintain backward compatibility with URL-based images
+- Use the same register/upload mechanism for consistency
+
+### 4. Security Considerations
+
+- Use secure temporary file creation to prevent path traversal attacks
+- Validate image data before creating temporary files
+- Set appropriate file permissions on temporary files
+- Implement cleanup mechanisms to prevent disk space exhaustion
+
+### 5. Cleanup Mechanism
+
+- Immediate cleanup after successful upload
+- Error handling to ensure cleanup even if upload fails
+- Periodic cleanup of orphaned temporary files (if any)
+
+## Implementation Plan
+
+### Step 1: Modify LinkedInService
+
+Update `backend/services/linkedin_service.py` to handle bytes data:
+
+```python
+def publish_post(self, access_token: str, user_id: str, text_content: str, image_url: str = None) -> dict:
+    # ... existing code ...
+    
+    if image_url:
+        if isinstance(image_url, bytes):
+            # Handle bytes data - create temporary file and upload
+            temp_file_path = self._create_temp_image_file(image_url)
+            try:
+                # Register upload
+                register_body = {
+                    "registerUploadRequest": {
+                        "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
+                        "owner": f"urn:li:person:{user_id}",
+                        "serviceRelationships": [{
+                            "relationshipType": "OWNER",
+                            "identifier": "urn:li:userGeneratedContent"
+                        }]
+                    }
+                }
+                
+                r = requests.post(
+                    "https://api.linkedin.com/v2/assets?action=registerUpload",
+                    headers=headers,
+                    json=register_body
+                )
+                
+                if r.status_code not in (200, 201):
+                    raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
+                
+                datar = r.json()["value"]
+                upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
+                asset_urn = datar["asset"]
+                
+                # Upload image from temporary file
+                upload_headers = {
+                    "Authorization": f"Bearer {access_token}",
+                    "X-Restli-Protocol-Version": "2.0.0",
+                    "Content-Type": "application/octet-stream"
+                }
+                
+                with open(temp_file_path, 'rb') as f:
+                    image_data = f.read()
+                    upload_response = requests.put(upload_url, headers=upload_headers, data=image_data)
+                    if upload_response.status_code not in (200, 201):
+                        raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
+                
+                # Create post with image
+                post_body = {
+                    "author": f"urn:li:person:{user_id}",
+                    "lifecycleState": "PUBLISHED",
+                    "specificContent": {
+                        "com.linkedin.ugc.ShareContent": {
+                            "shareCommentary": {"text": text_content},
+                            "shareMediaCategory": "IMAGE",
+                            "media": [{
+                                "status": "READY",
+                                "media": asset_urn,
+                                "description": {"text": "Post image"},
+                                "title": {"text": "Post image"}
+                            }]
+                        }
+                    },
+                    "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}
+                }
+            finally:
+                # Clean up temporary file
+                self._cleanup_temp_file(temp_file_path)
+        # ... rest of existing code for URL handling ...
+```
+
+### Step 2: Add Helper Methods
+
+Add the following helper methods to `LinkedInService`:
+
+```python
+import tempfile
+import os
+from typing import Optional
+
+def _create_temp_image_file(self, image_bytes: bytes) -> str:
+    """
+    Create a temporary file from image bytes.
+    
+    Args:
+        image_bytes: Image data as bytes
+        
+    Returns:
+        Path to the temporary file
+    """
+    # Create a temporary file
+    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.jpg')
+    temp_file_path = temp_file.name
+    
+    try:
+        # Write image bytes to the temporary file
+        temp_file.write(image_bytes)
+        temp_file.flush()
+    finally:
+        temp_file.close()
+    
+    return temp_file_path
+
+def _cleanup_temp_file(self, file_path: str) -> None:
+    """
+    Safely remove a temporary file.
+    
+    Args:
+        file_path: Path to the temporary file to remove
+    """
+    try:
+        if file_path and os.path.exists(file_path):
+            os.unlink(file_path)
+    except Exception as e:
+        # Log the error but don't fail the operation
+        import logging
+        logging.error(f"Failed to cleanup temporary file {file_path}: {str(e)}")
+```
+
+### Step 3: Update Error Handling
+
+Enhance error handling to ensure cleanup happens even if the upload fails:
+
+```python
+def publish_post(self, access_token: str, user_id: str, text_content: str, image_url: str = None) -> dict:
+    temp_file_path = None
+    try:
+        # ... existing implementation with temporary file creation ...
+    except Exception as e:
+        # Ensure cleanup happens even if an error occurs
+        if temp_file_path:
+            self._cleanup_temp_file(temp_file_path)
+        raise e
+```
+
+## Security Considerations
+
+1. **Secure Temporary File Creation**: Using `tempfile.NamedTemporaryFile` with `delete=False` creates files with secure random names in the system's temporary directory, preventing predictable file name attacks.
+
+2. **File Permissions**: The temporary files will have default system permissions, which are typically restricted to the creating user.
+
+3. **Input Validation**: Before creating temporary files, we should validate that the data is indeed image data and not malicious content.
+
+4. **Resource Management**: Implementing proper cleanup mechanisms prevents disk space exhaustion from orphaned temporary files.
+
+## Alternative Approaches
+
+### In-Memory Upload
+Instead of creating temporary files, we could upload directly from memory. However, this approach:
+- Requires modifying the existing LinkedIn upload mechanism
+- May have memory limitations for large images
+- Is more complex to implement safely
+
+### Streaming Upload
+Streaming the bytes directly to LinkedIn's upload endpoint:
+- More memory efficient
+- More complex implementation
+- Requires careful handling of HTTP streaming
+
+The temporary file approach is chosen for its simplicity and reliability while maintaining compatibility with the existing upload mechanism.
+
+## Testing Plan
+
+1. **Unit Tests**: Create tests for the new helper methods
+2. **Integration Tests**: Test the complete flow with bytes data
+3. **Error Handling Tests**: Verify cleanup happens even when uploads fail
+4. **Security Tests**: Validate temporary files are created securely
+
+## Deployment Considerations
+
+1. **File System Permissions**: Ensure the application has write access to the temporary directory
+2. **Disk Space Monitoring**: Monitor temporary directory for space issues
+3. **Cleanup Verification**: Verify that temporary files are properly cleaned up in all scenarios
+
+## Conclusion
+
+This solution addresses the LinkedIn image publishing issue by:
+1. Converting stored image bytes to a format compatible with LinkedIn's upload API through temporary files
+2. Implementing a secure temporary file storage mechanism
+3. Integrating seamlessly with the existing LinkedInService
+4. Addressing security considerations for temporary file storage
+5. Implementing proper cleanup mechanisms
+
+The approach maintains backward compatibility while adding the needed functionality to handle bytes-based image data.

backend/services/linkedin_service.py

@@ -2,6 +2,9 @@ from flask import current_app
 import requests
 from requests_oauthlib import OAuth2Session
 from urllib.parse import urlencode
+import tempfile
+import os
+import logging
 
 class LinkedInService:
     """Service for LinkedIn API integration."""
@@ -122,6 +125,43 @@ class LinkedInService:
             logger.error(f"🔗 [LinkedIn] Error type: {type(e)}")
             raise e
     
+    def _create_temp_image_file(self, image_bytes: bytes) -> str:
+        """
+        Create a temporary file from image bytes.
+        
+        Args:
+            image_bytes: Image data as bytes
+            
+        Returns:
+            Path to the temporary file
+        """
+        # Create a temporary file
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.jpg')
+        temp_file_path = temp_file.name
+        
+        try:
+            # Write image bytes to the temporary file
+            temp_file.write(image_bytes)
+            temp_file.flush()
+        finally:
+            temp_file.close()
+        
+        return temp_file_path
+    
+    def _cleanup_temp_file(self, file_path: str) -> None:
+        """
+        Safely remove a temporary file.
+        
+        Args:
+            file_path: Path to the temporary file to remove
+        """
+        try:
+            if file_path and os.path.exists(file_path):
+                os.unlink(file_path)
+        except Exception as e:
+            # Log the error but don't fail the operation
+            logging.error(f"Failed to cleanup temporary file {file_path}: {str(e)}")
+    
     def publish_post(self, access_token: str, user_id: str, text_content: str, image_url: str = None) -> dict:
         """
         Publish a post to LinkedIn.
@@ -130,11 +170,12 @@ class LinkedInService:
             access_token (str): LinkedIn access token
             user_id (str): LinkedIn user ID
             text_content (str): Post content
-            image_url (str, optional): Image URL (must be a URL, not bytes)
+            image_url (str or bytes, optional): Image URL or image bytes
             
         Returns:
             dict: Publish response
         """
+        temp_file_path = None
         url = "https://api.linkedin.com/v2/ugcPosts"
         headers = {
             "Authorization": f"Bearer {access_token}",
@@ -142,83 +183,150 @@ class LinkedInService:
             "Content-Type": "application/json"
         }
         
-        # Skip image handling if image_url is not a string (i.e., if it's bytes)
-        if image_url and isinstance(image_url, str):
-            # Handle image upload
-            register_body = {
-                "registerUploadRequest": {
-                    "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
-                    "owner": f"urn:li:person:{user_id}",
-                    "serviceRelationships": [{
-                        "relationshipType": "OWNER",
-                        "identifier": "urn:li:userGeneratedContent"
-                    }]
+        try:
+            if image_url and isinstance(image_url, bytes):
+                # Handle bytes data - create temporary file and upload
+                temp_file_path = self._create_temp_image_file(image_url)
+                
+                # Register upload
+                register_body = {
+                    "registerUploadRequest": {
+                        "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
+                        "owner": f"urn:li:person:{user_id}",
+                        "serviceRelationships": [{
+                            "relationshipType": "OWNER",
+                            "identifier": "urn:li:userGeneratedContent"
+                        }]
+                    }
                 }
-            }
-            
-            r = requests.post(
-                "https://api.linkedin.com/v2/assets?action=registerUpload",
-                headers=headers,
-                json=register_body
-            )
-            
-            if r.status_code not in (200, 201):
-                raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
-            
-            datar = r.json()["value"]
-            upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
-            asset_urn = datar["asset"]
-            
-            # Upload image
-            upload_headers = {
-                "Authorization": f"Bearer {access_token}",
-                "X-Restli-Protocol-Version": "2.0.0",
-                "Content-Type": "application/octet-stream"
-            }
-            
-            # Download image and upload to LinkedIn
-            image_response = requests.get(image_url)
-            if image_response.status_code == 200:
-                upload_response = requests.put(upload_url, headers=upload_headers, data=image_response.content)
-                if upload_response.status_code not in (200, 201):
-                    raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
-            
-            # Create post with image
-            post_body = {
-                "author": f"urn:li:person:{user_id}",
-                "lifecycleState": "PUBLISHED",
-                "specificContent": {
-                    "com.linkedin.ugc.ShareContent": {
-                        "shareCommentary": {"text": text_content},
-                        "shareMediaCategory": "IMAGE",
-                        "media": [{
-                            "status": "READY",
-                            "media": asset_urn,
-                            "description": {"text": "Post image"},
-                            "title": {"text": "Post image"}
+                
+                r = requests.post(
+                    "https://api.linkedin.com/v2/assets?action=registerUpload",
+                    headers=headers,
+                    json=register_body
+                )
+                
+                if r.status_code not in (200, 201):
+                    raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
+                
+                datar = r.json()["value"]
+                upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
+                asset_urn = datar["asset"]
+                
+                # Upload image from temporary file
+                upload_headers = {
+                    "Authorization": f"Bearer {access_token}",
+                    "X-Restli-Protocol-Version": "2.0.0",
+                    "Content-Type": "application/octet-stream"
+                }
+                
+                with open(temp_file_path, 'rb') as f:
+                    image_data = f.read()
+                    upload_response = requests.put(upload_url, headers=upload_headers, data=image_data)
+                    if upload_response.status_code not in (200, 201):
+                        raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
+                
+                # Create post with image
+                post_body = {
+                    "author": f"urn:li:person:{user_id}",
+                    "lifecycleState": "PUBLISHED",
+                    "specificContent": {
+                        "com.linkedin.ugc.ShareContent": {
+                            "shareCommentary": {"text": text_content},
+                            "shareMediaCategory": "IMAGE",
+                            "media": [{
+                                "status": "READY",
+                                "media": asset_urn,
+                                "description": {"text": "Post image"},
+                                "title": {"text": "Post image"}
+                            }]
+                        }
+                    },
+                    "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}
+                }
+            elif image_url and isinstance(image_url, str):
+                # Handle image upload for URL-based images
+                register_body = {
+                    "registerUploadRequest": {
+                        "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
+                        "owner": f"urn:li:person:{user_id}",
+                        "serviceRelationships": [{
+                            "relationshipType": "OWNER",
+                            "identifier": "urn:li:userGeneratedContent"
                         }]
                     }
-                },
-                "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}
-            }
-        else:
-            # Create text-only post
-            post_body = {
-                "author": f"urn:li:person:{user_id}",
-                "lifecycleState": "PUBLISHED",
-                "specificContent": {
-                    "com.linkedin.ugc.ShareContent": {
-                        "shareCommentary": {
-                            "text": text_content
-                        },
-                        "shareMediaCategory": "NONE"
+                }
+                
+                r = requests.post(
+                    "https://api.linkedin.com/v2/assets?action=registerUpload",
+                    headers=headers,
+                    json=register_body
+                )
+                
+                if r.status_code not in (200, 201):
+                    raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
+                
+                datar = r.json()["value"]
+                upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
+                asset_urn = datar["asset"]
+                
+                # Upload image
+                upload_headers = {
+                    "Authorization": f"Bearer {access_token}",
+                    "X-Restli-Protocol-Version": "2.0.0",
+                    "Content-Type": "application/octet-stream"
+                }
+                
+                # Download image and upload to LinkedIn
+                image_response = requests.get(image_url)
+                if image_response.status_code == 200:
+                    upload_response = requests.put(upload_url, headers=upload_headers, data=image_response.content)
+                    if upload_response.status_code not in (200, 201):
+                        raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
+                
+                # Create post with image
+                post_body = {
+                    "author": f"urn:li:person:{user_id}",
+                    "lifecycleState": "PUBLISHED",
+                    "specificContent": {
+                        "com.linkedin.ugc.ShareContent": {
+                            "shareCommentary": {"text": text_content},
+                            "shareMediaCategory": "IMAGE",
+                            "media": [{
+                                "status": "READY",
+                                "media": asset_urn,
+                                "description": {"text": "Post image"},
+                                "title": {"text": "Post image"}
+                            }]
+                        }
+                    },
+                    "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}
+                }
+            else:
+                # Create text-only post
+                post_body = {
+                    "author": f"urn:li:person:{user_id}",
+                    "lifecycleState": "PUBLISHED",
+                    "specificContent": {
+                        "com.linkedin.ugc.ShareContent": {
+                            "shareCommentary": {
+                                "text": text_content
+                            },
+                            "shareMediaCategory": "NONE"
+                        }
+                    },
+                    "visibility": {
+                        "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"
                     }
-                },
-                "visibility": {
-                    "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"
                 }
-            }
-        
-        response = requests.post(url, headers=headers, json=post_body)
-        response.raise_for_status()
-        return response.json()
+            
+            response = requests.post(url, headers=headers, json=post_body)
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            # Re-raise the exception to maintain existing behavior
+            raise e
+        finally:
+            # Clean up temporary file if it was created
+            if temp_file_path:
+                self._cleanup_temp_file(temp_file_path)

backend/tests/test_scheduler_image_integration.py

@@ -0,0 +1,310 @@
+#!/usr/bin/env python3
+"""
+Integration test for scheduling system with image handling.
+Tests the end-to-end workflow from scheduling through content generation to publishing,
+specifically for posts with images in both bytes and URL formats.
+"""
+
+import sys
+import os
+import unittest
+import uuid
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timedelta
+
+# Add the backend directory to the path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from backend.scheduler.apscheduler_service import APSchedulerService
+from backend.services.content_service import ContentService
+from backend.utils.image_utils import ensure_bytes_format
+
+
+class TestSchedulerImageIntegration(unittest.TestCase):
+    """Test cases for scheduler integration with image handling."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        self.user_id = "test_user_123"
+        self.schedule_id = str(uuid.uuid4())
+        self.social_account_id = "test_social_456"
+        
+        # Mock Flask app
+        self.mock_app = MagicMock()
+        self.mock_app.config = {
+            'SUPABASE_URL': 'test_url',
+            'SUPABASE_KEY': 'test_key'
+        }
+        
+        # Initialize scheduler service with mock app
+        self.scheduler_service = APSchedulerService(self.mock_app)
+        
+        # Mock Supabase client
+        self.mock_supabase = MagicMock()
+        self.scheduler_service.supabase_client = self.mock_supabase
+        
+        # Mock scheduler
+        self.mock_scheduler = MagicMock()
+        self.scheduler_service.scheduler = self.mock_scheduler
+        
+    def test_image_bytes_processing_in_content_generation(self):
+        """Test that image bytes are properly processed and stored during content generation."""
+        # Mock content service to return content with image bytes
+        test_content = "This is a test post with an image"
+        test_image_bytes = b"fake image bytes data"
+        
+        with patch('backend.scheduler.apscheduler_service.ContentService') as mock_content_service_class:
+            mock_content_service = MagicMock()
+            mock_content_service.generate_post_content.return_value = (test_content, test_image_bytes)
+            mock_content_service_class.return_value = mock_content_service
+            
+            # Mock database response for schedule lookup
+            self.mock_supabase.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [
+                {'id_social': self.social_account_id}
+            ]
+            
+            # Mock database response for content storage
+            mock_insert_response = MagicMock()
+            mock_insert_response.data = [{'id': 'test_post_789'}]
+            self.mock_supabase.table.return_value.insert.return_value.execute.return_value = mock_insert_response
+            
+            # Execute the content generation task
+            self.scheduler_service.generate_content_task(self.user_id, self.schedule_id)
+            
+            # Verify content service was called
+            mock_content_service.generate_post_content.assert_called_once_with(self.user_id)
+            
+            # Verify database insert was called with correct parameters
+            self.mock_supabase.table.assert_called_with("Post_content")
+            self.mock_supabase.table.return_value.insert.assert_called_once()
+            
+            # Get the actual call arguments to verify image handling
+            call_args = self.mock_supabase.table.return_value.insert.call_args
+            inserted_data = call_args[0][0]  # First positional argument
+            
+            # Verify text content is stored correctly
+            self.assertEqual(inserted_data['Text_content'], test_content)
+            
+            # Verify image data is stored correctly (should be bytes)
+            self.assertEqual(inserted_data['image_content_url'], test_image_bytes)
+            self.assertEqual(inserted_data['is_published'], False)
+            self.assertEqual(inserted_data['sched'], self.schedule_id)
+    
+    def test_image_url_processing_in_content_generation(self):
+        """Test that image URLs are properly stored during content generation."""
+        # Mock content service to return content with image URL
+        test_content = "This is a test post with an image URL"
+        test_image_url = "https://example.com/test-image.jpg"
+        
+        with patch('backend.scheduler.apscheduler_service.ContentService') as mock_content_service_class:
+            mock_content_service = MagicMock()
+            mock_content_service.generate_post_content.return_value = (test_content, test_image_url)
+            mock_content_service_class.return_value = mock_content_service
+            
+            # Mock database response for schedule lookup
+            self.mock_supabase.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [
+                {'id_social': self.social_account_id}
+            ]
+            
+            # Mock database response for content storage
+            mock_insert_response = MagicMock()
+            mock_insert_response.data = [{'id': 'test_post_789'}]
+            self.mock_supabase.table.return_value.insert.return_value.execute.return_value = mock_insert_response
+            
+            # Execute the content generation task
+            self.scheduler_service.generate_content_task(self.user_id, self.schedule_id)
+            
+            # Verify content service was called
+            mock_content_service.generate_post_content.assert_called_once_with(self.user_id)
+            
+            # Verify database insert was called with correct parameters
+            self.mock_supabase.table.assert_called_with("Post_content")
+            self.mock_supabase.table.return_value.insert.assert_called_once()
+            
+            # Get the actual call arguments to verify image handling
+            call_args = self.mock_supabase.table.return_value.insert.call_args
+            inserted_data = call_args[0][0]  # First positional argument
+            
+            # Verify text content is stored correctly
+            self.assertEqual(inserted_data['Text_content'], test_content)
+            
+            # Verify image URL is stored correctly
+            self.assertEqual(inserted_data['image_content_url'], test_image_url)
+            self.assertEqual(inserted_data['is_published'], False)
+            self.assertEqual(inserted_data['sched'], self.schedule_id)
+    
+    def test_image_base64_processing_in_content_generation(self):
+        """Test that base64 encoded images are properly converted to bytes during content generation."""
+        # Mock content service to return content with base64 image
+        test_content = "This is a test post with a base64 image"
+        test_base64_image = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg=="
+        expected_bytes = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\x0cIDAT\x08\xd7c\xfc\xff\x9f\xa1\x1e\x00\x07\x82\x02\x7f=\xc8H\xef\x00\x00\x00\x00IEND\xaeB`\x82'
+        
+        with patch('backend.scheduler.apscheduler_service.ContentService') as mock_content_service_class:
+            mock_content_service = MagicMock()
+            mock_content_service.generate_post_content.return_value = (test_content, test_base64_image)
+            mock_content_service_class.return_value = mock_content_service
+            
+            # Mock database response for schedule lookup
+            self.mock_supabase.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [
+                {'id_social': self.social_account_id}
+            ]
+            
+            # Mock database response for content storage
+            mock_insert_response = MagicMock()
+            mock_insert_response.data = [{'id': 'test_post_789'}]
+            self.mock_supabase.table.return_value.insert.return_value.execute.return_value = mock_insert_response
+            
+            # Execute the content generation task
+            self.scheduler_service.generate_content_task(self.user_id, self.schedule_id)
+            
+            # Verify content service was called
+            mock_content_service.generate_post_content.assert_called_once_with(self.user_id)
+            
+            # Verify database insert was called with correct parameters
+            self.mock_supabase.table.assert_called_with("Post_content")
+            self.mock_supabase.table.return_value.insert.assert_called_once()
+            
+            # Get the actual call arguments to verify image handling
+            call_args = self.mock_supabase.table.return_value.insert.call_args
+            inserted_data = call_args[0][0]  # First positional argument
+            
+            # Verify text content is stored correctly
+            self.assertEqual(inserted_data['Text_content'], test_content)
+            
+            # Verify base64 image was converted to bytes
+            self.assertEqual(inserted_data['image_content_url'], expected_bytes)
+            self.assertEqual(inserted_data['is_published'], False)
+            self.assertEqual(inserted_data['sched'], self.schedule_id)
+    
+    def test_publishing_with_image_bytes(self):
+        """Test that posts with image bytes are properly published."""
+        # Test data
+        test_post_id = "test_post_123"
+        test_content = "This is a test post for publishing"
+        test_image_bytes = b"fake image bytes data for publishing"
+        
+        # Mock database response for unpublished post lookup
+        self.mock_supabase.table.return_value.select.return_value.eq.return_value.eq.return_value.order.return_value.limit.return_value.execute.return_value.data = [
+            {
+                'id': test_post_id,
+                'Text_content': test_content,
+                'image_content_url': test_image_bytes,
+                'is_published': False
+            }
+        ]
+        
+        # Mock database response for schedule lookup
+        self.mock_supabase.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [
+            {
+                'Social_network': {
+                    'token': 'test_access_token',
+                    'sub': 'test_user_sub'
+                }
+            }
+        ]
+        
+        # Mock LinkedIn service
+        with patch('backend.scheduler.apscheduler_service.LinkedInService') as mock_linkedin_service_class:
+            mock_linkedin_service = MagicMock()
+            mock_linkedin_service.publish_post.return_value = {'success': True}
+            mock_linkedin_service_class.return_value = mock_linkedin_service
+            
+            # Mock database response for post update
+            mock_update_response = MagicMock()
+            mock_update_response.data = [{'id': test_post_id}]
+            self.mock_supabase.table.return_value.update.return_value.eq.return_value.execute.return_value = mock_update_response
+            
+            # Execute the publishing task
+            self.scheduler_service.publish_post_task(self.schedule_id)
+            
+            # Verify LinkedIn service was called with correct parameters
+            mock_linkedin_service.publish_post.assert_called_once_with(
+                'test_access_token',
+                'test_user_sub',
+                test_content,
+                test_image_bytes  # Bytes should be passed directly
+            )
+            
+            # Verify post status was updated in database
+            self.mock_supabase.table.return_value.update.assert_called_once_with({"is_published": True})
+            self.mock_supabase.table.return_value.update.return_value.eq.assert_called_once_with("id", test_post_id)
+    
+    def test_publishing_with_image_url(self):
+        """Test that posts with image URLs are properly published."""
+        # Test data
+        test_post_id = "test_post_456"
+        test_content = "This is a test post with URL for publishing"
+        test_image_url = "https://example.com/publish-test-image.jpg"
+        
+        # Mock database response for unpublished post lookup
+        self.mock_supabase.table.return_value.select.return_value.eq.return_value.eq.return_value.order.return_value.limit.return_value.execute.return_value.data = [
+            {
+                'id': test_post_id,
+                'Text_content': test_content,
+                'image_content_url': test_image_url,
+                'is_published': False
+            }
+        ]
+        
+        # Mock database response for schedule lookup
+        self.mock_supabase.table.return_value.select.return_value.eq.return_value.execute.return_value.data = [
+            {
+                'Social_network': {
+                    'token': 'test_access_token',
+                    'sub': 'test_user_sub'
+                }
+            }
+        ]
+        
+        # Mock LinkedIn service
+        with patch('backend.scheduler.apscheduler_service.LinkedInService') as mock_linkedin_service_class:
+            mock_linkedin_service = MagicMock()
+            mock_linkedin_service.publish_post.return_value = {'success': True}
+            mock_linkedin_service_class.return_value = mock_linkedin_service
+            
+            # Mock database response for post update
+            mock_update_response = MagicMock()
+            mock_update_response.data = [{'id': test_post_id}]
+            self.mock_supabase.table.return_value.update.return_value.eq.return_value.execute.return_value = mock_update_response
+            
+            # Execute the publishing task
+            self.scheduler_service.publish_post_task(self.schedule_id)
+            
+            # Verify LinkedIn service was called with correct parameters
+            mock_linkedin_service.publish_post.assert_called_once_with(
+                'test_access_token',
+                'test_user_sub',
+                test_content,
+                test_image_url  # URL should be passed directly
+            )
+            
+            # Verify post status was updated in database
+            self.mock_supabase.table.return_value.update.assert_called_once_with({"is_published": True})
+            self.mock_supabase.table.return_value.update.return_value.eq.assert_called_once_with("id", test_post_id)
+    
+    def test_ensure_bytes_format_utility(self):
+        """Test the ensure_bytes_format utility function with different input types."""
+        # Test with bytes input
+        test_bytes = b"test bytes data"
+        result = ensure_bytes_format(test_bytes)
+        self.assertEqual(result, test_bytes)
+        
+        # Test with base64 string input
+        test_base64 = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg=="
+        expected_bytes = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\x0cIDAT\x08\xd7c\xfc\xff\x9f\xa1\x1e\x00\x07\x82\x02\x7f=\xc8H\xef\x00\x00\x00\x00IEND\xaeB`\x82'
+        result = ensure_bytes_format(test_base64)
+        self.assertEqual(result, expected_bytes)
+        
+        # Test with URL string input
+        test_url = "https://example.com/image.jpg"
+        result = ensure_bytes_format(test_url)
+        self.assertEqual(result, test_url)
+        
+        # Test with None input
+        result = ensure_bytes_format(None)
+        self.assertIsNone(result)
+
+
+if __name__ == '__main__':
+    unittest.main()