-
Notifications
You must be signed in to change notification settings - Fork 89
Expand file tree
/
Copy pathtest_delete_tests.py
More file actions
150 lines (119 loc) · 4.71 KB
/
test_delete_tests.py
File metadata and controls
150 lines (119 loc) · 4.71 KB
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
"""
Unit tests for delete_tests Lambda function.
"""
import json
import os
import sys
from unittest.mock import Mock, patch
import pytest
from botocore.exceptions import ClientError
# Mock the logger before importing index
sys.modules["idp_common_pkg"] = Mock()
sys.modules["idp_common_pkg.logger"] = Mock()
# Mock boto3 before importing the Lambda module to prevent NoRegionError
# The Lambda creates boto3 clients at module level which requires AWS region
with patch("boto3.resource") as mock_resource, patch("boto3.client") as mock_client:
mock_resource.return_value = Mock()
mock_client.return_value = Mock()
# Add the lambda directory to the path for importing
lambda_path = os.path.join(
os.path.dirname(__file__), "../../../../src/lambda/delete_tests"
)
sys.path.insert(0, lambda_path)
import index # type: ignore[import-untyped] # noqa: E402
@pytest.mark.unit
@patch("index.lambda_client")
@patch("index.dynamodb")
def test_lambda_handler_success(mock_dynamodb, mock_lambda_client):
"""Test successful deletion of test runs."""
# Setup
mock_table = Mock()
mock_dynamodb.Table.return_value = mock_table
# Mock get_item responses
mock_table.get_item.side_effect = [
{"Item": {"Files": ["file1.pdf", "file2.pdf"]}},
{"Item": {"Files": ["file3.pdf"]}},
]
event = {"arguments": {"testRunIds": ["test1", "test2"]}}
context = Mock()
context.get.side_effect = lambda key: {
"TRACKING_TABLE_NAME": "test-table",
"DELETE_DOCUMENT_FUNCTION_NAME": "delete-func",
}[key]
# Execute
result = index.lambda_handler(event, context)
# Verify
assert result is True
assert mock_table.get_item.call_count == 2
assert mock_table.delete_item.call_count == 2
# Verify lambda invocation with all document keys
mock_lambda_client.invoke.assert_called_once()
call_args = mock_lambda_client.invoke.call_args
payload = json.loads(call_args[1]["Payload"])
expected_keys = ["test1/file1.pdf", "test1/file2.pdf", "test2/file3.pdf"]
assert payload["arguments"]["objectKeys"] == expected_keys
@pytest.mark.unit
@patch("index.lambda_client")
@patch("index.dynamodb")
def test_lambda_handler_test_run_not_found(mock_dynamodb, mock_lambda_client):
"""Test handling when test run is not found."""
mock_table = Mock()
mock_dynamodb.Table.return_value = mock_table
mock_table.get_item.return_value = {} # No Item key
event = {"arguments": {"testRunIds": ["nonexistent"]}}
context = Mock()
context.get.side_effect = lambda key: {
"TRACKING_TABLE_NAME": "test-table",
"DELETE_DOCUMENT_FUNCTION_NAME": "delete-func",
}[key]
result = index.lambda_handler(event, context)
assert result is False
mock_table.delete_item.assert_not_called()
mock_lambda_client.invoke.assert_not_called()
@pytest.mark.unit
@patch("index.lambda_client")
@patch("index.dynamodb")
def test_lambda_handler_no_files(mock_dynamodb, mock_lambda_client):
"""Test handling when test run has no files."""
mock_table = Mock()
mock_dynamodb.Table.return_value = mock_table
mock_table.get_item.return_value = {"Item": {}} # No Files key
event = {"arguments": {"testRunIds": ["test1"]}}
context = Mock()
context.get.side_effect = lambda key: {
"TRACKING_TABLE_NAME": "test-table",
"DELETE_DOCUMENT_FUNCTION_NAME": "delete-func",
}[key]
result = index.lambda_handler(event, context)
assert result is True
mock_table.delete_item.assert_called_once()
mock_lambda_client.invoke.assert_not_called()
@pytest.mark.unit
@patch("index.lambda_client")
@patch("index.dynamodb")
def test_lambda_handler_client_error(mock_dynamodb, mock_lambda_client):
"""Test handling of DynamoDB client errors."""
mock_table = Mock()
mock_dynamodb.Table.return_value = mock_table
mock_table.get_item.side_effect = ClientError(
{"Error": {"Code": "ResourceNotFoundException"}}, "GetItem"
)
event = {"arguments": {"testRunIds": ["test1"]}}
context = Mock()
context.get.side_effect = lambda key: {
"TRACKING_TABLE_NAME": "test-table",
"DELETE_DOCUMENT_FUNCTION_NAME": "delete-func",
}[key]
result = index.lambda_handler(event, context)
assert result is False
mock_lambda_client.invoke.assert_not_called()
@pytest.mark.unit
@patch("index.dynamodb")
def test_lambda_handler_missing_env_vars(mock_dynamodb):
"""Test handling of missing environment variables."""
event = {"arguments": {"testRunIds": ["test1"]}}
context = Mock()
context.get.return_value = None
# The actual error occurs when trying to create DynamoDB table with None name
result = index.lambda_handler(event, context)
assert result is False