admin_rb
131 lines
8.3 KiB
Python
131 lines
8.3 KiB
Python
from copy import deepcopy
|
|
from dataclasses import replace
|
|
import json,os,time
|
|
from pathlib import Path
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.auth.service import Auth
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui.core.reports import contract,data,operation_result,encoded,RUN_BYTES
|
|
from aim_webgui.core.protocol import response_result
|
|
from aim_webgui.reports import persist,Reports,presentation,TITLES
|
|
from aim_webgui.workflows import Workflows,presentation_status
|
|
from aim_webgui.errors import WebError
|
|
from evidence_fixtures import declared,sample,plan,result,job
|
|
|
|
@pytest.fixture
|
|
def local(tmp_path):
|
|
s=Settings(state_dir=tmp_path/'state',public_url='https://aim.example.test')
|
|
a=Auth(s);a.bootstrap()
|
|
with a.store.transaction()as db:db.execute('UPDATE users SET must_change_password=0');uid=db.execute('SELECT id FROM users').fetchone()[0]
|
|
return s,a,uid
|
|
|
|
@pytest.mark.parametrize('schema',sorted(TITLES))
|
|
def test_all_nine_real_schema_contracts_and_retained_views(local,schema):
|
|
s,a,u=local;decl=declared(schema);p=plan(decl);r=result(decl);ident=job(a,u,p,status='successful')
|
|
assert contract(decl)['schema']==schema
|
|
projected=response_result({'type':'response','api_version':'1.0','ok':True,'result':r},'execute',declaration=decl,request=p['core_request'])
|
|
with a.store.transaction()as db:
|
|
small=persist(db,s,ident,p,projected)
|
|
db.execute('UPDATE jobs SET core_result=? WHERE id=?',(json.dumps(small),ident))
|
|
index=Reports(s).index(u,ident);item=Reports(s).slot(u,ident)
|
|
assert index['recorded'] and item['status']=='available'
|
|
if schema=='checkmk_user_config_v1':assert item['retention']=='metadata_only' and 'sections'not in item['data']
|
|
else:assert item['data']==r['operation_result']['hosts']['test01.example']['data']
|
|
assert 'data'not in small['operation_result']['hosts']['test01.example']
|
|
view=presentation(item);assert isinstance(view['facts'],list)
|
|
token,session=a.new_session(u)
|
|
with TestClient(create_app(s),base_url=s.public_url)as c:
|
|
c.cookies.set(s.cookie_name,token)
|
|
for path in (f'/jobs/{ident}',f'/jobs/{ident}/reports',f'/api/v2/runs/{ident}/reports',f'/api/v2/runs/{ident}/report'):
|
|
response=c.get(path);assert response.status_code==200,response.text
|
|
|
|
@pytest.mark.parametrize('state',['missing','withheld','invalid','not_started','indeterminate'])
|
|
def test_unavailable_is_not_empty_data(local,state):
|
|
s,a,u=local;decl=declared();p=plan(decl);r=result(decl,availability=state,status='failed')
|
|
clean=operation_result(r['operation_result'],decl,targets=p['targets'],check=False)
|
|
assert clean['hosts']['test01.example']['data']is None
|
|
ident=job(a,u,p,status='failed')
|
|
with a.store.transaction()as db:persist(db,s,ident,p,r)
|
|
item=Reports(s).slot(u,ident);assert item['status']==state and item['data']is None
|
|
|
|
@pytest.mark.parametrize('mutation',[
|
|
lambda r:r.update(schema='wrong_v1'),lambda r:r.update(scope='global'),lambda r:r.update(check_mode=True),
|
|
lambda r:r['hosts'].update({'other.example':r['hosts']['test01.example']}),
|
|
lambda r:r['hosts']['test01.example']['data'].update(is_dc='false'),
|
|
lambda r:r['hosts']['test01.example']['data'].update(unrecognized='secret'),
|
|
lambda r:r['hosts']['test01.example'].update(status='withheld'),
|
|
lambda r:r['hosts']['test01.example'].update(error={'message':'RAW'})])
|
|
def test_mismatched_or_invalid_report_rejected(mutation):
|
|
decl=declared();value=result(decl)['operation_result'];mutation(value)
|
|
with pytest.raises(WebError):operation_result(value,decl,targets=['test01.example'],check=False)
|
|
|
|
def test_required_report_failure_preserves_native_success(local):
|
|
s,a,u=local;decl=declared();r=result(decl,availability='missing',status='failed');p=plan(decl)
|
|
out=response_result({'type':'response','api_version':'1.0','ok':False,'result':r},'execute',declaration=decl,request=p['core_request'])
|
|
assert out['stage']=='result_validation' and out['exit_code']==0
|
|
assert out['targets'][0]['outcome']=='successful' and presentation_status('failed',out)=='failed'
|
|
assert out['operation_result']['hosts']['test01.example']['status']=='missing'
|
|
|
|
def test_global_and_unknown_supported_schema():
|
|
decl=declared(scope='global');decl['schema']='future_capabilities_v1'
|
|
r=result(decl)['operation_result'];out=operation_result(r,decl,targets=['test01.example'],check=False)
|
|
assert out['hosts']=={} and out['global']['status']=='available'
|
|
|
|
@pytest.mark.parametrize('badshape',[{'$ref':'https://evil.invalid/schema'}, {'type':'string'}, {'type':'array','items':{'type':'boolean'},'maxItems':50000}])
|
|
def test_unsupported_schema_language_rejected(badshape):
|
|
decl=declared();decl['data_schema']=badshape
|
|
with pytest.raises(WebError):contract(decl)
|
|
|
|
def test_null_false_zero_and_redacted_metadata(local):
|
|
s,a,u=local;decl=declared('checkmk_user_config_v1');content=sample(decl['data_schema'])
|
|
content['sections']={'plugins':{'enabled':False,'count':0,'missing':None,'password':'[REDACTED]','normal':'fixture-config'}}
|
|
content['redacted_paths']=['sections.plugins.password'];p=plan(decl)
|
|
for full in (False,True):
|
|
cfg=replace(s,reports_retain_configuration=full);ident=job(a,u,p,status='successful')
|
|
with a.store.transaction()as db:persist(db,cfg,ident,p,result(decl,report_data=content))
|
|
item=Reports(cfg).slot(u,ident)
|
|
if full:assert item['data']==content
|
|
else:assert 'sections'not in item['data'] and item['data']['redacted_paths']==content['redacted_paths']
|
|
|
|
def test_secret_canary_and_nonfinite_rejected():
|
|
decl=declared('checkmk_user_config_v1');content=sample(decl['data_schema']);content['sections']={'value':'FIXTURE-SECRET'}
|
|
with pytest.raises(WebError):data(content,decl['data_schema'],secrets=['FIXTURE-SECRET'])
|
|
content['sections']={'password':'PLAIN'}
|
|
with pytest.raises(WebError):data(content,decl['data_schema'])
|
|
content['sections']={'value':float('nan')}
|
|
with pytest.raises(WebError):data(content,decl['data_schema'])
|
|
|
|
def test_deletion_and_owner_admin_scope(local):
|
|
s,a,admin=local;a.create_user('viewer','Synthetic-fixture-password-2026','viewer')
|
|
with a.store.transaction()as db:db.execute('UPDATE users SET must_change_password=0');viewer=db.execute("SELECT id FROM users WHERE username='viewer'").fetchone()[0]
|
|
decl=declared();p=plan(decl);ident=job(a,admin,p,status='failed')
|
|
with a.store.transaction()as db:persist(db,s,ident,p,result(decl))
|
|
with pytest.raises(WebError):Reports(s).index(viewer,ident)
|
|
assert Reports(s).host_links(viewer,'example','test01.example')==[]
|
|
assert len(Reports(s).host_links(admin,'example','test01.example'))==1
|
|
Workflows(s).delete_jobs(admin,[ident])
|
|
with a.store.read()as db:
|
|
assert db.execute('SELECT count(*) FROM job_operation_results').fetchone()[0]==0
|
|
assert db.execute('SELECT count(*) FROM job_operation_reports').fetchone()[0]==0
|
|
assert db.execute("SELECT count(*) FROM audit WHERE action='job-deleted'").fetchone()[0]==1
|
|
|
|
def test_json_escaped_html_and_lazy_report_content(local):
|
|
s,a,u=local;decl=declared('event_log_export_v1');content=sample(decl['data_schema']);content['files']=['<script>alert(1)</script>','javascript:alert(1)'];ident=job(a,u,plan(decl),status='successful')
|
|
with a.store.transaction()as db:
|
|
small=persist(db,s,ident,plan(decl),result(decl,report_data=content));db.execute('UPDATE jobs SET core_result=? WHERE id=?',(json.dumps(small),ident))
|
|
token,_=a.new_session(u)
|
|
with TestClient(create_app(s),base_url=s.public_url)as c:
|
|
c.cookies.set(s.cookie_name,token)
|
|
r=c.get('/jobs/'+ident+'/reports')
|
|
assert '<script>alert(1)</script>'not in r.text and '<script>'in r.text
|
|
assert 'href="javascript:'not in r.text
|
|
assert 'alert(1)'not in c.get('/api/v2/runs/'+ident).text
|
|
|
|
def test_smaller_local_budget_does_not_truncate(local):
|
|
s,a,u=local;decl=declared('event_log_export_v1');data=sample(decl['data_schema']);data['files']=['x'*1000]*90
|
|
ident=job(a,u,plan(decl),status='successful');s=replace(s,reports_max_bytes=65536)
|
|
with a.store.transaction()as db:persist(db,s,ident,plan(decl),result(decl,report_data=data))
|
|
item=Reports(s).slot(u,ident);assert item['status']=='available' and item['retention']=='not_retained_limit' and item['data']is None
|