admin_rb
105 lines
5.0 KiB
Python
105 lines
5.0 KiB
Python
import json
|
|
import concurrent.futures
|
|
import sqlite3
|
|
import time
|
|
from dataclasses import replace
|
|
import pytest
|
|
from aim_webgui.workflows import Workflows, presentation_status
|
|
from aim_webgui.errors import WebError
|
|
|
|
|
|
def reviewed(flow,admin,**kw):
|
|
return flow.review(admin,'example','debug_test_connection',['test01.example'],{},**kw)
|
|
|
|
|
|
|
|
def test_partial_success_presentation_uses_core_target_summary():
|
|
core={'target_summary':{'schema':'target_outcome_summary_v1','requested':25,'successful':24,'failed':0,'unreachable':1,'not_started':0,'indeterminate':0,'complete':True,'accounted':25}}
|
|
assert presentation_status('failed',core)=='partially_succeeded'
|
|
assert presentation_status('successful',core)=='successful'
|
|
core['target_summary']['successful']=0;core['target_summary']['failed']=24
|
|
assert presentation_status('failed',core)=='failed'
|
|
|
|
def test_one_run_without_saving(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
ident=f.queue_review(admin,r['review_id'],idempotency_key='OneRun1')
|
|
assert f.job(admin,ident)['status']=='queued'
|
|
assert f.plans(admin)==[]
|
|
assert f.queue_review(admin,r['review_id'],idempotency_key='OneRun1')==ident
|
|
with pytest.raises(WebError):f.queue_review(admin,r['review_id'],scheduled_at=int(time.time())+60,idempotency_key='OneRun1')
|
|
|
|
|
|
def test_optional_plan_names_and_collision(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
a=f.save_review(admin,r['review_id'],'');b=f.save_review(admin,r['review_id'],'')
|
|
assert a!=b and f.plan(admin,a)['name']!=f.plan(admin,b)['name']
|
|
p=f.save_review(admin,r['review_id'],'My Plan')
|
|
with pytest.raises(WebError) as e:f.save_review(admin,r['review_id'],' MY PLAN ')
|
|
assert e.value.code=='plan_name_exists'
|
|
assert f.plan(admin,p)['name']=='My Plan'
|
|
assert len(f.plans(admin))==3
|
|
|
|
|
|
def test_unicode_name_equivalence(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
f.save_review(admin,r['review_id'],'Plan A')
|
|
with pytest.raises(WebError):f.save_review(admin,r['review_id'],'Plan \uff21')
|
|
|
|
|
|
def test_concurrent_saves_do_not_overwrite(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
def save():
|
|
try:return f.save_review(admin,r['review_id'],'concurrent')
|
|
except WebError as e:return e.code
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex:results=list(ex.map(lambda _:save(),range(2)))
|
|
assert results.count('plan_name_exists')==1
|
|
assert len(f.plans(admin))==1
|
|
|
|
|
|
def test_stale_review_rejected(settings,admin,core_root):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
with (core_root/'inventories/example/hosts.yml').open('a') as file:file.write('# external edit\n')
|
|
with pytest.raises(WebError) as e:f.queue_review(admin,r['review_id'],idempotency_key='fresh1')
|
|
assert e.value.code=='plan_stale'
|
|
|
|
|
|
def test_mode_and_authentication_bound_to_revision(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin,check=False)
|
|
ident=f.queue_review(admin,r['review_id'],idempotency_key='mode1')
|
|
job=f.job(admin,ident)
|
|
assert job['mode']=='apply' and job['plan']['core_request']['check'] is False
|
|
assert job['plan']['authentication']['mode']=='inventory'
|
|
|
|
|
|
def test_pending_approval_when_enabled(settings,admin):
|
|
f=Workflows(replace(settings,execution_require_approval=True));r=reviewed(f,admin)
|
|
ident=f.queue_review(admin,r['review_id'],idempotency_key='approval1')
|
|
assert f.job(admin,ident)['status']=='pending'
|
|
with pytest.raises(WebError):f.job_action(admin,ident,'approve')
|
|
|
|
|
|
def test_core_opt_in_separate(settings,admin,core_root):
|
|
config=core_root/'scripts/aim.yml';config.write_text(config.read_text().replace('execution_enabled: true','execution_enabled: false'))
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
with pytest.raises(WebError) as e:f.queue_review(admin,r['review_id'],idempotency_key='disabled1')
|
|
assert e.value.code=='core_execution_disabled'
|
|
|
|
|
|
def test_retry_creates_fresh_job_and_bulk_delete_preserves_audit(settings,admin):
|
|
f=Workflows(settings);r=reviewed(f,admin)
|
|
one=f.queue_review(admin,r['review_id'],idempotency_key='retry1')
|
|
with f.store.transaction() as db:db.execute("UPDATE jobs SET status='failed' WHERE id=?",(one,))
|
|
two=f.retry_job(admin,one)
|
|
assert two!=one and f.job(admin,two)['status']=='queued'
|
|
with pytest.raises(WebError):f.delete_jobs(admin,[one,two])
|
|
f.delete_jobs(admin,[one])
|
|
with f.store.read() as db:assert db.execute("SELECT 1 FROM audit WHERE action='job-deleted'").fetchone()
|
|
|
|
|
|
def test_legacy_plan_requires_fresh_review(settings,admin):
|
|
f=Workflows(settings)
|
|
with f.store.transaction() as db:
|
|
db.execute('INSERT INTO plans(id,owner_id,name,customer,playbook,payload,created_at) VALUES(?,?,?,?,?,?,?)',('a'*32,admin,'legacy','example','debug_test_connection',json.dumps({'customer':'example','playbook':'debug_test_connection','targets':['test01.example'],'overrides':{}}),0))
|
|
assert f.plan(admin,'a'*32)['name']=='legacy'
|
|
assert 'core_request' not in f.plan(admin,'a'*32)['payload']
|