491 lines
13 KiB
Python
491 lines
13 KiB
Python
|
"""
|
||
|
|
Test script for Ticket Module
|
||
|
|
==============================
|
||
|
|
|
||
|
|
Tester database schema, funktioner, constraints og services.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import sys
|
||
|
|
from datetime import date, datetime
|
||
|
|
from decimal import Decimal
|
||
|
|
|
||
|
|
# Add parent directory to path
|
||
|
|
sys.path.insert(0, '/app')
|
||
|
|
|
||
|
|
# Initialize database pool before importing services
|
||
|
|
from app.core.database import init_db, execute_query, execute_insert, execute_update
|
||
|
|
from app.ticket.backend.ticket_service import TicketService
|
||
|
|
from app.ticket.backend.models import (
|
||
|
|
TTicketCreate,
|
||
|
|
TicketStatus,
|
||
|
|
TicketPriority,
|
||
|
|
TicketSource,
|
||
|
|
WorkType,
|
||
|
|
BillingMethod
|
||
|
|
)
|
||
|
|
|
||
|
|
def print_test(name: str, passed: bool, details: str = ""):
|
||
|
|
"""Print test result"""
|
||
|
|
emoji = "✅" if passed else "❌"
|
||
|
|
print(f"{emoji} {name}")
|
||
|
|
if details:
|
||
|
|
print(f" {details}")
|
||
|
|
print()
|
||
|
|
|
||
|
|
|
||
|
|
def test_tables_exist():
|
||
|
|
"""Test 1: Check all tables are created"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 1: Database Tables")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
tables = [
|
||
|
|
'tticket_metadata',
|
||
|
|
'tticket_tickets',
|
||
|
|
'tticket_comments',
|
||
|
|
'tticket_attachments',
|
||
|
|
'tticket_worklog',
|
||
|
|
'tticket_prepaid_cards',
|
||
|
|
'tticket_prepaid_transactions',
|
||
|
|
'tticket_email_log',
|
||
|
|
'tticket_audit_log'
|
||
|
|
]
|
||
|
|
|
||
|
|
for table in tables:
|
||
|
|
result = execute_query(
|
||
|
|
"""
|
||
|
|
SELECT EXISTS (
|
||
|
|
SELECT FROM information_schema.tables
|
||
|
|
WHERE table_name = %s
|
||
|
|
)
|
||
|
|
""",
|
||
|
|
(table,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
print_test(f"Table '{table}' exists", result['exists'])
|
||
|
|
|
||
|
|
# Check views
|
||
|
|
views = [
|
||
|
|
'tticket_open_tickets',
|
||
|
|
'tticket_billable_worklog',
|
||
|
|
'tticket_prepaid_balances',
|
||
|
|
'tticket_stats_by_status'
|
||
|
|
]
|
||
|
|
|
||
|
|
for view in views:
|
||
|
|
result = execute_query(
|
||
|
|
"""
|
||
|
|
SELECT EXISTS (
|
||
|
|
SELECT FROM information_schema.views
|
||
|
|
WHERE table_name = %s
|
||
|
|
)
|
||
|
|
""",
|
||
|
|
(view,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
print_test(f"View '{view}' exists", result['exists'])
|
||
|
|
|
||
|
|
|
||
|
|
def test_ticket_number_generation():
|
||
|
|
"""Test 2: Auto-generate ticket number"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 2: Ticket Number Generation")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Insert ticket without ticket_number (should auto-generate)
|
||
|
|
ticket_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_tickets (subject, description, status)
|
||
|
|
VALUES (%s, %s, %s)
|
||
|
|
""",
|
||
|
|
("Test Ticket", "Testing auto-generation", "open")
|
||
|
|
)
|
||
|
|
|
||
|
|
ticket = execute_query(
|
||
|
|
"SELECT ticket_number FROM tticket_tickets WHERE id = %s",
|
||
|
|
(ticket_id,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
# Check format: TKT-YYYYMMDD-XXX
|
||
|
|
import re
|
||
|
|
pattern = r'TKT-\d{8}-\d{3}'
|
||
|
|
matches = re.match(pattern, ticket['ticket_number'])
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Ticket number auto-generated",
|
||
|
|
matches is not None,
|
||
|
|
f"Generated: {ticket['ticket_number']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Test generating multiple tickets
|
||
|
|
ticket_id2 = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_tickets (subject, status)
|
||
|
|
VALUES (%s, %s)
|
||
|
|
""",
|
||
|
|
("Test Ticket 2", "open")
|
||
|
|
)
|
||
|
|
|
||
|
|
ticket2 = execute_query(
|
||
|
|
"SELECT ticket_number FROM tticket_tickets WHERE id = %s",
|
||
|
|
(ticket_id2,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Sequential ticket numbers",
|
||
|
|
ticket2['ticket_number'] > ticket['ticket_number'],
|
||
|
|
f"First: {ticket['ticket_number']}, Second: {ticket2['ticket_number']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_prepaid_card_constraints():
|
||
|
|
"""Test 3: Prepaid card constraints (1 active per customer)"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 3: Prepaid Card Constraints")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Create first active card for customer 1
|
||
|
|
card1_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_prepaid_cards
|
||
|
|
(customer_id, purchased_hours, price_per_hour, total_amount, status)
|
||
|
|
VALUES (%s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(1, Decimal('10.0'), Decimal('850.0'), Decimal('8500.0'), 'active')
|
||
|
|
)
|
||
|
|
|
||
|
|
card1 = execute_query(
|
||
|
|
"SELECT card_number, remaining_hours FROM tticket_prepaid_cards WHERE id = %s",
|
||
|
|
(card1_id,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"First prepaid card created",
|
||
|
|
card1 is not None,
|
||
|
|
f"Card: {card1['card_number']}, Balance: {card1['remaining_hours']} hours"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Try to create second active card for same customer (should fail due to UNIQUE constraint)
|
||
|
|
try:
|
||
|
|
card2_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_prepaid_cards
|
||
|
|
(customer_id, purchased_hours, price_per_hour, total_amount, status)
|
||
|
|
VALUES (%s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(1, Decimal('20.0'), Decimal('850.0'), Decimal('17000.0'), 'active')
|
||
|
|
)
|
||
|
|
print_test(
|
||
|
|
"Cannot create 2nd active card (UNIQUE constraint)",
|
||
|
|
False,
|
||
|
|
"ERROR: Constraint not enforced!"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
print_test(
|
||
|
|
"Cannot create 2nd active card (UNIQUE constraint)",
|
||
|
|
"unique" in str(e).lower() or "duplicate" in str(e).lower(),
|
||
|
|
f"Constraint enforced: {str(e)[:80]}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Create inactive card for same customer (should work)
|
||
|
|
card3_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_prepaid_cards
|
||
|
|
(customer_id, purchased_hours, price_per_hour, total_amount, status)
|
||
|
|
VALUES (%s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(1, Decimal('5.0'), Decimal('850.0'), Decimal('4250.0'), 'depleted')
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Can create inactive card for same customer",
|
||
|
|
card3_id is not None,
|
||
|
|
"Multiple inactive cards allowed"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_generated_column():
|
||
|
|
"""Test 4: Generated column (remaining_hours)"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 4: Generated Column (remaining_hours)")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Create card
|
||
|
|
card_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_prepaid_cards
|
||
|
|
(customer_id, purchased_hours, price_per_hour, total_amount, status)
|
||
|
|
VALUES (%s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(2, Decimal('20.0'), Decimal('850.0'), Decimal('17000.0'), 'active')
|
||
|
|
)
|
||
|
|
|
||
|
|
card = execute_query(
|
||
|
|
"SELECT purchased_hours, used_hours, remaining_hours FROM tticket_prepaid_cards WHERE id = %s",
|
||
|
|
(card_id,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Initial remaining_hours calculated",
|
||
|
|
card['remaining_hours'] == Decimal('20.0'),
|
||
|
|
f"Purchased: {card['purchased_hours']}, Used: {card['used_hours']}, Remaining: {card['remaining_hours']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Use some hours
|
||
|
|
execute_update(
|
||
|
|
"UPDATE tticket_prepaid_cards SET used_hours = %s WHERE id = %s",
|
||
|
|
(Decimal('5.5'), card_id)
|
||
|
|
)
|
||
|
|
|
||
|
|
card = execute_query(
|
||
|
|
"SELECT purchased_hours, used_hours, remaining_hours FROM tticket_prepaid_cards WHERE id = %s",
|
||
|
|
(card_id,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
expected = Decimal('14.5')
|
||
|
|
print_test(
|
||
|
|
"remaining_hours auto-updates",
|
||
|
|
card['remaining_hours'] == expected,
|
||
|
|
f"After using 5.5h: {card['remaining_hours']}h (expected: {expected}h)"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_ticket_service():
|
||
|
|
"""Test 5: Ticket Service business logic"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 5: Ticket Service")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Test create ticket
|
||
|
|
ticket_data = TTicketCreate(
|
||
|
|
subject="Test Service Ticket",
|
||
|
|
description="Testing TicketService",
|
||
|
|
priority=TicketPriority.HIGH,
|
||
|
|
customer_id=1,
|
||
|
|
source=TicketSource.MANUAL
|
||
|
|
)
|
||
|
|
|
||
|
|
ticket = TicketService.create_ticket(ticket_data, user_id=1)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"TicketService.create_ticket() works",
|
||
|
|
ticket is not None and 'id' in ticket,
|
||
|
|
f"Created ticket: {ticket['ticket_number']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Test status transition validation
|
||
|
|
is_valid, error = TicketService.validate_status_transition('open', 'in_progress')
|
||
|
|
print_test(
|
||
|
|
"Valid status transition (open → in_progress)",
|
||
|
|
is_valid and error is None,
|
||
|
|
"Transition allowed"
|
||
|
|
)
|
||
|
|
|
||
|
|
is_valid, error = TicketService.validate_status_transition('closed', 'open')
|
||
|
|
print_test(
|
||
|
|
"Invalid status transition (closed → open)",
|
||
|
|
not is_valid and error is not None,
|
||
|
|
f"Transition blocked: {error}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Test update status
|
||
|
|
try:
|
||
|
|
updated = TicketService.update_ticket_status(
|
||
|
|
ticket['id'],
|
||
|
|
'in_progress',
|
||
|
|
user_id=1,
|
||
|
|
note="Starting work"
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"TicketService.update_ticket_status() works",
|
||
|
|
updated['status'] == 'in_progress',
|
||
|
|
f"Status changed: {ticket['status']} → {updated['status']}"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
print_test(
|
||
|
|
"TicketService.update_ticket_status() works",
|
||
|
|
False,
|
||
|
|
f"ERROR: {e}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Test add comment
|
||
|
|
try:
|
||
|
|
comment = TicketService.add_comment(
|
||
|
|
ticket['id'],
|
||
|
|
"Test comment from service",
|
||
|
|
user_id=1,
|
||
|
|
is_internal=False
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"TicketService.add_comment() works",
|
||
|
|
comment is not None and 'id' in comment,
|
||
|
|
f"Comment added (ID: {comment['id']})"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
print_test(
|
||
|
|
"TicketService.add_comment() works",
|
||
|
|
False,
|
||
|
|
f"ERROR: {e}"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_worklog_creation():
|
||
|
|
"""Test 6: Worklog creation"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 6: Worklog")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Get a ticket
|
||
|
|
ticket = execute_query(
|
||
|
|
"SELECT id FROM tticket_tickets LIMIT 1",
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
if not ticket:
|
||
|
|
print_test("Worklog creation", False, "No tickets found to test worklog")
|
||
|
|
return
|
||
|
|
|
||
|
|
# Create worklog entry
|
||
|
|
worklog_id = execute_insert(
|
||
|
|
"""
|
||
|
|
INSERT INTO tticket_worklog
|
||
|
|
(ticket_id, work_date, hours, work_type, billing_method, status, user_id)
|
||
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(ticket['id'], date.today(), Decimal('2.5'), 'support', 'invoice', 'draft', 1)
|
||
|
|
)
|
||
|
|
|
||
|
|
worklog = execute_query(
|
||
|
|
"SELECT * FROM tticket_worklog WHERE id = %s",
|
||
|
|
(worklog_id,),
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Worklog entry created",
|
||
|
|
worklog is not None,
|
||
|
|
f"ID: {worklog_id}, Hours: {worklog['hours']}, Status: {worklog['status']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Check view
|
||
|
|
billable_count = execute_query(
|
||
|
|
"SELECT COUNT(*) as count FROM tticket_billable_worklog WHERE status = 'billable'",
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Billable worklog view accessible",
|
||
|
|
billable_count is not None,
|
||
|
|
f"Found {billable_count['count']} billable entries"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_audit_logging():
|
||
|
|
"""Test 7: Audit logging"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 7: Audit Logging")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Check if audit entries were created
|
||
|
|
audit_count = execute_query(
|
||
|
|
"SELECT COUNT(*) as count FROM tticket_audit_log",
|
||
|
|
fetchone=True
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"Audit log has entries",
|
||
|
|
audit_count['count'] > 0,
|
||
|
|
f"Found {audit_count['count']} audit entries"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Check recent audit entries
|
||
|
|
recent = execute_query(
|
||
|
|
"""
|
||
|
|
SELECT action, entity_type, created_at
|
||
|
|
FROM tticket_audit_log
|
||
|
|
ORDER BY created_at DESC
|
||
|
|
LIMIT 5
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
|
||
|
|
if recent:
|
||
|
|
print("Recent audit entries:")
|
||
|
|
for entry in recent:
|
||
|
|
print(f" - {entry['action']} on {entry['entity_type']} at {entry['created_at']}")
|
||
|
|
print()
|
||
|
|
|
||
|
|
|
||
|
|
def test_views():
|
||
|
|
"""Test 8: Database views"""
|
||
|
|
print("=" * 60)
|
||
|
|
print("TEST 8: Database Views")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# Test tticket_open_tickets view
|
||
|
|
open_tickets = execute_query(
|
||
|
|
"SELECT ticket_number, comment_count, age_hours FROM tticket_open_tickets LIMIT 5"
|
||
|
|
)
|
||
|
|
|
||
|
|
print_test(
|
||
|
|
"tticket_open_tickets view works",
|
||
|
|
open_tickets is not None,
|
||
|
|
f"Found {len(open_tickets)} open tickets"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Test tticket_stats_by_status view
|
||
|
|
stats = execute_query(
|
||
|
|
"SELECT status, ticket_count FROM tticket_stats_by_status"
|
||
|
|
)
|
||
|
|
|
||
|
|
if stats:
|
||
|
|
print("Ticket statistics by status:")
|
||
|
|
for stat in stats:
|
||
|
|
print(f" - {stat['status']}: {stat['ticket_count']} tickets")
|
||
|
|
print()
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
"""Run all tests"""
|
||
|
|
print("\n")
|
||
|
|
print("🎫 TICKET MODULE TEST SUITE")
|
||
|
|
print("=" * 60)
|
||
|
|
print()
|
||
|
|
|
||
|
|
# Initialize database pool
|
||
|
|
print("🔌 Initializing database connection...")
|
||
|
|
init_db()
|
||
|
|
print("✅ Database connected\n")
|
||
|
|
|
||
|
|
try:
|
||
|
|
test_tables_exist()
|
||
|
|
test_ticket_number_generation()
|
||
|
|
test_prepaid_card_constraints()
|
||
|
|
test_generated_column()
|
||
|
|
test_ticket_service()
|
||
|
|
test_worklog_creation()
|
||
|
|
test_audit_logging()
|
||
|
|
test_views()
|
||
|
|
|
||
|
|
print("=" * 60)
|
||
|
|
print("✅ ALL TESTS COMPLETED")
|
||
|
|
print("=" * 60)
|
||
|
|
print()
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"\n❌ TEST SUITE FAILED: {e}\n")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
sys.exit(1)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|