admin_rb
28 lines
1.2 KiB
Python
28 lines
1.2 KiB
Python
from dataclasses import replace
|
|
from pathlib import Path
|
|
import json
|
|
import time
|
|
from aim_webgui.auth.service import Auth
|
|
from aim_webgui.workflows import Workflows
|
|
from aim_webgui.worker import execute_claimed
|
|
from test_core_execution import native_fixture
|
|
|
|
|
|
def test_worker(settings,core_root):
|
|
native_fixture(core_root)
|
|
auth=Auth(settings);auth.bootstrap()
|
|
with auth.store.transaction()as db:
|
|
db.execute("UPDATE users SET must_change_password=0 WHERE username='admin'")
|
|
uid=db.execute("SELECT id FROM users WHERE username='admin'").fetchone()[0]
|
|
flow=Workflows(settings)
|
|
review=flow.review(uid,'example','debug_test_connection',['test01.example'],{})
|
|
ident=flow.queue_review(uid,review['review_id'],idempotency_key='worker-proof')
|
|
with flow.store.transaction()as db:
|
|
db.execute("UPDATE jobs SET status='running' WHERE id=?",(ident,))
|
|
execute_claimed(settings,ident)
|
|
result=flow.job(uid,ident)['core_result']
|
|
assert result['status']=='succeeded' and result['counts']['ok']==1
|
|
assert result['remote_work_may_have_started'] is True
|
|
assert not list(settings.state_dir.glob('.console-*'))
|
|
assert not flow.plans(uid)
|