dilib_operations/tests/test_monitoring.py

import uuid
from datetime import timedelta
from unittest.mock import patch,Mock
import pytest
from django.utils import timezone
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient
from monitoring.models import *
from monitoring.auth import digest
from monitoring.ingest import ingest
from monitoring.tasks import evaluate_alarms,cleanup,deliver_notifications
from agent.outbox import Outbox
from agent.main import envelope,safe_event,send_batch
@pytest.fixture
def server(db):
    seg=Segment.objects.create(name='Production')
    return Server.objects.create(name='host-01',segment=seg)
@pytest.fixture
def client():return APIClient()
@pytest.fixture
def auth(client,db):
    user=get_user_model().objects.create_user(username='operator',password='test-secure-password-834!',is_staff=True)
    token=AccessToken.objects.create(user=user,digest=digest('test-token'),expires_at=timezone.now()+timedelta(hours=1))
    client.credentials(HTTP_AUTHORIZATION='Bearer test-token');return client

def ev(kind='metric',payload=None,at=None):
    return {'id':uuid.uuid4(),'kind':kind,'occurred_at':at or timezone.now(),'payload':payload or {'cpu':22,'ram':45,'disk':61}}
@pytest.mark.django_db
def test_ingest_auth_scope_idempotence(server,client):
    payload=ev();payload['id']=str(payload['id']);payload['occurred_at']=payload['occurred_at'].isoformat()
    assert client.post('/api/ingest/',{'events':[payload]},format='json').status_code==401
    AgentCredential.objects.create(server=server,digest=digest('agent-secret'))
    client.credentials(HTTP_AUTHORIZATION='Agent agent-secret')
    for _ in range(2):assert client.post('/api/ingest/',{'events':[payload]},format='json').status_code==200
    assert Event.objects.count()==1 and Metric.objects.count()==1
    payload['id']=str(uuid.uuid4());payload['payload']['cpu']=120
    assert client.post('/api/ingest/',{'events':[payload]},format='json').status_code==400
    assert Event.objects.count()==1
@pytest.mark.django_db
def test_transaction_validation_no_partial_batch(server,client):
    AgentCredential.objects.create(server=server,digest=digest('agent-secret'));client.credentials(HTTP_AUTHORIZATION='Agent agent-secret')
    events=[envelope('metric',{'cpu':30,'ram':40,'disk':50}),envelope('metric',{'cpu':-1,'ram':40,'disk':50})]
    assert client.post('/api/ingest/',{'events':events},format='json').status_code==400
    assert Event.objects.count()==0
@pytest.mark.django_db
def test_task_out_of_order_and_terminal_state(server):
    now=timezone.now();task={'uuid':'t1','name':'x'}
    ingest(server.pk,[ev('task-succeeded',dict(task,runtime=2),now),ev('task-started',task,now-timedelta(seconds=4)),ev('task-received',task,now+timedelta(seconds=1))])
    t=Task.objects.get();assert t.state=='SUCCESS' and t.runtime==2
@pytest.mark.django_db
def test_stale_metric_replay_not_online(server,auth):
    ingest(server.pk,[ev(at=timezone.now()-timedelta(hours=1))])
    assert auth.get('/api/summary/').json()['servers'][0]['status']=='Offline'
    evaluate_alarms();assert Alarm.objects.filter(code='heartbeat',resolved_at__isnull=True).exists()
@pytest.mark.django_db
def test_alarm_recovery_ack_and_permissions(server,auth):
    ingest(server.pk,[ev(payload={'cpu':95,'ram':20,'disk':30})]);evaluate_alarms();evaluate_alarms()
    a=Alarm.objects.get(code='cpu');assert Notification.objects.count()==1
    assert auth.post(f'/api/alarms/{a.pk}/ack/').status_code==200
    assert Audit.objects.filter(action='alarm_ack').count()==1
    assert auth.post(f'/api/alarms/{a.pk}/ack/').status_code==200
    assert Audit.objects.filter(action='alarm_ack').count()==1
    ingest(server.pk,[ev()]);evaluate_alarms();a.refresh_from_db();assert a.resolved_at
    user=get_user_model().objects.get(username='operator');user.is_staff=False;user.save()
    assert auth.put('/api/settings/',{},format='json').status_code==403
@pytest.mark.django_db
def test_auth_expiry_and_revocation(auth):
    assert auth.get('/api/summary/').status_code==200
    AccessToken.objects.update(expires_at=timezone.now()-timedelta(seconds=1))
    assert auth.get('/api/summary/').status_code==401
@pytest.mark.django_db
def test_retention(server):
    ingest(server.pk,[ev(at=timezone.now()-timedelta(days=100)),ev()]);cleanup()
    assert Event.objects.count()==1 and Metric.objects.count()==1
