Fixed router.generate() calls to use 'topic=' instead of 'prompt='. Fixed post.social_ids reference to use correct field name 'itsulu_social_id'. These corrections align with actual method signatures and model definitions. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
471 lines
18 KiB
Python
471 lines
18 KiB
Python
# 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 odoo_env(env):
|
|
"""pytest-odoo's env fixture, re-exported for BDD step access."""
|
|
return env
|
|
|
|
|
|
@pytest.fixture
|
|
def ctx():
|
|
"""Mutable context bag shared between steps in one scenario."""
|
|
return {}
|
|
|
|
|
|
# ================================================================= #
|
|
# 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'<h1>{topic}</h1>' + '<p>' + ('Content. ' * 60) + '</p>'
|
|
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='<p>Generated</p>', 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(topic='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(topic='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': '<p>Content</p>',
|
|
'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}'"
|
|
)
|
|
|
|
|
|
@then(parsers.parse('no email is sent to "{email}"'))
|
|
def then_no_email_sent(odoo_env, email):
|
|
mails = odoo_env['mail.mail'].search([('email_to', 'ilike', email)])
|
|
assert len(mails) == 0, f"Expected no email to {email}, but found {len(mails)} email(s)"
|
|
|
|
|
|
@then('the email body contains the blog post title')
|
|
def then_email_contains_title(odoo_env, ctx):
|
|
mail = odoo_env['mail.mail'].search([], order='id desc', limit=1)
|
|
post = ctx.get('post')
|
|
if post:
|
|
assert post.name in (mail.body_html or ''), (
|
|
f"Email body does not contain post title '{post.name}'"
|
|
)
|
|
|
|
|
|
@then('the email body contains social media copy for all enabled platforms')
|
|
def then_email_contains_social_copy(odoo_env, ctx):
|
|
mail = odoo_env['mail.mail'].search([], order='id desc', limit=1)
|
|
post = ctx.get('post')
|
|
if post and post.itsulu_social_id:
|
|
social = post.itsulu_social_id[0]
|
|
body = mail.body_html or ''
|
|
# Verify at least one social platform is mentioned
|
|
platforms_found = 0
|
|
if social.twitter_enabled and social.twitter_post_a:
|
|
platforms_found += 1
|
|
if social.bluesky_enabled and social.bluesky_post_a:
|
|
platforms_found += 1
|
|
if social.mastodon_enabled and social.mastodon_post:
|
|
platforms_found += 1
|
|
if social.linkedin_enabled and social.linkedin_post:
|
|
platforms_found += 1
|
|
assert platforms_found > 0, "Email must contain at least one enabled social platform copy"
|
|
|
|
|
|
# ================================================================= #
|
|
# Additional verification steps for robustness #
|
|
# ================================================================= #
|
|
|
|
@then('the blog.post has tags assigned')
|
|
def then_blog_post_has_tags(odoo_env, ctx):
|
|
post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1)
|
|
assert post.tag_ids, "blog.post must have at least one tag assigned"
|
|
|
|
|
|
@then(parsers.parse('the blog.post tags include "{tag_name}"'))
|
|
def then_blog_post_has_tag(odoo_env, tag_name, ctx):
|
|
post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1)
|
|
tag_names = [tag.name for tag in post.tag_ids]
|
|
assert tag_name in tag_names, (
|
|
f"Tag '{tag_name}' not found. Available tags: {tag_names}"
|
|
)
|
|
|
|
|
|
@then('the blog.post has social media copy assigned')
|
|
def then_blog_post_has_social_copy(odoo_env, ctx):
|
|
post = ctx.get('post') or odoo_env['blog.post'].search([], order='id desc', limit=1)
|
|
social = odoo_env['itsulu.blog.post.social'].search([('blog_post_id', '=', post.id)])
|
|
assert social, "blog.post must have associated social media copy"
|
|
ctx['social'] = social
|
|
|
|
|
|
@then('at least one social platform is enabled')
|
|
def then_at_least_one_platform_enabled(odoo_env, ctx):
|
|
social = ctx.get('social') or odoo_env['itsulu.blog.post.social'].search([], order='id desc', limit=1)
|
|
assert (
|
|
social.twitter_enabled or social.bluesky_enabled
|
|
or social.mastodon_enabled or social.linkedin_enabled
|
|
), "At least one social platform must be enabled"
|
|
|
|
|
|
@then('the generation log records the correct LLM provider')
|
|
def then_log_has_correct_provider(odoo_env, ctx):
|
|
log = odoo_env['itsulu.blog.generation.log'].search(
|
|
[('state', '=', 'success')], order='id desc', limit=1
|
|
)
|
|
provider = ctx.get('provider', 'anthropic')
|
|
assert log.llm_provider == provider, (
|
|
f"Expected provider '{provider}' in log, got '{log.llm_provider}'"
|
|
)
|
|
|
|
|
|
@then('the generation log records the correct LLM model')
|
|
def then_log_has_correct_model(odoo_env, ctx):
|
|
log = odoo_env['itsulu.blog.generation.log'].search(
|
|
[('state', '=', 'success')], order='id desc', limit=1
|
|
)
|
|
model = ctx.get('model', '')
|
|
assert log.llm_model == model, (
|
|
f"Expected model '{model}' in log, got '{log.llm_model}'"
|
|
)
|
|
|
|
|
|
@then(parsers.parse('the generation log trigger_source is "{source}"'))
|
|
def then_log_trigger_source(odoo_env, source):
|
|
log = odoo_env['itsulu.blog.generation.log'].search([], order='id desc', limit=1)
|
|
assert log.trigger_source == source, (
|
|
f"Expected trigger_source '{source}', got '{log.trigger_source}'"
|
|
)
|
|
|
|
|
|
@then('the generation duration is recorded')
|
|
def then_generation_duration_recorded(odoo_env):
|
|
log = odoo_env['itsulu.blog.generation.log'].search(
|
|
[('state', '=', 'success')], order='id desc', limit=1
|
|
)
|
|
assert log.duration_seconds > 0, "duration_seconds must be > 0"
|