Spaces:
Running
Running
File size: 1,823 Bytes
7cc81cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_phase9_migration_is_additive_and_forces_batch_rls() -> None:
migration = (
ROOT / "app/social/migrations/0009_publishing_operations.sql"
).read_text()
normalized = migration.lower()
assert "drop table" not in normalized
assert "social_posts add column if not exists revision" in migration
assert "social_schedules add column if not exists revision" in migration
assert "create table if not exists social_publishing_batches" in migration
assert "create table if not exists social_publishing_batch_items" in migration
assert normalized.count("force row level security") >= 2
assert "current_setting('app.workspace_id'" in migration
def test_phase9_uses_existing_posts_schedules_jobs_and_worker() -> None:
service = (
ROOT / "app/social/services/publishing_operations_service.py"
).read_text()
scheduler = (ROOT / "app/social/workers/scheduler.py").read_text()
assert "PostRepository" in service
assert "JobRepository" in service
assert "SchedulingService" in service
assert "process_batch_items" in scheduler
assert "SocialJob(" not in service
def test_phase9_transports_expose_narrow_typed_operations() -> None:
api = (ROOT / "app/api/social.py").read_text()
mcp = (ROOT / "app/mcp/tools/social.py").read_text()
for route in (
'"/drafts"',
'"/posts/{post_id}/duplicate"',
'"/posts/{post_id}/reschedule"',
'"/calendar"',
'"/queue"',
'"/bulk"',
):
assert route in api
for tool in (
"publishing.list_queue",
"publishing.list_calendar",
"publishing.reschedule",
"publishing.duplicate",
"publishing.bulk",
):
assert tool in mcp
|