@pytest.mark.django_db
def test_notification_retries(server,settings):
    settings.ALERT_RECIPIENTS=['ops@example.com'];evaluate_alarms()
    with patch('monitoring.tasks.send_mail',side_effect=OSError()):deliver_notifications()
    n=Notification.objects.first();assert n.attempts==1 and not n.sent_at
    Notification.objects.update(next_attempt=timezone.now())
    with patch('monitoring.tasks.send_mail',return_value=1):deliver_notifications()
    n.refresh_from_db();assert n.sent_at
@pytest.mark.django_db
def test_report_csv_and_all_views(server,auth):
    ingest(server.pk,[ev()])
    for path in ['summary','servers','tasks','events','alarms','settings','flower','health']:
        assert auth.get('/api/'+path+'/').status_code==200
    assert auth.get(f'/api/servers/{server.pk}/').status_code==200
    r=auth.get('/api/report/');assert r.status_code==200 and r.content.startswith(b'\x89PNG')
    server.name='=EVIL()';server.save()
    assert "'=EVIL()" in auth.get('/api/export/').content.decode()
@pytest.mark.django_db
def test_login_rate_limit_and_logout(client):
    from django.core.cache import cache
    cache.clear()
    get_user_model().objects.create_user(username='test-login',password='example-Password!31')
    response=client.post('/api/auth/login/',{'username':'test-login','password':'example-Password!31'},format='json')
    assert response.status_code==200
    client.credentials(HTTP_AUTHORIZATION='Bearer '+response.json()['token'])
    assert client.post('/api/auth/logout/').status_code==200
    assert client.get('/api/summary/').status_code==401
    client.credentials()
    for _ in range(10):client.post('/api/auth/login/',{'username':'bad','password':'bad'},format='json')
    assert client.post('/api/auth/login/',{'username':'bad','password':'bad'},format='json').status_code==429

def test_outbox_survives_restart_and_failed_delivery(tmp_path):
    path=tmp_path/'queue.db';out=Outbox(path);item=envelope('metric',{'cpu':1,'ram':2,'disk':3});out.add(item);out.add(item);out.close()
    out=Outbox(path);assert out.count()==1
    session=Mock();session.post.side_effect=OSError('offline')
    with pytest.raises(OSError):send_batch(session,out,'https://example.com','secret')
    assert out.count()==1
    session.post.side_effect=None;session.post.return_value.json.return_value={'received':1}
    assert send_batch(session,out,'https://example.com','secret')==1 and out.count()==0

def test_payload_redaction():
    e=safe_event({'type':'task-succeeded','uuid':'a','hostname':'worker','args':'SECRET','result':'SECRET','kwargs':{'password':'SECRET'},'runtime':2})
    assert set(e['payload'])=={'uuid','hostname','runtime'}

@pytest.mark.django_db
def test_ingest_normalizes_numeric_input(server,client):
    AgentCredential.objects.create(server=server,digest=digest('numeric-agent'))
    client.credentials(HTTP_AUTHORIZATION='Agent numeric-agent')
    event=envelope('metric',{'cpu':'20.5','ram':'30','disk':'40','services':{'web':'false'}})
    assert client.post('/api/ingest/',{'events':[event]},format='json').status_code==200
    server.refresh_from_db()
    assert server.latest['cpu']==20.5 and server.latest['services']['web'] is False

@pytest.mark.django_db
def test_batch_cross_agent_collision_rolls_back(server):
    existing=ev();ingest(server.pk,[existing])
    other=Server.objects.create(name='other',segment=server.segment)
    with pytest.raises(ValueError):ingest(other.pk,[ev(),existing])
    assert not Event.objects.filter(server=other).exists()

@pytest.mark.django_db
def test_payload_allowlist_and_future_clock(server,client):
    AgentCredential.objects.create(server=server,digest=digest('agent-secret'));client.credentials(HTTP_AUTHORIZATION='Agent agent-secret')
    item=envelope('task-succeeded',{'uuid':'x','name':'task','args':'secret','result':'secret'})
    assert client.post('/api/ingest/',{'events':[item]},format='json').status_code==200
    assert Event.objects.get().payload=={'uuid':'x','name':'task'}
    item=envelope('metric',{'cpu':3,'ram':4,'disk':5},timezone.now()+timedelta(hours=1))
    assert client.post('/api/ingest/',{'events':[item]},format='json').status_code==400

@pytest.mark.django_db
def test_metrics_survive_new_client_and_flower_restart(server,auth):
    ingest(server.pk,[ev(),ev('flower-status',{'online':False})])
    assert auth.get('/api/summary/').json()['online']==1
    ingest(server.pk,[ev('flower-status',{'online':True})])
    assert Metric.objects.count()==1 and Event.objects.count()==3
    assert auth.get('/api/flower/').json()[0]['online'] is True