|
1 | 1 | from datetime import datetime |
2 | 2 |
|
3 | 3 | from api.controllers.devices import ( |
| 4 | + execute_bulk_operations, |
4 | 5 | execute_component_action, |
5 | 6 | get_component_actions, |
6 | 7 | get_device_config, |
@@ -320,3 +321,150 @@ async def execute(self, request): |
320 | 321 | assert data["ip"] == "192.168.1.100" |
321 | 322 | assert data["success"] is False |
322 | 323 | assert data["error"] == "Permission denied" |
| 324 | + |
| 325 | + def test_bulk_operations_update_successfully(self): |
| 326 | + from core.use_cases.bulk_operations import BulkOperationsUseCase |
| 327 | + |
| 328 | + class MockBulkOperationsUseCase(BulkOperationsUseCase): |
| 329 | + def __init__(self): |
| 330 | + pass # Skip parent constructor |
| 331 | + |
| 332 | + async def execute_bulk_update(self, device_ips, channel="stable"): |
| 333 | + return [ |
| 334 | + ActionResult( |
| 335 | + device_ip=ip, |
| 336 | + success=True, |
| 337 | + message="Update initiated", |
| 338 | + action_type="Update", |
| 339 | + ) |
| 340 | + for ip in device_ips |
| 341 | + ] |
| 342 | + |
| 343 | + with create_test_client( |
| 344 | + route_handlers=[execute_bulk_operations], |
| 345 | + dependencies={ |
| 346 | + "bulk_operations_use_case": Provide( |
| 347 | + lambda: MockBulkOperationsUseCase(), sync_to_thread=False |
| 348 | + ) |
| 349 | + }, |
| 350 | + ) as client: |
| 351 | + response = client.post( |
| 352 | + "/bulk", |
| 353 | + json={ |
| 354 | + "device_ips": ["192.168.1.100", "192.168.1.101"], |
| 355 | + "operation": "update", |
| 356 | + "channel": "beta", |
| 357 | + }, |
| 358 | + ) |
| 359 | + |
| 360 | + assert response.status_code == 200 |
| 361 | + data = response.json() |
| 362 | + assert len(data) == 2 |
| 363 | + assert all(result["success"] for result in data) |
| 364 | + assert all(result["operation"] == "update" for result in data) |
| 365 | + assert all(result["channel"] == "beta" for result in data) |
| 366 | + |
| 367 | + def test_bulk_operations_reboot_successfully(self): |
| 368 | + from core.use_cases.bulk_operations import BulkOperationsUseCase |
| 369 | + |
| 370 | + class MockBulkOperationsUseCase(BulkOperationsUseCase): |
| 371 | + def __init__(self): |
| 372 | + pass # Skip parent constructor |
| 373 | + |
| 374 | + async def execute_bulk_reboot(self, device_ips): |
| 375 | + return [ |
| 376 | + ActionResult( |
| 377 | + device_ip="192.168.1.100", |
| 378 | + success=True, |
| 379 | + message="Reboot initiated", |
| 380 | + action_type="Reboot", |
| 381 | + ) |
| 382 | + ] |
| 383 | + |
| 384 | + with create_test_client( |
| 385 | + route_handlers=[execute_bulk_operations], |
| 386 | + dependencies={ |
| 387 | + "bulk_operations_use_case": Provide( |
| 388 | + lambda: MockBulkOperationsUseCase(), sync_to_thread=False |
| 389 | + ) |
| 390 | + }, |
| 391 | + ) as client: |
| 392 | + response = client.post( |
| 393 | + "/bulk", json={"device_ips": ["192.168.1.100"], "operation": "reboot"} |
| 394 | + ) |
| 395 | + |
| 396 | + assert response.status_code == 200 |
| 397 | + data = response.json() |
| 398 | + assert len(data) == 1 |
| 399 | + assert data[0]["success"] |
| 400 | + assert data[0]["action_type"] == "Reboot" |
| 401 | + assert data[0]["operation"] == "reboot" |
| 402 | + |
| 403 | + def test_bulk_operations_factory_reset_successfully(self): |
| 404 | + from core.use_cases.bulk_operations import BulkOperationsUseCase |
| 405 | + |
| 406 | + class MockBulkOperationsUseCase(BulkOperationsUseCase): |
| 407 | + def __init__(self): |
| 408 | + pass # Skip parent constructor |
| 409 | + |
| 410 | + async def execute_bulk_factory_reset(self, device_ips): |
| 411 | + return [ |
| 412 | + ActionResult( |
| 413 | + device_ip="192.168.1.100", |
| 414 | + success=True, |
| 415 | + message="Factory reset initiated", |
| 416 | + action_type="FactoryReset", |
| 417 | + ) |
| 418 | + ] |
| 419 | + |
| 420 | + with create_test_client( |
| 421 | + route_handlers=[execute_bulk_operations], |
| 422 | + dependencies={ |
| 423 | + "bulk_operations_use_case": Provide( |
| 424 | + lambda: MockBulkOperationsUseCase(), sync_to_thread=False |
| 425 | + ) |
| 426 | + }, |
| 427 | + ) as client: |
| 428 | + response = client.post( |
| 429 | + "/bulk", |
| 430 | + json={"device_ips": ["192.168.1.100"], "operation": "factory_reset"}, |
| 431 | + ) |
| 432 | + |
| 433 | + assert response.status_code == 200 |
| 434 | + data = response.json() |
| 435 | + assert len(data) == 1 |
| 436 | + assert data[0]["success"] |
| 437 | + assert data[0]["action_type"] == "FactoryReset" |
| 438 | + assert data[0]["operation"] == "factory_reset" |
| 439 | + |
| 440 | + def test_bulk_operations_validation_errors(self): |
| 441 | + from core.use_cases.bulk_operations import BulkOperationsUseCase |
| 442 | + |
| 443 | + class MockBulkOperationsUseCase(BulkOperationsUseCase): |
| 444 | + def __init__(self): |
| 445 | + pass |
| 446 | + |
| 447 | + with create_test_client( |
| 448 | + route_handlers=[execute_bulk_operations], |
| 449 | + dependencies={ |
| 450 | + "bulk_operations_use_case": Provide( |
| 451 | + lambda: MockBulkOperationsUseCase(), sync_to_thread=False |
| 452 | + ) |
| 453 | + }, |
| 454 | + ) as client: |
| 455 | + # Test missing device_ips |
| 456 | + response = client.post("/bulk", json={"operation": "update"}) |
| 457 | + assert response.status_code == 400 |
| 458 | + assert "device_ips is required" in response.json()["detail"] |
| 459 | + |
| 460 | + # Test missing operation |
| 461 | + response = client.post("/bulk", json={"device_ips": ["192.168.1.100"]}) |
| 462 | + assert response.status_code == 400 |
| 463 | + assert "operation is required" in response.json()["detail"] |
| 464 | + |
| 465 | + # Test invalid operation |
| 466 | + response = client.post( |
| 467 | + "/bulk", json={"device_ips": ["192.168.1.100"], "operation": "invalid"} |
| 468 | + ) |
| 469 | + assert response.status_code == 400 |
| 470 | + assert "Unsupported operation: invalid" in response.json()["detail"] |
0 commit comments