itsulu-blog-publisher/addons/itsulu_blog_publisher/tests/test_blog_schedule.py

# addons/itsulu_blog_publisher/tests/test_blog_schedule.py
"""
Tests for models/blog_schedule.py
Behaviour: three configurable cron slots; each picks a topic, calls the LLM,
           creates a blog post, respects auto-publish, records a log entry.

RED PHASE — all tests FAIL until itsulu.blog.schedule model exists.
"""
from unittest.mock import patch, MagicMock
from odoo.tests import TransactionCase, tagged
from .factories import BlogPublisherFactory


@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'blog_schedule')
class TestBlogScheduleConfiguration(TransactionCase):
    """Verify that schedule slot records are configured correctly."""

    def setUp(self):
        super().setUp()
        self.factory = BlogPublisherFactory(self.env)
        self.blog = self.factory.blog(name='ITSulu Insights')

    def test_schedule_slot_is_created_with_correct_defaults(self):
        """
        1. Behaviour: a schedule slot stores its trigger time, blog, provider, and auto-publish flag
        2. ARRANGE: factory with explicit values
        3. ACT: create record
        4. ASSERT: all fields match
        5. FAIL reason: itsulu.blog.schedule model does not exist
        """
        # ACT
        slot = self.factory.blog_schedule(
            blog=self.blog,
            slot='morning',
            trigger_time=8.0,
            llm_provider='anthropic',
            llm_model='claude-sonnet-4-20250514',
            auto_publish=True,
        )

        # ASSERT
        self.assertEqual(slot.slot, 'morning')
        self.assertAlmostEqual(slot.trigger_time, 8.0)
        self.assertEqual(slot.blog_id.id, self.blog.id)
        self.assertEqual(slot.llm_provider, 'anthropic')
        self.assertTrue(slot.auto_publish)
        self.assertTrue(slot.active)

    def test_disabled_schedule_slot_has_active_false(self):
        """Setting active=False on a slot means it is skipped by the scheduler."""
        # ACT
        slot = self.factory.blog_schedule(blog=self.blog, active=False)

        # ASSERT
        self.assertFalse(slot.active)

    def test_three_distinct_slot_values_are_valid(self):
        """morning, afternoon, and evening are the three valid slot names."""
        for slot_name in ('morning', 'afternoon', 'evening'):
            slot = self.factory.blog_schedule(blog=self.blog, slot=slot_name)
            self.assertEqual(slot.slot, slot_name)


