Skip to content

Commented few codes related to post on X #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kamlesh0928 merged 2 commits into from
Sep 23, 2026
Merged

Commented few codes related to post on X #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0c379d1
Commented few codes related to post on X
kamlesh0928 Sep 23, 2026
1374a3e
Corrected the python testcases
kamlesh0928 Sep 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 7 additions & 6 deletions app/agents/orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@
from app.discovery.repository import mark_topic_status
from app.publishers.linkedin import LinkedInPublishError
from app.publishers.linkedin import publish_post as publish_linkedin_post
from app.publishers.x import XPublishError
from app.publishers.x import publish_post as publish_x_post

# from app.publishers.x import XPublishError # Uncomment - If added credits on X platform.
from app.utils.logger import setup_logger

logger = setup_logger("orchestrator")
Expand All @@ -32,10 +32,11 @@ async def run_pipeline() -> None:
except LinkedInPublishError:
logger.exception("Failed to publish to LinkedIn for topic '%s'", topic.title)

try:
publish_x_post(approved.x_content)
except XPublishError:
logger.exception("Failed to publish to X for topic '%s'", topic.title)
# Uncomment - If added credits on X platform.
# try:
# publish_x_post(approved.x_content)
# except XPublishError:
# logger.exception("Failed to publish to X for topic '%s'", topic.title)

return

Expand Down
46 changes: 24 additions & 22 deletions tests/agents/test_orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@
from app.agents import orchestrator
from app.database.models.topic import STATUS_REJECTED, STATUS_USED
from app.publishers.linkedin import LinkedInPublishError
from app.publishers.x import XPublishError

# from app.publishers.x import XPublishError # Uncomment - If added credits on X platform.


class _FakeTopic:
Expand All @@ -16,30 +17,30 @@ def __init__(self, id, title="Idea", rank=1):
@pytest.mark.asyncio
async def test_run_pipeline_stops_on_first_accepted_topic(mocker):
topics = [_FakeTopic(id=1), _FakeTopic(id=2)]
approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
approved = mocker.Mock(linkedin_content="Fake LinkedIn content") # Uncomment - If added credits on X platform. Add {x_content="Fake X content"} in the Mock parameter
mocker.patch("app.agents.orchestrator.get_topics", return_value=topics)
mock_generate_post = mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved])
mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status")
mock_publish_linkedin = mocker.patch("app.agents.orchestrator.publish_linkedin_post")
mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post")
# mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post") # Uncomment - If added credits on X platform.

await orchestrator.run_pipeline()

mock_generate_post.assert_awaited_once_with(topics[0])
mock_mark_status.assert_called_once_with(1, STATUS_USED)
mock_publish_linkedin.assert_called_once_with("Fake LinkedIn content")
mock_publish_x.assert_called_once_with("Fake X content")
# mock_publish_x.assert_called_once_with("Fake X content") # Uncomment - If added credits on X platform.


@pytest.mark.asyncio
async def test_run_pipeline_tries_next_topic_after_rejection(mocker):
topics = [_FakeTopic(id=1), _FakeTopic(id=2)]
approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
approved = mocker.Mock(linkedin_content="Fake LinkedIn content") # Uncomment - If added credits on X platform. Add {x_content="Fake X content"} in the Mock parameter
mocker.patch("app.agents.orchestrator.get_topics", return_value=topics)
mocker.patch("app.agents.orchestrator.generate_post", side_effect=[None, approved])
mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status")
mocker.patch("app.agents.orchestrator.publish_linkedin_post")
mocker.patch("app.agents.orchestrator.publish_x_post")
# mocker.patch("app.agents.orchestrator.publish_x_post") # Uncomment - If added credits on X platform.

await orchestrator.run_pipeline()

Expand All @@ -53,14 +54,14 @@ async def test_run_pipeline_tries_next_topic_after_rejection(mocker):
async def test_run_pipeline_regenerates_when_all_topics_rejected(mocker):
first_batch = [_FakeTopic(id=1)]
second_batch = [_FakeTopic(id=2)]
approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
approved = mocker.Mock(linkedin_content="Fake LinkedIn content") # Uncomment - If added credits on X platform. Add {x_content="Fake X content"} in the Mock parameter
mock_get_topics = mocker.patch(
"app.agents.orchestrator.get_topics", side_effect=[first_batch, second_batch]
)
mocker.patch("app.agents.orchestrator.generate_post", side_effect=[None, approved])
mocker.patch("app.agents.orchestrator.mark_topic_status")
mocker.patch("app.agents.orchestrator.publish_linkedin_post")
mocker.patch("app.agents.orchestrator.publish_x_post")
# mocker.patch("app.agents.orchestrator.publish_x_post") # Uncomment - If added credits on X platform.

await orchestrator.run_pipeline()

Expand All @@ -70,32 +71,33 @@ async def test_run_pipeline_regenerates_when_all_topics_rejected(mocker):
@pytest.mark.asyncio
async def test_run_pipeline_logs_and_continues_when_linkedin_publish_fails(mocker):
topics = [_FakeTopic(id=1)]
approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
approved = mocker.Mock(linkedin_content="Fake LinkedIn content") # Uncomment - If added credits on X platform. Add {x_content="Fake X content"} in the Mock parameter
mocker.patch("app.agents.orchestrator.get_topics", return_value=topics)
mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved])
mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status")
mocker.patch("app.agents.orchestrator.publish_linkedin_post", side_effect=LinkedInPublishError("boom"))
mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post")
# mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post") # Uncomment - If added credits on X platform.

await orchestrator.run_pipeline()

mock_mark_status.assert_called_once_with(1, STATUS_USED)
mock_publish_x.assert_called_once_with("Fake X content")
# mock_publish_x.assert_called_once_with("Fake X content") # Uncomment - If added credits on X platform.


@pytest.mark.asyncio
async def test_run_pipeline_logs_and_continues_when_x_publish_fails(mocker):
topics = [_FakeTopic(id=1)]
approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
mocker.patch("app.agents.orchestrator.get_topics", return_value=topics)
mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved])
mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status")
mocker.patch("app.agents.orchestrator.publish_linkedin_post")
mocker.patch("app.agents.orchestrator.publish_x_post", side_effect=XPublishError("boom"))
# Uncomment - If added credits on X platform.
# @pytest.mark.asyncio
# async def test_run_pipeline_logs_and_continues_when_x_publish_fails(mocker):
# topics = [_FakeTopic(id=1)]
# approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content")
# mocker.patch("app.agents.orchestrator.get_topics", return_value=topics)
# mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved])
# mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status")
# mocker.patch("app.agents.orchestrator.publish_linkedin_post")
# mocker.patch("app.agents.orchestrator.publish_x_post", side_effect=XPublishError("boom"))

await orchestrator.run_pipeline()
# await orchestrator.run_pipeline()

mock_mark_status.assert_called_once_with(1, STATUS_USED)
# mock_mark_status.assert_called_once_with(1, STATUS_USED)


@pytest.mark.asyncio
Expand Down
Loading