admin_rb
229 lines
12 KiB
Python
229 lines
12 KiB
Python
"""Read-only projections and mobile shell; no Ansible, keys or remote hosts."""
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
import json
|
|
import time
|
|
import uuid
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from aim_webgui.activity import Activity, HistoryFilter, host_url
|
|
from aim_webgui.explorer import Explorer
|
|
from aim_webgui.auth.service import Auth
|
|
from aim_webgui.config import Settings
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.errors import WebError
|
|
from aim_webgui.workflows import Workflows
|
|
|
|
COUNTS=('ok','changed','failures','unreachable','skipped','rescued','ignored')
|
|
|
|
@pytest.fixture
|
|
def local_auth(tmp_path):
|
|
s=Settings(state_dir=tmp_path/'state',public_url='https://aim.example.test').validate()
|
|
a=Auth(s);a.bootstrap()
|
|
a.create_user('operator','Fixture-only-long-password-2026','viewer')
|
|
a.create_user('other','Fixture-only-other-password-2026','viewer')
|
|
with a.store.transaction() as db:
|
|
db.execute('UPDATE users SET must_change_password=0')
|
|
return a
|
|
|
|
|
|
def uid(auth,name):
|
|
with auth.store.read() as db:return db.execute('SELECT id FROM users WHERE username=?',(name,)).fetchone()[0]
|
|
|
|
|
|
def add_job(auth,owner='operator',customer='example',book='debug_test_connection',hosts=None,status='successful',mode='apply',when=None,legacy=False):
|
|
hosts=hosts or {'test01.example':'successful'}
|
|
when=int(time.time())-60 if when is None else when
|
|
target_list=[];summ=dict.fromkeys(('successful','failed','unreachable','not_started','indeterminate'),0)
|
|
for host,outcome in hosts.items():
|
|
counts=dict.fromkeys(COUNTS,0);counts['ok']=4
|
|
if outcome=='unreachable':counts['unreachable']=1
|
|
if outcome=='failed':counts['failures']=1
|
|
if outcome=='successful':counts['changed']=1
|
|
summ[outcome]+=1
|
|
target_list.append({'host':host,'outcome':outcome,'counts':counts})
|
|
ident=uuid.uuid4().hex
|
|
result={'status':'failed' if status=='failed' else 'succeeded','targets':target_list,
|
|
'target_summary':{'schema':'target_outcome_summary_v1','requested':len(hosts),'accounted':len(hosts),'complete':True,**summ}}
|
|
plan={'customer':customer,'playbook':book,'targets':list(hosts),'overrides':{'ignored_fixture_field':'MUST-NOT-LEAVE-PROJECTION'}}
|
|
with auth.store.transaction() as db:
|
|
db.execute('''INSERT INTO jobs(id,owner_id,plan,mode,status,created_at,scheduled_at,finished_at,core_result)
|
|
VALUES(?,?,?,?,?,?,?,?,?)''',(ident,uid(auth,owner),json.dumps(plan),mode,status,when,when,
|
|
when if status not in ('queued','running','pending') else None,None if legacy else json.dumps(result)))
|
|
return ident
|
|
|
|
|
|
def test_mixed_target_semantics_and_no_task_count_inference(local_auth):
|
|
a=local_auth;ident=add_job(a,hosts={'test01.example':'successful','test02.example':'unreachable'},status='failed')
|
|
data=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(customer='example',days='all'))
|
|
assert data['jobs']==1 and data['samples']==2
|
|
assert data['counts']['successful']==1 and data['counts']['unreachable']==1
|
|
assert data['success_percent']==50 and data['eligible']==2
|
|
assert data['records'][0]['job_display_status']=='partially_succeeded'
|
|
assert 'MUST-NOT-LEAVE' not in repr(data)
|
|
|
|
|
|
def test_all_history_not_last_100_and_pagination(local_auth):
|
|
a=local_auth
|
|
for i in range(135):add_job(a,when=int(time.time())-1000-i)
|
|
data=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(host='test01.example',days='all',page=5))
|
|
assert data['jobs']==135 and data['samples']==135 and data['pages']==6
|
|
assert len(data['records'])==25
|
|
|
|
|
|
def test_visibility_for_totals_filters_and_matrix(local_auth):
|
|
a=local_auth
|
|
add_job(a);add_job(a,owner='other',customer='secret-customer',book='secret-book',hosts={'secret-host.example':'failed'},status='failed')
|
|
data=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(days='all'))
|
|
assert data['samples']==1 and 'secret-' not in repr(data)
|
|
admin=Activity(a.settings).report(uid(a,'admin'),HistoryFilter(days='all'))
|
|
assert admin['samples']==2
|
|
|
|
|
|
def test_mode_range_and_unknown_outcomes(local_auth):
|
|
a=local_auth
|
|
add_job(a,mode='check');add_job(a,legacy=True);add_job(a,status='queued')
|
|
add_job(a,hosts={'test01.example':'not_started'});add_job(a,hosts={'test01.example':'indeterminate'})
|
|
add_job(a,when=int(time.time())-86400*100)
|
|
data=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(days='30'))
|
|
assert data['samples']==4 and data['eligible']==0 and data['success_percent'] is None
|
|
assert data['counts']['unavailable']==1 and data['counts']['outstanding']==1
|
|
check=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(mode='check',days='all'))
|
|
assert check['samples']==1 and check['counts']['successful']==1
|
|
|
|
|
|
def test_customer_identity_deletion_and_bad_facts(local_auth):
|
|
a=local_auth
|
|
ident=add_job(a);add_job(a,customer='second')
|
|
s=Activity(a.settings)
|
|
assert s.report(uid(a,'operator'),HistoryFilter(customer='example',days='all'))['samples']==1
|
|
Workflows(a.settings).delete_jobs(uid(a,'operator'), [ident])
|
|
assert s.report(uid(a,'operator'),HistoryFilter(customer='example',days='all'))['samples']==0
|
|
ident=add_job(a)
|
|
with a.store.transaction() as db:db.execute('UPDATE jobs SET core_result=? WHERE id=?',(json.dumps({'targets':[],'target_summary':{}}),ident))
|
|
assert s.report(uid(a,'operator'),HistoryFilter(customer='example',days='all'))['counts']['unavailable']==1
|
|
|
|
|
|
def test_plan_visibility_is_owner_only(local_auth):
|
|
a=local_auth
|
|
with a.store.transaction() as db:
|
|
for name in ('admin','operator','other'):
|
|
db.execute('INSERT INTO plans(id,owner_id,name,customer,playbook,payload,created_at) VALUES(?,?,?,?,?,?,?)',
|
|
(uuid.uuid4().hex,uid(a,name),name+' plan','example','debug_test_connection',json.dumps({'targets':['test01.example']}),int(time.time())))
|
|
data=Activity(a.settings).report(uid(a,'admin'),HistoryFilter(customer='example',host='test01.example',days='all'))
|
|
assert [p['name'] for p in data['plans']]==['admin plan']
|
|
|
|
|
|
class FakeCore:
|
|
operations=[]
|
|
def __init__(self,*args,**kwargs):self.client=self
|
|
def request(self,operation,**kwargs):
|
|
self.operations.append(operation)
|
|
assert operation=='list_hosts'
|
|
return [{'name':h,'address':'192.0.2.'+str(i+1),'platforms':['linux']} for i,h in enumerate(('direct.example','shared.example','leaf.example'))]
|
|
def inventory_hierarchy(self,customer):
|
|
self.operations.append('inventory_hierarchy')
|
|
def node(name,path,hosts,children=None):return dict(name=name,path=path,hosts=hosts,children=children or [])
|
|
return {'schema':'inventory_hierarchy_v1','customer':customer,'hosts':['direct.example'],'groups':[
|
|
node('linux',['linux'],['shared.example'],[node('servers',['linux','servers'],['shared.example','leaf.example'])]),
|
|
node('lab',['lab'],[],[node('servers',['lab','servers'],['shared.example'])]),node('empty',['empty'],[])]}
|
|
|
|
|
|
def test_explorer_distinct_counts_paths_root_members_and_search(local_auth,monkeypatch):
|
|
monkeypatch.setattr('aim_webgui.explorer.CoreAdapter',FakeCore)
|
|
e=Explorer(local_auth.settings)
|
|
p=e.page('example')
|
|
assert p['snap']['host_count']==3 and len(p['direct_hosts'])==1
|
|
assert len(p['snap']['groups'][('linux',)]['members'])==2
|
|
assert len(p['snap']['hosts']['shared.example']['memberships'])==3
|
|
p=e.page('example',branch='["lab","servers"]')
|
|
assert p['selected']['path']==['lab','servers'] and len(p['direct_hosts'])==1
|
|
assert 'activity?host=' in p['direct_hosts'][0]['url']
|
|
assert e.page('example',q='shared')['search_count']==1
|
|
with pytest.raises(WebError):e.page('example',branch='["missing"]')
|
|
|
|
|
|
def test_explorer_map_is_bounded(local_auth,monkeypatch):
|
|
class Large(FakeCore):
|
|
def request(self,op,**kwargs):return []
|
|
def inventory_hierarchy(self,customer):return {'schema':'inventory_hierarchy_v1','customer':customer,'hosts':[],
|
|
'groups':[{'name':str(n),'path':[str(n)],'hosts':[],'children':[]}for n in range(50)]}
|
|
monkeypatch.setattr('aim_webgui.explorer.CoreAdapter',Large)
|
|
page=Explorer(local_auth.settings).page('example')
|
|
assert len(page['groups'])==12 and page['group_pages']==5 and len(page['graph_nodes'])<=49
|
|
assert page['groups_next']
|
|
|
|
|
|
def test_new_get_views_are_read_only_and_survive_core_outage(local_auth,monkeypatch):
|
|
a=local_auth;add_job(a);FakeCore.operations=[]
|
|
monkeypatch.setattr('aim_webgui.explorer.CoreAdapter',FakeCore)
|
|
token,_=a.new_session(uid(a,'operator'))
|
|
with TestClient(create_app(a.settings),base_url=a.settings.public_url) as c:
|
|
c.cookies.set(a.settings.cookie_name,token)
|
|
for path in ('/inventory/example/explore','/inventory/example/explore?view=map',
|
|
'/inventory/example/explore?q=shared',host_url('example','test01.example'),
|
|
'/insights','/api/v2/insights','/api/v2/activity?customer=example&host=test01.example',
|
|
'/api/v2/inventory/example/explore'):
|
|
r=c.get(path); assert r.status_code==200,(path,r.text[:2000])
|
|
assert r.headers['cache-control']=='no-store'
|
|
assert set(FakeCore.operations)=={'inventory_hierarchy','list_hosts'}
|
|
before=Activity(a.settings).report(uid(a,'operator'),HistoryFilter(days='all'))['jobs']
|
|
assert before==1
|
|
def down(*a,**k):raise WebError('core_offline','Do not echo sensitive path',503)
|
|
monkeypatch.setattr(FakeCore,'inventory_hierarchy',down)
|
|
r=c.get(host_url('example','test01.example'))
|
|
assert r.status_code==200 and 'Current inventory is unavailable' in r.text
|
|
assert 'Do not echo sensitive path' not in r.text
|
|
assert c.get('/inventory/example/explore').status_code==503
|
|
|
|
|
|
def test_filters_bad_input_and_anonymous_access(local_auth):
|
|
a=local_auth
|
|
for f in (HistoryFilter(mode='invalid'),HistoryFilter(page=0),HistoryFilter(days='-1'),HistoryFilter(outcome='fake'),HistoryFilter(q='x'*256)):
|
|
with pytest.raises(WebError):Activity(a.settings).report(uid(a,'operator'),f)
|
|
with TestClient(create_app(a.settings),base_url=a.settings.public_url,follow_redirects=False) as c:
|
|
assert c.get('/api/v2/insights').status_code==401
|
|
assert c.get('/insights').status_code==303
|
|
|
|
|
|
def test_menu_markup_and_escaped_history(local_auth,monkeypatch):
|
|
a=local_auth;add_job(a,book='<script>alert(1)</script>')
|
|
monkeypatch.setattr('aim_webgui.explorer.CoreAdapter',FakeCore)
|
|
token,_=a.new_session(uid(a,'operator'))
|
|
with TestClient(create_app(a.settings),base_url=a.settings.public_url) as c:
|
|
c.cookies.set(a.settings.cookie_name,token)
|
|
r=c.get('/insights?days=all')
|
|
assert '<script>alert(1)</script>' not in r.text
|
|
assert '<script>' in r.text
|
|
assert 'data-mobile-menu' in r.text and 'aria-controls="mobile-navigation"' in r.text
|
|
assert 'data-nav-forward' not in r.text and 'Swipe or use arrows' not in r.text
|
|
assert 'No terminal history collection' in r.text
|
|
|
|
|
|
def test_public_core_explorer_with_execution_disabled(settings):
|
|
# Real public Core discovery only; no Ansible or remote execution.
|
|
s=replace(settings,execution_enabled=False,credentials_enabled=False)
|
|
page=Explorer(s).page('example')
|
|
assert page['snap']['host_count']==2
|
|
assert ('linux','lab') in page['snap']['groups']
|
|
assert page['snap']['hosts']['test01.example']['memberships'][0]['label']=='linux / lab'
|
|
assert 'activity?host=test01.example' in page['snap']['hosts']['test01.example']['url']
|
|
|
|
|
|
def test_real_core_read_pages_with_empty_history(settings):
|
|
s=replace(settings,execution_enabled=False,credentials_enabled=False)
|
|
a=Auth(s);a.bootstrap()
|
|
with a.store.transaction() as db:db.execute('UPDATE users SET must_change_password=0')
|
|
token,_=a.new_session(uid(a,'admin'))
|
|
with TestClient(create_app(s),base_url=s.public_url) as c:
|
|
c.cookies.set(s.cookie_name,token)
|
|
for path in ('/inventory/example/explore',host_url('example','test01.example'),'/insights'):
|
|
r=c.get(path)
|
|
assert r.status_code==200,(path,r.text[:200])
|
|
r=c.get(host_url('example','old-host.example'))
|
|
assert 'Not in current inventory.' in r.text
|
|
assert 'No retained runs match' in r.text
|
|
with a.store.read() as db:
|
|
assert db.execute('SELECT COUNT(*) FROM jobs').fetchone()[0]==0
|
|
assert db.execute('SELECT COUNT(*) FROM run_reviews').fetchone()[0]==0
|