pi-weekly-newspaper/tests/test_scheduler.py

import json
import os
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo
from unittest.mock import MagicMock, patch

from PIL import Image as PILImage

from src.models import Article, Image
from src.scheduler import SchedulerManager


def test_scheduler_starts_fetch_job(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        mgr.start()
        jobs = mgr.scheduler.get_jobs()
        job_ids = [j.id for j in jobs]
        assert "rss_fetch" in job_ids
        mgr.shutdown()


def test_scheduler_update_fetch_interval(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        mgr.start()
        mgr.update_fetch_interval(2)
        job = mgr.scheduler.get_job("rss_fetch")
        assert job is not None
        assert job.trigger.interval.total_seconds() == 7200
        mgr.shutdown()


def test_scheduler_enable_auto_publish(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        mgr.start()
        mgr.enable_auto_publish(day_of_week="sun", hour=6, minute=0,
                                cover_method="text")
        job = mgr.scheduler.get_job("auto_publish")
        assert job is not None
        mgr.shutdown()


def test_scheduler_disable_auto_publish(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        mgr.start()
        mgr.enable_auto_publish(day_of_week="sun", hour=6, minute=0,
                                cover_method="text")
        mgr.disable_auto_publish()
        job = mgr.scheduler.get_job("auto_publish")
        assert job is None
        mgr.shutdown()


def test_scheduler_get_status(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        mgr.start()
        status = mgr.get_status()
        assert status["running"] is True
        assert "rss_fetch" in status
        mgr.shutdown()


def test_scheduler_starts_on_half_hour(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        
        # Test when current time is before :30
        with patch('src.scheduler.datetime') as mock_dt:
            # Use a date far in the future so APScheduler doesn't fast-forward it
            mock_dt.now.return_value = datetime(2030, 4, 6, 10, 15, 0)
            mgr.start()
            job = mgr.scheduler.get_job("rss_fetch")
            assert job.next_run_time.minute == 30
            assert job.next_run_time.hour == 10
            mgr.shutdown()
            
        mgr = SchedulerManager(app)
        # Test when current time is after :30
        with patch('src.scheduler.datetime') as mock_dt:
            mock_dt.now.return_value = datetime(2030, 4, 6, 10, 45, 0)
            mgr.start()
            job = mgr.scheduler.get_job("rss_fetch")
            assert job.next_run_time.minute == 0
            assert job.next_run_time.hour == 11
            mgr.shutdown()


def test_scheduler_get_status_formatting(app):
    with app.app_context():
        mgr = SchedulerManager(app)
        with patch('src.scheduler.datetime') as mock_dt:
            # Force it to start at exactly 10:30 in the future
            mock_dt.now.return_value = datetime(2030, 4, 6, 10, 15, 0)
            mgr.start()
            
            # The job's next_run_time will be timezone aware in APScheduler
            status = mgr.get_status()
            
            assert "rss_fetch" in status
            # Should look like "Apr 06, 2030 10:30 AM"
            assert "Apr 06, 2030 10:30 AM" in status["rss_fetch"]["next_run"]
            mgr.shutdown()


def test_auto_publish_passes_ordered_image_paths_to_generate_cover(app, tmp_path, db):
    """First image per article in pub_date order passed to generate_cover."""
    os.makedirs(tmp_path, exist_ok=True)
    img1 = str(tmp_path / "a1.jpg")
    img2 = str(tmp_path / "a2.jpg")
    for p in (img1, img2):
        PILImage.new("RGB", (100, 100), color="blue").save(p, format="JPEG")

    with app.app_context():
        a1 = Article(
            guid="g1",
            title="Earlier Article",
            author="A",
            pub_date=datetime(2026, 4, 6, 10, 0),
            categories=json.dumps(["Government"]),
            link="http://example.com/1",
            content_html="<p>x</p>",
        )
        a2 = Article(
            guid="g2",
            title="Later Article",
            author="B",
            pub_date=datetime(2026, 4, 7, 10, 0),
            categories=json.dumps(["Culture"]),
            link="http://example.com/2",
            content_html="<p>y</p>",
        )
        db.session.add_all([a1, a2])
        db.session.flush()
        db.session.add_all(
            [
                Image(
                    article_id=a1.id,
                    original_url="https://example.com/1.jpg",
                    local_path=img1,
                    width=100,
                    height=100,
                ),
                Image(
                    article_id=a2.id,
                    original_url="https://example.com/2.jpg",
                    local_path=img2,
                    width=100,
                    height=100,
                ),
            ]
        )
        db.session.commit()

    mock_cover = MagicMock(return_value="/fake/cover.jpg")
    mock_epub = MagicMock(return_value="/fake/issue.epub")

    with patch("src.scheduler.date") as mock_date, patch(
        "src.cover.generate_cover", mock_cover
    ), patch("src.epub_builder.build_epub", mock_epub):
        mock_date.today.return_value = date(2026, 4, 6)
        mgr = SchedulerManager(app)
        mgr._run_auto_publish()

    mock_cover.assert_called_once()
    args = mock_cover.call_args[0]
    assert args[0] == "text"
    assert args[4] == ["Earlier Article", "Later Article"]
    assert args[5] == [img1, img2]