trending/Octop
1
0
Fork 0
mirror of https://github.com/TencentCloud/Octop.git synced 2026-10-23 06:50:19 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
develop
Octop/tests/unit/proactive/test_scheduler.py
T
薄生andCursor 9738b7d591 fix(proactive): send care only after the chat has gone quiet (#2229) 
Random gaps that missed the care window were pinned to the opening two hours, and a push could land in the middle of a conversation. Wait for an idle chat inside care hours, including windows that cross midnight, and drop the message if the user starts talking while it is being written.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-20 14:33:31 +08:00

597 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Unit tests for ProactiveCareScheduler.
Coverage:
- compute_next_trigger: random scheduling inside active hours
- compute_next_trigger: defer to the next active window when outside active hours
- is_in_active_hours: active-window checks
- ProactiveCareScheduler: enabled=false should not trigger
- ProactiveCareScheduler: dynamic reschedule updates
- ProactiveCareConfigRepo: get/upsert/list_enabled
"""
from __future__ import annotations
import asyncio
from datetime import UTC, datetime, time, timedelta
from pathlib import Path
from unittest.mock import AsyncMock
from zoneinfo import ZoneInfo
import pytest
from octop.infra.db.migrate import run_migrations
from octop.infra.db.pool import SqlitePool
from octop.infra.db.repos.agents import AgentRepo
from octop.infra.db.repos.proactive_care_config import ProactiveCareConfig, ProactiveCareConfigRepo
from octop.infra.db.repos.users import UserRepo
from octop.infra.proactive.scheduler import (
ProactiveCareScheduler,
compute_next_trigger,
decide_send_timing,
is_in_active_hours,
)
from octop.infra.utils.ulid import new_ulid
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def db(tmp_path: Path) -> SqlitePool:
pool = SqlitePool(tmp_path / "test.db")
run_migrations(pool)
return pool
@pytest.fixture
def agent_id(db: SqlitePool) -> str:
uid = UserRepo(db).create(username="alice", password_hash="h", role="admin")
aid = new_ulid()
AgentRepo(db).create(agent_id=aid, user_id=uid, name="bot")
return aid
@pytest.fixture
def config_repo(db: SqlitePool) -> ProactiveCareConfigRepo:
return ProactiveCareConfigRepo(db)
# ---------------------------------------------------------------------------
# compute_next_trigger tests
# ---------------------------------------------------------------------------
def test_compute_next_trigger_in_active_hours():
"""Should return the candidate time directly when it is inside active hours."""
# Current time is UTC 10:00, active window is 09:00-22:00, interval is 1h.
now = datetime(2026, 7, 1, 10, 0, 0, tzinfo=UTC)
# Fix the random seed so the interval stays inside active hours.
import random
random.seed(42)
# Force the interval to 60 minutes, which remains inside active hours.
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 60) # Fixed 60 minutes.
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=1,
max_interval_hours=2,
)
# The result should remain inside active hours.
result_t = result.time().replace(second=0, microsecond=0)
assert time(9, 0) <= result_t < time(22, 0)
def test_compute_next_trigger_outside_active_hours_deferred():
"""Should defer to the next active window when the candidate is outside active hours."""
# Current time is UTC 14:00, active window is 09:00-22:00, interval is 10h.
now = datetime(2026, 7, 1, 14, 0, 0, tzinfo=UTC)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 600) # Fixed 600 minutes = 10h.
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=8,
max_interval_hours=24,
)
# The result should be inside active hours on the next day.
result_t = result.time().replace(second=0, microsecond=0)
assert time(9, 0) <= result_t < time(22, 0)
# The result should be on the next day.
assert result.date() > now.date()
def test_compute_next_trigger_already_past_active_hours():
"""Should defer to tomorrow when now and candidate time are past active hours."""
now = datetime(2026, 7, 1, 23, 0, 0, tzinfo=UTC)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 60) # 1h later = next-day 00:00.
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=1,
max_interval_hours=2,
)
result_t = result.time().replace(second=0, microsecond=0)
assert time(9, 0) <= result_t < time(22, 0)
def test_compute_next_trigger_returns_future():
"""compute_next_trigger should always return a time after now."""
now = datetime(2026, 7, 1, 12, 0, 0, tzinfo=UTC)
for _ in range(20):
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=8,
max_interval_hours=24,
)
assert result > now
# ---------------------------------------------------------------------------
# is_in_active_hours tests
# ---------------------------------------------------------------------------
def test_is_in_active_hours_true():
"""Should return True inside active hours."""
now = datetime(2026, 7, 1, 15, 30, 0, tzinfo=UTC)
assert is_in_active_hours(now, active_hours_start="09:00", active_hours_end="22:00") is True
def test_is_in_active_hours_false_before():
"""Should return False before active hours."""
now = datetime(2026, 7, 1, 8, 0, 0, tzinfo=UTC)
assert is_in_active_hours(now, active_hours_start="09:00", active_hours_end="22:00") is False
def test_is_in_active_hours_false_after():
"""Should return False after active hours."""
now = datetime(2026, 7, 1, 22, 30, 0, tzinfo=UTC)
assert is_in_active_hours(now, active_hours_start="09:00", active_hours_end="22:00") is False
def test_is_in_active_hours_with_timezone():
# 2026-07-01 01:00 UTC is 09:00 Asia/Shanghai.
now = datetime(2026, 7, 1, 1, 0, 0, tzinfo=UTC)
assert (
is_in_active_hours(
now,
active_hours_start="09:00",
active_hours_end="22:00",
timezone_name="Asia/Shanghai",
)
is True
)
def test_is_in_active_hours_boundary_start():
"""Should return True exactly at the active-window start."""
now = datetime(2026, 7, 1, 9, 0, 0, tzinfo=UTC)
assert is_in_active_hours(now, active_hours_start="09:00", active_hours_end="22:00") is True
def test_is_in_active_hours_boundary_end():
"""Should return False exactly at the active-window end."""
now = datetime(2026, 7, 1, 22, 0, 0, tzinfo=UTC)
assert is_in_active_hours(now, active_hours_start="09:00", active_hours_end="22:00") is False
def test_missed_window_waits_for_the_next_opening():
"""A gap that lands at night waits until care hours open, not two hours into them."""
now = datetime(2026, 7, 1, 23, 0, 0, tzinfo=UTC)
def randint(low: int, high: int) -> int:
# Minimum gap, then no opening jitter.
return low
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", randint)
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=1,
max_interval_hours=2,
)
assert result == datetime(2026, 7, 2, 9, 0, tzinfo=UTC)
def test_short_window_keeps_opening_jitter_inside():
now = datetime(2026, 7, 1, 23, 0, 0, tzinfo=UTC)
def randint(low: int, high: int) -> int:
return high
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", randint)
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="10:00",
min_interval_hours=1,
max_interval_hours=2,
)
result_t = result.time().replace(second=0, microsecond=0)
# Jitter is capped by the 60-minute window, so 09:00 + 120 cannot escape it.
assert time(9, 0) <= result_t < time(10, 0)
def test_config_body_allows_overnight_window():
from pydantic import ValidationError
from octop.api.routers.proactive_care import ProactiveCareConfigBody
cfg = ProactiveCareConfigBody(active_hours_start="22:00", active_hours_end="08:00")
assert cfg.active_hours_start == "22:00"
with pytest.raises(ValidationError):
ProactiveCareConfigBody(active_hours_start="09:00", active_hours_end="09:00")
def test_overnight_window_accepts_evening_and_morning():
evening = datetime(2026, 7, 1, 22, 30, tzinfo=UTC)
morning = datetime(2026, 7, 2, 3, 0, tzinfo=UTC)
midday = datetime(2026, 7, 1, 12, 0, tzinfo=UTC)
assert is_in_active_hours(evening, active_hours_start="22:00", active_hours_end="08:00")
assert is_in_active_hours(morning, active_hours_start="22:00", active_hours_end="08:00")
assert not is_in_active_hours(midday, active_hours_start="22:00", active_hours_end="08:00")
assert not is_in_active_hours(
datetime(2026, 7, 2, 8, 0, tzinfo=UTC),
active_hours_start="22:00",
active_hours_end="08:00",
)
def test_overnight_gap_lands_inside_the_night_window():
now = datetime(2026, 7, 1, 12, 0, tzinfo=UTC)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 60)
result = compute_next_trigger(
now=now,
active_hours_start="22:00",
active_hours_end="08:00",
min_interval_hours=1,
max_interval_hours=2,
)
assert is_in_active_hours(result, active_hours_start="22:00", active_hours_end="08:00")
assert result > now
def _unix(moment: datetime) -> int:
return int(moment.timestamp())
def test_recent_user_activity_defers_until_idle():
now = datetime(2026, 7, 1, 15, 0, tzinfo=UTC)
last_active = now - timedelta(minutes=30)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
last_active_at=_unix(last_active),
)
assert decision.send is False
assert decision.retry_at == last_active + timedelta(hours=4)
def test_own_care_push_does_not_reset_idle():
now = datetime(2026, 7, 1, 15, 0, tzinfo=UTC)
pushed = now - timedelta(hours=6)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
last_active_at=_unix(pushed) + 60,
last_push_at=_unix(pushed),
)
assert decision.send is True
def test_reply_after_care_restarts_idle():
now = datetime(2026, 7, 1, 15, 0, tzinfo=UTC)
pushed = now - timedelta(hours=6)
replied = now - timedelta(minutes=20)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
last_active_at=_unix(replied),
last_push_at=_unix(pushed),
)
assert decision.send is False
assert decision.retry_at == replied + timedelta(hours=4)
def test_busy_turn_waits_without_leaving_the_window():
now = datetime(2026, 7, 1, 15, 0, tzinfo=UTC)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
agent_busy=True,
)
assert decision.send is False
assert decision.retry_at == now + timedelta(minutes=15)
def test_outside_hours_waits_for_the_window_to_open():
now = datetime(2026, 7, 1, 23, 0, tzinfo=UTC)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 0)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
agent_busy=True,
)
assert decision.send is False
assert decision.retry_at == datetime(2026, 7, 2, 9, 0, tzinfo=UTC)
def test_idle_deadline_after_hours_moves_to_the_next_opening():
now = datetime(2026, 7, 1, 21, 0, tzinfo=UTC)
last_active = now - timedelta(minutes=30)
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 0)
decision = decide_send_timing(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
last_active_at=_unix(last_active),
)
assert decision.send is False
# Quiet at 00:30, which is outside 09:00–22:00, so wait until the window opens.
assert decision.retry_at == datetime(2026, 7, 2, 9, 0, tzinfo=UTC)
def test_compute_next_trigger_with_timezone():
now = datetime(2026, 7, 1, 1, 0, 0, tzinfo=UTC) # 09:00 Shanghai
with pytest.MonkeyPatch.context() as mp:
mp.setattr("random.randint", lambda a, b: 60) # 1 hour
result = compute_next_trigger(
now=now,
active_hours_start="09:00",
active_hours_end="22:00",
min_interval_hours=1,
max_interval_hours=2,
timezone_name="Asia/Shanghai",
)
local = result.astimezone(ZoneInfo("Asia/Shanghai"))
assert local.time().hour == 10
# ---------------------------------------------------------------------------
# ProactiveCareConfigRepo tests
# ---------------------------------------------------------------------------
def test_config_repo_get_default(agent_id: str, config_repo: ProactiveCareConfigRepo):
"""Should return defaults when no config exists."""
cfg = config_repo.get(agent_id)
assert cfg.agent_id == agent_id
assert cfg.enabled is True
assert cfg.active_hours_start == "09:00"
assert cfg.active_hours_end == "22:00"
assert cfg.min_interval_hours == 5
assert cfg.max_interval_hours == 24
def test_config_repo_upsert_and_get(agent_id: str, config_repo: ProactiveCareConfigRepo):
"""Should read the saved config after upsert."""
cfg = ProactiveCareConfig(
agent_id=agent_id,
enabled=True,
active_hours_start="10:00",
active_hours_end="20:00",
min_interval_hours=4,
max_interval_hours=12,
)
config_repo.upsert(cfg)
got = config_repo.get(agent_id)
assert got.enabled is True
assert got.active_hours_start == "10:00"
assert got.active_hours_end == "20:00"
assert got.min_interval_hours == 4
assert got.max_interval_hours == 12
def test_config_repo_upsert_update(agent_id: str, config_repo: ProactiveCareConfigRepo):
"""Repeated upsert should update the existing config."""
cfg1 = ProactiveCareConfig(agent_id=agent_id, enabled=True)
config_repo.upsert(cfg1)
cfg2 = ProactiveCareConfig(agent_id=agent_id, enabled=False, min_interval_hours=12)
config_repo.upsert(cfg2)
got = config_repo.get(agent_id)
assert got.enabled is False
assert got.min_interval_hours == 12
def test_config_repo_list_enabled(db: SqlitePool, config_repo: ProactiveCareConfigRepo):
"""list_enabled should return only enabled configs."""
uid = UserRepo(db).create(username="bob", password_hash="h", role="admin")
aid1 = new_ulid()
aid2 = new_ulid()
AgentRepo(db).create(agent_id=aid1, user_id=uid, name="bot1")
AgentRepo(db).create(agent_id=aid2, user_id=uid, name="bot2")
config_repo.upsert(ProactiveCareConfig(agent_id=aid1, enabled=True))
config_repo.upsert(ProactiveCareConfig(agent_id=aid2, enabled=False))
enabled = config_repo.list_enabled()
enabled_ids = {c.agent_id for c in enabled}
assert aid1 in enabled_ids
assert aid2 not in enabled_ids
def test_config_repo_list_enabled_includes_default_on_agents(
db: SqlitePool, config_repo: ProactiveCareConfigRepo
):
"""Agents with no config row default to enabled and must be scheduled."""
uid = UserRepo(db).create(username="carol", password_hash="h", role="admin")
aid_default = new_ulid()
aid_off = new_ulid()
AgentRepo(db).create(agent_id=aid_default, user_id=uid, name="default-on")
AgentRepo(db).create(agent_id=aid_off, user_id=uid, name="opted-out")
config_repo.upsert(ProactiveCareConfig(agent_id=aid_off, enabled=False))
enabled = config_repo.list_enabled()
enabled_ids = {c.agent_id for c in enabled}
assert aid_default in enabled_ids
assert aid_off not in enabled_ids
default_cfg = next(c for c in enabled if c.agent_id == aid_default)
assert default_cfg.enabled is True
assert default_cfg.min_interval_hours == 5
assert default_cfg.max_interval_hours == 24
# ---------------------------------------------------------------------------
# ProactiveCareScheduler tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_scheduler_cancel_stops_task(
agent_id: str,
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
):
"""cancel should stop the scheduled task."""
from octop.infra.db.repos.sessions import SessionRepo
care_service = AsyncMock()
session_repo = SessionRepo(db)
# Enable the config.
config_repo.upsert(ProactiveCareConfig(agent_id=agent_id, enabled=True))
scheduler = ProactiveCareScheduler(
care_service=care_service,
config_repo=config_repo,
session_repo=session_repo,
)
# Schedule manually.
scheduler._schedule(agent_id)
assert agent_id in scheduler._tasks
# Cancel.
scheduler.cancel(agent_id)
# Wait for the task to be cancelled.
await asyncio.sleep(0.01)
assert agent_id not in scheduler._tasks
@pytest.mark.asyncio
async def test_scheduler_reschedule_disabled(
agent_id: str,
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
):
"""reschedule should cancel the task when enabled=false."""
from octop.infra.db.repos.sessions import SessionRepo
care_service = AsyncMock()
session_repo = SessionRepo(db)
# Enable first.
config_repo.upsert(ProactiveCareConfig(agent_id=agent_id, enabled=True))
scheduler = ProactiveCareScheduler(
care_service=care_service,
config_repo=config_repo,
session_repo=session_repo,
)
scheduler._schedule(agent_id)
assert agent_id in scheduler._tasks
# Disable, then reschedule.
config_repo.upsert(ProactiveCareConfig(agent_id=agent_id, enabled=False))
scheduler.reschedule(agent_id)
await asyncio.sleep(0.01)
assert agent_id not in scheduler._tasks
@pytest.mark.asyncio
async def test_scheduler_start_all_no_enabled(
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
):
"""start_all should not create tasks when no agent is enabled."""
from octop.infra.db.repos.sessions import SessionRepo
care_service = AsyncMock()
session_repo = SessionRepo(db)
scheduler = ProactiveCareScheduler(
care_service=care_service,
config_repo=config_repo,
session_repo=session_repo,
)
await scheduler.start_all()
assert len(scheduler._tasks) == 0
@pytest.mark.asyncio
async def test_scheduler_start_all_includes_default_on_agent(
agent_id: str,
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
):
"""start_all should schedule agents that never saved a proactive-care row."""
from octop.infra.db.repos.sessions import SessionRepo
care_service = AsyncMock()
scheduler = ProactiveCareScheduler(
care_service=care_service,
config_repo=config_repo,
session_repo=SessionRepo(db),
)
await scheduler.start_all()
assert agent_id in scheduler._tasks
scheduler.cancel(agent_id)
await asyncio.sleep(0.01)
@pytest.mark.asyncio
async def test_shutdown_cancels_long_sleep_quickly(
agent_id: str,
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
) -> None:
"""A default-on loop sleeps for hours; shutdown must not wait that out."""
from octop.infra.db.repos.sessions import SessionRepo
scheduler = ProactiveCareScheduler(
care_service=AsyncMock(),
config_repo=config_repo,
session_repo=SessionRepo(db),
)
scheduler.ensure_scheduled(agent_id)
assert agent_id in scheduler._tasks
await asyncio.wait_for(scheduler.shutdown(), timeout=2)
assert scheduler._tasks == {}
@pytest.mark.asyncio
async def test_suspend_skips_new_schedules(
agent_id: str,
config_repo: ProactiveCareConfigRepo,
db: SqlitePool,
) -> None:
from octop.infra.db.repos.sessions import SessionRepo
scheduler = ProactiveCareScheduler(
care_service=AsyncMock(),
config_repo=config_repo,
session_repo=SessionRepo(db),
)
scheduler.suspend()
scheduler.ensure_scheduled(agent_id)
await scheduler.start_all()
assert scheduler._tasks == {}
Reference in New Issue View Git Blame Copy Permalink