|
21 | 21 | check_lighthouse_provider_connection_task, |
22 | 22 | generate_outputs_task, |
23 | 23 | perform_attack_paths_scan_task, |
| 24 | + perform_scan_task, |
24 | 25 | perform_scheduled_scan_task, |
25 | 26 | reaggregate_all_finding_group_summaries_task, |
26 | 27 | refresh_lighthouse_provider_models_task, |
@@ -1433,9 +1434,9 @@ def factory(*args, **kwargs): |
1433 | 1434 | ) |
1434 | 1435 |
|
1435 | 1436 | # Verify ASFF was NOT created for non-AWS provider |
1436 | | - assert ( |
1437 | | - "asff" not in created_writers |
1438 | | - ), "ASFF writer should NOT be created for non-AWS providers" |
| 1437 | + assert "asff" not in created_writers, ( |
| 1438 | + "ASFF writer should NOT be created for non-AWS providers" |
| 1439 | + ) |
1439 | 1440 | assert "csv" in created_writers, "CSV writer should be created" |
1440 | 1441 | assert "ocsf" in created_writers, "OCSF writer should be created" |
1441 | 1442 |
|
@@ -2454,6 +2455,51 @@ def _complete_scan(tenant_id, scan_id, provider_id): |
2454 | 2455 | == 1 |
2455 | 2456 | ) |
2456 | 2457 |
|
| 2458 | + def test_no_op_when_provider_does_not_exist(self, tenants_fixture): |
| 2459 | + """Return None without raising when the provider was already deleted.""" |
| 2460 | + tenant = tenants_fixture[0] |
| 2461 | + missing_provider_id = str(uuid.uuid4()) |
| 2462 | + task_id = str(uuid.uuid4()) |
| 2463 | + self._create_task_result(tenant.id, task_id) |
| 2464 | + |
| 2465 | + with ( |
| 2466 | + patch("tasks.tasks.perform_prowler_scan") as mock_scan, |
| 2467 | + patch("tasks.tasks._perform_scan_complete_tasks") as mock_complete_tasks, |
| 2468 | + self._override_task_request(perform_scheduled_scan_task, id=task_id), |
| 2469 | + ): |
| 2470 | + result = perform_scheduled_scan_task.run( |
| 2471 | + tenant_id=str(tenant.id), provider_id=missing_provider_id |
| 2472 | + ) |
| 2473 | + |
| 2474 | + assert result is None |
| 2475 | + mock_scan.assert_not_called() |
| 2476 | + mock_complete_tasks.assert_not_called() |
| 2477 | + |
| 2478 | + |
| 2479 | +@pytest.mark.django_db |
| 2480 | +class TestPerformScanTask: |
| 2481 | + """Unit tests for perform_scan_task.""" |
| 2482 | + |
| 2483 | + def test_no_op_when_provider_does_not_exist(self, tenants_fixture): |
| 2484 | + """Return None without raising when the provider was already deleted.""" |
| 2485 | + tenant = tenants_fixture[0] |
| 2486 | + missing_provider_id = str(uuid.uuid4()) |
| 2487 | + scan_id = str(uuid.uuid4()) |
| 2488 | + |
| 2489 | + with ( |
| 2490 | + patch("tasks.tasks.perform_prowler_scan") as mock_scan, |
| 2491 | + patch("tasks.tasks._perform_scan_complete_tasks") as mock_complete_tasks, |
| 2492 | + ): |
| 2493 | + result = perform_scan_task.run( |
| 2494 | + tenant_id=str(tenant.id), |
| 2495 | + scan_id=scan_id, |
| 2496 | + provider_id=missing_provider_id, |
| 2497 | + ) |
| 2498 | + |
| 2499 | + assert result is None |
| 2500 | + mock_scan.assert_not_called() |
| 2501 | + mock_complete_tasks.assert_not_called() |
| 2502 | + |
2457 | 2503 |
|
2458 | 2504 | @pytest.mark.django_db |
2459 | 2505 | class TestReaggregateAllFindingGroupSummaries: |
|
0 commit comments