Test body content for blog post.
', + 'website_meta_title': '', + 'website_meta_description': '', + 'website_meta_keywords': '', + } + defaults.update(kw) + return self.env['blog.post'].create(defaults) + + # ------------------------------------------------------------------ # + # itsulu_blog_publisher models # + # ------------------------------------------------------------------ # + + def blog_topic(self, **kw): + """itsulu.blog.topic — a planned topic in the queue.""" + defaults = { + 'name': f'Test Topic {self._seq()}', + 'priority': 'normal', + 'state': 'pending', + 'notes': '', + } + defaults.update(kw) + return self.env['itsulu.blog.topic'].create(defaults) + + def blog_schedule(self, blog=None, **kw): + """itsulu.blog.schedule — a cron slot configuration.""" + blog = blog or self.blog() + defaults = { + 'name': f'Test Schedule {self._seq()}', + 'slot': 'morning', + 'trigger_time': 8.0, # 08:00 + 'active': True, + 'blog_id': blog.id, + 'llm_provider': 'anthropic', + 'llm_model': 'claude-sonnet-4-20250514', + 'auto_publish': True, + } + defaults.update(kw) + return self.env['itsulu.blog.schedule'].create(defaults) + + def generation_log(self, blog_post=None, **kw): + """itsulu.blog.generation.log — one attempt record.""" + defaults = { + 'state': 'success', + 'trigger_source': 'manual', + 'llm_provider': 'anthropic', + 'llm_model': 'claude-sonnet-4-20250514', + 'tokens_used': 1500, + 'duration_seconds': 12.0, + 'error_message': '', + 'blog_post_id': blog_post.id if blog_post else False, + } + defaults.update(kw) + return self.env['itsulu.blog.generation.log'].create(defaults) + + def blog_post_social(self, blog_post=None, **kw): + """itsulu.blog.post.social — social media copy for one post.""" + blog_post = blog_post or self.blog_post() + defaults = { + 'blog_post_id': blog_post.id, + 'twitter_post_a': 'Twitter copy A placeholder.', + 'twitter_post_b': 'Twitter copy B placeholder.', + 'bluesky_post_a': 'BlueSky copy A placeholder.', + 'bluesky_post_b': 'BlueSky copy B placeholder.', + 'mastodon_post': 'Mastodon copy placeholder.', + 'linkedin_post': 'LinkedIn copy placeholder.', + 'sources_referenced': '', + } + defaults.update(kw) + return self.env['itsulu.blog.post.social'].create(defaults) diff --git a/test_bdd_steps.py b/test_bdd_steps.py new file mode 100644 index 0000000..94ae04b --- /dev/null +++ b/test_bdd_steps.py @@ -0,0 +1,362 @@ +# addons/itsulu_blog_publisher/tests/test_bdd_steps.py +""" +pytest-bdd step definitions for itsulu_blog_publisher feature files. +Links: blog_generation.feature, blog_scheduling.feature, + llm_provider_selection.feature, seo_population.feature, + notification_email.feature + +RED PHASE — all scenarios FAIL until implementation exists. +""" +import pytest +from unittest.mock import patch, MagicMock +from pytest_bdd import scenarios, given, when, then, parsers + +# Link all feature files +scenarios('../features/blog_generation.feature') +scenarios('../features/blog_scheduling.feature') +scenarios('../features/llm_provider_selection.feature') +scenarios('../features/seo_population.feature') +scenarios('../features/notification_email.feature') + + +# ================================================================= # +# Shared fixtures # +# ================================================================= # + +@pytest.fixture +def ctx(): + """Mutable context bag shared between steps in one scenario.""" + return {} + + +@pytest.fixture +def odoo_env(request): + """Provide the Odoo environment from the TransactionCase if available, + or from the pytest-odoo plugin's odoo_env fixture.""" + return request.getfixturevalue('odoo_env') + + +# ================================================================= # +# Feature: blog_generation — On-Demand Blog Generation # +# ================================================================= # + +@given('the Anthropic API key is configured in Settings') +def given_anthropic_api_key_configured(odoo_env): + odoo_env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', 'sk-ant-test-key' + ) + + +@given(parsers.parse('the blog "{blog_name}" exists in Odoo'), target_fixture='blog') +def given_blog_exists(odoo_env, blog_name): + blog = odoo_env['blog.blog'].search([('name', '=', blog_name)], limit=1) + if not blog: + blog = odoo_env['blog.blog'].create({'name': blog_name}) + return blog + + +@given('I am on the Blog Publisher backend form') +def given_on_backend_form(ctx): + # In a unit/integration context this is a no-op; Playwright covers UI navigation + ctx['location'] = 'backend_form' + + +@given(parsers.parse('I enter topic "{topic}"')) +def given_enter_topic(ctx, topic): + ctx['topic'] = topic + + +@given(parsers.parse('I select provider "{provider}" and model "{model}"')) +def given_select_provider_model(ctx, provider, model): + ctx['provider'] = provider + ctx['model'] = model + + +@given(parsers.parse('I set auto-publish to {auto_publish}')) +def given_set_auto_publish(ctx, auto_publish): + ctx['auto_publish'] = auto_publish.strip().lower() in ('true', 'yes', '1') + + +@given('the Anthropic API key is invalid') +def given_invalid_api_key(odoo_env): + odoo_env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', 'INVALID' + ) + + +def _make_mock_llm_response(topic='AI Trends'): + resp = MagicMock() + resp.text = f'' + ('Content. ' * 60) + '
' + resp.tokens_used = 850 + resp.title = topic + resp.meta_title = f'{topic} — Enterprise Guide 2026'[:60] + resp.meta_description = f'A comprehensive guide to {topic} for enterprise leaders.'[:155] + resp.meta_keywords = 'AI, enterprise, trends' + resp.tags = ['Enterprise AI', 'AI Trends'] + resp.social = MagicMock( + twitter_a=f'Read our post on {topic}: https://itsulu.com/blog/test', + twitter_b=f'More on {topic}: https://itsulu.com/blog/test', + bluesky_a=f'BlueSky: {topic} post A', + bluesky_b=f'BlueSky: {topic} post B', + mastodon=f'Mastodon post about {topic}.', + linkedin=f'LinkedIn post about {topic}. ' * 5, + ) + return resp + + +@when('I click "Generate Now"') +def when_click_generate_now(odoo_env, ctx): + """ + Calls the wizard/action that orchestrates blog generation. + The actual model method is action_generate_now on itsulu.blog.schedule + or a wizard model. We mock the LLM call. + """ + from odoo.addons.itsulu_blog_publisher.wizards.generate_now_wizard import GenerateNowWizard + + mock_resp = _make_mock_llm_response(ctx.get('topic', 'AI')) + with patch( + 'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate', + return_value=mock_resp, + ): + wizard = odoo_env['itsulu.blog.generate.wizard'].create({ + 'topic': ctx.get('topic', ''), + 'llm_provider': ctx.get('provider', 'anthropic'), + 'llm_model': ctx.get('model', 'claude-sonnet-4-20250514'), + 'auto_publish': ctx.get('auto_publish', True), + }) + try: + result = wizard.action_generate() + ctx['result'] = result + ctx['error'] = None + except Exception as exc: + ctx['result'] = None + ctx['error'] = exc + + +@then('a blog.post record is created with a non-empty title') +def then_blog_post_created_with_title(odoo_env, ctx): + post = odoo_env['blog.post'].search([], order='id desc', limit=1) + assert post, "No blog.post record was created" + assert post.name, "blog.post.name must not be empty" + ctx['post'] = post + + +@then(parsers.parse('the blog.post body_arch contains at least {min_chars:d} characters of HTML')) +def then_body_arch_length(odoo_env, ctx, min_chars): + post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1) + assert len(post.body_arch or '') >= min_chars, ( + f"blog.post body_arch has {len(post.body_arch or '')} chars, need >= {min_chars}" + ) + + +@then(parsers.parse('the blog.post is_published is {expected}')) +def then_blog_post_published_state(odoo_env, ctx, expected): + post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1) + expected_bool = expected.strip().lower() in ('true', 'yes') + assert post.is_published == expected_bool, ( + f"Expected is_published={expected_bool}, got {post.is_published}" + ) + + +@then('the SEO fields website_meta_title and website_meta_description are populated') +def then_seo_fields_populated(odoo_env, ctx): + post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1) + assert post.website_meta_title, "website_meta_title must not be empty" + assert post.website_meta_description, "website_meta_description must not be empty" + + +@then(parsers.parse('a generation log entry exists with state "{state}"')) +def then_generation_log_exists(odoo_env, state): + log = odoo_env['itsulu.blog.generation.log'].search( + [('state', '=', state)], order='id desc', limit=1 + ) + assert log, f"No generation log with state='{state}' found" + + +@then('no blog.post record is created') +def then_no_blog_post_created(odoo_env, ctx): + # If there was an error, result should be None + assert ctx.get('error') is not None or ctx.get('result') is None, ( + "A blog.post record was created but an error was expected" + ) + + +@then('the log contains a human-readable error message') +def then_log_has_error_message(odoo_env): + log = odoo_env['itsulu.blog.generation.log'].search( + [('state', '=', 'error')], order='id desc', limit=1 + ) + assert log.error_message, "Error log must have a non-empty error_message" + assert len(log.error_message) > 10, "Error message must be human-readable (>10 chars)" + + +@then('a "Retry" button is visible on the log record') +def then_retry_button_exists(odoo_env): + log = odoo_env['itsulu.blog.generation.log'].search( + [('state', '=', 'error')], order='id desc', limit=1 + ) + assert hasattr(log, 'action_retry'), "Log model must have action_retry() method" + + +# ================================================================= # +# Feature: llm_provider_selection # +# ================================================================= # + +@given(parsers.parse('provider is "{provider}" and model is "{model}"')) +def given_provider_and_model(ctx, provider, model): + ctx['provider'] = provider + ctx['model'] = model + + +@given(parsers.parse('the Ollama base URL is "{url}"')) +def given_ollama_base_url(odoo_env, url): + odoo_env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.ollama_base_url', url + ) + + +@when('the LLM router is called with a prompt') +def when_llm_router_called(odoo_env, ctx): + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_resp = MagicMock(text='Generated
', tokens_used=500) + + provider_map = { + 'anthropic': 'odoo.addons.itsulu_blog_publisher.services.anthropic_provider.AnthropicProvider.generate', + 'openai': 'odoo.addons.itsulu_blog_publisher.services.openai_provider.OpenAIProvider.generate', + 'gemini': 'odoo.addons.itsulu_blog_publisher.services.gemini_provider.GeminiProvider.generate', + 'ollama': 'odoo.addons.itsulu_blog_publisher.services.ollama_provider.OllamaProvider.generate', + } + provider = ctx.get('provider', 'anthropic') + patch_target = provider_map.get(provider) + + try: + if patch_target: + with patch(patch_target, return_value=mock_resp) as mock_gen: + router = LLMRouter(odoo_env, provider=provider, model=ctx.get('model', '')) + result = router.generate(prompt='Write a blog post') + ctx['result'] = result + ctx['mock_generate'] = mock_gen + else: + router = LLMRouter(odoo_env, provider=provider, model=ctx.get('model', '')) + ctx['result'] = router.generate(prompt='Write a blog post') + ctx['error'] = None + except Exception as exc: + ctx['result'] = None + ctx['error'] = exc + + +@then(parsers.parse('the router calls the Anthropic /v1/messages endpoint')) +def then_router_called_anthropic(ctx): + assert ctx.get('mock_generate') and ctx['mock_generate'].called + + +@then(parsers.parse('the router calls the OpenAI /v1/chat/completions endpoint')) +def then_router_called_openai(ctx): + assert ctx.get('mock_generate') and ctx['mock_generate'].called + + +@then(parsers.parse('the router calls the Google Gemini API endpoint')) +def then_router_called_gemini(ctx): + assert ctx.get('mock_generate') and ctx['mock_generate'].called + + +@then(parsers.parse('the router calls http://localhost:11434/api/chat')) +def then_router_called_ollama(ctx): + assert ctx.get('mock_generate') and ctx['mock_generate'].called + + +@then('returns a non-empty string response') +def then_non_empty_response(ctx): + result = ctx.get('result') + assert result and result.text, "LLM router must return a non-empty text response" + + +@then(parsers.parse('a UserError is raised with message containing "{msg_fragment}"')) +def then_user_error_raised(ctx, msg_fragment): + from odoo.exceptions import UserError + error = ctx.get('error') + assert error is not None, "Expected a UserError but no error was raised" + assert isinstance(error, UserError), f"Expected UserError, got {type(error)}" + assert msg_fragment.lower() in str(error).lower(), ( + f"UserError message '{error}' does not contain '{msg_fragment}'" + ) + + +@then(parsers.parse('the generation log record contains tokens_used > {min_tokens:d}')) +def then_tokens_used_recorded(odoo_env, min_tokens): + log = odoo_env['itsulu.blog.generation.log'].search( + [('state', '=', 'success')], order='id desc', limit=1 + ) + assert log, "No success log found" + assert log.tokens_used > min_tokens, ( + f"tokens_used={log.tokens_used} must be > {min_tokens}" + ) + + +# ================================================================= # +# Feature: notification_email # +# ================================================================= # + +@given(parsers.parse('the notification email recipient is "{email}"')) +def given_notification_recipient(odoo_env, email): + odoo_env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.notification_emails', email + ) + + +@given('a blog post was generated and auto-published') +def given_published_blog_post(odoo_env, ctx): + blog = odoo_env['blog.blog'].search([('name', '=', 'ITSulu Insights')], limit=1) + if not blog: + blog = odoo_env['blog.blog'].create({'name': 'ITSulu Insights'}) + post = odoo_env['blog.post'].create({ + 'name': 'Prompt Governance & AI Scaling', + 'blog_id': blog.id, + 'is_published': True, + 'body_arch': 'Content
', + 'website_meta_title': 'AI Governance 2026', + 'website_meta_description': 'Learn AI governance frameworks.', + }) + odoo_env['itsulu.blog.post.social'].create({ + 'blog_post_id': post.id, + 'twitter_post_a': 'Twitter A', + 'twitter_post_b': 'Twitter B', + 'bluesky_post_a': 'BlueSky A', + 'bluesky_post_b': 'BlueSky B', + 'mastodon_post': 'Mastodon post', + 'linkedin_post': 'LinkedIn post about AI governance frameworks in enterprise.', + }) + log = odoo_env['itsulu.blog.generation.log'].create({ + 'blog_post_id': post.id, + 'state': 'success', + 'trigger_source': 'manual', + 'llm_provider': 'anthropic', + 'llm_model': 'claude-sonnet-4-20250514', + 'tokens_used': 900, + 'duration_seconds': 10.5, + }) + ctx['post'] = post + ctx['log'] = log + + +@when('the generation completes') +def when_generation_completes(odoo_env, ctx): + log = ctx.get('log') + if log: + log.send_notification_email() + + +@then(parsers.parse('exactly one email is sent to "{email}"')) +def then_exactly_one_email_sent(odoo_env, email): + mails = odoo_env['mail.mail'].search([('email_to', 'ilike', email)], order='id desc') + assert len(mails) == 1, f"Expected 1 email to {email}, found {len(mails)}" + + +@then(parsers.parse('the email subject matches "{subject_pattern}"')) +def then_email_subject_matches(odoo_env, subject_pattern): + mail = odoo_env['mail.mail'].search([], order='id desc', limit=1) + # Replace template vars with loose matching + core = subject_pattern.split('{')[0].strip() + assert core in (mail.subject or ''), ( + f"Subject '{mail.subject}' does not start with '{core}'" + ) diff --git a/test_blog_generation_log.py b/test_blog_generation_log.py new file mode 100644 index 0000000..9656e81 --- /dev/null +++ b/test_blog_generation_log.py @@ -0,0 +1,150 @@ +# addons/itsulu_blog_publisher/tests/test_blog_generation_log.py +""" +Tests for models/blog_generation_log.py +Behaviour: every generation attempt (success or failure) is recorded with + full metadata; failed logs expose a Retry action. + +RED PHASE — all tests FAIL until itsulu.blog.generation.log model exists. +""" +from odoo.tests import TransactionCase, tagged +from .factories import BlogPublisherFactory + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'generation_log') +class TestBlogGenerationLogCreation(TransactionCase): + """Verify that generation log records capture the correct metadata.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.factory.blog(name='ITSulu Insights') + + def test_successful_log_record_is_created_with_correct_fields(self): + """ + 1. Behaviour: a success log captures provider, model, tokens, duration, post link + 2. ARRANGE: a blog post and a log factory call + 3. ACT: create the log + 4. ASSERT: all fields populated, state='success' + 5. FAIL reason: model does not exist + """ + # ARRANGE + blog_post = self.factory.blog_post(blog=self.blog) + + # ACT + log = self.factory.generation_log( + blog_post=blog_post, + state='success', + llm_provider='anthropic', + llm_model='claude-sonnet-4-20250514', + tokens_used=1500, + duration_seconds=11.7, + trigger_source='manual', + ) + + # ASSERT + self.assertEqual(log.state, 'success') + self.assertEqual(log.llm_provider, 'anthropic') + self.assertEqual(log.llm_model, 'claude-sonnet-4-20250514') + self.assertEqual(log.tokens_used, 1500) + self.assertAlmostEqual(log.duration_seconds, 11.7, places=1) + self.assertEqual(log.blog_post_id.id, blog_post.id) + self.assertEqual(log.trigger_source, 'manual') + + def test_error_log_record_stores_human_readable_error_message(self): + """Error logs must store the full error message for operator diagnosis.""" + # ARRANGE + ACT + log = self.factory.generation_log( + state='error', + error_message='Anthropic API returned 401 Unauthorized — check API key.', + blog_post_id=False, + ) + + # ASSERT + self.assertEqual(log.state, 'error') + self.assertIn('401', log.error_message) + + def test_error_log_has_no_linked_blog_post(self): + """When generation fails, no blog post is created, so log.blog_post_id is False.""" + # ACT + log = self.factory.generation_log(state='error', blog_post_id=False) + + # ASSERT + self.assertFalse(log.blog_post_id) + + def test_log_trigger_source_can_be_scheduled(self): + """Logs triggered by a cron job record trigger_source='scheduled'.""" + # ACT + log = self.factory.generation_log(trigger_source='scheduled', state='success') + + # ASSERT + self.assertEqual(log.trigger_source, 'scheduled') + + def test_log_trigger_source_records_slot_name_for_scheduled_runs(self): + """Scheduled logs record which slot triggered them (morning/afternoon/evening).""" + # ACT + log = self.factory.generation_log( + trigger_source='scheduled', + schedule_slot='morning', + state='success', + ) + + # ASSERT + self.assertEqual(log.schedule_slot, 'morning') + + def test_tokens_used_defaults_to_zero_for_error_logs(self): + """Error logs that never reached the LLM should have tokens_used=0.""" + # ACT + log = self.factory.generation_log(state='error', tokens_used=0) + + # ASSERT + self.assertEqual(log.tokens_used, 0) + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'generation_log') +class TestBlogGenerationLogRetry(TransactionCase): + """Verify that failed logs expose a working Retry action.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + + def test_error_log_action_retry_returns_wizard_action(self): + """ + Calling action_retry() on an error log must return an ir.actions.act_window + or similar action dict so the UI can open the Generate Now wizard. + """ + # ARRANGE + log = self.factory.generation_log( + state='error', + error_message='Timeout', + ) + + # ACT + result = log.action_retry() + + # ASSERT + self.assertIsInstance(result, dict, + "action_retry() must return an action dict") + self.assertIn(result.get('type', ''), [ + 'ir.actions.act_window', + 'ir.actions.client', + 'ir.actions.server', + ]) + + def test_success_log_does_not_expose_retry_button(self): + """ + A success log should NOT have action_retry() callable + (or it should raise, or return False — not open a wizard). + This prevents accidental duplicate generation. + """ + # ARRANGE + blog_post = self.factory.blog_post() + log = self.factory.generation_log(state='success', blog_post=blog_post) + + # ASSERT — retry on success log should either not exist or be a no-op + if hasattr(log, 'action_retry'): + result = log.action_retry() + self.assertFalse(result, + "action_retry() on a success log must return False or no-op") diff --git a/test_blog_post_social.py b/test_blog_post_social.py new file mode 100644 index 0000000..d435c23 --- /dev/null +++ b/test_blog_post_social.py @@ -0,0 +1,330 @@ +# addons/itsulu_blog_publisher/tests/test_blog_post_social.py +""" +Tests for: + - SEO field population on blog.post after generation + - models/blog_post_social.py (itsulu.blog.post.social) + - Notification email content and recipients + +RED PHASE — all tests FAIL until models and email logic exist. +""" +from unittest.mock import patch, MagicMock +from odoo.tests import TransactionCase, tagged +from .factories import BlogPublisherFactory + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'seo') +class TestSEOPopulation(TransactionCase): + """Verify that all SEO fields are correctly populated after blog post generation.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.factory.blog(name='ITSulu Insights') + + def test_generated_post_has_non_empty_meta_title(self): + """ + 1. Behaviour: website_meta_title is set after generation + 2. ARRANGE: a blog post with SEO fields populated by factory + 3. ACT: check field + 4. ASSERT: non-empty, ≤ 60 chars + 5. FAIL reason: field population logic does not exist + """ + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, + website_meta_title='AI Governance Frameworks for Enterprises 2026', + ) + + # ASSERT + self.assertTrue(post.website_meta_title, + "website_meta_title must not be empty after generation") + self.assertLessEqual( + len(post.website_meta_title), 60, + f"SEO title too long: {len(post.website_meta_title)} chars (max 60)" + ) + + def test_generated_post_has_non_empty_meta_description(self): + """website_meta_description is set and ≤ 155 characters.""" + # ARRANGE + desc = 'Learn how leading enterprises implement AI governance frameworks in 2026.' + post = self.factory.blog_post( + blog=self.blog, + website_meta_description=desc, + ) + + # ASSERT + self.assertTrue(post.website_meta_description) + self.assertLessEqual(len(post.website_meta_description), 155) + + def test_generated_post_has_meta_keywords(self): + """website_meta_keywords is populated with comma-separated keywords.""" + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, + website_meta_keywords='AI governance, enterprise AI, compliance', + ) + + # ASSERT + self.assertTrue(post.website_meta_keywords) + keywords = [k.strip() for k in post.website_meta_keywords.split(',')] + self.assertGreaterEqual(len(keywords), 2, + "At least 2 keywords must be present") + + def test_generated_post_has_at_least_two_tags(self): + """blog.post.tag_ids must contain at least 2 blog.tag records.""" + # ARRANGE + tag1 = self.factory.blog_tag(name='Enterprise AI') + tag2 = self.factory.blog_tag(name='AI Governance') + post = self.factory.blog_post(blog=self.blog) + post.write({'tag_ids': [(6, 0, [tag1.id, tag2.id])]}) + + # ASSERT + self.assertGreaterEqual(len(post.tag_ids), 2) + + def test_seo_title_is_not_identical_to_post_title(self): + """ + The SEO meta title should be an optimised variant, not a verbatim copy + of the post title (which would waste the SEO field). + """ + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, + name='AI Trends 2026', + website_meta_title='Top AI Governance Trends Reshaping Enterprises in 2026', + ) + + # ASSERT + self.assertNotEqual( + post.name, post.website_meta_title, + "SEO meta title must be an optimised variant, not identical to the post title" + ) + + def test_new_tags_from_llm_are_created_automatically(self): + """ + If the LLM returns a tag name that doesn't exist in blog.tag, + the tag is created automatically and linked to the post. + """ + # ARRANGE — ensure tag doesn't exist + new_tag_name = 'BrandNewTagXYZ99' + self.assertFalse( + self.env['blog.tag'].search([('name', '=', new_tag_name)]), + "Pre-condition: tag must not exist before this test" + ) + + # ACT — call the helper that creates/fetches tags by name + tags = self.env['blog.tag'].get_or_create_tags([new_tag_name]) + + # ASSERT + self.assertTrue(tags, "get_or_create_tags() must return a non-empty recordset") + created = self.env['blog.tag'].search([('name', '=', new_tag_name)]) + self.assertTrue(created, "The new tag must exist in the DB after get_or_create_tags()") + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'social') +class TestBlogPostSocialModel(TransactionCase): + """Verify the itsulu.blog.post.social model stores all platform copy correctly.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.factory.blog(name='ITSulu Insights') + + def test_social_record_is_linked_one_to_one_with_blog_post(self): + """Each blog post has at most one social record.""" + # ARRANGE + post = self.factory.blog_post(blog=self.blog) + + # ACT + social = self.factory.blog_post_social(blog_post=post) + + # ASSERT + self.assertEqual(social.blog_post_id.id, post.id) + + def test_twitter_posts_stored_and_retrievable(self): + """twitter_post_a and twitter_post_b are stored correctly.""" + # ARRANGE + post = self.factory.blog_post(blog=self.blog) + social = self.factory.blog_post_social( + blog_post=post, + twitter_post_a='80% of enterprises manage prompts manually. Read why: https://itsulu.com/blog/1', + twitter_post_b='Manual prompt engineering = technical debt. Fix it: https://itsulu.com/blog/1', + ) + + # ASSERT + self.assertIn('80%', social.twitter_post_a) + self.assertIn('technical debt', social.twitter_post_b) + + def test_twitter_posts_are_within_character_limit(self): + """X/Twitter posts (excluding URL) must be ≤ 280 chars.""" + # ARRANGE + post = self.factory.blog_post(blog=self.blog) + social = self.factory.blog_post_social(blog_post=post) + + # ACT — strip the URL for character count + url = 'https://itsulu.com/blog/the-itsulu-blog-1/99' + post_a_without_url = social.twitter_post_a.replace(url, '').strip() + post_b_without_url = social.twitter_post_b.replace(url, '').strip() + + # ASSERT + self.assertLessEqual(len(post_a_without_url), 280, + "Twitter Post A exceeds 280 character limit") + self.assertLessEqual(len(post_b_without_url), 280, + "Twitter Post B exceeds 280 character limit") + + def test_mastodon_post_is_within_character_limit(self): + """Mastodon/Fediverse post must be ≤ 500 characters.""" + post = self.factory.blog_post(blog=self.blog) + social = self.factory.blog_post_social(blog_post=post) + self.assertLessEqual(len(social.mastodon_post), 500) + + def test_linkedin_post_is_at_least_150_characters(self): + """LinkedIn copy should be substantive — minimum 150 characters.""" + post = self.factory.blog_post(blog=self.blog) + social = self.factory.blog_post_social( + blog_post=post, + linkedin_post=( + 'The era of manual prompt engineering is ending. ' + 'Enterprises with governance frameworks report 34% higher AI satisfaction ' + 'and 76% fewer hallucinations. Learn the 3 critical shifts.' + ), + ) + self.assertGreaterEqual(len(social.linkedin_post), 150) + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'notification_email') +class TestNotificationEmail(TransactionCase): + """ + Verify the post-generation notification email is sent with the correct + structure matching the [ITSulu Insights] template in the uploaded .eml. + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.factory.blog(name='ITSulu Insights') + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.notification_emails', + 'nicholasr@itsulu.com', + ) + + def test_notification_email_is_sent_after_auto_publish(self): + """ + 1. Behaviour: one email sent to the configured recipient when post is published + 2. ARRANGE: published post + social copy + 3. ACT: call send_notification_email() on the log or the schedule model + 4. ASSERT: mail.mail (or mail.message) created for the recipient + 5. FAIL reason: send_notification_email() does not exist + """ + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, + name='Prompt Governance & AI Scaling', + website_published=True, + is_published=True, + ) + social = self.factory.blog_post_social(blog_post=post) + log = self.factory.generation_log(blog_post=post, state='success') + + mail_count_before = self.env['mail.mail'].search_count([]) + + # ACT + log.send_notification_email() + + # ASSERT + mail_count_after = self.env['mail.mail'].search_count([]) + self.assertEqual(mail_count_after, mail_count_before + 1, + "Exactly one mail.mail record must be created") + + sent_mail = self.env['mail.mail'].search([], order='id desc', limit=1) + self.assertIn('nicholasr@itsulu.com', sent_mail.email_to or sent_mail.recipient_ids.mapped('email')) + + def test_notification_email_subject_matches_expected_format(self): + """Email subject: '[ITSulu Insights] Blog Post Published: {title} - {date}'""" + # ARRANGE + import datetime + post = self.factory.blog_post( + blog=self.blog, + name='Prompt Governance & AI Scaling', + is_published=True, + ) + social = self.factory.blog_post_social(blog_post=post) + log = self.factory.generation_log(blog_post=post, state='success') + + # ACT + log.send_notification_email() + + # ASSERT + sent_mail = self.env['mail.mail'].search([], order='id desc', limit=1) + today = datetime.date.today().strftime('%B %d, %Y').replace(' 0', ' ') + expected_subject_start = '[ITSulu Insights] Blog Post Published:' + self.assertIn(expected_subject_start, sent_mail.subject) + self.assertIn('Prompt Governance', sent_mail.subject) + + def test_notification_email_body_contains_all_social_platforms(self): + """Email body must contain sections for X, BlueSky, Mastodon, and LinkedIn.""" + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, name='AI Trends', is_published=True + ) + social = self.factory.blog_post_social( + blog_post=post, + twitter_post_a='Twitter A copy', + twitter_post_b='Twitter B copy', + bluesky_post_a='BlueSky A copy', + bluesky_post_b='BlueSky B copy', + mastodon_post='Mastodon copy', + linkedin_post='LinkedIn copy about AI trends in enterprise.', + ) + log = self.factory.generation_log(blog_post=post, state='success') + + # ACT + log.send_notification_email() + + # ASSERT + sent_mail = self.env['mail.mail'].search([], order='id desc', limit=1) + body = sent_mail.body_html or '' + for expected_fragment in [ + 'Twitter', 'BlueSky', 'Mastodon', 'LinkedIn', + 'Twitter A copy', 'LinkedIn copy', + ]: + self.assertIn(expected_fragment, body, + f"Email body missing: {expected_fragment}") + + def test_notification_email_body_contains_post_url(self): + """Email body must include a clickable link to the published post.""" + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, name='Cloud Migration Strategies', is_published=True + ) + self.factory.blog_post_social(blog_post=post) + log = self.factory.generation_log(blog_post=post, state='success') + + # ACT + log.send_notification_email() + + # ASSERT + sent_mail = self.env['mail.mail'].search([], order='id desc', limit=1) + self.assertIn('itsulu.com', sent_mail.body_html or '', + "Email body must contain the blog post URL") + + def test_notification_email_is_not_sent_for_draft_posts(self): + """No email is sent when the post is left as a draft (is_published=False).""" + # ARRANGE + post = self.factory.blog_post( + blog=self.blog, name='Draft Post', is_published=False + ) + self.factory.blog_post_social(blog_post=post) + log = self.factory.generation_log(blog_post=post, state='success') + + mail_count_before = self.env['mail.mail'].search_count([]) + + # ACT + log.send_notification_email() + + # ASSERT + mail_count_after = self.env['mail.mail'].search_count([]) + self.assertEqual(mail_count_after, mail_count_before, + "No email must be sent for a draft (unpublished) post") diff --git a/test_blog_schedule.py b/test_blog_schedule.py new file mode 100644 index 0000000..16ff604 --- /dev/null +++ b/test_blog_schedule.py @@ -0,0 +1,251 @@ +# 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.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.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. + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.blog = cls.factory.blog(name='ITSulu Insights') + cls.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 = ( + '' + ('Enterprise AI paragraph. ' * 30) + '
' + ) + 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 included in the prompt sent to the LLM + call_args = mock_gen.call_args + prompt_used = call_args[1].get('prompt') or call_args[0][0] + self.assertIn('Kubernetes Cost Optimisation', prompt_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") diff --git a/test_blog_topic.py b/test_blog_topic.py new file mode 100644 index 0000000..33fe732 --- /dev/null +++ b/test_blog_topic.py @@ -0,0 +1,113 @@ +# addons/itsulu_blog_publisher/tests/test_blog_topic.py +""" +Tests for models/blog_topic.py +Behaviour: topic queue management — create, prioritise, mark used, + pop next topic, fall back gracefully when queue is empty. + +RED PHASE — all tests FAIL until itsulu.blog.topic model exists. +""" +from odoo.tests import TransactionCase, tagged +from odoo.exceptions import ValidationError +from .factories import BlogPublisherFactory + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'blog_topic') +class TestBlogTopicQueueManagement(TransactionCase): + """Verify that the topic queue picks topics in priority order.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + + def test_topic_is_created_with_pending_state(self): + """ + 1. Behaviour: newly created topic starts in state='pending' + 2. ARRANGE: nothing — default factory values + 3. ACT: create a topic record + 4. ASSERT: state == 'pending' + 5. FAIL reason: model itsulu.blog.topic does not exist + """ + # ARRANGE + ACT + topic = self.factory.blog_topic(name='AI Ethics in Healthcare') + + # ASSERT + self.assertEqual(topic.state, 'pending') + + def test_get_next_topic_returns_highest_priority_pending_topic(self): + """ + get_next_topic() on the model returns the pending topic + with the highest priority (urgent > high > normal > low). + """ + # ARRANGE + low = self.factory.blog_topic(name='Low priority topic', priority='low') + normal = self.factory.blog_topic(name='Normal priority topic', priority='normal') + urgent = self.factory.blog_topic(name='Urgent topic', priority='urgent') + high = self.factory.blog_topic(name='High priority topic', priority='high') + + # ACT + next_topic = self.env['itsulu.blog.topic'].get_next_topic() + + # ASSERT + self.assertEqual(next_topic.id, urgent.id, + "get_next_topic() must return the urgent-priority topic first") + + def test_get_next_topic_returns_none_when_queue_is_empty(self): + """get_next_topic() returns False (empty recordset) when no pending topics exist.""" + # ARRANGE — mark all pending topics as used to clear the queue + self.env['itsulu.blog.topic'].search([('state', '=', 'pending')]).write( + {'state': 'used'} + ) + + # ACT + result = self.env['itsulu.blog.topic'].get_next_topic() + + # ASSERT + self.assertFalse(result, "get_next_topic() must return empty recordset when queue is empty") + + def test_mark_topic_as_used_changes_state(self): + """Calling topic.mark_used() sets state to 'used'.""" + # ARRANGE + topic = self.factory.blog_topic(name='DevOps best practices') + + # ACT + topic.mark_used() + + # ASSERT + self.assertEqual(topic.state, 'used') + + def test_used_topic_is_excluded_from_next_topic_selection(self): + """A used topic is never returned by get_next_topic().""" + # ARRANGE + used_topic = self.factory.blog_topic( + name='Already published topic', priority='urgent', state='used' + ) + pending_topic = self.factory.blog_topic( + name='Not yet published', priority='normal', state='pending' + ) + + # ACT + next_topic = self.env['itsulu.blog.topic'].get_next_topic() + + # ASSERT + self.assertNotEqual(next_topic.id, used_topic.id) + self.assertEqual(next_topic.id, pending_topic.id) + + def test_topic_name_cannot_be_empty(self): + """Creating a topic with an empty name raises ValidationError.""" + with self.assertRaises((ValidationError, Exception)): + self.factory.blog_topic(name='') + + def test_topic_can_be_linked_to_a_specific_blog(self): + """A topic can optionally specify which blog it should be published to.""" + # ARRANGE + blog = self.factory.blog(name='ITSulu Insights') + + # ACT + topic = self.factory.blog_topic( + name='Kubernetes for SMBs', + blog_id=blog.id, + ) + + # ASSERT + self.assertEqual(topic.blog_id.id, blog.id) diff --git a/test_llm_router.py b/test_llm_router.py new file mode 100644 index 0000000..639854f --- /dev/null +++ b/test_llm_router.py @@ -0,0 +1,182 @@ +# addons/itsulu_blog_publisher/tests/test_llm_router.py +""" +Tests for services/llm_router.py +Behaviour: route LLM calls to the correct provider, record token usage, + raise UserError on unknown or misconfigured providers. + +RED PHASE — all tests here FAIL until llm_router and provider services exist. +""" +from unittest.mock import patch, MagicMock +from odoo.tests import TransactionCase, tagged +from odoo.exceptions import UserError +from .factories import BlogPublisherFactory + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'llm_router') +class TestLLMRouterProviderDispatch(TransactionCase): + """Verify that the LLM router dispatches to the correct backend provider.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + # Store a known-valid stub API key in ir.config_parameter + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', 'sk-ant-test-key' + ) + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.openai_api_key', 'sk-openai-test-key' + ) + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.gemini_api_key', 'gemini-test-key' + ) + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.ollama_base_url', 'http://localhost:11434' + ) + + # --- Provider dispatch --- + + def test_anthropic_provider_calls_anthropic_endpoint(self): + """ + 1. Behaviour: router with provider='anthropic' calls AnthropicProvider.generate() + 2. ARRANGE: router instantiated with provider='anthropic' + 3. ACT: router.generate(prompt='Write a blog post about AI') + 4. ASSERT: AnthropicProvider.generate was called once; result is non-empty string + 5. FAIL reason: LLMRouter class does not exist yet + """ + # ARRANGE + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_response = MagicMock() + mock_response.text = 'Content here...
' + mock_response.tokens_used = 800 + + with patch( + 'odoo.addons.itsulu_blog_publisher.services.anthropic_provider' + '.AnthropicProvider.generate', + return_value=mock_response, + ) as mock_generate: + # ACT + router = LLMRouter(self.env, provider='anthropic', model='claude-sonnet-4-20250514') + result = router.generate(prompt='Write a blog post about AI trends') + + # ASSERT + mock_generate.assert_called_once() + self.assertTrue(result.text, "Router must return non-empty text from Anthropic") + + def test_openai_provider_calls_openai_endpoint(self): + """Router with provider='openai' delegates to OpenAIProvider.generate().""" + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_response = MagicMock() + mock_response.text = 'OpenAI generated content
' + mock_response.tokens_used = 600 + + with patch( + 'odoo.addons.itsulu_blog_publisher.services.openai_provider' + '.OpenAIProvider.generate', + return_value=mock_response, + ) as mock_generate: + router = LLMRouter(self.env, provider='openai', model='gpt-4o') + result = router.generate(prompt='Write a blog post about cloud computing') + + mock_generate.assert_called_once() + self.assertTrue(result.text) + + def test_gemini_provider_calls_gemini_endpoint(self): + """Router with provider='gemini' delegates to GeminiProvider.generate().""" + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_response = MagicMock() + mock_response.text = 'Gemini generated content
' + mock_response.tokens_used = 700 + + with patch( + 'odoo.addons.itsulu_blog_publisher.services.gemini_provider' + '.GeminiProvider.generate', + return_value=mock_response, + ) as mock_generate: + router = LLMRouter(self.env, provider='gemini', model='gemini-2.0-flash') + result = router.generate(prompt='Write a blog post about automation') + + mock_generate.assert_called_once() + self.assertTrue(result.text) + + def test_ollama_provider_calls_local_api_endpoint(self): + """Router with provider='ollama' delegates to OllamaProvider.generate().""" + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_response = MagicMock() + mock_response.text = 'Mistral generated content
' + mock_response.tokens_used = 500 + + with patch( + 'odoo.addons.itsulu_blog_publisher.services.ollama_provider' + '.OllamaProvider.generate', + return_value=mock_response, + ) as mock_generate: + router = LLMRouter(self.env, provider='ollama', model='mistral') + result = router.generate(prompt='Write a blog post about open source AI') + + mock_generate.assert_called_once() + self.assertTrue(result.text) + + def test_unknown_provider_raises_user_error(self): + """ + Router with unrecognised provider must raise UserError immediately, + before any network call is made. + """ + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + + with self.assertRaises(UserError) as ctx: + router = LLMRouter(self.env, provider='unknown_provider', model='some-model') + router.generate(prompt='This should fail') + + self.assertIn('provider not configured', str(ctx.exception).lower()) + + def test_missing_api_key_raises_user_error(self): + """Router raises UserError when required API key param is absent.""" + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + # Remove the key + self.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', '' + ) + with self.assertRaises(UserError) as ctx: + router = LLMRouter(self.env, provider='anthropic', model='claude-sonnet-4-20250514') + router.generate(prompt='Fail fast') + + self.assertIn('api key', str(ctx.exception).lower()) + # Restore for other tests + self.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', 'sk-ant-test-key' + ) + + +@tagged('post_install', '-at_install', 'itsulu_blog_publisher', 'llm_router') +class TestLLMRouterTokenLogging(TransactionCase): + """Verify that token usage is captured from provider responses.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.factory = BlogPublisherFactory(cls.env) + cls.env['ir.config_parameter'].sudo().set_param( + 'itsulu_blog_publisher.anthropic_api_key', 'sk-ant-test-key' + ) + + def test_router_response_includes_tokens_used(self): + """ + The LLMResponse object returned by router.generate() must expose + tokens_used as a positive integer when the API returns usage data. + """ + from odoo.addons.itsulu_blog_publisher.services.llm_router import LLMRouter + mock_response = MagicMock() + mock_response.text = 'Content
' + mock_response.tokens_used = 1234 + + with patch( + 'odoo.addons.itsulu_blog_publisher.services.anthropic_provider' + '.AnthropicProvider.generate', + return_value=mock_response, + ): + router = LLMRouter(self.env, provider='anthropic', model='claude-sonnet-4-20250514') + result = router.generate(prompt='Token test') + + self.assertGreater(result.tokens_used, 0) + self.assertEqual(result.tokens_used, 1234)