snapshot: preserve reliability hardening and Workdock identity pass

backend/workflows/tests/test_app_registry_permissions.py

@@ -0,0 +1,50 @@
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+
+from workflows.app_registry import build_portal_app_sections, ensure_portal_app_configs
+from workflows.models import PortalAppConfig
+from workflows.roles import ROLE_ADMIN, ROLE_IT_STAFF, ROLE_PLATFORM_OWNER, ROLE_STAFF, ROLE_SUPER_ADMIN, assign_user_role
+
+
+class AppRegistryPermissionTests(TestCase):
+    def setUp(self):
+        user_model = get_user_model()
+        self.platform_owner = user_model.objects.create_user(username='platform_owner_case', password='secret123')
+        assign_user_role(self.platform_owner, ROLE_PLATFORM_OWNER)
+
+        self.super_admin = user_model.objects.create_user(username='super_admin_case', password='secret123')
+        assign_user_role(self.super_admin, ROLE_SUPER_ADMIN)
+
+        self.admin = user_model.objects.create_user(username='admin_case', password='secret123')
+        assign_user_role(self.admin, ROLE_ADMIN)
+
+        self.it_staff = user_model.objects.create_user(username='it_staff_case', password='secret123')
+        assign_user_role(self.it_staff, ROLE_IT_STAFF)
+
+        self.staff = user_model.objects.create_user(username='staff_case', password='secret123')
+        assign_user_role(self.staff, ROLE_STAFF)
+
+        ensure_portal_app_configs()
+
+    def _visible_keys(self, user):
+        sections = build_portal_app_sections(user)
+        return {app['key'] for section in sections for app in section['apps']}
+
+    def test_onboarding_and_offboarding_visible_to_staff_by_default(self):
+        keys = self._visible_keys(self.staff)
+        self.assertIn('onboarding', keys)
+        self.assertIn('offboarding', keys)
+
+    def test_trial_management_is_platform_only(self):
+        self.assertIn('trial_management', self._visible_keys(self.platform_owner))
+        self.assertNotIn('trial_management', self._visible_keys(self.super_admin))
+        self.assertNotIn('trial_management', self._visible_keys(self.admin))
+
+    def test_requests_dashboard_can_be_hidden_from_staff_via_registry(self):
+        config = PortalAppConfig.objects.get(key='requests_dashboard')
+        config.visible_to_staff = False
+        config.save(update_fields=['visible_to_staff', 'updated_at'])
+
+        self.assertNotIn('requests_dashboard', self._visible_keys(self.staff))
+        self.assertIn('requests_dashboard', self._visible_keys(self.it_staff))
+

backend/workflows/tests/test_async_task_logging.py

@@ -0,0 +1,52 @@
+from datetime import date
+from pathlib import Path
+from unittest.mock import patch
+
+from django.test import TestCase, override_settings
+from django.utils import timezone
+
+from workflows.models import AsyncTaskLog, OnboardingRequest, ScheduledWelcomeEmail
+from workflows.tasks import process_onboarding_request, send_scheduled_welcome_email
+
+
+@override_settings(PDF_OUTPUT_DIR=Path('/tmp/onoff_test_pdfs'))
+class AsyncTaskLoggingTests(TestCase):
+    def setUp(self):
+        self.onboarding = OnboardingRequest.objects.create(
+            full_name='Task Failure',
+            gender='herr',
+            job_title='Engineer',
+            department='IT-Service',
+            work_email='task.failure@tub.co',
+            contract_start=date(2026, 11, 1),
+            employment_type='unbefristet',
+            onboarded_by_email='requester@tub.co',
+            agreement='accepted',
+        )
+
+    @patch('workflows.tasks._generate_onboarding_pdf', side_effect=RuntimeError('pdf failed'))
+    def test_failed_onboarding_task_creates_failed_async_log(self, _mock_generate_pdf):
+        with self.assertRaises(RuntimeError):
+            process_onboarding_request(self.onboarding.id)
+
+        log = AsyncTaskLog.objects.filter(task_name='process_onboarding_request').latest('id')
+        self.assertEqual(log.status, 'failed')
+        self.assertEqual(log.target_id, self.onboarding.id)
+        self.assertIn('pdf failed', log.error_message)
+
+    @patch('workflows.tasks._send_templated_email', side_effect=RuntimeError('smtp failed'))
+    def test_failed_welcome_email_creates_failed_async_log(self, _mock_send):
+        scheduled = ScheduledWelcomeEmail.objects.create(
+            onboarding_request=self.onboarding,
+            recipient_email='task.failure@tub.co',
+            send_at=timezone.now(),
+            status='scheduled',
+        )
+
+        with self.assertRaises(RuntimeError):
+            send_scheduled_welcome_email(scheduled.id, True)
+
+        log = AsyncTaskLog.objects.filter(task_name='send_scheduled_welcome_email').latest('id')
+        self.assertEqual(log.status, 'failed')
+        self.assertEqual(log.target_id, scheduled.id)
+        self.assertIn('smtp failed', log.error_message)

