mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
Backend Developer 1
fb293a787d
Audited x_post_service.py, video_post_service.py, release_proposal.py, and heartbeat_mutex.py for the pre-lock DB-write anti-pattern (a session write that happens before the SET NX / HeartbeatMutex acquire returns a token, letting a losing racer's stale write clobber a winner's committed state). XPostService.approve, VideoPostService.approve, and ReleaseProposalService.approve/reject already implement the correct validate-pure-pre-lock, apply-under-lock pattern (the XPostService fix already shipped per CHANGELOG.md: "X edited_body write deferred into the single-flight lock (M5)"). HeartbeatMutex holds no AsyncSession at all, so the anti-pattern is structurally inapplicable there. Adds a genuine cross-session concurrency regression test to test_x_post_service.py (a real second DB connection, not an in-process mock) mirroring VideoPostService's existing cross-session test, proving a concurrently-committed post survives and the CEO's edited body never lands on the just-posted row.
905 lines
34 KiB
Python
905 lines
34 KiB
Python
"""XPostService coverage: approve posts (idempotent), reject cancels.
|
|
|
|
Mirrors the release-proposal service tests. The Redis lock helpers are
|
|
patched (no live Redis in tests, matching the project's ``_no_live_redis``
|
|
fixture) so approve exercises the real post + status-transition path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
from contextlib import contextmanager
|
|
from typing import TYPE_CHECKING, cast
|
|
from unittest.mock import AsyncMock, patch
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from roboco.config import settings as cfg
|
|
from roboco.db.tables import AgentTable, ProjectTable, TaskTable
|
|
from roboco.foundation import identity as _foundation
|
|
from roboco.foundation.policy.content import markers
|
|
from roboco.models.base import (
|
|
AgentRole,
|
|
AgentStatus,
|
|
Complexity,
|
|
Team,
|
|
)
|
|
from roboco.models.base import TaskNature as TN
|
|
from roboco.models.base import TaskStatus as TS
|
|
from roboco.models.base import TaskType as TT
|
|
from roboco.services.task import (
|
|
X_FEATURE_SOURCE,
|
|
X_POST_SOURCE,
|
|
X_REPLY_SOURCE,
|
|
TaskService,
|
|
)
|
|
from roboco.services.x_client import XClient, XMention, XPostResult
|
|
from roboco.services.x_post_service import (
|
|
TaskAlreadyCompletedError,
|
|
XPostBodyTooLongError,
|
|
XPostExecuteResult,
|
|
XPostService,
|
|
get_x_post_service,
|
|
)
|
|
from sqlalchemy.ext.asyncio import (
|
|
AsyncEngine,
|
|
AsyncSession,
|
|
async_sessionmaker,
|
|
create_async_engine,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterator
|
|
from uuid import UUID
|
|
|
|
SYSTEM_UUID = _foundation.AGENTS["system"].uuid
|
|
SECRETARY_UUID = _foundation.AGENTS["secretary-1"].uuid
|
|
ONE = 1
|
|
TWO = 2
|
|
|
|
|
|
@contextmanager
|
|
def _lock_free() -> Iterator[None]:
|
|
"""Patch XPostService's lock helpers so approve/reject exercise the real
|
|
post/cancel path without touching the (test-blocked) Redis."""
|
|
with (
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
yield
|
|
|
|
|
|
class _StubClient(XClient):
|
|
def __init__(self, *, posted: bool = True, tweet_id: str = "999") -> None:
|
|
self._posted = posted
|
|
self._tweet_id = tweet_id
|
|
self.calls: list[str] = []
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return True
|
|
|
|
async def post_tweet(self, text: str) -> XPostResult:
|
|
self.calls.append(text)
|
|
if not self._posted:
|
|
return XPostResult(posted=False, tweet_id=None, detail="rejected by X")
|
|
return XPostResult(posted=True, tweet_id=self._tweet_id, detail="posted")
|
|
|
|
async def fetch_mentions(
|
|
self, since_id: str | None, max_results: int
|
|
) -> list[XMention]:
|
|
_ = (since_id, max_results)
|
|
return []
|
|
|
|
|
|
async def _seed_draft(
|
|
session: AsyncSession, *, source: str = X_POST_SOURCE, body: str = "Draft body"
|
|
) -> TaskTable:
|
|
for uuid, slug, role in (
|
|
(SYSTEM_UUID, "system", AgentRole.SYSTEM),
|
|
(SECRETARY_UUID, "secretary-1", AgentRole.SECRETARY),
|
|
):
|
|
if await session.get(AgentTable, uuid) is None:
|
|
session.add(
|
|
AgentTable(
|
|
id=uuid,
|
|
name=slug,
|
|
slug=slug,
|
|
role=role,
|
|
team=None,
|
|
status=AgentStatus.ACTIVE,
|
|
model_config={},
|
|
system_prompt="x",
|
|
capabilities=[],
|
|
permissions={},
|
|
metrics={},
|
|
)
|
|
)
|
|
await session.flush()
|
|
project = ProjectTable(
|
|
id=uuid4(),
|
|
name="RoboCo",
|
|
slug=f"roboco-{uuid4().hex[:6]}",
|
|
git_url="https://example.com/roboco.git",
|
|
assigned_cell=Team.BACKEND,
|
|
created_by=SYSTEM_UUID,
|
|
)
|
|
session.add(project)
|
|
await session.flush()
|
|
task = TaskTable(
|
|
id=uuid4(),
|
|
title="X draft",
|
|
description=body,
|
|
acceptance_criteria=["CEO approves or rejects"],
|
|
status=TS.PENDING,
|
|
priority=2,
|
|
task_type=TT.ADMINISTRATIVE,
|
|
nature=TN.NON_TECHNICAL,
|
|
estimated_complexity=Complexity.LOW,
|
|
project_id=project.id,
|
|
created_by=SYSTEM_UUID,
|
|
assigned_to=SECRETARY_UUID,
|
|
team=Team.MAIN_PM,
|
|
source=source,
|
|
confirmed_by_human=False,
|
|
)
|
|
session.add(task)
|
|
await session.flush()
|
|
markers.set_x_draft_body(task, body)
|
|
await session.flush()
|
|
return task
|
|
|
|
|
|
_FEATURE_SLUG = "org-memory"
|
|
_FEATURE_TITLE = "Organizational Memory Loop"
|
|
|
|
|
|
async def _seed_feature_draft(
|
|
session: AsyncSession,
|
|
*,
|
|
wants_video: bool = True,
|
|
video_script: str = "",
|
|
body: str = "Draft body",
|
|
) -> TaskTable:
|
|
"""An X_FEATURE_SOURCE draft carrying the x_feature_ref marker
|
|
``propose_feature_spotlight`` stamps (Task 4, 2026-07-09 pipeline fixes):
|
|
slug/title always, plus wants_video/video_script when a companion video
|
|
was requested at authoring time."""
|
|
task = await _seed_draft(session, source=X_FEATURE_SOURCE, body=body)
|
|
markers.set_x_feature_ref(
|
|
task,
|
|
{
|
|
"slug": _FEATURE_SLUG,
|
|
"title": _FEATURE_TITLE,
|
|
"wants_video": wants_video,
|
|
"video_script": video_script,
|
|
},
|
|
)
|
|
await session.flush()
|
|
return task
|
|
|
|
|
|
def _enable_video(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(cfg, "video_engine_enabled", True)
|
|
monkeypatch.setattr(cfg, "video_on_spotlight", True)
|
|
|
|
|
|
def _svc(session: AsyncSession) -> XPostService:
|
|
return get_x_post_service(session)
|
|
|
|
|
|
def _id(task: TaskTable) -> UUID:
|
|
"""The ORM id typed as stdlib ``uuid.UUID`` for service-call sites."""
|
|
return cast("UUID", task.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_posts_and_completes(db_session: AsyncSession) -> None:
|
|
task = await _seed_draft(db_session)
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
assert result.tweet_id == "999"
|
|
assert client.calls == ["Draft body"]
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.COMPLETED
|
|
assert markers.get_x_posted_tweet_id(task) == "999"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_is_idempotent_second_call_is_noop(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
task = await _seed_draft(db_session)
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
svc = _svc(db_session)
|
|
first = await svc.approve(_id(task))
|
|
second = await svc.approve(_id(task))
|
|
assert first is not None
|
|
assert first.status == "posted"
|
|
assert second is not None
|
|
assert second.status == "already_posted"
|
|
assert second.tweet_id == "999"
|
|
# The X client was called exactly once — the second approve never re-posts.
|
|
assert client.calls == ["Draft body"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_with_edited_body_posts_the_edit(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
task = await _seed_draft(db_session, body="Original")
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task), "Edited body")
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
assert client.calls == ["Edited body"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_rejects_edited_body_over_280_chars(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
task = await _seed_draft(db_session)
|
|
with pytest.raises(XPostBodyTooLongError):
|
|
await _svc(db_session).approve(_id(task), "x" * 281)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_no_credentials_result(db_session: AsyncSession) -> None:
|
|
task = await _seed_draft(db_session)
|
|
no_creds_client = _StubClient()
|
|
|
|
class _Null(XClient):
|
|
@property
|
|
def configured(self) -> bool:
|
|
return False
|
|
|
|
async def post_tweet(self, text: str) -> XPostResult:
|
|
_ = text
|
|
return XPostResult(posted=False, tweet_id=None, detail="no creds")
|
|
|
|
async def fetch_mentions(
|
|
self, since_id: str | None, max_results: int
|
|
) -> list[XMention]:
|
|
_ = (since_id, max_results)
|
|
return []
|
|
|
|
_ = no_creds_client
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=_Null()),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "no_credentials"
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.PENDING # never advanced without credentials
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_post_failed_keeps_task_open(db_session: AsyncSession) -> None:
|
|
task = await _seed_draft(db_session)
|
|
client = _StubClient(posted=False)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "post_failed"
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.PENDING
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_rechecks_completed_under_lock_and_never_reposts(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""The double-post guard: a concurrent approve wins the lock, posts, and
|
|
commits COMPLETED after our pre-lock read. Once we acquire the lock, the
|
|
in-lock re-read must see COMPLETED and short-circuit — never re-posting."""
|
|
task = await _seed_draft(db_session)
|
|
client = _StubClient()
|
|
|
|
async def _win_the_race(_self: XPostService, _key: str) -> str:
|
|
# Simulate the concurrent winner: the row is COMPLETED (posted) by the
|
|
# time we hold the lock, exactly as the in-lock expire()+re-fetch sees.
|
|
markers.set_x_posted_tweet_id(task, "111")
|
|
task.status = TS.COMPLETED
|
|
await db_session.flush()
|
|
return "tok"
|
|
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", _win_the_race),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "already_posted"
|
|
assert result.tweet_id == "111"
|
|
# The tweet was never posted a second time.
|
|
assert client.calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_refuses_already_rejected_draft(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""The chokepoint guard: approving a CANCELLED (already-rejected) draft
|
|
refuses and never calls the X client — the reproduced bug (a stale
|
|
Approve after reject re-posting)."""
|
|
task = await _seed_draft(db_session)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(task), "not on-brand")
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "already_rejected"
|
|
assert client.calls == []
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.CANCELLED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_rechecks_cancelled_under_lock_and_never_posts(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""TOCTOU parity with the COMPLETED re-check: a concurrent reject cancels
|
|
the draft after our pre-lock read; the in-lock re-read must see CANCELLED
|
|
and short-circuit — never posting a rejected draft."""
|
|
task = await _seed_draft(db_session)
|
|
client = _StubClient()
|
|
|
|
async def _win_the_race(_self: XPostService, _key: str) -> str:
|
|
task.status = TS.CANCELLED
|
|
await db_session.flush()
|
|
return "tok"
|
|
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", _win_the_race),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "already_rejected"
|
|
assert client.calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_concurrent_lock_held_returns_in_progress(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
task = await _seed_draft(db_session)
|
|
with patch.object(XPostService, "_acquire_lock", AsyncMock(return_value=None)):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "already_in_progress"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_unknown_task_returns_none(db_session: AsyncSession) -> None:
|
|
result = await _svc(db_session).approve(uuid4())
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_records_reason_and_cancels(db_session: AsyncSession) -> None:
|
|
task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
updated = await _svc(db_session).reject(
|
|
_id(task), "Tone doesn't match our voice"
|
|
)
|
|
assert updated is not None
|
|
assert updated.status == TS.CANCELLED
|
|
assert markers.get_x_reject_reason(updated) == "Tone doesn't match our voice"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_refused_while_lock_held_by_concurrent_approve(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""A concurrent approve holds the post lock (mid-tweet-POST); reject must
|
|
fail closed instead of racing a CANCEL under it — previously reject()
|
|
never even attempted the lock, so it could commit CANCELLED to a draft a
|
|
concurrent approve was about to mark COMPLETED, or clobber the approve's
|
|
outcome depending on commit ordering."""
|
|
task = await _seed_draft(db_session)
|
|
with patch.object(XPostService, "_acquire_lock", AsyncMock(return_value=None)):
|
|
result = await _svc(db_session).reject(_id(task), "not relevant")
|
|
assert result is None
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.PENDING
|
|
assert markers.get_x_reject_reason(task) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_open_posts_excludes_terminal(db_session: AsyncSession) -> None:
|
|
open_task = await _seed_draft(db_session)
|
|
rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(rejected_task), "not relevant")
|
|
open_posts = await _svc(db_session).list_open_posts()
|
|
ids = {t.id for t in open_posts}
|
|
assert open_task.id in ids
|
|
assert rejected_task.id not in ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_posts_feature_spotlight_draft(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""The feature-spotlight source rides the same generic post path as
|
|
x_post/x_reply; it only branches for the best-effort video hook below
|
|
(a no-op here since this draft carries no x_feature_ref marker)."""
|
|
task = await _seed_draft(db_session, source=X_FEATURE_SOURCE)
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
assert result.tweet_id == "999"
|
|
assert client.calls == ["Draft body"]
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.COMPLETED
|
|
assert markers.get_x_posted_tweet_id(task) == "999"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_open_posts_includes_feature_spotlight_source(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
task = await _seed_draft(db_session, source=X_FEATURE_SOURCE)
|
|
open_posts = await _svc(db_session).list_open_posts()
|
|
ids = {t.id for t in open_posts}
|
|
assert task.id in ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_completed_raises(db_session: AsyncSession) -> None:
|
|
"""An already-posted (COMPLETED) draft is live on X; rejecting it would
|
|
lie "cancelled (never posted)" while the tweet is public."""
|
|
task = await _seed_draft(db_session)
|
|
markers.set_x_posted_tweet_id(task, "999")
|
|
task.status = TS.COMPLETED
|
|
await db_session.flush()
|
|
with pytest.raises(TaskAlreadyCompletedError):
|
|
await _svc(db_session).reject(_id(task), "nope")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_post_history_excludes_open_drafts(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
open_task = await _seed_draft(db_session)
|
|
rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(rejected_task), "not relevant")
|
|
history = await _svc(db_session).list_post_history()
|
|
ids = {t.id for t in history}
|
|
assert rejected_task.id in ids
|
|
assert open_task.id not in ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_post_history_newest_acted_first(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(rejected_task), "not relevant")
|
|
posted_task = await _seed_draft(db_session)
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
await _svc(db_session).approve(_id(posted_task))
|
|
history = await _svc(db_session).list_post_history()
|
|
ids = [t.id for t in history]
|
|
assert ids.index(posted_task.id) < ids.index(rejected_task.id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_post_history_includes_marker_fields(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
posted_task = await _seed_draft(db_session)
|
|
client = _StubClient(tweet_id="777")
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
await _svc(db_session).approve(_id(posted_task))
|
|
rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(rejected_task), "off-brand tone")
|
|
|
|
history = await _svc(db_session).list_post_history()
|
|
by_id = {t.id: t for t in history}
|
|
assert markers.get_x_posted_tweet_id(by_id[posted_task.id]) == "777"
|
|
assert markers.get_x_reject_reason(by_id[rejected_task.id]) == "off-brand tone"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_post_history_respects_limit(db_session: AsyncSession) -> None:
|
|
tasks = []
|
|
for _ in range(3):
|
|
t = await _seed_draft(db_session, source=X_REPLY_SOURCE)
|
|
with _lock_free():
|
|
await _svc(db_session).reject(_id(t), "not relevant")
|
|
tasks.append(t)
|
|
history = await _svc(db_session).list_post_history(limit=2)
|
|
assert len(history) == TWO
|
|
ids = {t.id for t in history}
|
|
assert tasks[2].id in ids
|
|
assert tasks[1].id in ids
|
|
assert tasks[0].id not in ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_does_not_flush_edited_body_before_lock(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""The CEO's edit must land on the re-read locked row INSIDE the critical
|
|
section, after the COMPLETED check — never on the pre-lock row. A concurrent
|
|
approve that already posted (locked row COMPLETED) must not have this edit
|
|
overwrite the just-posted task's stored body."""
|
|
task = await _seed_draft(db_session, body="Original")
|
|
original_body = markers.get_x_draft_body(task)
|
|
|
|
async def _already_posted_locked(
|
|
_self: XPostService, _task_id: UUID, _task: TaskTable, _trimmed: str | None
|
|
) -> XPostExecuteResult:
|
|
return XPostExecuteResult(
|
|
status="already_posted",
|
|
tweet_id="111",
|
|
detail="this draft was already posted",
|
|
)
|
|
|
|
with (
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch.object(XPostService, "_approve_locked", _already_posted_locked),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task), "new body")
|
|
assert result is not None
|
|
assert result.status == "already_posted"
|
|
await db_session.refresh(task)
|
|
assert markers.get_x_draft_body(task) == original_body
|
|
|
|
|
|
async def _fresh_session(url: str) -> tuple[AsyncSession, AsyncEngine]:
|
|
"""A session on a brand-new engine/connection (caller disposes)."""
|
|
engine = create_async_engine(url, future=True)
|
|
factory = async_sessionmaker(
|
|
bind=engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
return factory(), engine
|
|
|
|
|
|
async def _dispose(session: AsyncSession, engine: AsyncEngine) -> None:
|
|
with contextlib.suppress(Exception):
|
|
await session.rollback()
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_concurrent_edit_does_not_clobber_a_committed_post(
|
|
db_session: AsyncSession, _test_database_url: str
|
|
) -> None:
|
|
"""Redis mutex pre-lock write audit regression: a genuinely concurrent
|
|
approve (a real second session/connection, not an in-process mock) posts
|
|
+ commits COMPLETED in the window between our pre-lock read and our lock
|
|
acquisition. The in-lock re-read must see that committed state and the
|
|
CEO's edited body must never land on the just-posted row — proving the
|
|
fix holds across sessions, not merely within one, mirroring
|
|
VideoPostService's identical cross-session regression test."""
|
|
task = await _seed_draft(db_session, body="Original")
|
|
task_id = _id(task)
|
|
await db_session.commit() # externally visible to the "concurrent" session below
|
|
|
|
real_get = TaskService.get
|
|
injected = False
|
|
|
|
async def _get_then_inject_concurrent_post(
|
|
self: TaskService, tid: UUID
|
|
) -> TaskTable | None:
|
|
"""Fires once, right after the outer pre-lock read — the exact
|
|
window between our read and our own (would-be) pre-lock write."""
|
|
nonlocal injected
|
|
result = await real_get(self, tid)
|
|
if not injected:
|
|
injected = True
|
|
other, other_engine = await _fresh_session(_test_database_url)
|
|
try:
|
|
other_task = await other.get(TaskTable, tid)
|
|
assert other_task is not None
|
|
markers.set_x_posted_tweet_id(other_task, "concurrent-999")
|
|
other_task.status = TS.COMPLETED
|
|
await other.commit()
|
|
finally:
|
|
await _dispose(other, other_engine)
|
|
return result
|
|
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(TaskService, "get", _get_then_inject_concurrent_post),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
):
|
|
result = await _svc(db_session).approve(task_id, "Edited body")
|
|
|
|
assert result is not None
|
|
assert result.status == "already_posted"
|
|
assert result.tweet_id == "concurrent-999"
|
|
# No double-post: the concurrently-committed tweet wins, ours never fires.
|
|
assert client.calls == []
|
|
|
|
fresh, fresh_engine = await _fresh_session(_test_database_url)
|
|
try:
|
|
final = await fresh.get(TaskTable, task_id)
|
|
assert final is not None
|
|
assert final.status == TS.COMPLETED
|
|
assert markers.get_x_posted_tweet_id(final) == "concurrent-999"
|
|
# The edit must never have landed on the just-posted row.
|
|
assert markers.get_x_draft_body(final) == "Original"
|
|
finally:
|
|
await _dispose(fresh, fresh_engine)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Spotlight video hook (Task 4, 2026-07-09 pipeline fixes): moved from
|
|
# authoring time (propose_feature_spotlight) to this posted-success branch so
|
|
# a ux-dev never burns a cycle on a spotlight the CEO then rejects.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_with_video_opens_video_task(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session, video_script="Custom voiceover script")
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
video_engine.open_video_task.assert_awaited_once()
|
|
kwargs = video_engine.open_video_task.call_args.kwargs
|
|
assert kwargs["occasion"] == "spotlight org-memory"
|
|
assert kwargs["platforms"] == ["x", "tiktok"]
|
|
assert kwargs["script"] == "Custom voiceover script"
|
|
assert kwargs["brief"] == "Organizational Memory Loop: Draft body"
|
|
# The spotlight's own project scopes the video authoring — without it the
|
|
# video authored against the deployment-anchor project regardless.
|
|
assert kwargs["project_id"] == task.project_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_video_falls_back_to_brief_script(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""No explicit video_script -> script falls back to the brief, mirroring
|
|
the fallback the authoring-time hook used to do."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session)
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
kwargs = video_engine.open_video_task.call_args.kwargs
|
|
expected_brief = "Organizational Memory Loop: Draft body"
|
|
assert kwargs["script"] == expected_brief
|
|
assert kwargs["brief"] == expected_brief
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_reapprove_does_not_reopen_video(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Idempotent re-approve: the second call short-circuits on the already-
|
|
COMPLETED check before ever reaching _post/_open_spotlight_video again."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session)
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
svc = _svc(db_session)
|
|
first = await svc.approve(_id(task))
|
|
second = await svc.approve(_id(task))
|
|
assert first is not None
|
|
assert first.status == "posted"
|
|
assert second is not None
|
|
assert second.status == "already_posted"
|
|
video_engine.open_video_task.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_plain_x_post_never_opens_video(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A plain x_post draft carries no x_feature_ref, so the source check
|
|
alone keeps the video hook from ever firing for it."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_draft(db_session, source=X_POST_SOURCE)
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
video_engine.open_video_task.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_feature_spotlight_with_wants_video_opens_none(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Rejecting a spotlight draft never posts, so the video hook (which only
|
|
fires from the posted-success branch of _post) never runs either."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session)
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
_lock_free(),
|
|
):
|
|
updated = await _svc(db_session).reject(_id(task), "not on-brand")
|
|
assert updated is not None
|
|
assert updated.status == TS.CANCELLED
|
|
video_engine.open_video_task.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_video_flags_off_skips(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setattr(cfg, "video_engine_enabled", False)
|
|
monkeypatch.setattr(cfg, "video_on_spotlight", False)
|
|
task = await _seed_feature_draft(db_session)
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
video_engine.open_video_task.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_without_wants_video_skips(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Flags on but the draft's author didn't request a video (wants_video
|
|
absent/False on the marker) -> no video task, distinct from the
|
|
flags-off case above."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session, wants_video=False)
|
|
client = _StubClient()
|
|
video_engine = AsyncMock()
|
|
video_engine.open_video_task = AsyncMock(return_value=None)
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
return_value=video_engine,
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
video_engine.open_video_task.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_feature_spotlight_video_failure_does_not_break_post(
|
|
db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Best-effort: a video-engine blow-up must not affect the already-
|
|
succeeded post."""
|
|
_enable_video(monkeypatch)
|
|
task = await _seed_feature_draft(db_session)
|
|
client = _StubClient()
|
|
with (
|
|
patch("roboco.services.x_post_service.build_x_client", return_value=client),
|
|
patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
|
|
patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
|
|
patch(
|
|
"roboco.services.video_engine.get_video_engine",
|
|
side_effect=RuntimeError("video-engine boom"),
|
|
),
|
|
):
|
|
result = await _svc(db_session).approve(_id(task))
|
|
assert result is not None
|
|
assert result.status == "posted"
|
|
await db_session.refresh(task)
|
|
assert task.status == TS.COMPLETED
|