Consultoría AS
85bda6abcf
Testing: - Add pytest configuration (pytest.ini) - Add test fixtures (tests/conftest.py) - Add ContentGenerator tests (13 tests) - Add ContentScheduler tests (16 tests) - Add PublisherManager tests (16 tests) - All 45 tests passing Production Docker: - Add docker-compose.prod.yml with healthchecks, resource limits - Add Dockerfile.prod with multi-stage build, non-root user - Add nginx.prod.conf with SSL, rate limiting, security headers - Add .env.prod.example template Maintenance Scripts: - Add backup.sh for database and media backups - Add restore.sh for database restoration - Add cleanup.sh for log rotation and Docker cleanup - Add healthcheck.sh with Telegram alerts Documentation: - Add DEPLOY.md with complete deployment guide Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
279 lines
9.6 KiB
Python
279 lines
9.6 KiB
Python
"""
|
|
Tests for PublisherManager service.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
|
|
|
|
class TestPublisherManager:
|
|
"""Tests for the PublisherManager class."""
|
|
|
|
@pytest.fixture
|
|
def mock_publishers(self):
|
|
"""Create mock publishers."""
|
|
x_publisher = MagicMock()
|
|
x_publisher.char_limit = 280
|
|
x_publisher.validate_content.return_value = True
|
|
x_publisher.client = MagicMock()
|
|
x_publisher.publish = AsyncMock(return_value=MagicMock(
|
|
success=True, post_id="123", url="https://x.com/post/123"
|
|
))
|
|
|
|
threads_publisher = MagicMock()
|
|
threads_publisher.char_limit = 500
|
|
threads_publisher.validate_content.return_value = True
|
|
threads_publisher.access_token = "token"
|
|
threads_publisher.user_id = "user123"
|
|
threads_publisher.publish = AsyncMock(return_value=MagicMock(
|
|
success=True, post_id="456", url="https://threads.net/post/456"
|
|
))
|
|
|
|
fb_publisher = MagicMock()
|
|
fb_publisher.char_limit = 63206
|
|
fb_publisher.validate_content.return_value = True
|
|
fb_publisher.access_token = "token"
|
|
fb_publisher.page_id = "page123"
|
|
fb_publisher.publish = AsyncMock(return_value=MagicMock(
|
|
success=True, post_id="789"
|
|
))
|
|
|
|
ig_publisher = MagicMock()
|
|
ig_publisher.char_limit = 2200
|
|
ig_publisher.validate_content.return_value = True
|
|
ig_publisher.access_token = "token"
|
|
ig_publisher.account_id = "acc123"
|
|
ig_publisher.publish = AsyncMock(return_value=MagicMock(
|
|
success=True, post_id="101"
|
|
))
|
|
|
|
return {
|
|
"x": x_publisher,
|
|
"threads": threads_publisher,
|
|
"facebook": fb_publisher,
|
|
"instagram": ig_publisher
|
|
}
|
|
|
|
@pytest.fixture
|
|
def manager(self, mock_publishers):
|
|
"""Create a PublisherManager with mocked publishers."""
|
|
with patch('app.publishers.manager.XPublisher', return_value=mock_publishers["x"]), \
|
|
patch('app.publishers.manager.ThreadsPublisher', return_value=mock_publishers["threads"]), \
|
|
patch('app.publishers.manager.FacebookPublisher', return_value=mock_publishers["facebook"]), \
|
|
patch('app.publishers.manager.InstagramPublisher', return_value=mock_publishers["instagram"]):
|
|
|
|
from app.publishers.manager import PublisherManager, Platform
|
|
mgr = PublisherManager()
|
|
# Override with mocks
|
|
mgr._publishers = {
|
|
Platform.X: mock_publishers["x"],
|
|
Platform.THREADS: mock_publishers["threads"],
|
|
Platform.FACEBOOK: mock_publishers["facebook"],
|
|
Platform.INSTAGRAM: mock_publishers["instagram"],
|
|
}
|
|
return mgr
|
|
|
|
def test_init(self, manager):
|
|
"""Test manager initialization."""
|
|
assert manager._publishers is not None
|
|
assert len(manager._publishers) == 4
|
|
|
|
def test_get_publisher(self, manager):
|
|
"""Test getting a specific publisher."""
|
|
from app.publishers.manager import Platform
|
|
|
|
publisher = manager.get_publisher(Platform.X)
|
|
assert publisher is not None
|
|
|
|
def test_get_available_platforms(self, manager):
|
|
"""Test getting available platforms."""
|
|
available = manager.get_available_platforms()
|
|
|
|
assert isinstance(available, list)
|
|
assert "x" in available
|
|
assert "threads" in available
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_single_platform(self, manager):
|
|
"""Test publishing to a single platform."""
|
|
from app.publishers.manager import Platform
|
|
|
|
result = await manager.publish(
|
|
platform=Platform.X,
|
|
content="Test post #Testing"
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.post_id == "123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_content_too_long(self, manager, mock_publishers):
|
|
"""Test that too long content fails validation."""
|
|
from app.publishers.manager import Platform
|
|
|
|
mock_publishers["x"].validate_content.return_value = False
|
|
|
|
result = await manager.publish(
|
|
platform=Platform.X,
|
|
content="x" * 300 # Exceeds 280 limit
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "límite" in result.error_message.lower() or "excede" in result.error_message.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_unsupported_platform(self, manager):
|
|
"""Test publishing to unsupported platform."""
|
|
result = await manager.publish(
|
|
platform=MagicMock(value="unsupported"),
|
|
content="Test"
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "no soportada" in result.error_message.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_to_multiple_parallel(self, manager):
|
|
"""Test publishing to multiple platforms in parallel."""
|
|
from app.publishers.manager import Platform
|
|
|
|
result = await manager.publish_to_multiple(
|
|
platforms=[Platform.X, Platform.THREADS],
|
|
content="Multi-platform test #Test",
|
|
parallel=True
|
|
)
|
|
|
|
assert result.success is True
|
|
assert len(result.successful_platforms) >= 1
|
|
assert "x" in result.results
|
|
assert "threads" in result.results
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_to_multiple_sequential(self, manager):
|
|
"""Test publishing to multiple platforms sequentially."""
|
|
from app.publishers.manager import Platform
|
|
|
|
result = await manager.publish_to_multiple(
|
|
platforms=[Platform.X, Platform.FACEBOOK],
|
|
content="Sequential test",
|
|
parallel=False
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_to_multiple_with_dict_content(self, manager):
|
|
"""Test publishing with platform-specific content."""
|
|
from app.publishers.manager import Platform
|
|
|
|
content = {
|
|
"x": "Short post for X #X",
|
|
"threads": "Longer post for Threads with more details #Threads"
|
|
}
|
|
|
|
result = await manager.publish_to_multiple(
|
|
platforms=[Platform.X, Platform.THREADS],
|
|
content=content
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_with_image_meta_platforms(self, manager, mock_publishers):
|
|
"""Test that Meta platforms get public image URL."""
|
|
from app.publishers.manager import Platform
|
|
|
|
with patch('app.publishers.manager.image_upload') as mock_upload:
|
|
mock_upload.upload_from_path = AsyncMock(
|
|
return_value="https://imgbb.com/image.jpg"
|
|
)
|
|
|
|
result = await manager.publish(
|
|
platform=Platform.THREADS,
|
|
content="Post with image",
|
|
image_path="/local/image.jpg"
|
|
)
|
|
|
|
mock_upload.upload_from_path.assert_called_once_with("/local/image.jpg")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_connection(self, manager, mock_publishers):
|
|
"""Test connection testing."""
|
|
from app.publishers.manager import Platform
|
|
|
|
mock_publishers["x"].client.get_me.return_value = MagicMock(
|
|
data=MagicMock(username="testuser", name="Test", id=123)
|
|
)
|
|
|
|
result = await manager.test_connection(Platform.X)
|
|
|
|
assert result["platform"] == "x"
|
|
assert result["configured"] is True
|
|
assert result["connected"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_all_connections(self, manager, mock_publishers):
|
|
"""Test testing all connections."""
|
|
mock_publishers["x"].client.get_me.return_value = MagicMock(
|
|
data=MagicMock(username="test", name="Test", id=123)
|
|
)
|
|
|
|
results = await manager.test_all_connections()
|
|
|
|
assert len(results) == 4
|
|
assert "x" in results
|
|
assert "threads" in results
|
|
|
|
|
|
class TestMultiPublishResult:
|
|
"""Tests for MultiPublishResult dataclass."""
|
|
|
|
def test_successful_platforms(self):
|
|
"""Test getting successful platforms."""
|
|
from app.publishers.manager import MultiPublishResult
|
|
from app.publishers.base import PublishResult
|
|
|
|
results = {
|
|
"x": PublishResult(success=True, post_id="123"),
|
|
"threads": PublishResult(success=False, error_message="Error"),
|
|
"facebook": PublishResult(success=True, post_id="456")
|
|
}
|
|
|
|
multi = MultiPublishResult(success=True, results=results, errors=[])
|
|
|
|
assert set(multi.successful_platforms) == {"x", "facebook"}
|
|
|
|
def test_failed_platforms(self):
|
|
"""Test getting failed platforms."""
|
|
from app.publishers.manager import MultiPublishResult
|
|
from app.publishers.base import PublishResult
|
|
|
|
results = {
|
|
"x": PublishResult(success=True, post_id="123"),
|
|
"threads": PublishResult(success=False, error_message="Error"),
|
|
}
|
|
|
|
multi = MultiPublishResult(success=True, results=results, errors=[])
|
|
|
|
assert multi.failed_platforms == ["threads"]
|
|
|
|
|
|
class TestPlatformEnum:
|
|
"""Tests for Platform enum."""
|
|
|
|
def test_platform_values(self):
|
|
"""Test platform enum values."""
|
|
from app.publishers.manager import Platform
|
|
|
|
assert Platform.X.value == "x"
|
|
assert Platform.THREADS.value == "threads"
|
|
assert Platform.FACEBOOK.value == "facebook"
|
|
assert Platform.INSTAGRAM.value == "instagram"
|
|
|
|
def test_platform_is_string_enum(self):
|
|
"""Test platform enum is string."""
|
|
from app.publishers.manager import Platform
|
|
|
|
assert isinstance(Platform.X, str)
|
|
assert Platform.X == "x"
|