Spaces:
Paused
Paused
| #!/usr/bin/env python3 | |
| """ | |
| Comprehensive test to validate notes implementation according to requirements: | |
| 1. NoteID should be sequence generated for same customer (1, 2, 3, 4...) | |
| 2. Date should come from UI, not default today date | |
| 3. CustomerTypeID should be 1 (default) | |
| """ | |
| import requests | |
| import json | |
| from datetime import datetime, timedelta | |
| import sys | |
| # Test configuration | |
| BASE_URL = "http://localhost:8000" # Adjust if your API runs on different port | |
| TEST_CUSTOMER_ID = 16425 # From the attached image | |
| def print_separator(title=""): | |
| print("=" * 60) | |
| if title: | |
| print(f" {title}") | |
| print("=" * 60) | |
| def test_customer_notes_sequence_and_date(): | |
| """ | |
| Test that validates: | |
| 1. NoteID sequence generation (1, 2, 3, 4... per customer) | |
| 2. Date comes from UI input | |
| 3. CustomerTypeID defaults to 1 | |
| """ | |
| print_separator("Customer Notes Sequence and Date Validation") | |
| # Get current notes to understand existing state | |
| print(f"π Getting existing notes for customer {TEST_CUSTOMER_ID}") | |
| response = requests.get(f"{BASE_URL}/api/v1/notes/customers/{TEST_CUSTOMER_ID}") | |
| if response.status_code != 200: | |
| print(f"β Failed to get existing notes: {response.status_code} - {response.text}") | |
| return False | |
| existing_notes = response.json() | |
| print(f"β Customer {TEST_CUSTOMER_ID} currently has {len(existing_notes)} notes") | |
| if existing_notes: | |
| note_ids = [note.get('note_id') for note in existing_notes if note.get('note_id')] | |
| max_note_id = max(note_ids) if note_ids else 0 | |
| print(f"π Existing NoteIDs: {sorted(note_ids)} (max: {max_note_id})") | |
| else: | |
| max_note_id = 0 | |
| print("π No existing notes found") | |
| # Test data with specific dates from UI | |
| test_notes_data = [ | |
| { | |
| "notes": "Test note 1 - entering customer notes on 23rd Nov 25 845pm", | |
| "employee_id": "1", | |
| "customer_type_id": 1, | |
| "date": "2025-11-23T20:45:00" # Specific date from UI | |
| }, | |
| { | |
| "notes": "Test note 2 - second note entry", | |
| "employee_id": "2", | |
| "customer_type_id": 1, | |
| "date": "2025-11-24T10:30:00" # Different date from UI | |
| }, | |
| { | |
| "notes": "Test note 3 - third note entering as on 25th Nov", | |
| "employee_id": "3", | |
| "customer_type_id": 1, | |
| "date": "2025-11-25T15:39:00" # Another specific date | |
| }, | |
| { | |
| "notes": "Test note 4 - fourth note as on 26th", | |
| "employee_id": "4", | |
| "customer_type_id": 1 | |
| # No date provided - should use current time | |
| } | |
| ] | |
| created_note_ids = [] | |
| expected_sequence_start = max_note_id + 1 | |
| print(f"\nπ Creating {len(test_notes_data)} test notes...") | |
| # Create test notes | |
| for i, note_data in enumerate(test_notes_data, 1): | |
| print(f"\nπ Creating test note {i}:") | |
| print(f" Content: {note_data['notes'][:50]}...") | |
| print(f" Employee ID: {note_data['employee_id']}") | |
| print(f" Date: {note_data.get('date', 'Current time')}") | |
| response = requests.post( | |
| f"{BASE_URL}/api/v1/notes/customers/{TEST_CUSTOMER_ID}", | |
| json=note_data, | |
| headers={"Content-Type": "application/json"} | |
| ) | |
| if response.status_code == 201: | |
| result = response.json() | |
| created_note_ids.append(result['id']) | |
| print(f" β Created successfully with NoteID2: {result['id']}") | |
| else: | |
| print(f" β Failed to create: {response.status_code} - {response.text}") | |
| return False | |
| # Retrieve and validate created notes | |
| print(f"\nπ Retrieving and validating created notes...") | |
| response = requests.get(f"{BASE_URL}/api/v1/notes/customers/{TEST_CUSTOMER_ID}") | |
| if response.status_code != 200: | |
| print(f"β Failed to retrieve notes: {response.status_code}") | |
| return False | |
| all_notes = response.json() | |
| our_notes = [note for note in all_notes if note['note_id2'] in created_note_ids] | |
| our_notes.sort(key=lambda x: x['note_id2']) # Sort by creation order | |
| print(f"β Found {len(our_notes)} of our created notes") | |
| # Validate each note | |
| validation_results = [] | |
| for i, note in enumerate(our_notes): | |
| print(f"\nπ Validating Note {i+1} (NoteID2: {note['note_id2']}):") | |
| print(f" NoteID: {note.get('note_id', 'NULL')}") | |
| print(f" Content: {note['notes']}") | |
| print(f" Employee ID: {note['employee_id']}") | |
| print(f" Customer Type ID: {note['customer_type_id']}") | |
| print(f" Date: {note['date']}") | |
| # Check NoteID sequence | |
| expected_note_id = expected_sequence_start + i | |
| actual_note_id = note.get('note_id') | |
| if actual_note_id == expected_note_id: | |
| print(f" β NoteID sequence correct: {actual_note_id}") | |
| validation_results.append(True) | |
| else: | |
| print(f" β NoteID sequence wrong: expected {expected_note_id}, got {actual_note_id}") | |
| validation_results.append(False) | |
| # Check CustomerTypeID | |
| if note['customer_type_id'] == 1: | |
| print(f" β CustomerTypeID correct: 1") | |
| validation_results.append(True) | |
| else: | |
| print(f" β CustomerTypeID wrong: expected 1, got {note['customer_type_id']}") | |
| validation_results.append(False) | |
| # Check date handling (for notes with specific dates) | |
| if i < len(test_notes_data) and 'date' in test_notes_data[i]: | |
| expected_date = test_notes_data[i]['date'] | |
| note_date = note['date'] | |
| # Convert to comparable format (just date part) | |
| expected_dt = datetime.fromisoformat(expected_date.replace('T', ' ')) | |
| note_dt = datetime.fromisoformat(note_date.replace('T', ' ')) | |
| # Check if dates match (within reasonable tolerance) | |
| time_diff = abs((note_dt - expected_dt).total_seconds()) | |
| if time_diff < 60: # Within 1 minute tolerance | |
| print(f" β Date from UI correctly used") | |
| validation_results.append(True) | |
| else: | |
| print(f" β Date not from UI: expected ~{expected_date}, got {note_date}") | |
| validation_results.append(False) | |
| else: | |
| print(f" β Uses current time (no date specified)") | |
| validation_results.append(True) | |
| # Test update functionality | |
| if our_notes: | |
| print(f"\nπ Testing note update functionality...") | |
| note_to_update = our_notes[0] | |
| update_data = { | |
| "notes": "Updated test note content", | |
| "employee_id": "99", | |
| "customer_type_id": 1 | |
| } | |
| response = requests.put( | |
| f"{BASE_URL}/api/v1/notes/customers/{TEST_CUSTOMER_ID}/{note_to_update['note_id2']}", | |
| json=update_data, | |
| headers={"Content-Type": "application/json"} | |
| ) | |
| if response.status_code == 200: | |
| updated_note = response.json() | |
| print(f"β Note updated successfully") | |
| print(f" New content: {updated_note['notes']}") | |
| print(f" New employee ID: {updated_note['employee_id']}") | |
| validation_results.append(True) | |
| else: | |
| print(f"β Failed to update note: {response.status_code} - {response.text}") | |
| validation_results.append(False) | |
| # Clean up created notes | |
| print(f"\nπ§Ή Cleaning up test notes...") | |
| for note_id in created_note_ids: | |
| response = requests.delete(f"{BASE_URL}/api/v1/notes/customers/{TEST_CUSTOMER_ID}/{note_id}") | |
| if response.status_code == 204: | |
| print(f"β Deleted note {note_id}") | |
| else: | |
| print(f"β οΈ Failed to delete note {note_id}: {response.status_code}") | |
| success_rate = sum(validation_results) / len(validation_results) * 100 if validation_results else 0 | |
| success = success_rate >= 90 # 90% pass rate | |
| print(f"\nπ Validation Results:") | |
| print(f" Total checks: {len(validation_results)}") | |
| print(f" Passed: {sum(validation_results)}") | |
| print(f" Failed: {len(validation_results) - sum(validation_results)}") | |
| print(f" Success rate: {success_rate:.1f}%") | |
| return success | |
| def test_api_endpoints(): | |
| """Test that all API endpoints are working""" | |
| print_separator("API Endpoints Test") | |
| endpoints = [ | |
| ("GET", f"/api/v1/notes/customers/{TEST_CUSTOMER_ID}", "List customer notes"), | |
| ("GET", "/api/v1/notes/projects/1", "List project notes"), | |
| ("GET", "/api/v1/notes/bidders/1", "List bidder notes"), | |
| ] | |
| results = [] | |
| for method, endpoint, description in endpoints: | |
| print(f"π Testing {method} {endpoint} - {description}") | |
| if method == "GET": | |
| response = requests.get(f"{BASE_URL}{endpoint}") | |
| if response.status_code in [200, 404]: # 404 is OK if no data exists | |
| print(f" β {response.status_code} - OK") | |
| results.append(True) | |
| else: | |
| print(f" β {response.status_code} - {response.text}") | |
| results.append(False) | |
| return all(results) | |
| def main(): | |
| """Run all validation tests""" | |
| print_separator("Notes Implementation Validation") | |
| print("π Starting comprehensive notes validation...") | |
| print(f"π Testing against: {BASE_URL}") | |
| print(f"π€ Test Customer ID: {TEST_CUSTOMER_ID}") | |
| try: | |
| # Test API availability | |
| print(f"\nπ Checking API availability...") | |
| response = requests.get(f"{BASE_URL}/docs") | |
| if response.status_code == 200: | |
| print("β API is running") | |
| else: | |
| print("β API not accessible") | |
| return False | |
| # Run tests | |
| test1_result = test_api_endpoints() | |
| test2_result = test_customer_notes_sequence_and_date() | |
| print_separator("FINAL RESULTS") | |
| results = { | |
| "API Endpoints": test1_result, | |
| "Notes Implementation": test2_result | |
| } | |
| for test_name, result in results.items(): | |
| status = "β PASS" if result else "β FAIL" | |
| print(f"{test_name}: {status}") | |
| overall_success = all(results.values()) | |
| print(f"\nπ― Overall Result: {'β ALL TESTS PASSED' if overall_success else 'β SOME TESTS FAILED'}") | |
| if overall_success: | |
| print("\nπ Notes implementation is working correctly!") | |
| print("β NoteID sequence generation implemented") | |
| print("β Date parameter from UI supported") | |
| print("β CustomerTypeID defaults to 1") | |
| print("β All CRUD operations functional") | |
| else: | |
| print("\nβ οΈ Some issues found - check the test output above") | |
| return overall_success | |
| except requests.exceptions.ConnectionError: | |
| print(f"β Cannot connect to API at {BASE_URL}") | |
| print("π‘ Make sure your API server is running") | |
| return False | |
| except Exception as e: | |
| print(f"β Unexpected error: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False | |
| if __name__ == "__main__": | |
| success = main() | |
| sys.exit(0 if success else 1) |