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
|
import boto3
from moto import mock_aws
from obfuscator.read import CSVReader
import pytest
reader = CSVReader(log_level="DEBUG")
def test_empty_csv_should_return_no_content():
content = ""
result = reader.read_string(content)
expected = []
assert result == expected
def test_csv_with_header_only_should_return_no_content():
content = "student_id,name,course\n"
result = reader.read_string(content)
expected = []
assert result == expected
def test_csv_with_valid_data():
content = (
"student_id,name,course\n"
"1234,Student 1,Course 1\n"
"5678,Student 2,Course 2\n"
)
result = reader.read_string(content)
expected = [
{"student_id": "1234", "name": "Student 1", "course": "Course 1"},
{"student_id": "5678", "name": "Student 2", "course": "Course 2"},
]
assert result == expected
def test_csv_with_quoted_fields_should_run_as_expected():
content = (
"student_id,name,course\n"
'1234,"Student 1","Course 1"\n'
'5678,"Student 2","Course 2"\n'
)
result = reader.read_string(content)
expected = [
{"student_id": "1234", "name": "Student 1", "course": "Course 1"},
{"student_id": "5678", "name": "Student 2", "course": "Course 2"},
]
assert result == expected
def setup_s3(s3_client, bucket: str, key: str, content: str):
s3_client.create_bucket(
Bucket=bucket,
CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
)
s3_client.put_object(Bucket=bucket, Key=key, Body=content)
@pytest.fixture(autouse=True)
def s3_client():
with mock_aws():
yield boto3.client("s3", "eu-west-2")
def test_read_s3_valid_csv_returns_expected():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "test-bucket"
key = "data/mock.csv"
csv_content = (
"student_id,name,course\n"
"1234,Student 1,Course 1\n"
"5678,Student 2,Course 2\n"
)
setup_s3(s3, bucket, key, csv_content)
path = f"s3://{bucket}/{key}"
data = reader.read_s3(path)
expected = [
{"student_id": "1234", "name": "Student 1", "course": "Course 1"},
{"student_id": "5678", "name": "Student 2", "course": "Course 2"},
]
assert data == expected
def test_read_s3_empty_csv_returns_empty_list():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "empty-bucket"
key = "data/empty.csv"
csv_content = "student_id,name,course\n"
setup_s3(s3, bucket, key, csv_content)
path = f"s3://{bucket}/{key}"
data = reader.read_s3(path)
assert data == []
def test_read_s3_nonexistent_bucket_raises_exception():
with mock_aws():
bucket = "nonexistent-bucket"
key = "data/mock.csv"
path = f"s3://{bucket}/{key}"
with pytest.raises(Exception):
reader.read_s3(path)
def test_read_s3_nonexistent_key_raises_exception():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "test-bucket"
s3.create_bucket(
Bucket=bucket,
CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
)
key = "data/nonexistent.csv"
path = f"s3://{bucket}/{key}"
with pytest.raises(Exception):
reader.read_s3(path)
def test_read_s3_malformed_csv_returns_expected():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "test-bucket"
key = "data/malformed.csv"
csv_content = "1234,Student 1,Course 1\n" "5678,Student 2,Course 2\n"
setup_s3(s3, bucket, key, csv_content)
path = f"s3://{bucket}/{key}"
data = reader.read_s3(path)
expected = [{"1234": "5678", "Student 1": "Student 2", "Course 1": "Course 2"}]
assert data == expected
def test_read_s3_csv_with_extra_empty_lines():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "test-bucket"
key = "data/extra_lines.csv"
csv_content = (
"student_id,name,course\n"
"1234,Student 1,Course 1\n"
"\n"
"5678,Student 2,Course 2\n"
"\n"
)
setup_s3(s3, bucket, key, csv_content)
path = f"s3://{bucket}/{key}"
data = reader.read_s3(path)
expected = [
{"student_id": "1234", "name": "Student 1", "course": "Course 1"},
{"student_id": "5678", "name": "Student 2", "course": "Course 2"},
]
assert data == expected
def test_read_s3_csv_with_whitespace_in_fields():
with mock_aws():
s3 = boto3.client("s3", region_name="eu-west-2")
bucket = "test-bucket"
key = "data/whitespace.csv"
csv_content = (
"student_id, name , course \n"
" 1234 , Student 1 , Course 1 \n"
"5678,Student 2,Course 2\n"
)
setup_s3(s3, bucket, key, csv_content)
path = f"s3://{bucket}/{key}"
data = reader.read_s3(path)
expected = [
{"student_id": " 1234 ", " name ": " Student 1 ", " course ": " Course 1 "},
{"student_id": "5678", " name ": "Student 2", " course ": "Course 2"},
]
assert data == expected
|
