vickytechkey
722f15a89e
All checks were successful
Deploy Beta (NATIVE) / deploy (push) Successful in 29s
745 lines
29 KiB
Python
745 lines
29 KiB
Python
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from rest_framework import status
|
|
from rest_framework.test import APIClient
|
|
|
|
from verifier.engine.base import BaseVendor, VendorResult, VerificationStatus
|
|
from verifier.engine.registry import VendorRegistry, registry
|
|
from verifier.engine.orchestrator import VerificationEngine
|
|
from verifier.vendors.codeplus import CodeplusVendor
|
|
from verifier.models import VerificationTransaction, VendorAttemptLog
|
|
|
|
|
|
class MockFailingVendor(BaseVendor):
|
|
"""Simulates a primary vendor that encounters an API error/failure."""
|
|
name = "vendor_alpha"
|
|
priority = 10
|
|
is_active = True
|
|
supported_documents = ["pan", "aadhaar"]
|
|
|
|
def execute_verification(self, document_type, data, files=None):
|
|
return VendorResult(
|
|
vendor_name=self.name,
|
|
is_success=False,
|
|
status=VerificationStatus.ERROR,
|
|
error_code="VENDOR_API_DOWNTIME",
|
|
error_message="Primary vendor Alpha API returned HTTP 503 Service Unavailable",
|
|
retryable=True,
|
|
)
|
|
|
|
|
|
class MockSucceedingVendor(BaseVendor):
|
|
"""Simulates a fallback vendor that successfully completes verification."""
|
|
name = "vendor_beta"
|
|
priority = 20
|
|
is_active = True
|
|
supported_documents = ["pan", "aadhaar", "driving_license"]
|
|
|
|
def execute_verification(self, document_type, data, files=None):
|
|
return VendorResult(
|
|
vendor_name=self.name,
|
|
is_success=True,
|
|
status=VerificationStatus.SUCCESS,
|
|
data={
|
|
"document_number": data.get("pan_number", "ABCDE1234F"),
|
|
"full_name": "Test User",
|
|
"verification_status": "VALID",
|
|
},
|
|
)
|
|
|
|
|
|
class VerificationEngineWaterfallTests(TestCase):
|
|
"""Tests the core failover / waterfall mechanism of the VerificationEngine."""
|
|
|
|
def setUp(self):
|
|
self.custom_registry = VendorRegistry()
|
|
self.vendor_alpha = MockFailingVendor()
|
|
self.vendor_beta = MockSucceedingVendor()
|
|
|
|
self.custom_registry.register(self.vendor_alpha)
|
|
self.custom_registry.register(self.vendor_beta)
|
|
self.engine = VerificationEngine(vendor_registry=self.custom_registry)
|
|
|
|
def test_waterfall_falls_back_when_first_vendor_errors(self):
|
|
"""When vendor_alpha throws an error, engine must fall back to vendor_beta and succeed."""
|
|
result = self.engine.verify(
|
|
document_type="pan",
|
|
data={"pan_number": "ABCDE1234F", "name": "Test User"},
|
|
)
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.successful_vendor, "vendor_beta")
|
|
self.assertEqual(len(result.attempts), 2)
|
|
|
|
# Check attempt 1 (failed)
|
|
attempt_1 = result.attempts[0]
|
|
self.assertEqual(attempt_1["vendor_name"], "vendor_alpha")
|
|
self.assertFalse(attempt_1["is_success"])
|
|
self.assertEqual(attempt_1["error_code"], "VENDOR_API_DOWNTIME")
|
|
|
|
# Check attempt 2 (succeeded)
|
|
attempt_2 = result.attempts[1]
|
|
self.assertEqual(attempt_2["vendor_name"], "vendor_beta")
|
|
self.assertTrue(attempt_2["is_success"])
|
|
self.assertEqual(attempt_2["data"]["verification_status"], "VALID")
|
|
|
|
def test_unsupported_document_type_returns_error(self):
|
|
"""When no registered vendor supports the document, engine returns appropriate error."""
|
|
result = self.engine.verify(
|
|
document_type="unsupported_doc",
|
|
data={"id": "12345"},
|
|
)
|
|
self.assertFalse(result.is_success)
|
|
self.assertIn("No active vendors", result.error_message)
|
|
|
|
def test_preferred_vendor_overrides_priority(self):
|
|
"""Specifying preferred_vendor places it at the front of the chain."""
|
|
vendors = self.custom_registry.get_vendors_for_document("pan", preferred_vendor="vendor_beta")
|
|
self.assertEqual(vendors[0].name, "vendor_beta")
|
|
self.assertEqual(vendors[1].name, "vendor_alpha")
|
|
|
|
|
|
class CodeplusVendorTests(TestCase):
|
|
"""Tests Codeplus vendor adapter configuration and handling."""
|
|
|
|
def setUp(self):
|
|
self.codeplus = CodeplusVendor(
|
|
config={
|
|
"base_url": "https://apihub.services",
|
|
"client_api_id": "CAPI-YYR0YJIE0VMTDENY",
|
|
"client_api_secret": "test_secret_123",
|
|
}
|
|
)
|
|
|
|
def test_codeplus_metadata(self):
|
|
self.assertEqual(self.codeplus.name, "codeplus")
|
|
self.assertTrue(self.codeplus.supports("pan"))
|
|
self.assertTrue(self.codeplus.supports("aadhaar"))
|
|
self.assertTrue(self.codeplus.supports("driving_license"))
|
|
|
|
def test_codeplus_headers(self):
|
|
headers = self.codeplus.get_headers()
|
|
self.assertEqual(headers["X-Client-Api-ID"], "CAPI-YYR0YJIE0VMTDENY")
|
|
self.assertEqual(headers["X-Client-Api-Secret"], "test_secret_123")
|
|
self.assertEqual(headers["Content-Type"], "application/json")
|
|
|
|
def test_codeplus_pan_v5_success(self):
|
|
"""Test parsing valid PAN response via v5 Recommended API."""
|
|
mock_response_payload = {
|
|
"success": True,
|
|
"message": "PAN verification completed",
|
|
"data": {
|
|
"transactionId": "TXN_1787136465045_demo12345",
|
|
"service": "PAN_VERIFICATION",
|
|
"success": True,
|
|
"resultCode": 101,
|
|
"status": "SUCCESS",
|
|
"message": "PAN verified successfully",
|
|
"providerTransactionId": "TXN_1787136465045_demo12345",
|
|
"pan": "ABCDE1234F",
|
|
"panType": "Individual",
|
|
"fullName": "RAHUL KUMAR",
|
|
"firstName": "RAHUL",
|
|
"middleName": None,
|
|
"lastName": "KUMAR",
|
|
"gender": "male",
|
|
"dob": "15/06/1995",
|
|
"aadhaarNumber": "XXXXXXXX1234",
|
|
"aadhaarLinked": True,
|
|
"addressDetails": {
|
|
"building_name": "12/45",
|
|
"locality": "Central Area",
|
|
"street_name": "MG Road",
|
|
"pincode": "641001",
|
|
"city": "COIMBATORE",
|
|
"state": "Tamil Nadu",
|
|
"country": "India",
|
|
},
|
|
"clientRefNum": "REQ_1787136465034_demo12345",
|
|
},
|
|
"requestId": "REQ_1787136465034_demo12345",
|
|
"timestamp": "2026-08-19T10:47:45.735Z",
|
|
"statusCode": 200,
|
|
"status": "SUCCESS",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("pan", {"pan": "ABCDE1234F"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["full_name"], "RAHUL KUMAR")
|
|
self.assertEqual(result.data["dob"], "15/06/1995")
|
|
self.assertEqual(result.data["pan_type"], "Individual")
|
|
self.assertEqual(result.data["address_details"]["city"], "COIMBATORE")
|
|
self.assertTrue(result.data["aadhaar_linked"])
|
|
|
|
def test_codeplus_pan_advanced_success(self):
|
|
"""Test parsing valid PAN response via v4 Advanced API (HTTP 200, code 1000)."""
|
|
mock_response_payload = {
|
|
"request_id": "8a42f1c7-63d9-4e15-b820-7c91a5d34628",
|
|
"transaction_id": "d5b73e91-2a46-4c08-9f61-83e2b7a51439",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1000",
|
|
"message": "Valid PAN.",
|
|
"pan_data": {
|
|
"document_type": "PAN",
|
|
"document_id": "ABCDE1234F",
|
|
"name": "RAHUL KUMAR",
|
|
"first_name": "RAHUL",
|
|
"last_name": "KUMAR",
|
|
"category": "P",
|
|
"category_type": "Individual",
|
|
"is_individual": True,
|
|
"individual_tax_compliance_status": "Operative",
|
|
"aadhaar_linked": True,
|
|
},
|
|
},
|
|
"timestamp": 1787031203055,
|
|
"path": "/pan-api/fetch-advanced",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("pan_advanced", {"pan": "ABCDE1234F"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["name"], "RAHUL KUMAR")
|
|
self.assertEqual(result.data["document_id"], "ABCDE1234F")
|
|
self.assertTrue(result.data["aadhaar_linked"])
|
|
|
|
def test_codeplus_pan_advanced_error_nonexistent(self):
|
|
"""Test parsing non-existent PAN (code 1004)."""
|
|
mock_error_payload = {
|
|
"request_id": "23bc5fa5-a02e-4246-aefc-3c455bcd1853",
|
|
"transaction_id": "542f956c-db4f-48f5-8d6b-d1877a0b930a",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1004",
|
|
"message": "PAN does not exist.",
|
|
},
|
|
"timestamp": 1787031223795,
|
|
"path": "/pan-api/fetch-advanced",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 400
|
|
def json(self):
|
|
return mock_error_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("pan_advanced", {"pan": "INVALID999Z"})
|
|
|
|
self.assertFalse(result.is_success)
|
|
self.assertEqual(result.error_code, "1004")
|
|
self.assertEqual(result.error_message, "PAN does not exist.")
|
|
|
|
def test_codeplus_pan_detailed_success(self):
|
|
"""Test parsing valid Detailed PAN response (HTTP 200, code 1000 with address/email/phone)."""
|
|
mock_response_payload = {
|
|
"request_id": "7f3a1c92-5b84-4d21-a7e6-91c2f0d84b56",
|
|
"transaction_id": "b6e8d4a1-2c73-49f5-9a10-5f7c3e821d64",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1000",
|
|
"message": "Valid PAN.",
|
|
"pan_data": {
|
|
"document_type": "PAN",
|
|
"document_id": "ABCDE1234F",
|
|
"name": "RAHUL KUMAR",
|
|
"first_name": "RAHUL",
|
|
"last_name": "KUMAR",
|
|
"category": "P",
|
|
"date_of_birth": "1995-06-15",
|
|
"masked_aadhaar_number": "XXXXXXXX1234",
|
|
"address_data": {
|
|
"line_1": "12/45",
|
|
"city": "COIMBATORE",
|
|
"state": "TAMIL NADU",
|
|
"pincode": "641001",
|
|
},
|
|
"email": "ra******@example.com",
|
|
"phone": "98XXXXXX10",
|
|
"gender": "MALE",
|
|
"aadhaar_linked": True,
|
|
"taxpayer_profile": True,
|
|
},
|
|
},
|
|
"timestamp": 1787028660489,
|
|
"path": "/pan-api/fetch-detailed",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("pan_detailed", {"pan_number": "ABCDE1234F"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["name"], "RAHUL KUMAR")
|
|
self.assertEqual(result.data["email"], "ra******@example.com")
|
|
self.assertEqual(result.data["phone"], "98XXXXXX10")
|
|
self.assertEqual(result.data["date_of_birth"], "1995-06-15")
|
|
self.assertEqual(result.data["address_data"]["city"], "COIMBATORE")
|
|
|
|
def test_codeplus_pan_detailed_error(self):
|
|
"""Test parsing invalid Detailed PAN error (HTTP 400 with error block)."""
|
|
mock_error_payload = {
|
|
"request_id": "b89afa37-8282-4cf7-827f-985700f84ab8",
|
|
"transaction_id": "d796efe7-26e2-41c0-a1e7-2f7387b2d5aa",
|
|
"status": 400,
|
|
"error": {
|
|
"code": "INVALID_PAN",
|
|
"message": "Invalid PAN number.",
|
|
},
|
|
"timestamp": 1787028631589,
|
|
"path": "/pan-api/fetch-detailed",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 400
|
|
def json(self):
|
|
return mock_error_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("pan_detailed", {"pan_number": "INVALID999Z"})
|
|
|
|
self.assertFalse(result.is_success)
|
|
self.assertEqual(result.error_code, "INVALID_PAN")
|
|
self.assertEqual(result.error_message, "Invalid PAN number.")
|
|
|
|
def test_codeplus_bank_verify_success(self):
|
|
"""Test successful Bank Account verification."""
|
|
mock_response_payload = {
|
|
"request_id": "b1a2c3d4-e5f6-7890-abcd-ef1234567890",
|
|
"transaction_id": "tx-bank-998877",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1000",
|
|
"message": "Bank account verified successfully.",
|
|
"bank_data": {
|
|
"account_number": "123456789012",
|
|
"ifsc": "HDFC0001234",
|
|
"name_at_bank": "PRIYA SHARMA",
|
|
"micr": "641240002",
|
|
"bank_name": "HDFC BANK",
|
|
"branch": "MAIN BRANCH",
|
|
"city": "COIMBATORE",
|
|
"state": "TAMIL NADU",
|
|
"pincode": "641001",
|
|
"account_exists": True,
|
|
},
|
|
},
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify(
|
|
"bank_account",
|
|
{"account_number": "123456789012", "ifsc": "HDFC0001234"},
|
|
)
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["beneficiary_name"], "PRIYA SHARMA")
|
|
self.assertEqual(result.data["bank_name"], "HDFC BANK")
|
|
self.assertEqual(result.data["micr"], "641240002")
|
|
self.assertEqual(result.data["city"], "COIMBATORE")
|
|
|
|
def test_codeplus_bank_verify_missing_fields(self):
|
|
"""Test bank verification validation for missing account or IFSC."""
|
|
result_missing_acc = self.codeplus.verify("bank_account", {"ifsc": "HDFC0001234"})
|
|
self.assertFalse(result_missing_acc.is_success)
|
|
self.assertEqual(result_missing_acc.error_code, "MISSING_ACCOUNT_NUMBER")
|
|
|
|
result_missing_ifsc = self.codeplus.verify("bank_account", {"account_number": "123456"})
|
|
self.assertFalse(result_missing_ifsc.is_success)
|
|
self.assertEqual(result_missing_ifsc.error_code, "MISSING_IFSC")
|
|
|
|
def test_codeplus_fetch_gst_success(self):
|
|
"""Test successful fetch GST by PAN."""
|
|
mock_response_payload = {
|
|
"request_id": "b27c9087-f061-4d38-b7aa-96498f8e667e",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1002",
|
|
"message": "Found 3 records",
|
|
"results": [
|
|
{
|
|
"document_type": "GSTIN",
|
|
"document_id": "20AXXXXXXXXXXX",
|
|
"status": "Active",
|
|
"state": "Jharkhand",
|
|
"state_code": "20",
|
|
},
|
|
{
|
|
"document_type": "GSTIN",
|
|
"document_id": "27AXXXXXXXXXXX",
|
|
"status": "Active",
|
|
"state": "Maharashtra",
|
|
"state_code": "27",
|
|
},
|
|
{
|
|
"document_type": "GSTIN",
|
|
"document_id": "29AXXXXXXXXXXX",
|
|
"status": "Active",
|
|
"state": "Karnataka",
|
|
"state_code": "29",
|
|
},
|
|
],
|
|
},
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("gst_by_pan", {"pan": "ABCDE1234F"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["records_count"], 3)
|
|
self.assertEqual(len(result.data["results"]), 3)
|
|
self.assertEqual(result.data["results"][0]["document_id"], "20AXXXXXXXXXXX")
|
|
|
|
def test_codeplus_fetch_gst_missing_pan(self):
|
|
"""Test validation for missing PAN in GST by PAN."""
|
|
result = self.codeplus.verify("gst_by_pan", {})
|
|
self.assertFalse(result.is_success)
|
|
self.assertEqual(result.error_code, "MISSING_PAN")
|
|
|
|
def test_codeplus_fetch_msme_no_record(self):
|
|
"""Test MSME by PAN returning code 1011 (No record found)."""
|
|
mock_response_payload = {
|
|
"request_id": "cd7a7566-3122-4942-a110-d2b0ab81224d",
|
|
"transaction_id": "87186c96-f51d-436b-9fce-3a9d1ab8c44c",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1011",
|
|
"message": "No record found.",
|
|
},
|
|
"timestamp": 1741349527562,
|
|
"path": "/msme-api/udyam/fetch-by-pan",
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("msme", {"pan": "ABCDE1234F"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertFalse(result.data["has_msme"])
|
|
self.assertEqual(result.data["code"], "1011")
|
|
self.assertEqual(result.data["message"], "No record found.")
|
|
|
|
def test_codeplus_fetch_msme_missing_pan(self):
|
|
"""Test validation for missing PAN in MSME."""
|
|
result = self.codeplus.verify("msme", {})
|
|
self.assertFalse(result.is_success)
|
|
self.assertEqual(result.error_code, "MISSING_PAN")
|
|
|
|
def test_codeplus_gst_verify_v5_success(self):
|
|
"""Test successful GST verification v5."""
|
|
mock_response_payload = {
|
|
"success": True,
|
|
"message": "GST verification completed",
|
|
"data": {
|
|
"service": "GST_VERIFICATION",
|
|
"gstNumber": "27ABCDE1234F1Z5",
|
|
"legalName": "ABCDE PRIVATE LIMITED",
|
|
"registrationStatus": "ACTIVE",
|
|
},
|
|
"statusCode": 200,
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_response_payload
|
|
|
|
from unittest.mock import patch
|
|
with patch.object(self.codeplus.session, "post", return_value=MockResponse()):
|
|
result = self.codeplus.verify("gstin", {"gst_number": "27ABCDE1234F1Z5"})
|
|
|
|
self.assertTrue(result.is_success)
|
|
self.assertEqual(result.status, VerificationStatus.SUCCESS)
|
|
self.assertEqual(result.data["gst_number"], "27ABCDE1234F1Z5")
|
|
self.assertEqual(result.data["legal_name"], "ABCDE PRIVATE LIMITED")
|
|
self.assertEqual(result.data["registration_status"], "ACTIVE")
|
|
|
|
def test_codeplus_gst_verify_v5_missing_gst(self):
|
|
"""Test validation for missing GST number."""
|
|
result = self.codeplus.verify("gstin", {})
|
|
self.assertFalse(result.is_success)
|
|
self.assertEqual(result.error_code, "MISSING_GST_NUMBER")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class VerificationAPITests(TestCase):
|
|
"""Tests the DRF API endpoints with waterfall execution and persistence."""
|
|
|
|
def setUp(self):
|
|
self.client = APIClient()
|
|
|
|
def test_vendor_list_api(self):
|
|
url = reverse("verifier:list_vendors")
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertIn("vendors", response.data)
|
|
vendor_names = [v["name"] for v in response.data["vendors"]]
|
|
self.assertIn("codeplus", vendor_names)
|
|
|
|
def test_verify_api_validation_error(self):
|
|
url = reverse("verifier:verify_document")
|
|
response = self.client.post(url, {}, format="json")
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertFalse(response.data["is_success"])
|
|
|
|
def test_verify_api_database_persistence(self):
|
|
"""Ensure full transaction and attempt logs (including raw response and client response) are saved in DB."""
|
|
from unittest.mock import patch
|
|
from verifier.vendors import codeplus_vendor
|
|
|
|
mock_v5_response = {
|
|
"success": True,
|
|
"message": "PAN verification completed",
|
|
"data": {
|
|
"transactionId": "TXN_TEST_DB_123",
|
|
"service": "PAN_VERIFICATION",
|
|
"success": True,
|
|
"resultCode": 101,
|
|
"status": "SUCCESS",
|
|
"pan": "ABCDE1234F",
|
|
"fullName": "DATABASE PERSISTENCE USER",
|
|
"dob": "01/01/1990",
|
|
"aadhaarLinked": True,
|
|
},
|
|
"requestId": "REQ_TEST_DB_123",
|
|
"statusCode": 200,
|
|
}
|
|
|
|
class MockResponse:
|
|
status_code = 200
|
|
def json(self):
|
|
return mock_v5_response
|
|
|
|
url = reverse("verifier:verify_document")
|
|
with patch.object(codeplus_vendor.session, "post", return_value=MockResponse()):
|
|
res = self.client.post(
|
|
url,
|
|
{
|
|
"document_type": "pan",
|
|
"data": {"pan": "ABCDE1234F"},
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(res.status_code, status.HTTP_200_OK)
|
|
ref_id = res.data["reference_id"]
|
|
|
|
# Verify transaction in PostgreSQL DB
|
|
tx = VerificationTransaction.objects.get(reference_id=ref_id)
|
|
self.assertEqual(tx.document_type, "pan")
|
|
self.assertEqual(tx.status, "SUCCESS")
|
|
self.assertEqual(tx.successful_vendor, "codeplus")
|
|
self.assertEqual(tx.final_data["full_name"], "DATABASE PERSISTENCE USER")
|
|
self.assertEqual(tx.response_payload["reference_id"], str(ref_id))
|
|
self.assertEqual(tx.attempts_count, 1)
|
|
|
|
# Verify attempt log in DB
|
|
attempts = tx.attempts.all()
|
|
self.assertEqual(attempts.count(), 1)
|
|
attempt_1 = attempts.first()
|
|
self.assertEqual(attempt_1.vendor_name, "codeplus")
|
|
self.assertTrue(attempt_1.is_success)
|
|
self.assertEqual(attempt_1.raw_response["statusCode"], 200)
|
|
|
|
# Verify GET /api/transactions/<ref_id>/
|
|
detail_url = reverse("verifier:transaction_detail", kwargs={"reference_id": ref_id})
|
|
detail_res = self.client.get(detail_url)
|
|
self.assertEqual(detail_res.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(detail_res.data["reference_id"], str(ref_id))
|
|
self.assertEqual(len(detail_res.data["attempts"]), 1)
|
|
|
|
# Verify GET /api/transactions/ (List endpoint)
|
|
list_url = reverse("verifier:transaction_list")
|
|
list_res = self.client.get(list_url)
|
|
self.assertEqual(list_res.status_code, status.HTTP_200_OK)
|
|
self.assertGreaterEqual(list_res.data["count"], 1)
|
|
|
|
def test_portal_view_renders(self):
|
|
"""Test GET / and /api/portal/ render the verification playground portal."""
|
|
# Test root path
|
|
res_root = self.client.get("/")
|
|
self.assertEqual(res_root.status_code, status.HTTP_200_OK)
|
|
self.assertContains(res_root, "Tradhox Document Verification Studio")
|
|
self.assertContains(res_root, "PAN Verification (v5 Recommended)")
|
|
|
|
# Test /api/portal/
|
|
res_portal = self.client.get(reverse("verifier:portal"))
|
|
self.assertEqual(res_portal.status_code, status.HTTP_200_OK)
|
|
self.assertContains(res_portal, "Test Execution Console")
|
|
|
|
def test_seller_pan_gst_verification_success(self):
|
|
"""Test composite seller PAN verification, name matching, and GST pull."""
|
|
from verifier.views import compare_names
|
|
# Test name matcher directly
|
|
m1 = compare_names("RAHUL KUMAR", "RAHUL KUMAR")
|
|
self.assertTrue(m1["matched"])
|
|
self.assertEqual(m1["score"], 100.0)
|
|
|
|
m2 = compare_names("Mr. Rahul Kumar", "RAHUL KUMAR")
|
|
self.assertTrue(m2["matched"])
|
|
|
|
m_diff = compare_names("PRIYA SHARMA", "ROHIT SHARMA")
|
|
self.assertFalse(m_diff["matched"])
|
|
|
|
# Test POST /api/verify/seller/ with mocked vendor responses
|
|
mock_pan_resp = {
|
|
"statusCode": 200,
|
|
"data": {
|
|
"name": "RAHUL KUMAR",
|
|
"pan_type": "Individual",
|
|
"status": "VALID",
|
|
"aadhaar_seeding_status": "LINKED",
|
|
},
|
|
}
|
|
mock_gst_resp = {
|
|
"request_id": "req-gst-1",
|
|
"status": 200,
|
|
"data": {
|
|
"code": "1002",
|
|
"message": "Found 1 records",
|
|
"results": [
|
|
{
|
|
"document_type": "GSTIN",
|
|
"document_id": "27ABCDE1234F1Z5",
|
|
"status": "Active",
|
|
"state": "Maharashtra",
|
|
"state_code": "27",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
class MockResp:
|
|
def __init__(self, d):
|
|
self.d = d
|
|
self.status_code = 200
|
|
def json(self):
|
|
return self.d
|
|
|
|
codeplus_vendor = registry.get_vendor("codeplus")
|
|
from unittest.mock import patch
|
|
with patch.object(codeplus_vendor.session, "post") as mock_post:
|
|
mock_post.side_effect = [MockResp(mock_pan_resp), MockResp(mock_gst_resp)]
|
|
|
|
url = reverse("verifier:verify_seller_pan_gst")
|
|
res = self.client.post(
|
|
url,
|
|
{
|
|
"seller_name": "RAHUL KUMAR",
|
|
"pan": "ABCDE1234F",
|
|
"gstin": "27ABCDE1234F1Z5",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(res.status_code, status.HTTP_200_OK)
|
|
self.assertTrue(res.data["is_success"])
|
|
self.assertEqual(res.data["status"], "SUCCESS")
|
|
self.assertTrue(res.data["data"]["name_matched"])
|
|
self.assertEqual(res.data["data"]["primary_gstin"], "27ABCDE1234F1Z5")
|
|
self.assertTrue(res.data["data"]["gstin_matched"])
|
|
self.assertEqual(len(res.data["data"]["gst_records"]), 1)
|
|
|
|
# Check DB persistence
|
|
tx = VerificationTransaction.objects.get(reference_id=res.data["reference_id"])
|
|
self.assertEqual(tx.document_type, "SELLER_PAN_GST")
|
|
self.assertEqual(tx.status, "SUCCESS")
|
|
self.assertEqual(tx.final_data["verified_pan_name"], "RAHUL KUMAR")
|
|
|
|
def test_seller_pan_gst_name_mismatch(self):
|
|
"""Test seller PAN verification when entered seller name does not match PAN holder."""
|
|
mock_pan_resp = {
|
|
"statusCode": 200,
|
|
"data": {
|
|
"name": "ROHIT SHARMA",
|
|
"pan_type": "Individual",
|
|
"status": "VALID",
|
|
},
|
|
}
|
|
mock_gst_resp = {"status": 200, "data": {"code": "1011", "message": "No record found."}}
|
|
|
|
class MockResp:
|
|
def __init__(self, d):
|
|
self.d = d
|
|
self.status_code = 200
|
|
def json(self):
|
|
return self.d
|
|
|
|
codeplus_vendor = registry.get_vendor("codeplus")
|
|
from unittest.mock import patch
|
|
with patch.object(codeplus_vendor.session, "post") as mock_post:
|
|
mock_post.side_effect = [MockResp(mock_pan_resp), MockResp(mock_gst_resp)]
|
|
|
|
url = reverse("verifier:verify_seller_pan_gst")
|
|
res = self.client.post(
|
|
url,
|
|
{
|
|
"seller_name": "PRIYA VERMA",
|
|
"pan": "ABCDE1234F",
|
|
},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(res.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)
|
|
self.assertFalse(res.data["is_success"])
|
|
self.assertEqual(res.data["status"], "NAME_MISMATCH")
|
|
self.assertFalse(res.data["data"]["name_matched"])
|
|
self.assertIn("PAN Name Mismatch", res.data["error_message"])
|
|
|
|
# DB persistence check
|
|
tx = VerificationTransaction.objects.get(reference_id=res.data["reference_id"])
|
|
self.assertEqual(tx.status, "NAME_MISMATCH")
|
|
|
|
|
|
|