From 65fd09c06d4e5ab7ac7a0a683a18b2e1dbb572eb Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Mon, 27 Apr 2026 16:09:35 -0700 Subject: [PATCH] feat: add BlogPostStrategy wrapping Directus publish_blog_post() --- app/services/platforms/blog_post.py | 70 ++++++++++++++++ tests/services/platforms/test_blog_post.py | 93 ++++++++++++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 app/services/platforms/blog_post.py create mode 100644 tests/services/platforms/test_blog_post.py diff --git a/app/services/platforms/blog_post.py b/app/services/platforms/blog_post.py new file mode 100644 index 0000000..ae10aa9 --- /dev/null +++ b/app/services/platforms/blog_post.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import json +import logging + +from app.services.directus import publish_blog_post +from app.services.platforms.base import PostingStrategy, PostResult + +logger = logging.getLogger(__name__) + +_BLOG_BASE_URL = "https://circuitforge.tech/blog" + + +class BlogPostStrategy(PostingStrategy): + """Publish a blog post to the CircuitForge website via Directus. + + target: arbitrary label string (e.g. "blog/main") — used for logging only. + title: post title + body: post body (Markdown) + extra: optional dict with keys: + slug (str) — Directus URL slug; generated from title if absent + tags (list|str) — tag list or JSON-encoded list string + seo_description (str) — meta description + """ + + campaign_type = "blog_post" + + def execute( + self, + *, + target: str, + title: str, + body: str, + flair: str | None = None, + extra: dict | None = None, + ) -> PostResult: + extra = extra or {} + + slug = extra.get("slug") or None + seo_description = extra.get("seo_description") or None + + # tags may be stored as a JSON string in the DB; decode it + raw_tags = extra.get("tags") + tags: list[str] | None = None + if isinstance(raw_tags, str): + try: + tags = json.loads(raw_tags) + except json.JSONDecodeError: + logger.warning("Could not parse tags JSON: %r", raw_tags) + elif isinstance(raw_tags, list): + tags = raw_tags + + item = publish_blog_post( + title=title, + body=body, + slug=slug, + tags=tags, + seo_description=seo_description, + ) + + published_slug = item.get("slug", slug or "") + url = f"{_BLOG_BASE_URL}/{published_slug}" + + return PostResult( + url=url, + metadata={"directus_id": item.get("id"), "slug": published_slug}, + ) + + def supports_dupe_guard(self) -> bool: + return False diff --git a/tests/services/platforms/test_blog_post.py b/tests/services/platforms/test_blog_post.py new file mode 100644 index 0000000..946429a --- /dev/null +++ b/tests/services/platforms/test_blog_post.py @@ -0,0 +1,93 @@ +import json +from unittest.mock import patch + +import pytest + +from app.services.platforms.blog_post import BlogPostStrategy +from app.services.platforms.base import PostResult + + +def _make_directus_response(slug: str = "test-post", post_id: int = 42) -> dict: + return {"id": post_id, "slug": slug, "title": "Test Post", "status": "published"} + + +def test_campaign_type(): + assert BlogPostStrategy.campaign_type == "blog_post" + + +def test_does_not_support_dupe_guard(): + assert BlogPostStrategy().supports_dupe_guard() is False + + +def test_execute_publishes_immediately(): + mock_response = _make_directus_response(slug="my-post") + + with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub: + strategy = BlogPostStrategy() + result = strategy.execute( + target="blog/main", + title="My Test Post", + body="# Hello\n\nThis is the post body.", + ) + + mock_pub.assert_called_once_with( + title="My Test Post", + body="# Hello\n\nThis is the post body.", + slug=None, + tags=None, + seo_description=None, + ) + assert isinstance(result, PostResult) + assert result.url == "https://circuitforge.tech/blog/my-post" + assert result.metadata["directus_id"] == 42 + + +def test_execute_passes_extra_fields(): + mock_response = _make_directus_response(slug="custom-slug") + + with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub: + strategy = BlogPostStrategy() + result = strategy.execute( + target="blog/main", + title="Custom Slug Post", + body="Body content here.", + extra={ + "slug": "custom-slug", + "tags": ["self-hosting", "peregrine"], + "seo_description": "A short description for SEO.", + }, + ) + + mock_pub.assert_called_once_with( + title="Custom Slug Post", + body="Body content here.", + slug="custom-slug", + tags=["self-hosting", "peregrine"], + seo_description="A short description for SEO.", + ) + assert result.url == "https://circuitforge.tech/blog/custom-slug" + + +def test_execute_parses_tags_from_json_string(): + """Tags stored as JSON string in campaign_variants.tags should be decoded.""" + mock_response = _make_directus_response(slug="json-tags-post") + + with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub: + strategy = BlogPostStrategy() + strategy.execute( + target="blog/main", + title="JSON Tags Post", + body="Body.", + extra={"tags": '["self-hosting","kiwi"]'}, # stored as JSON string + ) + + call_args = mock_pub.call_args + assert call_args.kwargs["tags"] == ["self-hosting", "kiwi"] + + +def test_execute_raises_on_directus_error(): + with patch("app.services.platforms.blog_post.publish_blog_post", + side_effect=RuntimeError("Directus 502")): + strategy = BlogPostStrategy() + with pytest.raises(RuntimeError, match="Directus 502"): + strategy.execute(target="blog/main", title="T", body="B")