46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""modules/job/service.py 暂停/恢复任务：update_user 应取当前操作人，而非创建人。"""
|
|
import json
|
|
import types
|
|
|
|
from modules.job import service as job_service
|
|
from modules.job.models import Job
|
|
|
|
|
|
def _make_request(user_id=5, realname='测试用户'):
|
|
req = types.SimpleNamespace()
|
|
req.state = types.SimpleNamespace(user_id=user_id, realname=realname)
|
|
return req
|
|
|
|
|
|
def _make_job(sqlite_db, status=1, create_user=1):
|
|
job = Job(job_name='任务A', status=status, cron_expression='0 * * * *',
|
|
execute_policy=1, is_sync=0, create_user=create_user)
|
|
job.save()
|
|
sqlite_db.commit()
|
|
return job
|
|
|
|
|
|
def _body(resp):
|
|
return json.loads(resp.body)
|
|
|
|
|
|
def test_pause_job_sets_update_user_to_operator(monkeypatch, sqlite_db):
|
|
monkeypatch.setattr(job_service, 'pause_job_in_scheduler', lambda jid: True)
|
|
job = _make_job(sqlite_db, status=1, create_user=1)
|
|
|
|
resp = job_service.pause_job_by_id(_make_request(user_id=5, realname='张三'), job.id)
|
|
assert _body(resp)['code'] == 0
|
|
assert job.status == 2
|
|
# 更新人 = 操作人姓名（张三），而非创建人（1）
|
|
assert job.update_user == '张三'
|
|
|
|
|
|
def test_resume_job_sets_update_user_to_operator(monkeypatch, sqlite_db):
|
|
monkeypatch.setattr(job_service, 'add_job_to_scheduler', lambda data: True)
|
|
job = _make_job(sqlite_db, status=2, create_user=1)
|
|
|
|
resp = job_service.resume_job_by_id(_make_request(user_id=5, realname='张三'), job.id)
|
|
assert _body(resp)['code'] == 0
|
|
assert job.status == 1
|
|
assert job.update_user == '张三'
|