Merge pull request #6 from tekdi/claude/frappe-hrms-cv-upload-011CV5La1ED2RTd2Ycx4vWcQ

Add multipart upload testing to test script

Author: coolbung

services/hrms-tools/tests/test_service.py

@@ -5,16 +5,18 @@
 This script tests the CV Analysis Service locally by:
 1. Checking service health
 2. Creating a sample CV
-3. Sending it for analysis
+3. Sending it for analysis (base64 or multipart upload)
 4. Validating the response
 
 Usage:
-    python test_service.py [--url URL] [--provider PROVIDER]
+    python test_service.py [--url URL] [--provider PROVIDER] [--method METHOD]
 
 Examples:
     python test_service.py
     python test_service.py --url http://localhost:8000
     python test_service.py --provider anthropic
+    python test_service.py --method multipart
+    python test_service.py --method both
 """
 
 import requests
@@ -206,7 +208,7 @@ def test_health(base_url: str) -> bool:
 
 def test_analyze(base_url: str, llm_provider: str = "auto") -> bool:
     """
-    Test the CV analysis endpoint
+    Test the CV analysis endpoint (Base64 method)
 
     Args:
         base_url: Base URL of the service
@@ -215,7 +217,7 @@ def test_analyze(base_url: str, llm_provider: str = "auto") -> bool:
     Returns:
         True if successful, False otherwise
     """
-    print(f"\n{Colors.BOLD}Testing CV Analysis Endpoint{Colors.END}")
+    print(f"\n{Colors.BOLD}Testing CV Analysis Endpoint (Base64){Colors.END}")
     print(f"{Colors.BLUE}URL: {base_url}/api/v1/analyze{Colors.END}")
 
     # Create sample CV
@@ -339,6 +341,146 @@ def test_analyze(base_url: str, llm_provider: str = "auto") -> bool:
         return False
 
 
+def test_analyze_multipart(base_url: str, llm_provider: str = "auto") -> bool:
+    """
+    Test the CV analysis endpoint (Multipart upload method)
+
+    Args:
+        base_url: Base URL of the service
+        llm_provider: LLM provider to use
+
+    Returns:
+        True if successful, False otherwise
+    """
+    print(f"\n{Colors.BOLD}Testing CV Analysis Endpoint (Multipart Upload){Colors.END}")
+    print(f"{Colors.BLUE}URL: {base_url}/api/v1/analyze-upload{Colors.END}")
+
+    # Create sample CV
+    print("Creating sample CV...")
+    cv_bytes = create_sample_cv()
+    print(f"  CV size: {len(cv_bytes)} bytes")
+
+    # Prepare position framework and company criteria as JSON strings
+    position_framework = {
+        "role_title": "Senior Backend Engineer",
+        "key_requirements": [
+            "5+ years Python experience",
+            "Microservices architecture expertise",
+            "Database design and optimization",
+            "REST API development",
+            "Cloud infrastructure experience"
+        ],
+        "scoring_weights": {
+            "technical_skills": 40,
+            "experience": 30,
+            "education": 15,
+            "cultural_fit": 15
+        },
+        "must_have_skills": ["Python", "REST API", "Database Design"],
+        "nice_to_have_skills": ["Docker", "Kubernetes", "AWS"],
+        "experience_years_required": 5
+    }
+
+    company_criteria = {
+        "company_name": "ACME Corp",
+        "values": ["Innovation", "Collaboration", "Ownership", "Excellence"],
+        "evaluation_guidelines": "Focus on problem-solving ability, architectural thinking, and team collaboration. Value candidates who demonstrate continuous learning and technical depth.",
+        "disqualifiers": ["Less than 3 years experience", "No backend development experience"],
+        "preferred_backgrounds": ["Computer Science degree", "Startup experience"]
+    }
+
+    # Prepare multipart form data
+    files = {
+        'cv_file': ('john_doe_cv.pdf', cv_bytes, 'application/pdf')
+    }
+
+    data = {
+        'position_framework': json.dumps(position_framework),
+        'company_criteria': json.dumps(company_criteria),
+        'llm_provider': llm_provider,
+        'prompt_version': 'v1',
+        'analysis_depth': 'detailed'
+    }
+
+    print(f"Sending multipart analysis request (Provider: {llm_provider})...")
+
+    try:
+        response = requests.post(
+            f"{base_url}/api/v1/analyze-upload",
+            files=files,
+            data=data,
+            timeout=120
+        )
+        response.raise_for_status()
+
+        result = response.json()
+
+        print(f"\n{Colors.GREEN}✓ Analysis completed successfully{Colors.END}\n")
+
+        # Display results
+        print(f"{Colors.BOLD}=== ANALYSIS RESULTS ==={Colors.END}\n")
+
+        print(f"{Colors.BOLD}Analysis ID:{Colors.END} {result['analysis_id']}")
+        print(f"{Colors.BOLD}Overall Score:{Colors.END} {result['overall_score']}/100")
+        print(f"{Colors.BOLD}Recommendation:{Colors.END} {result['recommendation']}")
+
+        # Section Scores
+        print(f"\n{Colors.BOLD}Section Scores:{Colors.END}")
+        for section in result['section_scores']:
+            print(f"\n  {Colors.BLUE}{section['section']}{Colors.END}")
+            print(f"    Score: {section['score']}/100 (Weight: {section['weight']}%)")
+            print(f"    Weighted: {section['weighted_score']:.2f}")
+            print(f"    Rationale: {section['rationale'][:100]}...")
+
+        # Key Strengths
+        print(f"\n{Colors.BOLD}Key Strengths:{Colors.END}")
+        for i, strength in enumerate(result['key_strengths'], 1):
+            print(f"  {i}. {strength}")
+
+        # Critical Gaps
+        print(f"\n{Colors.BOLD}Critical Gaps:{Colors.END}")
+        if result['critical_gaps']:
+            for i, gap in enumerate(result['critical_gaps'], 1):
+                print(f"  {i}. {gap}")
+        else:
+            print("  None identified")
+
+        # Follow-up Questions
+        print(f"\n{Colors.BOLD}Follow-up Questions:{Colors.END}")
+        for i, question in enumerate(result['follow_up_questions'], 1):
+            print(f"  {i}. {question}")
+
+        # Metadata
+        print(f"\n{Colors.BOLD}Analysis Metadata:{Colors.END}")
+        metadata = result['metadata']
+        print(f"  Provider: {metadata['llm_provider']}")
+        print(f"  Model: {metadata['model']}")
+        print(f"  Tokens Used: {metadata.get('tokens_used', 'N/A')}")
+        print(f"  Processing Time: {metadata['processing_time_ms']}ms")
+        print(f"  CV Pages: {metadata.get('cv_pages', 'N/A')}")
+
+        # Save full response to file
+        output_file = "test_analysis_result_multipart.json"
+        with open(output_file, 'w') as f:
+            json.dump(result, f, indent=2, default=str)
+        print(f"\n{Colors.GREEN}Full response saved to: {output_file}{Colors.END}")
+
+        return True
+
+    except requests.exceptions.RequestException as e:
+        print(f"{Colors.RED}✗ Analysis failed: {e}{Colors.END}")
+        if hasattr(e, 'response') and e.response is not None:
+            try:
+                error_detail = e.response.json()
+                print(f"  Error details: {json.dumps(error_detail, indent=2)}")
+            except:
+                print(f"  Response text: {e.response.text[:500]}")
+        return False
+    except Exception as e:
+        print(f"{Colors.RED}✗ Unexpected error: {e}{Colors.END}")
+        return False
+
+
 def main():
     """Main test runner"""
     parser = argparse.ArgumentParser(
@@ -349,7 +491,9 @@ def main():
   python test_service.py
   python test_service.py --url http://localhost:8000
   python test_service.py --provider anthropic
-  python test_service.py --url http://localhost:8000 --provider openai
+  python test_service.py --method multipart
+  python test_service.py --url http://localhost:8000 --provider openai --method base64
+  python test_service.py --method both
         """
     )
     parser.add_argument(
@@ -363,6 +507,12 @@ def main():
         choices=['auto', 'openai', 'anthropic', 'gemini'],
         help='LLM provider to use (default: auto)'
     )
