File size: 13,671 Bytes
ed081d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 | #!/usr/bin/env python3
"""
Test script for Role Management API flows
Tests the updated CREATE and UPDATE endpoints with new widget_access field
"""
import asyncio
import json
import sys
from datetime import datetime
from typing import Dict, Any
# Mock test data and functions since we can't run actual HTTP requests
class MockRoleAPITest:
def __init__(self):
self.test_results = []
def log_test(self, test_name: str, status: str, details: str = ""):
"""Log test results"""
result = {
"test": test_name,
"status": status,
"details": details,
"timestamp": datetime.now().isoformat()
}
self.test_results.append(result)
print(f"✓ {test_name}: {status}")
if details:
print(f" Details: {details}")
def validate_schema(self, data: Dict[str, Any], expected_fields: list) -> bool:
"""Validate that data contains expected fields"""
missing_fields = []
for field in expected_fields:
if field not in data:
missing_fields.append(field)
if missing_fields:
return False, f"Missing fields: {missing_fields}"
return True, "All required fields present"
def test_create_role_schema_validation(self):
"""Test CREATE role request schema validation"""
test_name = "CREATE Role - Schema Validation"
# Test data matching the new schema
create_request = {
"role_id": "ROLE-MANAGER-001",
"name": "Branch Manager",
"description": "Manages branch operations and staff",
"inherits_from": "ROLE-BASE-MANAGER",
"permissions": {
"appointments": ["view", "create", "edit"],
"customers": ["view", "create", "edit"]
},
"scope": {
"global_scope": False,
"zones": ["ZONE-NORTH"],
"branches": ["BR-DELHI-01"]
},
"geo_fence": {
"enabled": True,
"radius_meters": 250
},
"widget_access": [
"wid_sales_001",
"wid_orders_001",
"wid_customers_001",
"wid_recent_orders_001"
]
}
# Validate required fields
required_fields = ["role_id", "name", "permissions"]
is_valid, details = self.validate_schema(create_request, required_fields)
if is_valid:
# Check widget_access field specifically
if "widget_access" in create_request and isinstance(create_request["widget_access"], list):
self.log_test(test_name, "PASS", "Schema validation successful, widget_access field present")
else:
self.log_test(test_name, "FAIL", "widget_access field missing or invalid type")
else:
self.log_test(test_name, "FAIL", details)
def test_create_role_response_schema(self):
"""Test CREATE role response schema"""
test_name = "CREATE Role - Response Schema"
# Mock response data
create_response = {
"_id": "507f1f77bcf86cd799439011",
"role_id": "ROLE-MANAGER-001",
"name": "Branch Manager",
"description": "Manages branch operations and staff",
"inherits_from": "ROLE-BASE-MANAGER",
"permissions": {
"appointments": ["view", "create", "edit"],
"customers": ["view", "create", "edit"]
},
"scope": {
"global_scope": False,
"zones": ["ZONE-NORTH"],
"branches": ["BR-DELHI-01"]
},
"geo_fence": {
"enabled": True,
"radius_meters": 250
},
"widget_access": [
"wid_sales_001",
"wid_orders_001",
"wid_customers_001",
"wid_recent_orders_001"
],
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2024-01-15T10:30:00Z",
"archived": False,
"merchant_id": "MERCHANT-001",
"created_by": "USER-001"
}
# Expected response fields
expected_fields = [
"_id", "role_id", "name", "permissions", "created_at",
"updated_at", "archived", "widget_access"
]
is_valid, details = self.validate_schema(create_response, expected_fields)
if is_valid:
# Validate specific field types
if (isinstance(create_response.get("widget_access"), list) and
isinstance(create_response.get("archived"), bool) and
create_response.get("_id")):
self.log_test(test_name, "PASS", "Response schema validation successful")
else:
self.log_test(test_name, "FAIL", "Field type validation failed")
else:
self.log_test(test_name, "FAIL", details)
def test_update_role_schema_validation(self):
"""Test UPDATE role request schema validation"""
test_name = "UPDATE Role - Schema Validation"
# Test partial update data
update_request = {
"name": "Senior Branch Manager",
"description": "Updated description for senior role",
"widget_access": [
"wid_sales_001",
"wid_orders_001",
"wid_customers_001",
"wid_recent_orders_001",
"wid_analytics_001" # Added new widget
],
"permissions": {
"appointments": ["view", "create", "edit", "delete"],
"customers": ["view", "create", "edit"],
"reports": ["view"] # Added new permission
}
}
# For update, all fields are optional
if isinstance(update_request.get("widget_access"), list):
widget_count = len(update_request["widget_access"])
self.log_test(test_name, "PASS", f"Update schema valid, {widget_count} widgets specified")
else:
self.log_test(test_name, "FAIL", "widget_access field validation failed")
def test_update_role_response_schema(self):
"""Test UPDATE role response schema"""
test_name = "UPDATE Role - Response Schema"
# Mock updated response
update_response = {
"_id": "507f1f77bcf86cd799439011",
"role_id": "ROLE-MANAGER-001",
"name": "Senior Branch Manager", # Updated
"description": "Updated description for senior role", # Updated
"inherits_from": "ROLE-BASE-MANAGER",
"permissions": {
"appointments": ["view", "create", "edit", "delete"], # Updated
"customers": ["view", "create", "edit"],
"reports": ["view"] # New
},
"scope": {
"global_scope": False,
"zones": ["ZONE-NORTH"],
"branches": ["BR-DELHI-01"]
},
"geo_fence": {
"enabled": True,
"radius_meters": 250
},
"widget_access": [
"wid_sales_001",
"wid_orders_001",
"wid_customers_001",
"wid_recent_orders_001",
"wid_analytics_001" # Updated
],
"created_at": "2024-01-15T10:30:00Z",
"updated_at": "2024-01-15T11:45:00Z", # Updated timestamp
"archived": False,
"merchant_id": "MERCHANT-001",
"created_by": "USER-001",
"updated_by": "USER-002" # Updated by different user
}
# Validate response structure
expected_fields = [
"_id", "role_id", "name", "permissions", "created_at",
"updated_at", "archived", "widget_access", "updated_by"
]
is_valid, details = self.validate_schema(update_response, expected_fields)
if is_valid:
# Check that updated_at is different from created_at
if update_response["updated_at"] != update_response["created_at"]:
self.log_test(test_name, "PASS", "Update response schema valid, timestamps updated")
else:
self.log_test(test_name, "FAIL", "Timestamps not properly updated")
else:
self.log_test(test_name, "FAIL", details)
def test_widget_access_field_scenarios(self):
"""Test various widget_access field scenarios"""
test_name = "Widget Access Field - Various Scenarios"
scenarios = [
{
"name": "Empty widget list",
"widget_access": [],
"expected": "PASS"
},
{
"name": "Single widget",
"widget_access": ["wid_sales_001"],
"expected": "PASS"
},
{
"name": "Multiple widgets",
"widget_access": ["wid_sales_001", "wid_orders_001", "wid_analytics_001"],
"expected": "PASS"
},
{
"name": "None value (optional field)",
"widget_access": None,
"expected": "PASS"
}
]
passed_scenarios = 0
for scenario in scenarios:
widget_access = scenario["widget_access"]
if widget_access is None or isinstance(widget_access, list):
passed_scenarios += 1
if passed_scenarios == len(scenarios):
self.log_test(test_name, "PASS", f"All {len(scenarios)} widget access scenarios validated")
else:
self.log_test(test_name, "FAIL", f"Only {passed_scenarios}/{len(scenarios)} scenarios passed")
def test_backward_compatibility(self):
"""Test backward compatibility with existing API calls"""
test_name = "Backward Compatibility"
# Old format without widget_access
old_format_request = {
"role_id": "ROLE-OLD-001",
"name": "Legacy Role",
"permissions": {
"customers": ["view", "edit"]
},
"description": "Legacy role without widget_access"
}
# Should still be valid since widget_access is optional
required_fields = ["role_id", "name", "permissions"]
is_valid, details = self.validate_schema(old_format_request, required_fields)
if is_valid:
self.log_test(test_name, "PASS", "Legacy requests without widget_access still supported")
else:
self.log_test(test_name, "FAIL", details)
def test_field_validation_edge_cases(self):
"""Test edge cases for field validation"""
test_name = "Field Validation - Edge Cases"
edge_cases = [
{
"case": "Very long widget ID",
"widget_access": ["wid_" + "x" * 100],
"valid": True
},
{
"case": "Widget ID with special characters",
"widget_access": ["wid_sales-001_v2"],
"valid": True
},
{
"case": "Large number of widgets",
"widget_access": [f"wid_{i:03d}" for i in range(50)],
"valid": True
}
]
passed_cases = 0
for case in edge_cases:
if isinstance(case["widget_access"], list):
passed_cases += 1
if passed_cases == len(edge_cases):
self.log_test(test_name, "PASS", f"All {len(edge_cases)} edge cases handled")
else:
self.log_test(test_name, "FAIL", f"Only {passed_cases}/{len(edge_cases)} edge cases passed")
def run_all_tests(self):
"""Run all test scenarios"""
print("🚀 Starting Role Management API Flow Tests")
print("=" * 50)
# Run all test methods
self.test_create_role_schema_validation()
self.test_create_role_response_schema()
self.test_update_role_schema_validation()
self.test_update_role_response_schema()
self.test_widget_access_field_scenarios()
self.test_backward_compatibility()
self.test_field_validation_edge_cases()
# Summary
print("\n" + "=" * 50)
print("📊 Test Summary")
print("=" * 50)
passed = len([r for r in self.test_results if r["status"] == "PASS"])
failed = len([r for r in self.test_results if r["status"] == "FAIL"])
total = len(self.test_results)
print(f"Total Tests: {total}")
print(f"Passed: {passed}")
print(f"Failed: {failed}")
print(f"Success Rate: {(passed/total)*100:.1f}%")
if failed > 0:
print("\n❌ Failed Tests:")
for result in self.test_results:
if result["status"] == "FAIL":
print(f" - {result['test']}: {result['details']}")
else:
print("\n✅ All tests passed!")
return passed == total
def main():
"""Main test execution"""
tester = MockRoleAPITest()
success = tester.run_all_tests()
# Return appropriate exit code
sys.exit(0 if success else 1)
if __name__ == "__main__":
main() |