mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-06 23:12:22 +02:00
pewdiepie-archdaemon
2e8413a54a
Snapshot current maintainer-preview application changes and regression fixtures for integration into lab. Excludes local runtime data, evaluation outputs and source backups. Focused Python regression selection: 140 passed; full suite not certified.
174 lines
7.7 KiB
Python
174 lines
7.7 KiB
Python
import json
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from core import database
|
|
from src.tools.system import do_manage_tasks
|
|
|
|
|
|
@pytest.fixture
|
|
def storage(monkeypatch, tmp_path):
|
|
engine = create_engine('sqlite:///' + str(tmp_path / 'tasks.db'))
|
|
database.Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine)
|
|
monkeypatch.setattr(database, 'SessionLocal', factory)
|
|
monkeypatch.setattr('src.task_scheduler._utcnow', lambda: datetime(2026, 10, 4))
|
|
yield factory
|
|
engine.dispose()
|
|
|
|
|
|
async def call(**args):
|
|
return await do_manage_tasks(json.dumps(args), owner='fixture')
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cron_create_edit_resume_and_invalid_edit_rollback(storage):
|
|
result = await call(action='create', name='Digest', prompt='Summarize technology news.',
|
|
schedule='cron', cron_expression='15 9 * * 1,3,5')
|
|
assert result['exit_code'] == 0, result
|
|
task_id = result['task_id']
|
|
|
|
def saved():
|
|
with storage() as db:
|
|
return db.get(database.ScheduledTask, task_id)
|
|
|
|
assert saved().next_run == datetime(2026, 10, 5, 9, 15)
|
|
assert saved().cron_expression == '15 9 * * 1,3,5'
|
|
assert '15 9 * * 1,3,5' in (await call(action='list'))['response']
|
|
edited = await call(action='edit', task_id=task_id, cron_expression='45 8 * * 2,4')
|
|
assert edited['exit_code'] == 0, edited
|
|
assert saved().next_run == datetime(2026, 10, 6, 8, 45)
|
|
for action in ('pause', 'resume'):
|
|
assert (await call(action=action, task_id=task_id))['exit_code'] == 0
|
|
assert saved().next_run == datetime(2026, 10, 6, 8, 45)
|
|
rejected = await call(action='edit', task_id=task_id, name='Must not stick', cron_expression='bad')
|
|
assert rejected['exit_code'] == 1
|
|
assert saved().name == 'Digest'
|
|
assert saved().cron_expression == '45 8 * * 2,4'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('expression', [None, '', 'bad', '99 99 * * *'])
|
|
async def test_invalid_cron_does_not_create_inert_task(storage, expression):
|
|
result = await call(action='create', name='Invalid', prompt='Test', schedule='cron',
|
|
cron_expression=expression)
|
|
assert result['exit_code'] == 1
|
|
with storage() as db:
|
|
assert db.query(database.ScheduledTask).count() == 0
|
|
|
|
|
|
def test_compact_schema_exposes_cron():
|
|
from src.clean_agent_preview import compact_schemas
|
|
from src.tool_schemas import FUNCTION_TOOL_SCHEMAS
|
|
schema = next(s for s in FUNCTION_TOOL_SCHEMAS if s['function']['name'] == 'manage_tasks')
|
|
props = compact_schemas([schema], model='Ajax')[0]['function']['parameters']['properties']
|
|
assert 'cron' in props['schedule']['enum']
|
|
assert 'cron_expression' in props
|
|
assert '0=Sunday' in props['cron_expression']['description']
|
|
assert 'scheduled_day' not in props
|
|
assert props['day_of_month']['minimum'] == 1
|
|
assert props['day_of_month']['maximum'] == 31
|
|
assert 'UTC' in props['scheduled_time']['description']
|
|
assert 'weekdays' in props
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_named_weekdays_create_and_edit_preserve_actual_clock(storage):
|
|
created = await call(action='create', name='Digest', prompt='News',
|
|
weekdays=['monday', 'wednesday', 'friday'], scheduled_time='09:15')
|
|
assert created['exit_code'] == 0, created
|
|
with storage() as db:
|
|
row = db.get(database.ScheduledTask, created['task_id'])
|
|
assert row.schedule == 'cron'
|
|
assert row.cron_expression == '15 9 * * 1,3,5'
|
|
# Direct-cron tasks may have a different legacy scheduled_time field.
|
|
raw = await call(action='create', name='Cron', prompt='News', schedule='cron',
|
|
cron_expression='45 8 * * 1,3,5')
|
|
edited = await call(action='edit', task_id=raw['task_id'], weekdays=['tuesday', 'thursday'])
|
|
assert edited['exit_code'] == 0, edited
|
|
with storage() as db:
|
|
row = db.get(database.ScheduledTask, raw['task_id'])
|
|
assert row.cron_expression == '45 8 * * 2,4'
|
|
assert row.next_run == datetime(2026, 10, 6, 8, 45)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('extra', [
|
|
{'weekdays': []}, {'weekdays': ['noday']}, {'weekdays': 'monday'},
|
|
{'weekdays': ['monday'], 'scheduled_time': '25:00'},
|
|
{'weekdays': ['monday'], 'scheduled_time': None},
|
|
{'weekdays': ['monday'], 'cron_expression': '0 9 * * *'},
|
|
{'weekdays': ['monday'], 'scheduled_day': 0},
|
|
{'weekdays': ['monday'], 'trigger_type': 'event'},
|
|
{'weekdays': ['monday'], 'schedule': 'once'},
|
|
])
|
|
async def test_invalid_weekday_schedule_is_not_saved(storage, extra):
|
|
result = await call(**{'action': 'create', 'name': 'Invalid', 'prompt': 'Test',
|
|
'scheduled_time': '09:15', **extra})
|
|
assert result['exit_code'] == 1
|
|
with storage() as db:
|
|
assert db.query(database.ScheduledTask).count() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_month_day_create_and_edit(storage):
|
|
result = await call(action='create', name='Monthly', prompt='Review budget',
|
|
day_of_month=15, scheduled_time='08:00')
|
|
assert result['exit_code'] == 0, result
|
|
with storage() as db:
|
|
row = db.get(database.ScheduledTask, result['task_id'])
|
|
assert row.schedule == 'monthly' and row.scheduled_day == 15
|
|
assert row.next_run == datetime(2026, 10, 15, 8)
|
|
assert (await call(action='edit', task_id=result['task_id'], day_of_month=20))['exit_code'] == 0
|
|
with storage() as db:
|
|
assert db.get(database.ScheduledTask, result['task_id']).next_run == datetime(2026, 10, 20, 8)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('day', [0, 32, True, '15', None])
|
|
async def test_invalid_month_day_is_not_saved(storage, day):
|
|
result = await call(action='create', name='Invalid', prompt='Test', day_of_month=day)
|
|
assert result['exit_code'] == 1
|
|
with storage() as db:
|
|
assert db.query(database.ScheduledTask).count() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('schedule', ['daily', 'weekly', 'monthly', 'cron'])
|
|
async def test_recurring_schedule_does_not_silently_ignore_one_off_date(storage, schedule):
|
|
result = await call(action='create', name='Invalid', prompt='Test', schedule=schedule,
|
|
scheduled_date='2099-01-15T08:00:00Z')
|
|
assert result['exit_code'] == 1
|
|
assert 'scheduled_date is only' in result['error']
|
|
with storage() as db:
|
|
assert db.query(database.ScheduledTask).count() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('expression', ['15 9 * * 1,3,5', '15 9 15 * *', '0,30 8-10 * * 2,4'])
|
|
async def test_time_only_edit_changes_cron_clock_not_calendar_fields(storage, expression):
|
|
result = await call(action='create', name='Digest', prompt='News', schedule='cron',
|
|
cron_expression=expression)
|
|
edited = await call(action='edit', task_id=result['task_id'], scheduled_time='10:30')
|
|
assert edited['exit_code'] == 0, edited
|
|
with storage() as db:
|
|
row = db.get(database.ScheduledTask, result['task_id'])
|
|
assert row.cron_expression.split() == ['30', '10', *expression.split()[2:]]
|
|
assert row.next_run.hour == 10 and row.next_run.minute == 30
|
|
assert row.name == 'Digest' and row.prompt == 'News'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_cron_retime_rolls_back_all_edits(storage):
|
|
result = await call(action='create', name='Digest', prompt='News', schedule='cron',
|
|
cron_expression='15 9 * * 1,3,5')
|
|
edited = await call(action='edit', task_id=result['task_id'], scheduled_time='25:00', name='Wrong')
|
|
assert edited['exit_code'] == 1
|
|
with storage() as db:
|
|
row = db.get(database.ScheduledTask, result['task_id'])
|
|
assert row.name == 'Digest' and row.cron_expression == '15 9 * * 1,3,5'
|
|
assert row.next_run == datetime(2026, 10, 5, 9, 15)
|