test_api_media.py•29.7 kB
"""
Tests for LinkedIn Media API.
Tests image and video upload methods with mocked HTTP requests.
"""
from unittest.mock import AsyncMock, MagicMock, mock_open, patch
import pytest
from chuk_mcp_linkedin.api import LinkedInAPIError, LinkedInClient, LinkedInConfig
@pytest.fixture
def configured_client():
"""Create a configured LinkedIn client for testing"""
config = LinkedInConfig(
linkedin_access_token="test_token",
linkedin_person_urn="urn:li:person:test123",
)
return LinkedInClient(config=config)
@pytest.fixture
def mock_image_init_response():
"""Mock image upload initialization response"""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"value": {
"uploadUrl": "https://upload.example.com/image",
"image": "urn:li:image:123456",
}
}
mock_response.raise_for_status = MagicMock()
return mock_response
@pytest.fixture
def mock_video_init_response():
"""Mock video upload initialization response"""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"value": {
"video": "urn:li:video:123456",
"uploadInstructions": [
{
"uploadUrl": "https://upload.example.com/video",
}
],
"uploadToken": "test-token",
}
}
mock_response.raise_for_status = MagicMock()
return mock_response
@pytest.fixture
def mock_upload_success_response():
"""Mock successful upload response"""
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.raise_for_status = MagicMock()
return mock_response
@pytest.fixture
def mock_file_stat():
"""Mock file stat for file size"""
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024 # 1MB
return mock_stat
class TestUploadImage:
"""Test upload_image method"""
@pytest.mark.asyncio
async def test_upload_image_success(
self,
configured_client,
mock_image_init_response,
mock_upload_success_response,
mock_file_stat,
):
"""Test successful image upload"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_client.post.return_value = mock_image_init_response
mock_client.put.return_value = mock_upload_success_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake image data")):
with patch("mimetypes.guess_type", return_value=("image/jpeg", None)):
result = await configured_client.upload_image("test.jpg")
assert result == "urn:li:image:123456"
assert mock_client.post.call_count == 1
assert mock_client.put.call_count == 1
@pytest.mark.asyncio
async def test_upload_image_with_alt_text(
self,
configured_client,
mock_image_init_response,
mock_upload_success_response,
mock_file_stat,
):
"""Test image upload with alt text"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_client.post.return_value = mock_image_init_response
mock_client.put.return_value = mock_upload_success_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake image data")):
with patch("mimetypes.guess_type", return_value=("image/jpeg", None)):
result = await configured_client.upload_image(
"test.jpg", alt_text="Test image"
)
assert result == "urn:li:image:123456"
@pytest.mark.asyncio
async def test_upload_image_without_credentials(self):
"""Test image upload fails without credentials"""
config = LinkedInConfig(linkedin_access_token=None, linkedin_person_urn=None)
client = LinkedInClient(config=config)
with pytest.raises(LinkedInAPIError, match="not configured"):
await client.upload_image("test.jpg")
@pytest.mark.asyncio
async def test_upload_image_init_fails(self, configured_client):
"""Test image upload when initialization fails"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_error_response = MagicMock()
mock_error_response.status_code = 400
mock_error_response.json.return_value = {"message": "Invalid request"}
def raise_for_status():
from httpx import HTTPStatusError
raise HTTPStatusError(
"400 Bad Request", request=MagicMock(), response=mock_error_response
)
mock_error_response.raise_for_status = raise_for_status
mock_client.post.return_value = mock_error_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.stat") as mock_stat:
mock_stat.return_value.st_size = 1024 * 1024
with pytest.raises(
LinkedInAPIError, match="Failed to (upload|initialize) image"
):
await configured_client.upload_image("test.jpg")
@pytest.mark.asyncio
async def test_upload_image_upload_fails(
self, configured_client, mock_image_init_response, mock_file_stat
):
"""Test image upload when file upload fails"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_error_response = MagicMock()
mock_error_response.status_code = 500
mock_error_response.text = "Server Error"
mock_client.post.return_value = mock_image_init_response
mock_client.put.return_value = mock_error_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake image data")):
with patch("mimetypes.guess_type", return_value=("image/jpeg", None)):
with pytest.raises(LinkedInAPIError, match="Failed to upload image"):
await configured_client.upload_image("test.jpg")
class TestUploadVideo:
"""Test upload_video method"""
@pytest.mark.asyncio
async def test_upload_video_success(
self,
configured_client,
mock_video_init_response,
mock_upload_success_response,
mock_file_stat,
):
"""Test successful video upload"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
# Mock finalize response
mock_finalize_response = MagicMock()
mock_finalize_response.status_code = 200
# Return init response first, then finalize response for subsequent posts
mock_client.post.side_effect = [mock_video_init_response, mock_finalize_response]
mock_client.put.return_value = mock_upload_success_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with patch("mimetypes.guess_type", return_value=("video/mp4", None)):
with patch("asyncio.sleep", return_value=None):
result = await configured_client.upload_video("test.mp4")
assert result == "urn:li:video:123456"
assert mock_client.post.call_count == 2 # init + finalize
assert mock_client.put.call_count == 1
@pytest.mark.asyncio
async def test_upload_video_with_metadata(
self,
configured_client,
mock_video_init_response,
mock_upload_success_response,
mock_file_stat,
):
"""Test video upload with title"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
# Mock finalize response
mock_finalize_response = MagicMock()
mock_finalize_response.status_code = 200
mock_client.post.side_effect = [mock_video_init_response, mock_finalize_response]
mock_client.put.return_value = mock_upload_success_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with patch("mimetypes.guess_type", return_value=("video/mp4", None)):
with patch("asyncio.sleep", return_value=None):
result = await configured_client.upload_video(
"test.mp4", title="Video Title"
)
assert result == "urn:li:video:123456"
@pytest.mark.asyncio
async def test_upload_video_without_credentials(self):
"""Test video upload fails without credentials"""
config = LinkedInConfig(linkedin_access_token=None, linkedin_person_urn=None)
client = LinkedInClient(config=config)
with pytest.raises(LinkedInAPIError, match="not configured"):
await client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_init_fails(self, configured_client):
"""Test video upload when initialization fails"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_error_response = MagicMock()
mock_error_response.status_code = 400
mock_error_response.json.return_value = {"message": "Invalid request"}
def raise_for_status():
from httpx import HTTPStatusError
raise HTTPStatusError(
"400 Bad Request", request=MagicMock(), response=mock_error_response
)
mock_error_response.raise_for_status = raise_for_status
mock_client.post.return_value = mock_error_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.stat") as mock_stat:
mock_stat.return_value.st_size = 1024 * 1024
with pytest.raises(
LinkedInAPIError, match="Failed to (upload|initialize) video"
):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_upload_fails(
self, configured_client, mock_video_init_response, mock_file_stat
):
"""Test video upload when file upload fails"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_error_response = MagicMock()
mock_error_response.status_code = 500
mock_error_response.text = "Server Error"
mock_client.post.return_value = mock_video_init_response
mock_client.put.return_value = mock_error_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.stat", return_value=mock_file_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with patch("mimetypes.guess_type", return_value=("video/mp4", None)):
with pytest.raises(LinkedInAPIError, match="Failed to upload video"):
await configured_client.upload_video("test.mp4")
class TestImageValidation:
"""Test image validation and error cases"""
@pytest.mark.asyncio
async def test_upload_image_file_not_found(self, configured_client):
"""Test image upload when file doesn't exist"""
with patch("pathlib.Path.exists", return_value=False):
with pytest.raises(LinkedInAPIError, match="File not found"):
await configured_client.upload_image("nonexistent.jpg")
@pytest.mark.asyncio
async def test_upload_image_unsupported_type(self, configured_client):
"""Test image upload with unsupported file type"""
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".bmp"):
with pytest.raises(LinkedInAPIError, match="Unsupported file type"):
await configured_client.upload_image("test.bmp")
@pytest.mark.asyncio
async def test_upload_image_file_too_large(self, configured_client):
"""Test image upload when file is too large"""
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".jpg"):
mock_stat = MagicMock()
mock_stat.st_size = 15 * 1024 * 1024 # 15MB (over 10MB limit)
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(LinkedInAPIError, match="File too large"):
await configured_client.upload_image("large.jpg")
@pytest.mark.asyncio
async def test_upload_image_http_error_init(self, configured_client):
"""Test image upload when HTTP error occurs during init"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
import httpx
mock_client.post.side_effect = httpx.HTTPError("Connection failed")
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".jpg"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(
LinkedInAPIError, match="HTTP error during upload initialization"
):
await configured_client.upload_image("test.jpg")
@pytest.mark.asyncio
async def test_upload_image_http_error_upload(
self, configured_client, mock_image_init_response
):
"""Test image upload when HTTP error occurs during file upload"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
import httpx
mock_client.post.return_value = mock_image_init_response
mock_client.put.side_effect = httpx.HTTPError("Upload failed")
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".jpg"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with patch("builtins.open", mock_open(read_data=b"fake image data")):
with patch("mimetypes.guess_type", return_value=(None, None)):
with pytest.raises(
LinkedInAPIError, match="HTTP error during file upload"
):
await configured_client.upload_image("test.jpg")
@pytest.mark.asyncio
async def test_upload_image_mime_type_fallback(
self, configured_client, mock_image_init_response, mock_upload_success_response
):
"""Test image upload with mime type fallback when guess_type returns None"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_client.post.return_value = mock_image_init_response
mock_client.put.return_value = mock_upload_success_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".jpg"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with patch("builtins.open", mock_open(read_data=b"fake image data")):
with patch("mimetypes.guess_type", return_value=(None, None)):
result = await configured_client.upload_image("test.jpg")
assert result == "urn:li:image:123456"
class TestVideoValidation:
"""Test video validation and error cases"""
@pytest.mark.asyncio
async def test_upload_video_file_not_found(self, configured_client):
"""Test video upload when file doesn't exist"""
with patch("pathlib.Path.exists", return_value=False):
with pytest.raises(LinkedInAPIError, match="File not found"):
await configured_client.upload_video("nonexistent.mp4")
@pytest.mark.asyncio
async def test_upload_video_unsupported_type(self, configured_client):
"""Test video upload with unsupported file type"""
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".avi"):
with pytest.raises(LinkedInAPIError, match="Unsupported file type"):
await configured_client.upload_video("test.avi")
@pytest.mark.asyncio
async def test_upload_video_file_too_small(self, configured_client):
"""Test video upload when file is too small"""
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 50 * 1024 # 50KB (under 75KB minimum)
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(LinkedInAPIError, match="Video too small"):
await configured_client.upload_video("small.mp4")
@pytest.mark.asyncio
async def test_upload_video_file_too_large(self, configured_client):
"""Test video upload when file is too large"""
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 600 * 1024 * 1024 # 600MB (over 500MB limit)
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(LinkedInAPIError, match="Video too large"):
await configured_client.upload_video("large.mp4")
@pytest.mark.asyncio
async def test_upload_video_no_upload_instructions(self, configured_client):
"""Test video upload when no upload instructions are returned"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"value": {
"video": "urn:li:video:123456",
"uploadInstructions": [], # Empty instructions
"uploadToken": "test-token",
}
}
mock_client.post.return_value = mock_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(LinkedInAPIError, match="No upload instructions"):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_missing_key_in_response(self, configured_client):
"""Test video upload when response is missing expected keys"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"value": {
# Missing "video" key
"uploadInstructions": [{"uploadUrl": "https://example.com"}],
"uploadToken": "test-token",
}
}
mock_client.post.return_value = mock_response
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(LinkedInAPIError, match="Unexpected response structure"):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_http_error_init(self, configured_client):
"""Test video upload when HTTP error occurs during init"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
import httpx
mock_client.post.side_effect = httpx.HTTPError("Connection failed")
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with pytest.raises(
LinkedInAPIError, match="HTTP error during upload initialization"
):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_http_error_upload(
self, configured_client, mock_video_init_response
):
"""Test video upload when HTTP error occurs during file upload"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
import httpx
mock_client.post.return_value = mock_video_init_response
mock_client.put.side_effect = httpx.HTTPError("Upload failed")
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with pytest.raises(
LinkedInAPIError, match="HTTP error during file upload"
):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_http_error_finalize(
self, configured_client, mock_video_init_response
):
"""Test video upload when HTTP error occurs during finalization"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_upload_response = MagicMock()
mock_upload_response.status_code = 201
mock_upload_response.headers = {"ETag": '"test-etag"'}
import httpx
mock_client.put.return_value = mock_upload_response
# First post is init (succeeds), second post is finalize (fails with HTTPError)
mock_client.post.side_effect = [
mock_video_init_response,
httpx.HTTPError("Finalize failed"),
]
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with pytest.raises(
LinkedInAPIError, match="HTTP error during video finalization"
):
await configured_client.upload_video("test.mp4")
@pytest.mark.asyncio
async def test_upload_video_finalize_fails(self, configured_client, mock_video_init_response):
"""Test video upload when finalization returns error status"""
with patch("httpx.AsyncClient") as mock_client_class:
mock_client = AsyncMock()
mock_upload_response = MagicMock()
mock_upload_response.status_code = 201
mock_upload_response.headers = {"ETag": '"test-etag"'}
mock_finalize_response = MagicMock()
mock_finalize_response.status_code = 400
mock_finalize_response.text = "Finalization failed"
mock_client.put.return_value = mock_upload_response
mock_client.post.side_effect = [mock_video_init_response, mock_finalize_response]
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
with patch("pathlib.Path.exists", return_value=True):
with patch("pathlib.Path.suffix", new_callable=lambda: ".mp4"):
mock_stat = MagicMock()
mock_stat.st_size = 1024 * 1024
with patch("pathlib.Path.stat", return_value=mock_stat):
with patch("builtins.open", mock_open(read_data=b"fake video data")):
with pytest.raises(
LinkedInAPIError, match="Failed to finalize video upload"
):
await configured_client.upload_video("test.mp4")