import base64 from unittest.mock import patch from django.contrib.auth import get_user_model from django.core.files.base import ContentFile from django.test import Client, TestCase from rest_framework.authtoken.models import Token from rest_framework.test import APIClient, APITestCase from accounts.kyc_review import apply_kyc_review, sync_user_kyc_verified from accounts.models import KYCDocument, TransporterProfile, UserRole User = get_user_model() _TINY_PNG = ( 'data:image/png;base64,' + base64.b64encode( b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01' b'\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89' b'\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n-\xdb' b'\x00\x00\x00\x00IEND\xaeB`\x82' ).decode() ) def _png_content(name='doc.png'): raw = base64.b64decode(_TINY_PNG.split(',', 1)[1]) return ContentFile(raw, name=name) class KycReviewHelperTests(TestCase): def setUp(self): self.driver = User.objects.create_user( 'kyc-driver@test.com', email='kyc-driver@test.com', password='secret', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') self.profile = TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, ) self.admin = User.objects.create_superuser( 'kyc-admin@test.com', email='kyc-admin@test.com', password='secret', ) def _make_doc(self, doc_type, **kwargs): doc = KYCDocument( user=self.driver, document_type=doc_type, **kwargs, ) doc.file.save('front.png', _png_content(), save=False) if doc_type in KYCDocument.TRANSPORTER_TWO_SIDED_TYPES: doc.file_back.save('back.png', _png_content('back.png'), save=False) doc.save() return doc @patch('api.notify.notify_users') def test_reject_requires_notes_and_notifies(self, mock_notify): doc = self._make_doc(KYCDocument.DocumentType.PASSPORT_COPY) with self.assertRaises(ValueError): apply_kyc_review( doc, review_status=KYCDocument.ReviewStatus.REJECTED, review_notes='', reviewer=self.admin, ) apply_kyc_review( doc, review_status=KYCDocument.ReviewStatus.REJECTED, review_notes='Blurry passport photo', reviewer=self.admin, ) doc.refresh_from_db() self.assertEqual(doc.review_status, KYCDocument.ReviewStatus.REJECTED) self.assertFalse(doc.verified) self.assertEqual(doc.review_notes, 'Blurry passport photo') mock_notify.assert_called_once() args, kwargs = mock_notify.call_args self.assertEqual(args[0], [self.driver.id]) self.assertEqual(args[1], 'DOCUMENT_REJECTED') self.assertEqual(args[2]['reason'], 'Blurry passport photo') self.assertIn('Passport', kwargs['body']) @patch('api.notify.notify_users') def test_approve_and_sync_profile_when_all_required(self, _mock_notify): docs = [] for doc_type in KYCDocument.DRIVER_PERSON_KYC_TYPES: docs.append(self._make_doc(doc_type)) self.assertFalse(self.profile.documents_verified) for doc in docs[:-1]: apply_kyc_review( doc, review_status=KYCDocument.ReviewStatus.APPROVED, reviewer=self.admin, notify=False, ) self.profile.refresh_from_db() self.assertFalse(self.profile.documents_verified) apply_kyc_review( docs[-1], review_status=KYCDocument.ReviewStatus.APPROVED, reviewer=self.admin, notify=False, ) self.profile.refresh_from_db() self.assertTrue(self.profile.documents_verified) @patch('api.notify.notify_users') def test_reject_clears_profile_verified(self, _mock_notify): for doc_type in KYCDocument.DRIVER_PERSON_KYC_TYPES: doc = self._make_doc(doc_type) apply_kyc_review( doc, review_status=KYCDocument.ReviewStatus.APPROVED, reviewer=self.admin, notify=False, ) self.profile.refresh_from_db() self.assertTrue(self.profile.documents_verified) rejected = KYCDocument.objects.get( user=self.driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, ) apply_kyc_review( rejected, review_status=KYCDocument.ReviewStatus.REJECTED, review_notes='Expired license', reviewer=self.admin, ) self.profile.refresh_from_db() self.assertFalse(self.profile.documents_verified) class KycAdminReviewViewTests(TestCase): def setUp(self): self.admin = User.objects.create_superuser( 'admin-kyc@test.com', email='admin-kyc@test.com', password='secret', ) self.driver = User.objects.create_user( 'drv-kyc@test.com', email='drv-kyc@test.com', password='secret', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, ) self.doc = KYCDocument(user=self.driver, document_type=KYCDocument.DocumentType.PASSPORT_COPY) self.doc.file.save('p.png', _png_content(), save=False) self.doc.save() self.client = Client() self.client.force_login(self.admin) @patch('api.notify.notify_users') def test_ajax_reject_without_notes_fails(self, mock_notify): response = self.client.post( f'/kyc-documents/{self.doc.id}/verified', {'review_status': 'REJECTED', 'review_notes': ''}, HTTP_X_REQUESTED_WITH='XMLHttpRequest', ) self.assertEqual(response.status_code, 400) mock_notify.assert_not_called() @patch('api.notify.notify_users') def test_ajax_approve_ok(self, mock_notify): response = self.client.post( f'/kyc-documents/{self.doc.id}/verified', {'review_status': 'APPROVED', 'review_notes': ''}, HTTP_X_REQUESTED_WITH='XMLHttpRequest', ) self.assertEqual(response.status_code, 200) body = response.json() self.assertTrue(body['success']) self.assertEqual(body['document']['review_status'], 'APPROVED') mock_notify.assert_not_called() class KycResubmitApiTests(APITestCase): def setUp(self): self.driver = User.objects.create_user( 'resubmit@test.com', email='resubmit@test.com', password='secret', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, ) self.doc = KYCDocument( user=self.driver, document_type=KYCDocument.DocumentType.PASSPORT_COPY, review_status=KYCDocument.ReviewStatus.REJECTED, review_notes='Bad scan', verified=False, passport_number='RESUBMIT-1', ) self.doc.file.save('old.png', _png_content(), save=False) self.doc.save() token, _ = Token.objects.get_or_create(user=self.driver) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') def test_list_includes_review_fields(self): response = self.client.get('/api/transporter/documents/') self.assertEqual(response.status_code, 200) row = response.json()['data'][0] self.assertEqual(row['review_status'], 'REJECTED') self.assertEqual(row['review_notes'], 'Bad scan') self.assertEqual(row['passport_number'], 'RESUBMIT-1') def test_reupload_resets_to_pending(self): response = self.client.post( '/api/transporter/documents/', { 'document_type': 'PASSPORT_COPY', 'file': _TINY_PNG, 'passport_number': 'RESUBMIT-1', }, format='json', ) self.assertEqual(response.status_code, 201, response.content) data = response.json()['data'] self.assertEqual(data['review_status'], 'PENDING') self.assertEqual(data['review_notes'], '') self.doc.refresh_from_db() self.assertEqual(self.doc.review_status, KYCDocument.ReviewStatus.PENDING) self.assertEqual(self.doc.review_notes, '')