itsulu-blog-publisher/addons/itsulu_blog_publisher/tests/test_bdd_steps.py
Nicholas Riegel c039b5f0cb release: v0.4.8 — CI pipeline green + Odoo 17 fixes (squash of !1)
Squash-merge of fix/ci-pipeline-corrections. Drives the full test suite
to 69/69 green on the ITSulu K8s cluster and fixes two production bugs.

Production fixes:
- Email template migrated from dead Odoo Mako (${}/% for) to Odoo 17
  inline_template ({{ }}) + qweb body (type="html", t-out/t-foreach/t-if).
  Notification emails previously rendered raw code in the subject/body.
- _create_blog_post now writes 'content': llm_response.body_html — every
  auto-generated post was publishing empty.
- Removed duplicate itsulu_social_id field (startup warning).

Testing & infra:
- CI pipeline corrected (stage order, DB auth, junit artifact, addons path).
- E2E moved to ephemeral jobs in the itsulu-testing K8s namespace.
- Test code brought up to Odoo 17 (mail rendering, blog.post.content,
  pytest-bdd env fixture, _render_field).

Versioning:
- Introduce MAJOR.MINOR.PATCH scheme, VERSION file, scripts/bump-version.sh,
  CHANGELOG.md; first release v0.4.8. CLAUDE.md §15 documents the process.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-05-30 10:58:57 -04:00

532 lines
21 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
import odoo
from odoo.api import Environment
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(request):
"""Odoo environment for BDD step definitions.
pytest-odoo 2.x does not expose an 'env' pytest fixture — env only
exists as self.env inside TransactionCase subclasses. We build one
directly from the registry and roll back after each scenario.
"""
db = request.config.getoption('--odoo-database')
registry = odoo.registry(db)
with registry.cursor() as cr:
env = Environment(cr, odoo.SUPERUSER_ID, {})
yield env
cr.rollback()
@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, ctx):
odoo_env['ir.config_parameter'].sudo().set_param(
'itsulu_blog_publisher.anthropic_api_key', 'INVALID'
)
ctx['api_key_invalid'] = True
def _make_mock_llm_response(topic='AI Trends'):
resp = MagicMock()
# _create_blog_post writes llm_response.body_html into blog.post.content,
# so body_html must be a real string (not an auto-MagicMock) and long
# enough to satisfy the 500-char content assertion.
body = f'<h1>{topic}</h1>' + '<p>' + ('Content. ' * 60) + '</p>'
resp.text = body
resp.body_html = body
resp.raw_text = body
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 BlogGenerateWizard as GenerateNowWizard
from odoo.exceptions import UserError
# When the scenario marks the API key invalid, the LLM call must fail so
# run_generation records an error log and creates no post.
if ctx.get('api_key_invalid'):
patch_kwargs = {'side_effect': UserError('Authentication failed: invalid API key.')}
else:
patch_kwargs = {'return_value': _make_mock_llm_response(ctx.get('topic', 'AI'))}
with patch(
'odoo.addons.itsulu_blog_publisher.services.llm_router.LLMRouter.generate',
**patch_kwargs,
):
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 content 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.content or '') >= min_chars, (
f"blog.post content has {len(post.content 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
)
def _valid_llm_json():
"""Return a JSON string with all fields LLMRouter._parse_response requires."""
import json
return json.dumps({
'title': 'Generated Blog Post',
'body_html': '<p>' + ('Generated content. ' * 30) + '</p>',
'meta_title': 'Generated SEO Title',
'meta_description': 'Generated meta description for the post.',
'meta_keywords': 'ai, testing, blog',
'tags': ['AI', 'Testing'],
'social': {
'twitter_a': 'Tweet A', 'twitter_b': 'Tweet B',
'bluesky_a': 'BlueSky A', 'bluesky_b': 'BlueSky B',
'mastodon': 'Mastodon post', 'linkedin': 'LinkedIn post',
},
'sources': [],
})
@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
# The router parses provider_response.text as JSON, so the mock must
# return a valid JSON blob with all required fields — not raw HTML.
mock_resp = MagicMock(text=_valid_llm_json(), 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)
# The router requires the chosen provider's own credential before it
# dispatches to provider.generate(). The Background only sets the Anthropic
# key, so configure the relevant credential for the provider under test.
_config = {
'openai': ('itsulu_blog_publisher.openai_api_key', 'sk-openai-test'),
'gemini': ('itsulu_blog_publisher.gemini_api_key', 'gemini-test-key'),
'ollama': ('itsulu_blog_publisher.ollama_base_url', 'http://localhost:11434'),
}
if provider in _config:
key, val = _config[provider]
odoo_env['ir.config_parameter'].sudo().set_param(key, val)
try:
if patch_target:
with patch(patch_target, return_value=mock_resp) as mock_gen:
# Record the mock before generate() so assertions can see it
# even if generate() raises.
ctx['mock_generate'] = mock_gen
router = LLMRouter(odoo_env, provider=provider, model=ctx.get('model', ''))
ctx['result'] = router.generate(topic='Write a blog post')
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.body_html, "LLM router must return a non-empty body_html 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(ctx, min_tokens):
# The router returns an LLMResponse carrying tokens_used; the schedule/wizard
# is what persists it to a generation log. At router level we verify the
# response records token usage.
result = ctx.get('result')
assert result is not None, "No LLM result captured"
assert result.tokens_used > min_tokens, (
f"Expected tokens_used > {min_tokens}, got {result.tokens_used}"
)
# ================================================================= #
# 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,
'content': '<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"