backend/workflows/tests/test_backup_reliability.py

@@ -0,0 +1,67 @@
+import json
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+from django.core.management import call_command
+from django.test import TestCase, override_settings
+from django.utils import timezone
+
+from workflows.backup_ops import latest_backup_health_snapshot
+
+
+class BackupReliabilityTests(TestCase):
+    @override_settings(BACKUP_OUTPUT_DIR=tempfile.gettempdir())
+    def test_latest_backup_health_reports_missing_when_no_bundle_exists(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with override_settings(BACKUP_OUTPUT_DIR=tmpdir):
+                snapshot = latest_backup_health_snapshot()
+
+        self.assertEqual(snapshot['status'], 'missing')
+        self.assertTrue(snapshot['is_stale'])
+
+    def test_latest_backup_health_reports_stale_when_verification_is_old(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            backup_dir = Path(tmpdir) / 'backup_20260326_010101'
+            backup_dir.mkdir(parents=True)
+            (backup_dir / 'db.dump').write_text('db', encoding='utf-8')
+            (backup_dir / 'media.tar.gz').write_text('media', encoding='utf-8')
+            (backup_dir / 'backup_meta.json').write_text(
+                json.dumps(
+                    {
+                        'created_at': timezone.now().isoformat(),
+                        'verify_status': 'verified',
+                        'verified_at': (timezone.now() - timezone.timedelta(hours=72)).isoformat(),
+                    }
+                ),
+                encoding='utf-8',
+            )
+
+            with override_settings(BACKUP_OUTPUT_DIR=tmpdir):
+                snapshot = latest_backup_health_snapshot(stale_after_hours=48)
+
+        self.assertEqual(snapshot['status'], 'stale')
+        self.assertEqual(snapshot['bundle_name'], 'backup_20260326_010101')
+
+    @patch('workflows.management.commands.verify_latest_backup.verify_backup_bundle')
+    @patch('workflows.management.commands.verify_latest_backup.list_backup_bundles')
+    def test_verify_latest_backup_command_uses_existing_latest_bundle(self, list_bundles, verify_bundle):
+        list_bundles.return_value = [{'name': 'backup_20260326_020202'}]
+        verify_bundle.return_value = {'name': 'backup_20260326_020202', 'summary': 'ok'}
+
+        call_command('verify_latest_backup')
+
+        verify_bundle.assert_called_once_with('backup_20260326_020202')
+
+    @patch('workflows.management.commands.verify_latest_backup.verify_backup_bundle')
+    @patch('workflows.management.commands.verify_latest_backup.create_backup_bundle')
+    @patch('workflows.management.commands.verify_latest_backup.list_backup_bundles')
+    def test_verify_latest_backup_command_can_create_when_missing(self, list_bundles, create_bundle, verify_bundle):
+        list_bundles.return_value = []
+        create_bundle.return_value = {'name': 'backup_20260326_030303'}
+        verify_bundle.return_value = {'name': 'backup_20260326_030303', 'summary': 'ok'}
+
+        call_command('verify_latest_backup', create_if_missing=True)
+
+        create_bundle.assert_called_once()
+        verify_bundle.assert_called_once_with('backup_20260326_030303')

backend/workflows/tests/test_nextcloud_service.py

@@ -22,10 +22,12 @@ class NextcloudServiceTests(TestCase):
         NEXTCLOUD_USERNAME='u',
         NEXTCLOUD_PASSWORD='p',
     )
+    @patch('workflows.services.requests.request')
     @patch('workflows.services.requests.put')
-    def test_upload_calls_webdav_and_accepts_201(self, mock_put):
+    def test_upload_calls_webdav_and_accepts_201(self, mock_put, mock_request):
         temp_file = Path('/tmp/nextcloud_mock_upload.txt')
         temp_file.write_text('hello', encoding='utf-8')
+        mock_request.return_value.status_code = 201
         mock_put.return_value.status_code = 201
 
         try:
@@ -45,8 +47,9 @@ class NextcloudServiceTests(TestCase):
         NEXTCLOUD_USERNAME='env-user',
         NEXTCLOUD_PASSWORD='env-pass',
     )
+    @patch('workflows.services.requests.request')
     @patch('workflows.services.requests.put')