@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'blog_schedule')
class TestBlogScheduleExecution(TransactionCase):
    """
    Verify that schedule slot execution creates a blog post and log.
    LLM calls are mocked — we are testing orchestration, not the LLM.
    """

    def setUp(self):
        super().setUp()
        self.factory = BlogPublisherFactory(self.env)
        self.blog = self.factory.blog(name='ITSulu Insights')
        self.env['ir.config_parameter'].sudo().set_param(
            'itsulu_blog_publisher.anthropic_api_key', 'sk-ant-test-key'
        )

    def _make_mock_llm_response(self):
        resp = MagicMock()
        resp.text = (
            '<h1>AI Governance in 2026</h1>'
            '<p>' + ('Enterprise AI paragraph. ' * 30) + '</p>'
        )
        # _create_blog_post writes body_html into blog.post.content; it must be a
        # real string. On Odoo 14 an unset MagicMock attr fails psycopg2 adaptation
        # ("can't adapt type 'MagicMock'"); on 17 it silently stringified.
        resp.body_html = resp.text
        resp.raw_text = resp.text
        resp.tokens_used = 900
        resp.title = 'AI Governance in 2026'
        resp.meta_title = 'AI Governance Frameworks for Enterprises 2026'
        resp.meta_description = 'Learn how leading enterprises implement AI governance in 2026.'
        resp.meta_keywords = 'AI governance, enterprise AI, compliance'
        resp.tags = ['AI Governance', 'Enterprise AI', 'Digital Transformation']
        resp.social = MagicMock(
            twitter_a='Short X post A.',
            twitter_b='Short X post B.',
            bluesky_a='BlueSky post A.',
            bluesky_b='BlueSky post B.',
            mastodon='Mastodon post.',
            linkedin='LinkedIn long-form post about AI governance.',
        )
        return resp

    def test_active_slot_run_creates_blog_post(self):
        """
        Calling slot.run_generation() creates exactly one blog.post
        linked to the slot's configured blog.
        """
        # ARRANGE
        slot = self.factory.blog_schedule(blog=self.blog, slot='morning', active=True)
        post_count_before = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ):
            # ACT
            slot.run_generation()

        # ASSERT
        post_count_after = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )
        self.assertEqual(post_count_after, post_count_before + 1)

    def test_active_slot_run_creates_generation_log_with_success_state(self):
        """run_generation() on an active slot writes a success log entry."""
        # ARRANGE
        slot = self.factory.blog_schedule(blog=self.blog, slot='afternoon', active=True)

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ):
            slot.run_generation()

        # ASSERT
        log = self.env['itsulu.blog.generation.log'].search(
            [('trigger_source', '=', 'scheduled'), ('schedule_slot', '=', 'afternoon')],
            limit=1,
        )
        self.assertTrue(log, "A generation log entry must exist for the afternoon slot")
        self.assertEqual(log.state, 'success')

    def test_inactive_slot_run_does_not_create_blog_post(self):
        """Calling run_generation() on an inactive slot is a no-op."""
        # ARRANGE
        slot = self.factory.blog_schedule(blog=self.blog, slot='evening', active=False)
        post_count_before = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ) as mock_generate:
            slot.run_generation()

        # ASSERT
        mock_generate.assert_not_called()
        post_count_after = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )
        self.assertEqual(post_count_after, post_count_before)

    def test_slot_picks_next_topic_from_queue_when_available(self):
        """run_generation() uses the next pending topic from the queue."""
        # ARRANGE
        topic = self.factory.blog_topic(
            name='Kubernetes Cost Optimisation', priority='urgent', state='pending'
        )
        slot = self.factory.blog_schedule(blog=self.blog, active=True)
        mock_resp = self._make_mock_llm_response()

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=mock_resp,
        ) as mock_gen:
            slot.run_generation()

        # ASSERT — the topic text was passed to the LLM
        self.assertTrue(mock_gen.called, "LLMRouter.generate() must be called")
        call_kwargs = mock_gen.call_args[1]
        topic_used = call_kwargs.get('topic', '')
        self.assertIn('Kubernetes Cost Optimisation', topic_used)
        # And the topic is now marked used
        self.assertEqual(topic.state, 'used')

    def test_slot_falls_back_to_llm_chosen_topic_when_queue_is_empty(self):
        """When queue is empty, generation still runs with an LLM-chosen topic."""
        # ARRANGE — clear all pending topics
        self.env['itsulu.blog.topic'].search([('state', '=', 'pending')]).write(
            {'state': 'used'}
        )
        slot = self.factory.blog_schedule(blog=self.blog, active=True)
        post_count_before = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ):
            slot.run_generation()

        # ASSERT — a post was still created
        post_count_after = self.env['blog.post'].search_count(
            [('blog_id', '=', self.blog.id)]
        )
        self.assertGreater(post_count_after, post_count_before)
        # And the log notes the fallback
        log = self.env['itsulu.blog.generation.log'].search(
            [('trigger_source', '=', 'scheduled')], limit=1
        )
        self.assertIn('llm', (log.topic_source or '').lower())

    def test_auto_publish_true_publishes_the_created_blog_post(self):
        """When auto_publish=True on the slot, the created post is published immediately."""
        # ARRANGE
        slot = self.factory.blog_schedule(blog=self.blog, active=True, auto_publish=True)

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ):
            slot.run_generation()

        # ASSERT
        latest_post = self.env['blog.post'].search(
            [('blog_id', '=', self.blog.id)], order='id desc', limit=1
        )
        self.assertTrue(latest_post.is_published,
                        "Post must be published when auto_publish=True")

    def test_auto_publish_false_leaves_the_created_blog_post_as_draft(self):
        """When auto_publish=False, the created post is left as a draft."""
        # ARRANGE
        slot = self.factory.blog_schedule(blog=self.blog, active=True, auto_publish=False)

        with patch(
            'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
            return_value=self._make_mock_llm_response(),
        ):
            slot.run_generation()

        # ASSERT
        latest_post = self.env['blog.post'].search(
            [('blog_id', '=', self.blog.id)], order='id desc', limit=1
        )
        self.assertFalse(latest_post.is_published,
                         "Post must remain draft when auto_publish=False")