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>
260 lines
9.7 KiB
Python
260 lines
9.7 KiB
Python
"""
|
|
Tests for ContentScheduler service.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
class TestContentScheduler:
|
|
"""Tests for the ContentScheduler class."""
|
|
|
|
@pytest.fixture
|
|
def mock_db_session(self):
|
|
"""Create a mock database session."""
|
|
mock_session = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.first.return_value = None # No existing posts
|
|
mock_query.all.return_value = []
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_session.query.return_value = mock_query
|
|
return mock_session
|
|
|
|
@pytest.fixture
|
|
def scheduler(self, mock_db_session):
|
|
"""Create a ContentScheduler with mocked database."""
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
from app.services.scheduler import ContentScheduler
|
|
return ContentScheduler()
|
|
|
|
def test_init(self, scheduler):
|
|
"""Test scheduler initialization."""
|
|
assert scheduler.posting_times is not None
|
|
assert "x" in scheduler.posting_times
|
|
assert "threads" in scheduler.posting_times
|
|
|
|
def test_get_next_available_slot_weekday(self, scheduler, mock_db_session, fixed_datetime):
|
|
"""Test getting next slot on a weekday."""
|
|
# Monday 10:00
|
|
weekday = datetime(2024, 6, 17, 10, 0, 0)
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.get_next_available_slot("x", after=weekday)
|
|
|
|
assert result is not None
|
|
assert result > weekday
|
|
|
|
def test_get_next_available_slot_weekend(self, scheduler, mock_db_session):
|
|
"""Test getting next slot on a weekend."""
|
|
# Saturday 10:00
|
|
weekend = datetime(2024, 6, 15, 10, 0, 0)
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.get_next_available_slot("x", after=weekend)
|
|
|
|
assert result is not None
|
|
|
|
def test_get_next_available_slot_late_night(self, scheduler, mock_db_session):
|
|
"""Test that late night moves to next day."""
|
|
# 11 PM
|
|
late_night = datetime(2024, 6, 17, 23, 0, 0)
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.get_next_available_slot("x", after=late_night)
|
|
|
|
# Should be next day
|
|
assert result.date() > late_night.date()
|
|
|
|
def test_get_available_slots(self, scheduler, mock_db_session, fixed_datetime):
|
|
"""Test getting all available slots."""
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
slots = scheduler.get_available_slots(
|
|
platform="x",
|
|
start_date=fixed_datetime,
|
|
days=3
|
|
)
|
|
|
|
assert len(slots) > 0
|
|
for slot in slots:
|
|
assert slot.platform == "x"
|
|
assert slot.available is True
|
|
|
|
def test_schedule_post(self, scheduler, mock_db_session):
|
|
"""Test scheduling a post."""
|
|
mock_post = MagicMock()
|
|
mock_post.id = 1
|
|
mock_post.platforms = ["x"]
|
|
mock_post.status = "draft"
|
|
|
|
mock_db_session.query.return_value.filter.return_value.first.return_value = mock_post
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.schedule_post(
|
|
post_id=1,
|
|
scheduled_at=datetime(2024, 6, 20, 12, 0, 0)
|
|
)
|
|
|
|
assert result == datetime(2024, 6, 20, 12, 0, 0)
|
|
assert mock_post.status == "scheduled"
|
|
mock_db_session.commit.assert_called_once()
|
|
|
|
def test_schedule_post_auto_time(self, scheduler, mock_db_session):
|
|
"""Test scheduling with auto-selected time."""
|
|
mock_post = MagicMock()
|
|
mock_post.id = 1
|
|
mock_post.platforms = ["x"]
|
|
|
|
# First call returns the post, second returns None (no conflicts)
|
|
mock_db_session.query.return_value.filter.return_value.first.side_effect = [
|
|
mock_post, None, None, None, None, None, None
|
|
]
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.schedule_post(post_id=1)
|
|
|
|
assert result is not None
|
|
assert mock_post.scheduled_at is not None
|
|
|
|
def test_schedule_post_not_found(self, scheduler, mock_db_session):
|
|
"""Test scheduling a non-existent post."""
|
|
mock_db_session.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
scheduler.schedule_post(post_id=999)
|
|
|
|
def test_reschedule_post(self, scheduler, mock_db_session):
|
|
"""Test rescheduling a post."""
|
|
mock_post = MagicMock()
|
|
mock_post.id = 1
|
|
mock_post.status = "scheduled"
|
|
|
|
mock_db_session.query.return_value.filter.return_value.first.return_value = mock_post
|
|
|
|
new_time = datetime(2024, 6, 25, 14, 0, 0)
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.reschedule_post(post_id=1, new_time=new_time)
|
|
|
|
assert result is True
|
|
assert mock_post.scheduled_at == new_time
|
|
mock_db_session.commit.assert_called_once()
|
|
|
|
def test_reschedule_published_post_fails(self, scheduler, mock_db_session):
|
|
"""Test that published posts cannot be rescheduled."""
|
|
mock_post = MagicMock()
|
|
mock_post.status = "published"
|
|
|
|
mock_db_session.query.return_value.filter.return_value.first.return_value = mock_post
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.reschedule_post(
|
|
post_id=1,
|
|
new_time=datetime(2024, 6, 25, 14, 0, 0)
|
|
)
|
|
|
|
assert result is False
|
|
|
|
def test_cancel_scheduled(self, scheduler, mock_db_session):
|
|
"""Test canceling a scheduled post."""
|
|
mock_post = MagicMock()
|
|
mock_post.id = 1
|
|
mock_post.status = "scheduled"
|
|
mock_post.scheduled_at = datetime(2024, 6, 20, 12, 0, 0)
|
|
|
|
mock_db_session.query.return_value.filter.return_value.first.return_value = mock_post
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.cancel_scheduled(post_id=1)
|
|
|
|
assert result is True
|
|
assert mock_post.status == "draft"
|
|
assert mock_post.scheduled_at is None
|
|
|
|
def test_get_calendar(self, scheduler, mock_db_session):
|
|
"""Test getting calendar view."""
|
|
mock_posts = [
|
|
MagicMock(
|
|
id=1,
|
|
content="Test post 1",
|
|
platforms=["x"],
|
|
status="scheduled",
|
|
scheduled_at=datetime(2024, 6, 17, 12, 0, 0),
|
|
content_type="tip"
|
|
),
|
|
MagicMock(
|
|
id=2,
|
|
content="Test post 2",
|
|
platforms=["threads"],
|
|
status="scheduled",
|
|
scheduled_at=datetime(2024, 6, 17, 14, 0, 0),
|
|
content_type="product"
|
|
)
|
|
]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter.return_value = mock_query
|
|
mock_query.order_by.return_value = mock_query
|
|
mock_query.all.return_value = mock_posts
|
|
mock_db_session.query.return_value = mock_query
|
|
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
result = scheduler.get_calendar(
|
|
start_date=datetime(2024, 6, 15),
|
|
end_date=datetime(2024, 6, 20)
|
|
)
|
|
|
|
assert "2024-06-17" in result
|
|
assert len(result["2024-06-17"]) == 2
|
|
|
|
def test_auto_fill_calendar(self, scheduler, mock_db_session, fixed_datetime):
|
|
"""Test auto-filling calendar with suggested slots."""
|
|
with patch('app.services.scheduler.SessionLocal', return_value=mock_db_session):
|
|
slots = scheduler.auto_fill_calendar(
|
|
start_date=fixed_datetime,
|
|
days=3,
|
|
platforms=["x", "threads"]
|
|
)
|
|
|
|
assert len(slots) > 0
|
|
# Should be sorted by datetime
|
|
for i in range(1, len(slots)):
|
|
assert slots[i].datetime >= slots[i-1].datetime
|
|
|
|
|
|
class TestOptimalTimes:
|
|
"""Tests for optimal posting times configuration."""
|
|
|
|
def test_x_has_weekday_times(self):
|
|
"""Test that X platform has weekday times defined."""
|
|
from app.data.content_templates import OPTIMAL_POSTING_TIMES
|
|
|
|
assert "x" in OPTIMAL_POSTING_TIMES
|
|
assert "weekday" in OPTIMAL_POSTING_TIMES["x"]
|
|
assert len(OPTIMAL_POSTING_TIMES["x"]["weekday"]) > 0
|
|
|
|
def test_all_platforms_have_times(self):
|
|
"""Test all platforms have posting times."""
|
|
from app.data.content_templates import OPTIMAL_POSTING_TIMES
|
|
|
|
expected_platforms = ["x", "threads", "instagram", "facebook"]
|
|
|
|
for platform in expected_platforms:
|
|
assert platform in OPTIMAL_POSTING_TIMES
|
|
assert "weekday" in OPTIMAL_POSTING_TIMES[platform]
|
|
assert "weekend" in OPTIMAL_POSTING_TIMES[platform]
|
|
|
|
def test_time_format(self):
|
|
"""Test that times are in correct HH:MM format."""
|
|
from app.data.content_templates import OPTIMAL_POSTING_TIMES
|
|
|
|
import re
|
|
time_pattern = re.compile(r'^([01]?[0-9]|2[0-3]):[0-5][0-9]$')
|
|
|
|
for platform, times in OPTIMAL_POSTING_TIMES.items():
|
|
for day_type in ["weekday", "weekend"]:
|
|
for time_str in times.get(day_type, []):
|
|
assert time_pattern.match(time_str), f"Invalid time format: {time_str}"
|