58 lines
3.3 KiB
Python
58 lines
3.3 KiB
Python
|
import io
|
||
|
|
from pathlib import Path
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import time
|
||
|
|
import unittest
|
||
|
|
from urllib.parse import parse_qs, urlencode, urlsplit
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
|
||
|
|
from alert_ack import Store
|
||
|
|
from alert_identity import Login
|
||
|
|
from alert_service import Router
|
||
|
|
from test_identity_policy import Issuer
|
||
|
|
|
||
|
|
|
||
|
|
class ServiceTests(unittest.TestCase):
|
||
|
|
def test_browser_login_confirmation_and_logout(self):
|
||
|
|
with tempfile.TemporaryDirectory() as tmp:
|
||
|
|
token = Path(tmp) / 'webhook-token'
|
||
|
|
token.write_text('w' * 32)
|
||
|
|
store = Store(Path(tmp) / 'state.db')
|
||
|
|
issuer = Issuer()
|
||
|
|
login = Login('https://issuer.example', 'https://telemetry.example', transport=issuer)
|
||
|
|
router = Router({'origin': 'https://telemetry.example', 'webhook_token_file': str(token)}, store,
|
||
|
|
login=login, policy=lambda *args: dict(id='fixture-decision', effect='allow', expires_at=time.time() + 30))
|
||
|
|
query = urlencode(dict(fingerprint='0123456789abcdef', starts_at='2026-09-28T00:00:00Z'))
|
||
|
|
def call(path, method='GET', cookie='', query=query, body=b''):
|
||
|
|
env = dict(PATH_INFO=path, REQUEST_METHOD=method, HTTP_COOKIE=cookie, QUERY_STRING=query,
|
||
|
|
CONTENT_LENGTH=str(len(body)), HTTP_ORIGIN='https://telemetry.example')
|
||
|
|
env['wsgi.input'] = io.BytesIO(body)
|
||
|
|
result = {}
|
||
|
|
output = b''.join(router(env, lambda status, headers: result.update(status=status, headers=headers)))
|
||
|
|
return result, output.decode()
|
||
|
|
response, _ = call('/ack/alerts')
|
||
|
|
self.assertEqual(response['status'], '303 See Other')
|
||
|
|
headers = dict(response['headers'])
|
||
|
|
browser_cookie = headers['Set-Cookie'].split(';')[0]
|
||
|
|
params = parse_qs(urlsplit(headers['Location']).query)
|
||
|
|
issuer.nonce = params['nonce'][0]
|
||
|
|
response, _ = call('/ack/auth/callback', cookie=browser_cookie,
|
||
|
|
query=urlencode(dict(state=params['state'][0], code='fixture-code')))
|
||
|
|
self.assertEqual(response['status'], '303 See Other')
|
||
|
|
cookies = [value for key, value in response['headers'] if key == 'Set-Cookie']
|
||
|
|
session_cookie = next(c.split(';')[0] for c in cookies if c.startswith('__Host-rtel-session='))
|
||
|
|
actor = login.session(session_cookie.split('=', 1)[1])
|
||
|
|
identity = store.receive({'version': '4', 'receiver': 'railiance-admin-email', 'alerts': [
|
||
|
|
{'status': 'firing', 'fingerprint': '0123456789abcdef', 'startsAt': '2026-09-28T00:00:00Z',
|
||
|
|
'labels': {'alertname': 'TestFailure', 'owner': 'railiance-telemetry'}}]}, time.time())[0]
|
||
|
|
response, page = call('/ack/alerts', cookie=session_cookie)
|
||
|
|
self.assertIn('Acknowledge receipt', page)
|
||
|
|
self.assertIsNone(store.get(identity)['occurred_at'])
|
||
|
|
form = urlencode(dict(csrf=actor.csrf)).encode()
|
||
|
|
response, page = call('/ack/alerts', 'POST', session_cookie, body=form)
|
||
|
|
self.assertIn('Receipt acknowledged', page)
|
||
|
|
response, _ = call('/ack/logout', 'POST', session_cookie, query='', body=form)
|
||
|
|
self.assertEqual(response['status'], '200 OK')
|
||
|
|
self.assertIsNone(login.session(session_cookie.split('=', 1)[1]))
|