admin_rb
92 lines
4.9 KiB
Python
92 lines
4.9 KiB
Python
"""HTTP/HTMX flows against the real core metadata/prepare protocol, no execution."""
|
|
import json
|
|
import re
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from aim_webgui.app import create_app
|
|
from aim_webgui.workflows import Workflows
|
|
|
|
@pytest.fixture
|
|
def browser(auth,admin):
|
|
token,session=auth.new_session(admin)
|
|
with TestClient(create_app(auth.settings),base_url=auth.settings.public_url,follow_redirects=False) as client:
|
|
client.cookies.set(auth.settings.cookie_name,token)
|
|
client.headers.update({'Origin':auth.settings.public_url,'X-CSRF-Token':session['csrf']})
|
|
yield client
|
|
|
|
|
|
def test_html_new_run_review_queue_and_save_optional(browser,auth,admin):
|
|
response=browser.get('/plan?customer=example&playbook=debug_test_connection')
|
|
assert response.status_code==200, response.text
|
|
assert 'Load customer SSH key' in response.text
|
|
response=browser.post('/_partials/preflight',data={'customer':'example','playbook':'debug_test_connection',
|
|
'targets':'test01.example','mode':'check','key_mode':'none'},headers={'HX-Request':'true'})
|
|
assert response.status_code==200,response.text
|
|
assert 'Run once' in response.text and 'Save as a reusable plan (optional)' in response.text
|
|
review=re.search(r'name="review_id" value="([a-f0-9]+)"',response.text).group(1)
|
|
assert 'name="name"' in response.text and not re.search('name="name"[^>]+required',response.text)
|
|
response=browser.post('/jobs',data={'review_id':review,'submission_key':review,'confirm':'reviewed'})
|
|
assert response.status_code==303,response.text
|
|
path=response.headers['location']
|
|
response=browser.get(path)
|
|
assert response.status_code==200,response.text
|
|
assert 'queued' in response.text
|
|
assert Workflows(auth.settings).plans(admin)==[]
|
|
saved=browser.post('/plans',data={'review_id':review,'name':''})
|
|
assert saved.status_code==303,saved.text
|
|
assert len(Workflows(auth.settings).plans(admin))==1
|
|
title=Workflows(auth.settings).plans(admin)[0]['name']
|
|
conflict=browser.post('/plans',data={'review_id':review,'name':title.upper()})
|
|
assert conflict.status_code==409
|
|
assert 'nothing was overwritten' in conflict.text
|
|
assert len(Workflows(auth.settings).plans(admin))==1
|
|
|
|
|
|
def test_json_one_run_idempotency_and_api_version(browser,auth,admin):
|
|
r=browser.post('/api/v2/preflight',json={'customer':'example','playbook':'debug_test_connection','targets':['test01.example'],'overrides':{}})
|
|
assert r.status_code==200,r.text
|
|
data={'review_id':r.json()['review_id'],'confirm':True}
|
|
r=browser.post('/api/v2/runs',json=data,headers={'Idempotency-Key':'test-run-once'})
|
|
assert r.status_code==202,r.text
|
|
ident=r.json()['id']
|
|
repeat=browser.post('/api/v2/runs',json=data,headers={'Idempotency-Key':'test-run-once'})
|
|
assert repeat.json()['id']==ident
|
|
assert browser.get('/api/v2/runs/'+ident).json()['plan']['core_request']['check'] is True
|
|
assert browser.get('/api/v1/runs/'+ident).status_code==404
|
|
|
|
|
|
def test_security_and_private_review(browser,auth,admin):
|
|
response=browser.post('/api/v2/preflight',headers={'Origin':'https://evil.invalid'},json={})
|
|
assert response.status_code==403
|
|
browser.headers.pop('X-CSRF-Token')
|
|
assert browser.post('/jobs',data={}).status_code==403
|
|
assert browser.get('/users').status_code==200
|
|
auth.create_user('viewer','Long-fixture-password-2026','viewer')
|
|
with auth.store.transaction()as db:
|
|
db.execute("UPDATE users SET must_change_password=0 WHERE username='viewer'")
|
|
user=db.execute("SELECT id FROM users WHERE username='viewer'").fetchone()[0]
|
|
token,session=auth.new_session(user)
|
|
browser.cookies.set(auth.settings.cookie_name,token)
|
|
browser.headers['X-CSRF-Token']=session['csrf']
|
|
assert browser.get('/users').status_code==403
|
|
assert browser.post('/api/v2/runs',json={'review_id':'not-mine','confirm':True},headers={'Idempotency-Key':'x'}).status_code in(403,409)
|
|
|
|
|
|
def test_core_outage_keeps_local_login_recovery(auth,admin,monkeypatch):
|
|
from aim_webgui.adapters.core_v1 import CoreAdapter
|
|
from aim_webgui.errors import WebError
|
|
def down(*args,**kwargs):raise WebError('executor_unavailable','Offline',503)
|
|
monkeypatch.setattr(CoreAdapter,'customers',down)
|
|
with TestClient(create_app(auth.settings),base_url=auth.settings.public_url)as client:
|
|
assert client.get('/login').status_code==200
|
|
assert client.get('/healthz').status_code==200
|
|
assert client.get('/readyz').status_code==503
|
|
|
|
|
|
def test_retained_overviews_and_admin_pages(browser):
|
|
for path in ('/','/customers','/customers/example/hosts','/playbooks','/jobs','/plans','/users','/setup','/audit','/system','/password'):
|
|
response=browser.get(path)
|
|
assert response.status_code==200,(path,response.status_code,response.text[:500])
|
|
response=browser.get('/users')
|
|
assert response.text.count('Scoped execution grants')==1
|