3806c7d011
- Added migration 025 for the Ticket System, creating tables for tickets, comments, attachments, worklogs, prepaid cards, and audit logs. - Introduced migration 026 to add ticket-related permissions to the auth system and assign them to user groups. - Developed a test suite for the Ticket Module, validating database schema, ticket number generation, prepaid card constraints, service logic, worklog creation, audit logging, and views.
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()
|