124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
from datetime import timedelta
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import Client, TestCase
|
|
from django.utils import timezone
|
|
|
|
from workflows.models import AsyncTaskLog
|
|
from workflows.roles import ROLE_ADMIN, ROLE_STAFF, assign_user_role
|
|
|
|
|
|
class ObservabilityUITests(TestCase):
|
|
def setUp(self):
|
|
user_model = get_user_model()
|
|
self.admin = user_model.objects.create_user(
|
|
username='ops_admin',
|
|
email='ops-admin@example.com',
|
|
password='secret123',
|
|
)
|
|
assign_user_role(self.admin, ROLE_ADMIN)
|
|
|
|
self.staff = user_model.objects.create_user(
|
|
username='ops_staff',
|
|
email='ops-staff@example.com',
|
|
password='secret123',
|
|
)
|
|
assign_user_role(self.staff, ROLE_STAFF)
|
|
|
|
def _create_log(self, *, status: str, task_name: str, target_label: str, error_message: str = '') -> AsyncTaskLog:
|
|
log = AsyncTaskLog.objects.create(
|
|
task_name=task_name,
|
|
status=status,
|
|
target_type='request',
|
|
target_id=1,
|
|
target_label=target_label,
|
|
error_message=error_message,
|
|
)
|
|
AsyncTaskLog.objects.filter(id=log.id).update(
|
|
started_at=timezone.now() - timedelta(hours=2),
|
|
finished_at=timezone.now() - timedelta(hours=1, minutes=45),
|
|
)
|
|
return AsyncTaskLog.objects.get(id=log.id)
|
|
|
|
def test_home_shows_operations_overview_for_admin(self):
|
|
self._create_log(
|
|
status='failed',
|
|
task_name='send_scheduled_welcome_email',
|
|
target_label='Request A',
|
|
error_message='smtp failed hard',
|
|
)
|
|
self._create_log(
|
|
status='succeeded',
|
|
task_name='process_onboarding_request',
|
|
target_label='Request B',
|
|
)
|
|
self._create_log(
|
|
status='started',
|
|
task_name='process_offboarding_request',
|
|
target_label='Request C',
|
|
)
|
|
|
|
client = Client()
|
|
client.force_login(self.admin)
|
|
response = client.get('/', HTTP_HOST='localhost')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, 'Operations Overview')
|
|
self.assertContains(response, 'Fehlgeschlagene Jobs (24h)')
|
|
self.assertContains(response, '<strong class="is-error">1</strong>', html=True)
|
|
self.assertContains(response, 'send_scheduled_welcome_email')
|
|
self.assertContains(response, 'Backup-Status')
|
|
|
|
def test_home_hides_operations_overview_for_staff(self):
|
|
self._create_log(
|
|
status='failed',
|
|
task_name='process_onboarding_request',
|
|
target_label='Request A',
|
|
error_message='pdf failed',
|
|
)
|
|
|
|
client = Client()
|
|
client.force_login(self.staff)
|
|
response = client.get('/', HTTP_HOST='localhost')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotContains(response, 'Operations Overview')
|
|
self.assertNotContains(response, 'Job Monitor öffnen')
|
|
|
|
def test_job_monitor_summary_shows_recent_counts(self):
|
|
self._create_log(
|
|
status='failed',
|
|
task_name='process_onboarding_request',
|
|
target_label='Request A',
|
|
error_message='pdf failed',
|
|
)
|
|
self._create_log(
|
|
status='succeeded',
|
|
task_name='process_offboarding_request',
|
|
target_label='Request B',
|
|
)
|
|
self._create_log(
|
|
status='started',
|
|
task_name='send_scheduled_welcome_email',
|
|
target_label='Request C',
|
|
)
|
|
|
|
client = Client()
|
|
client.force_login(self.admin)
|
|
response = client.get('/admin-tools/jobs/', HTTP_HOST='localhost')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, 'Fehlgeschlagene Jobs (24h)')
|
|
self.assertContains(response, 'Erfolgreiche Jobs (24h)')
|
|
self.assertContains(response, 'Offene Starts (24h)')
|
|
self.assertContains(response, 'Zuletzt fehlgeschlagen')
|
|
self.assertContains(response, 'pdf failed')
|
|
|
|
def test_job_monitor_requires_capability(self):
|
|
client = Client()
|
|
client.force_login(self.staff)
|
|
|
|
response = client.get('/admin-tools/jobs/', HTTP_HOST='localhost')
|
|
|
|
self.assertEqual(response.status_code, 302)
|