From d069b6459a79b22a4babeab7e5d1de1d76e3b8fa Mon Sep 17 00:00:00 2001 From: Rashed Alothman Date: Mon, 22 Dec 2025 09:33:40 +0300 Subject: [PATCH] feat: add task filtering, search, priority, and statistics --- Project_testing_files/Test.py | 320 +++++++++++++++++++++------------- app.py | 89 ++++++++-- 2 files changed, 268 insertions(+), 141 deletions(-) diff --git a/Project_testing_files/Test.py b/Project_testing_files/Test.py index e53aa439..46912a32 100644 --- a/Project_testing_files/Test.py +++ b/Project_testing_files/Test.py @@ -3,7 +3,7 @@ BASE_URL = 'http://localhost:5000/homepage/api/tasks' -# Color codes for terminal output (optional, works on most terminals) +# Color codes for terminal output GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' @@ -31,36 +31,55 @@ def print_info(message): print(f"\n{BLUE}{'='*70}") print("TMS (Task Management System) - Complete API Test Suite") +print("Testing CRUD + Filtering + Sorting + Priority") print(f"{'='*70}{RESET}\n") # ============================================================================ -# TEST 1: Add Multiple Tasks +# PART 1: BASIC CRUD TESTS # ============================================================================ -print_test_header(1, "Creating Multiple Tasks") + +print_test_header(1, "Creating Tasks with Different Priorities") +priorities = ['low', 'medium', 'high', 'urgent'] try: - for i in range(1, 4): + for priority in priorities: response = requests.post( f'{BASE_URL}/add_Tasks', - json={'description': f'Task {i}'} + json={'description': f'Task with {priority} priority', 'priority': priority} ) if response.status_code == 201: result = response.json() task_id = result['task']['id'] task_ids.append(task_id) - print_success(f"Created task with ID: {task_id}") - print_info(f"Description: {result['task']['description']}") + print_success(f"Created {priority} priority task (ID: {task_id})") passed += 1 else: - print_error(f"Failed to create task {i}: {response.status_code}") + print_error(f"Failed to create {priority} task: {response.status_code}") failed += 1 except Exception as e: print_error(f"Exception occurred: {e}") failed += 1 -# ============================================================================ -# TEST 2: Verify All Tasks Created -# ============================================================================ -print_test_header(2, "Retrieving All Tasks") +print_test_header(2, "Creating Tasks Without Priority (Should Default to 'low')") +try: + response = requests.post( + f'{BASE_URL}/add_Tasks', + json={'description': 'Task without priority specified'} + ) + if response.status_code == 201: + result = response.json() + task_id = result['task']['id'] + task_ids.append(task_id) + print_success(f"Created task with default priority") + print_info(f"Priority: {result['task']['priority']}") + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +print_test_header(3, "Retrieving All Tasks") try: response = requests.get(f'{BASE_URL}') if response.status_code == 200: @@ -68,7 +87,8 @@ def print_info(message): print_success(f"Retrieved {len(all_tasks)} tasks") for task in all_tasks: status = "✓ Done" if task.get('completed') else "○ Pending" - print_info(f"{status} ID: {task['id'][:8]}... | {task['description']}") + priority = task.get('priority', 'N/A') + print_info(f"{status} [{priority}] {task['description'][:40]}") passed += 1 else: print_error(f"Failed to retrieve tasks: {response.status_code}") @@ -78,9 +98,109 @@ def print_info(message): failed += 1 # ============================================================================ -# TEST 3: Update Task Description +# PART 2: FILTERING TESTS # ============================================================================ -print_test_header(3, "Updating Task Description") + +print_test_header(4, "Filtering: Get Only Pending Tasks") +try: + response = requests.get(f'{BASE_URL}?completed=false') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + all_pending = all(not task['completed'] for task in tasks) + if all_pending: + print_success(f"Retrieved {len(tasks)} pending tasks (all correct)") + else: + print_error("Some completed tasks were included") + failed += 1 + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +print_test_header(5, "Filtering: Get High Priority Tasks") +try: + response = requests.get(f'{BASE_URL}?priority=high') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + all_high = all(task['priority'] == 'high' for task in tasks) + if all_high: + print_success(f"Retrieved {len(tasks)} high priority tasks (all correct)") + else: + print_error("Some non-high priority tasks were included") + failed += 1 + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +print_test_header(6, "Filtering: Combine Filters (Pending + Urgent)") +try: + response = requests.get(f'{BASE_URL}?completed=false&priority=urgent') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + correct = all(not task['completed'] and task['priority'] == 'urgent' for task in tasks) + if correct: + print_success(f"Retrieved {len(tasks)} pending urgent tasks (correct)") + else: + print_error("Filter combination didn't work correctly") + failed += 1 + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +# ============================================================================ +# PART 3: SORTING TESTS +# ============================================================================ + +print_test_header(7, "Sorting: By Creation Date (Newest First)") +try: + response = requests.get(f'{BASE_URL}?sort=created_at') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + print_success(f"Retrieved {len(tasks)} tasks sorted by creation date") + print_info("First 3 tasks (newest first):") + for task in tasks[:3]: + print_info(f" - {task['description'][:40]} (created: {task['created_at']})") + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +print_test_header(8, "Sorting: By Priority (Urgent > High > Medium > Low)") +try: + response = requests.get(f'{BASE_URL}?sort=priority') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + print_success(f"Retrieved {len(tasks)} tasks sorted by priority") + print_info("Priority order:") + for task in tasks: + print_info(f" - [{task['priority']}] {task['description'][:40]}") + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + +# ============================================================================ +# PART 4: UPDATE AND DELETE TESTS +# ============================================================================ + +print_test_header(9, "Update: Change Task Description") if task_ids: try: update_id = task_ids[0] @@ -97,7 +217,7 @@ def print_info(message): print_info(f"New description: {result['task']['description']}") passed += 1 else: - print_error(f"Update failed: {response.status_code} - {response.json()}") + print_error(f"Update failed: {response.status_code}") failed += 1 except Exception as e: print_error(f"Exception occurred: {e}") @@ -106,10 +226,7 @@ def print_info(message): print_error("No tasks available to update") failed += 1 -# ============================================================================ -# TEST 4: Mark Task as Completed -# ============================================================================ -print_test_header(4, "Marking Task as Completed") +print_test_header(10, "Update: Mark Task as Completed") if len(task_ids) > 1: try: complete_id = task_ids[1] @@ -135,61 +252,7 @@ def print_info(message): print_error("Not enough tasks to test completion") failed += 1 -# ============================================================================ -# TEST 5: Update Both Description and Status -# ============================================================================ -print_test_header(5, "Updating Description AND Completion Status") -if len(task_ids) > 2: - try: - update_id = task_ids[2] - response = requests.patch( - f'{BASE_URL}/updated_task', - json={ - 'id': update_id, - 'description': 'Completely updated task', - 'completed': True - } - ) - if response.status_code == 200: - result = response.json() - print_success(f"Updated task {update_id[:8]}... fully") - print_info(f"Description: {result['task']['description']}") - print_info(f"Completed: {result['task']['completed']}") - passed += 1 - else: - print_error(f"Failed: {response.status_code}") - failed += 1 - except Exception as e: - print_error(f"Exception occurred: {e}") - failed += 1 -else: - print_error("Not enough tasks for this test") - failed += 1 - -# ============================================================================ -# TEST 6: View Updated Tasks -# ============================================================================ -print_test_header(6, "Verifying Updates") -try: - response = requests.get(f'{BASE_URL}') - if response.status_code == 200: - all_tasks = response.json().get('Tasks', []) - print_success(f"Retrieved {len(all_tasks)} tasks after updates") - for task in all_tasks: - status = "✓ Done" if task.get('completed') else "○ Pending" - print_info(f"{status} ID: {task['id'][:8]}... | {task['description']}") - passed += 1 - else: - print_error(f"Failed to retrieve: {response.status_code}") - failed += 1 -except Exception as e: - print_error(f"Exception occurred: {e}") - failed += 1 - -# ============================================================================ -# TEST 7: Delete a Task -# ============================================================================ -print_test_header(7, "Deleting a Task") +print_test_header(11, "Delete: Remove a Task") if task_ids: try: delete_id = task_ids[0] @@ -212,39 +275,17 @@ def print_info(message): failed += 1 # ============================================================================ -# TEST 8: Try to Update Non-Existent Task (Should Fail) +# PART 5: ERROR HANDLING TESTS # ============================================================================ -print_test_header(8, "Error Handling: Update Non-Existent Task") -try: - response = requests.patch( - f'{BASE_URL}/updated_task', - json={ - 'id': 'nonexistent-uuid-1234', - 'description': 'This should fail' - } - ) - if response.status_code == 404: - print_success("Correctly returned 404 for non-existent task") - print_info(f"Response: {response.json()}") - passed += 1 - else: - print_error(f"Wrong status code: {response.status_code} (expected 404)") - failed += 1 -except Exception as e: - print_error(f"Exception occurred: {e}") - failed += 1 -# ============================================================================ -# TEST 9: Try to Update Without ID (Should Fail) -# ============================================================================ -print_test_header(9, "Error Handling: Update Without ID") +print_test_header(12, "Error: Add Task with Invalid Priority") try: - response = requests.patch( - f'{BASE_URL}/updated_task', - json={'description': 'No ID provided'} + response = requests.post( + f'{BASE_URL}/add_Tasks', + json={'description': 'Task with bad priority', 'priority': 'super-mega-ultra'} ) if response.status_code == 400: - print_success("Correctly returned 400 for missing ID") + print_success("Correctly rejected invalid priority") print_info(f"Response: {response.json()}") passed += 1 else: @@ -254,14 +295,14 @@ def print_info(message): print_error(f"Exception occurred: {e}") failed += 1 -# ============================================================================ -# TEST 10: Try to Delete Non-Existent Task (Should Fail) -# ============================================================================ -print_test_header(10, "Error Handling: Delete Non-Existent Task") +print_test_header(13, "Error: Update Non-Existent Task") try: - response = requests.delete( - f'{BASE_URL}/delete_task', - json={'id': 'fake-uuid-9999'} + response = requests.patch( + f'{BASE_URL}/updated_task', + json={ + 'id': 'fake-uuid-9999', + 'description': 'This should fail' + } ) if response.status_code == 404: print_success("Correctly returned 404 for non-existent task") @@ -274,10 +315,7 @@ def print_info(message): print_error(f"Exception occurred: {e}") failed += 1 -# ============================================================================ -# TEST 11: Try to Delete Without ID (Should Fail) -# ============================================================================ -print_test_header(11, "Error Handling: Delete Without ID") +print_test_header(14, "Error: Delete Without ID") try: response = requests.delete( f'{BASE_URL}/delete_task', @@ -294,14 +332,11 @@ def print_info(message): print_error(f"Exception occurred: {e}") failed += 1 -# ============================================================================ -# TEST 12: Try to Add Task Without Description (Should Fail) -# ============================================================================ -print_test_header(12, "Error Handling: Add Task Without Description") +print_test_header(15, "Error: Add Task Without Description") try: response = requests.post( f'{BASE_URL}/add_Tasks', - json={'title': 'Wrong field'} + json={'priority': 'high'} # Missing description ) if response.status_code == 400: print_success("Correctly returned 400 for missing description") @@ -314,6 +349,26 @@ def print_info(message): print_error(f"Exception occurred: {e}") failed += 1 +# ============================================================================ +# PART 6: ADVANCED FILTER COMBINATIONS +# ============================================================================ + +print_test_header(16, "Advanced: Filter Completed + Sort by Priority") +try: + response = requests.get(f'{BASE_URL}?completed=true&sort=priority') + if response.status_code == 200: + tasks = response.json().get('Tasks', []) + print_success(f"Retrieved {len(tasks)} completed tasks sorted by priority") + for task in tasks[:5]: + print_info(f" [{task['priority']}] {task['description'][:40]}") + passed += 1 + else: + print_error(f"Failed: {response.status_code}") + failed += 1 +except Exception as e: + print_error(f"Exception occurred: {e}") + failed += 1 + # ============================================================================ # FINAL STATE # ============================================================================ @@ -324,9 +379,24 @@ def print_info(message): remaining = response.json().get('Tasks', []) if remaining: print_info(f"Total remaining: {len(remaining)} tasks") + + # Group by priority + by_priority = {} for task in remaining: - status = "✓ Done" if task.get('completed') else "○ Pending" - print_info(f"{status} ID: {task['id'][:8]}... | {task['description']}") + priority = task.get('priority', 'unknown') + by_priority[priority] = by_priority.get(priority, 0) + 1 + + print_info("Breakdown by priority:") + for priority in ['urgent', 'high', 'medium', 'low']: + count = by_priority.get(priority, 0) + if count > 0: + print_info(f" - {priority.capitalize()}: {count} tasks") + + # Show completion stats + completed_count = sum(1 for task in remaining if task['completed']) + print_info(f"\nCompletion status:") + print_info(f" - Completed: {completed_count}") + print_info(f" - Pending: {len(remaining) - completed_count}") else: print_info("No tasks remaining in system") else: @@ -346,7 +416,13 @@ def print_info(message): if failed == 0: print(f"\n{GREEN}{'='*70}") - print(" ALL TESTS PASSED! Your API is working perfectly!") + print("ALL TESTS PASSED!") + print("Your TMS API is working perfectly with all new features!") + print("- CRUD operations") + print("- Priority system") + print("- Filtering by completion and priority") + print("- Sorting by date and priority") + print("- Comprehensive error handling") print(f"{'='*70}{RESET}\n") sys.exit(0) else: diff --git a/app.py b/app.py index 46a8fcaf..d1225bc7 100644 --- a/app.py +++ b/app.py @@ -26,6 +26,8 @@ answer_for_data_not_found = 'Invalid or missing data' +error_massage_for_try_except_Exception_in_jsonify_fromat='An unexpected error occurred' +error_massage_for_database = 'Database error occurred' # Initialize Flask app app = Flask(__name__, template_folder='templates', static_folder='static') # Database configuration @@ -46,7 +48,9 @@ class Task(db.Model): description: Mapped[str] = mapped_column(String(255), nullable=False) due_date: Mapped[Optional[datetime]] = mapped_column(db.DateTime, nullable=True) completed: Mapped[bool] = mapped_column(db.Boolean, default=False) + priority: Mapped[str] = mapped_column(String(10),default="low") created_at: Mapped[datetime] = mapped_column(db.DateTime, default=datetime.utcnow) + # Representation method for debugging def __repr__(self) -> str: @@ -60,6 +64,7 @@ def to_dict(self): 'description': self.description, 'Due Date': self.due_date.strftime('%Y-%m-%d %H:%M:%S') if self.due_date else None, 'completed': self.completed, + 'priority': self.priority, 'created_at': self.created_at.strftime('%Y-%m-%d %H:%M:%S')} @app.route('/') @@ -84,26 +89,72 @@ def about(): @app.route('/homepage/api/tasks', methods=['GET']) def get_tasks(): - all_tasks = Task.query.all() - return jsonify({'Tasks': [task.to_dict() for task in all_tasks]}) + try: + completed_param = request.args.get('completed') + + priority_param = request.args.get('priority') + + sort_by = request.args.get('sort','created_at') + + query = Task.query + + if completed_param is not None: + completed_bool = completed_param.lower() == 'true' + query = query.filter_by(completed=completed_bool) + + + if priority_param : + query = query.filter_by(priority=priority_param) + + if sort_by == 'created_at': + query = query.order_by(Task.created_at.desc()) + elif sort_by == 'due_date': + query = query.order_by(Task.due_date.asc().nullslast()) + elif sort_by == 'priority': + priority_order = ['urgent', 'high', 'medium', 'low'] + query = query.order_by(db.case({p: i for i, p in enumerate(priority_order)}, value=Task.priority)) + + tasks = query.all() + return jsonify({'Tasks': [task.to_dict() for task in tasks]}) + except SQLAlchemyError as e: + logger.error(f"Database error while Looking and sorting the tasks: {str(e)}") + return jsonify({'error': error_massage_for_database}), 500 + except Exception as e: + logger.error(f'there was error in showing Tasks: {str(e)}') + return jsonify({'error': error_massage_for_try_except_Exception_in_jsonify_fromat}), 500 @app.route('/homepage/api/tasks/add_Tasks',methods=['POST']) def add_task_api(): - - data = request.get_json() - - if not data or 'description' not in data: - return jsonify({'error': answer_for_data_not_found}), 400 - - description = data.get('description') - - new_task = Task( - description = description, - due_date = datetime.now() - ) - db.session.add(new_task) - db.session.commit() - return jsonify({'message':'task added','task':new_task.to_dict()}),201 + try: + data = request.get_json() + + if not data or 'description' not in data: + return jsonify({'error': answer_for_data_not_found}), 400 + + description = data.get('description') + + priority = data.get('priority', 'low').lower() + + valid_priorities = ['low', 'medium', 'high', 'urgent'] + + if priority not in valid_priorities: + return jsonify({'error': f'Priority must be one of: {valid_priorities}'}), 400 + + new_task = Task( + description = description, + due_date = datetime.now(), + priority = priority + ) + db.session.add(new_task) + db.session.commit() + return jsonify({'message':'task added','task':new_task.to_dict()}),201 + except SQLAlchemyError as e: + db.session.rollback() + logger.error(f"Database error while adding task: {str(e)}") + return jsonify({'error': error_massage_for_database}), 500 + except Exception as e: + logger.error(f"Unexpected error while adding task: {str(e)}") + return jsonify({'error': error_massage_for_try_except_Exception_in_jsonify_fromat}), 500 @app.route('/homepage/api/tasks/delete_task',methods=['DELETE']) def delete_task(): @@ -131,10 +182,10 @@ def delete_task(): except SQLAlchemyError as e: db.session.rollback() logger.error(f"Database error while deleting task: {str(e)}") - return jsonify({'error': 'Database error occurred'}), 500 + return jsonify({'error': error_massage_for_database}), 500 except Exception as e: logger.error(f"Unexpected error while deleting task: {str(e)}") - return jsonify({'error': 'An unexpected error occurred'}), 500 + return jsonify({'error': error_massage_for_try_except_Exception_in_jsonify_fromat}), 500 @app.route('/homepage/api/tasks/updated_task',methods=["PATCH"])