-    def test_upload_prefers_workflowconfig_overrides(self, mock_put):
+    def test_upload_prefers_workflowconfig_overrides(self, mock_put, mock_request):
         WorkflowConfig.objects.update_or_create(
             name='Default',
             defaults={
@@ -59,6 +62,7 @@ class NextcloudServiceTests(TestCase):
         )
         temp_file = Path('/tmp/nextcloud_override_upload.txt')
         temp_file.write_text('hello', encoding='utf-8')
+        mock_request.return_value.status_code = 201
         mock_put.return_value.status_code = 201
 
         try:

backend/workflows/tests/test_request_id_logging.py

@@ -0,0 +1,31 @@
+from django.contrib.auth import get_user_model
+from django.test import Client, TestCase
+
+
+class RequestIDMiddlewareTests(TestCase):
+    def setUp(self):
+        user_model = get_user_model()
+        self.user = user_model.objects.create_user(
+            username='request_id_user',
+            password='secret123',
+            email='requestid@tub.co',
+        )
+
+    def test_response_contains_request_id_header(self):
+        client = Client(HTTP_HOST='127.0.0.1')
+        client.force_login(self.user)
+
+        response = client.get('/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn('X-Request-ID', response.headers)
+        self.assertTrue(response.headers['X-Request-ID'])
+
+    def test_incoming_request_id_is_preserved(self):
+        client = Client(HTTP_HOST='127.0.0.1', HTTP_X_REQUEST_ID='external-request-123')
+        client.force_login(self.user)
+
+        response = client.get('/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers['X-Request-ID'], 'external-request-123')

backend/workflows/tests/test_trial_lifecycle.py

@@ -0,0 +1,73 @@
+from django.contrib.auth import get_user_model
+from django.test import Client, TestCase
+from django.utils import timezone
+
+from workflows.branding import get_portal_trial_config
+from workflows.roles import ROLE_PLATFORM_OWNER, ROLE_STAFF, assign_user_role
+from workflows.services import is_email_test_mode, is_nextcloud_enabled
+
+
+class TrialLifecycleTests(TestCase):
+    def setUp(self):
+        user_model = get_user_model()
+
+        self.platform_owner = user_model.objects.create_user(username='trial_platform_owner', password='secret123')
+        assign_user_role(self.platform_owner, ROLE_PLATFORM_OWNER)
+
+        self.staff = user_model.objects.create_user(username='trial_staff', password='secret123')
+        assign_user_role(self.staff, ROLE_STAFF)
+
+        self.trial = get_portal_trial_config()
+        self.original_values = (
+            self.trial.is_trial_mode,
+            self.trial.trial_started_at,
+            self.trial.trial_expires_at,
+            self.trial.restrict_production_integrations,
+            self.trial.auto_cleanup_enabled,
+        )
+
+    def tearDown(self):
+        (
+            self.trial.is_trial_mode,
+            self.trial.trial_started_at,
+            self.trial.trial_expires_at,
+            self.trial.restrict_production_integrations,
+            self.trial.auto_cleanup_enabled,
+        ) = self.original_values
+        self.trial.save()
+
+    def test_staff_is_blocked_after_trial_expiry(self):
+        self.trial.is_trial_mode = True
+        self.trial.trial_started_at = timezone.now() - timezone.timedelta(days=10)
+        self.trial.trial_expires_at = timezone.now() - timezone.timedelta(days=1)
+        self.trial.save()
+
+        client = Client(HTTP_HOST='127.0.0.1')
+        client.force_login(self.staff)
+        response = client.get('/requests/')
+
+        self.assertEqual(response.status_code, 403)
+        self.assertIn('Trial abgelaufen', response.content.decode('utf-8'))
+
+    def test_platform_owner_keeps_access_after_trial_expiry(self):
+        self.trial.is_trial_mode = True
+        self.trial.trial_started_at = timezone.now() - timezone.timedelta(days=10)
+        self.trial.trial_expires_at = timezone.now() - timezone.timedelta(days=1)
+        self.trial.save()
+
+        client = Client(HTTP_HOST='127.0.0.1')
+        client.force_login(self.platform_owner)
+        response = client.get('/requests/')
+
+        self.assertEqual(response.status_code, 200)
+
+    def test_trial_restriction_forces_safe_integration_modes(self):
+        self.trial.is_trial_mode = True
+        self.trial.trial_started_at = timezone.now() - timezone.timedelta(days=1)
+        self.trial.trial_expires_at = timezone.now() + timezone.timedelta(days=2)
+        self.trial.restrict_production_integrations = True
+        self.trial.save()
+
+        self.assertFalse(is_nextcloud_enabled())
+        self.assertTrue(is_email_test_mode())
+