ab-ms-core / tests /unit /test_notes_validation.py
PupaClic
Add comprehensive unit tests for various endpoints and features
7ff20d3
#!/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)