+    parser.add_argument(
+        '--method',
+        default='multipart',
+        choices=['base64', 'multipart', 'both'],
+        help='Upload method to test (default: multipart, recommended for efficiency)'
+    )
     parser.add_argument(
         '--health-only',
         action='store_true',
@@ -379,6 +529,7 @@ def main():
 
     print(f"Service URL: {args.url}")
     print(f"LLM Provider: {args.provider}")
+    print(f"Upload Method: {args.method}")
 
     # Test health endpoint
     health_ok = test_health(args.url)
@@ -394,21 +545,34 @@ def main():
         print(f"\n{Colors.GREEN}Health check only - skipping analysis test{Colors.END}")
         sys.exit(0)
 
-    # Test analysis endpoint
-    analysis_ok = test_analyze(args.url, args.provider)
+    # Test analysis endpoint(s)
+    base64_ok = True
+    multipart_ok = True
+
+    if args.method in ['base64', 'both']:
+        base64_ok = test_analyze(args.url, args.provider)
+
+    if args.method in ['multipart', 'both']:
+        multipart_ok = test_analyze_multipart(args.url, args.provider)
 
     # Summary
     print(f"\n{Colors.BOLD}{'=' * 60}{Colors.END}")
     print(f"{Colors.BOLD}Test Summary{Colors.END}")
     print(f"{'=' * 60}")
 
     health_status = f"{Colors.GREEN}PASS{Colors.END}" if health_ok else f"{Colors.RED}FAIL{Colors.END}"
-    analysis_status = f"{Colors.GREEN}PASS{Colors.END}" if analysis_ok else f"{Colors.RED}FAIL{Colors.END}"
-
     print(f"Health Check: {health_status}")
-    print(f"CV Analysis: {analysis_status}")
 
-    if health_ok and analysis_ok:
+    if args.method in ['base64', 'both']:
+        base64_status = f"{Colors.GREEN}PASS{Colors.END}" if base64_ok else f"{Colors.RED}FAIL{Colors.END}"
+        print(f"CV Analysis (Base64): {base64_status}")
+
+    if args.method in ['multipart', 'both']:
+        multipart_status = f"{Colors.GREEN}PASS{Colors.END}" if multipart_ok else f"{Colors.RED}FAIL{Colors.END}"
+        print(f"CV Analysis (Multipart): {multipart_status}")
+
+    all_ok = health_ok and base64_ok and multipart_ok
+    if all_ok:
         print(f"\n{Colors.GREEN}{Colors.BOLD}✓ All tests passed!{Colors.END}")
         sys.exit(0)
     else: