Completed
Pull Request — master (#32)
by Philip
01:30
created

DelimitedReaderFixture._clean()   A

Complexity

Conditions 2

Size

Total Lines 4

Duplication

Lines 0
Ratio 0 %

Importance

Changes 1
Bugs 0 Features 0
Metric Value
cc 2
dl 0
loc 4
rs 10
c 1
b 0
f 0
1
import csv
0 ignored issues
show
Coding Style introduced by
This module should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
2
import io
3
import os
4
import unittest
5
import textwrap
6
import zipfile
7
from collections import namedtuple
8
from datetime import date, datetime
9
from tempfile import NamedTemporaryFile
10
11
from foil.fileio import (concatenate_streams, DelimitedReader,
12
                         DelimitedSubsetReader, TextReader, ZipReader)
13
14
15
class MockDialect(csv.Dialect):
0 ignored issues
show
Coding Style introduced by
This class should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
16
    delimiter = '|'
17
    quotechar = '"'
18
    doublequote = True
19
    skipinitialspace = False
20
    lineterminator = '\n'
21
    quoting = csv.QUOTE_MINIMAL
22
23
24
def delimited_text():
0 ignored issues
show
Coding Style introduced by
This function should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
25
    file_content = textwrap.dedent(r"""
26
"NAME"|"CLASS"|"DATE"|"ASSIGNMENT"|"SCORE"|"AVERAGE"
27
"Dave"|"American History"|2015-03-02|"QUIZ 1"|82|82.0
28
"Dave"|"American History"|2015-04-04|"QUIZ 2"|91|86.5
29
"Dave"|"American History"|2015-04-20|"Mid-term"|77|83.333
30
""").strip()
31
    return file_content
32
33
34
def data_records(fields):
0 ignored issues
show
Coding Style introduced by
This function should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
35
    Record = namedtuple('Record', fields)
0 ignored issues
show
Coding Style Naming introduced by
The name Record does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
36
    records = [
37
        Record('Dave', 'American History', date(2015, 3, 2), 'QUIZ 1', 82.0, 82.0),
38
        Record('Dave', 'American History', date(2015, 4, 4), 'QUIZ 2', 91.0, 86.5),
39
        Record('Dave', 'American History', date(2015, 4, 20), 'Mid-term', 77.0, 83.333)]
40
41
    return records
42
43
44
def partial_data_records(fields):
0 ignored issues
show
Coding Style introduced by
This function should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
45
    Record = namedtuple('Record', fields)
0 ignored issues
show
Coding Style Naming introduced by
The name Record does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
46
    records = [
47
        Record('Dave', 'American History', 82.0),
48
        Record('Dave', 'American History', 86.5),
49
        Record('Dave', 'American History', 83.333)]
50
51
    return records
52
53
54
def single_field_records(fields):
55
    Record = namedtuple('Record', fields)
0 ignored issues
show
Coding Style Naming introduced by
The name Record does not conform to the variable naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
56
    records = [Record('Dave'), Record('Dave'), Record('Dave')]
57
58
    return records
59
60
61
def parse_date(date_str):
62
    if date_str is '':
63
        return None
64
    else:
65
        return datetime.strptime(date_str, '%Y-%m-%d').date()
66
67
68
class ReaderFixture(object):
0 ignored issues
show
Coding Style introduced by
This class should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
69
    encoding = 'UTF-8'
70
71
    @classmethod
72
    def setUpClass(cls):
0 ignored issues
show
Coding Style Naming introduced by
The name setUpClass does not conform to the method naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
73
        cls._prep()
74
75
    @classmethod
76
    def _prep(cls):
77
        pass
78
79
    @classmethod
80
    def _clean(cls):
81
        pass
82
83
    @classmethod
84
    def tearDownClass(cls):
0 ignored issues
show
Coding Style Naming introduced by
The name tearDownClass does not conform to the method naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
85
        cls._clean()
86
87
        if os.path.exists(cls.path):
0 ignored issues
show
Bug introduced by
The Class ReaderFixture does not seem to have a member named path.

This check looks for calls to members that are non-existent. These calls will fail.

The member could have been renamed or removed.

Loading history...
88
            os.unlink(cls.path)
0 ignored issues
show
Bug introduced by
The Class ReaderFixture does not seem to have a member named path.

This check looks for calls to members that are non-existent. These calls will fail.

The member could have been renamed or removed.

Loading history...
89
90
91
class DelimitedReaderFixture(ReaderFixture):
0 ignored issues
show
Coding Style introduced by
This class should have a docstring.

The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods:

class SomeClass:
    def some_method(self):
        """Do x and return foo."""

If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions.

Loading history...
92
    zip_filename = 'delimited_file1.txt'
93
94
    @classmethod
95
    def _prep(cls):
96
        cls.dialect = MockDialect()
97
        file_content = delimited_text()
98
        with NamedTemporaryFile(prefix='delim_', suffix='.txt', delete=False) as tmp:
99
            with open(tmp.name, 'w', encoding=cls.encoding) as text_file:
100
                text_file.write(file_content)
101
            cls.path = tmp.name
102
103
        file_content_bytes = bytes(file_content, cls.encoding)
104
        with NamedTemporaryFile(prefix='zipped_', suffix='.zip', delete=False) as tmp:
105
            with zipfile.ZipFile(tmp.name, mode='w') as myzip:
106
                myzip.writestr(cls.zip_filename, file_content_bytes)
107
            cls.zip_path = tmp.name
108
109
    @classmethod
110
    def _clean(cls):
111
        if os.path.exists(cls.zip_path):
112
            os.unlink(cls.zip_path)
113
114
115
class TestTextReader(ReaderFixture, unittest.TestCase):
116
    @classmethod
117
    def _prep(cls):
118
        file_content = 'hello\nworld\n'
119
        with NamedTemporaryFile(prefix='text_', suffix='.txt', delete=False) as tmp:
120
            with open(tmp.name, 'w', encoding=cls.encoding) as text_file:
121
                text_file.write(file_content)
122
            cls.path = tmp.name
123
124
    def test_text_reader(self):
125
        reader = TextReader(self.path, 'UTF-8')
126
127
        expected = ['hello', 'world']
128
        result = list(reader)
129
130
        self.assertEqual(expected, result)
131
132
133
class TestDelimitedReader(DelimitedReaderFixture, unittest.TestCase):
134
    def setUp(self):
135
        self.fields = ['NAME', 'CLASS', 'DATE', 'ASSIGNMENT', 'SCORE', 'AVERAGE']
136
        self.converters = [str, str, parse_date, str, float, float]
137
        self.expected = data_records(self.fields)
138
139
    def test_stream_reader(self):
140
        stream = io.StringIO(delimited_text())
141
        reader = DelimitedReader(stream, dialect=self.dialect,
142
                                 fields=self.fields, converters=self.converters)
143
144
        result = list(reader)
145
146
        self.assertSequenceEqual(self.expected, result)
147
148
    def test_from_file(self):
149
        reader = DelimitedReader.from_file(path=self.path,
150
                                           encoding=self.encoding,
151
                                           dialect=self.dialect,
152
                                           fields=self.fields,
153
                                           converters=self.converters)
154
155
        result = list(reader)
156
157
        self.assertSequenceEqual(self.expected, result)
158
159
    def test_from_zipfile(self):
160
        reader = DelimitedReader.from_zipfile(path=self.zip_path,
161
                                              filename=self.zip_filename,
162
                                              encoding=self.encoding,
163
                                              dialect=self.dialect,
164
                                              fields=self.fields,
165
                                              converters=self.converters)
166
167
        result = list(reader)
168
169
        self.assertSequenceEqual(self.expected, result)
170
171
    def test_line_number(self):
172
        # counts skipped header line
173
        stream = io.StringIO(delimited_text())
174
        reader = DelimitedReader(stream, dialect=self.dialect, fields=self.fields,
175
                                 converters=self.converters)
176
        next(reader)
177
        next(reader)
178
179
        self.assertEqual(reader.file_line_number, 3)
180
181
182
class TestDelimitedSubsetReader(DelimitedReaderFixture, unittest.TestCase):
183
    def setUp(self):
184
        self.maxDiff = None
0 ignored issues
show
Coding Style Naming introduced by
The name maxDiff does not conform to the attribute naming conventions ([a-z_][a-z0-9_]{2,30}$).

This check looks for invalid names for a range of different identifiers.

You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements.

If your project includes a Pylint configuration file, the settings contained in that file take precedence.

To find out more about Pylint, please refer to their site.

Loading history...
185
        self.headers = ['NAME', 'CLASS', 'DATE', 'ASSIGNMENT', 'SCORE', 'AVERAGE']
186
        self.fields = ['NAME', 'CLASS', 'AVERAGE']
187
        self.field_index = [self.headers.index(field) for field in self.fields]
188
        self.converters = [str, str, float]
189
        self.expected = partial_data_records(self.fields)
190
191
    def test_stream_reader(self):
192
        stream = io.StringIO(delimited_text())
193
        reader = DelimitedSubsetReader(stream,
194
                                       dialect=self.dialect,
195
                                       fields=self.fields,
196
                                       converters=self.converters,
197
                                       field_index=self.field_index)
198
199
        result = list(reader)
200
201
        self.assertSequenceEqual(self.expected, result)
202
203
    def test_from_file(self):
204
        reader = DelimitedSubsetReader.from_file(path=self.path,
205
                                                 encoding=self.encoding,
206
                                                 dialect=self.dialect,
207
                                                 fields=self.fields,
208
                                                 converters=self.converters,
209
                                                 field_index=self.field_index)
210
211
        result = list(reader)
212
213
        self.assertSequenceEqual(self.expected, result)
214
215
    def test_from_zipfile(self):
216
        reader = DelimitedSubsetReader.from_zipfile(path=self.zip_path,
217
                                                    filename=self.zip_filename,
218
                                                    encoding=self.encoding,
219
                                                    dialect=self.dialect,
220
                                                    fields=self.fields,
221
                                                    converters=self.converters,
222
                                                    field_index=self.field_index)
223
224
        result = list(reader)
225
226
        self.assertSequenceEqual(self.expected, result)
227
228
    def test_single_field(self):
229
        stream = io.StringIO(delimited_text())
230
        field_index = [0]
231
        fields = [self.fields[index] for index in field_index]
232
        converters = [self.converters[index] for index in field_index]
233
234
        reader = DelimitedSubsetReader(stream,
235
                                       dialect=self.dialect,
236
                                       fields=fields,
237
                                       converters=converters,
238
                                       field_index=field_index)
239
240
        expected = single_field_records(fields)
241
        result = list(reader)
242
243
        self.assertEqual(expected, result)
244
245
246
class TestZipReader(ReaderFixture, unittest.TestCase):
247
248
    filename = 'sample_file.txt'
249
    file_content = 'abc\neasy\n123.\n'
250
251
    @classmethod
252
    def _prep(cls):
253
        cls.file_content_bytes = bytes(cls.file_content, encoding=cls.encoding)
254
        with NamedTemporaryFile(prefix='zipped_', suffix='.zip', delete=False) as tmp:
255
            with zipfile.ZipFile(tmp.name, mode='w') as myzip:
256
                    myzip.writestr(cls.filename, cls.file_content_bytes)
0 ignored issues
show
Coding Style introduced by
The indentation here looks off. 16 spaces were expected, but 20 were found.
Loading history...
257
            cls.path = tmp.name
258
259
    def test_read(self):
260
        result = ZipReader(self.path, self.filename).read(self.encoding)
261
262
        self.assertEqual(self.file_content, result)
263
264
    def test_readlines(self):
265
        line_gen = ZipReader(self.path, self.filename).readlines(self.encoding)
266
        expected = ['abc', 'easy', '123.']
267
268
        self.assertEqual(expected, list(line_gen))
269
270
271
class TestConcatenateStreams(unittest.TestCase):
272
    def test_concatenate_streams(self):
273
        streams = [[1, 2, 3], ['a', 'b', 'c']]
274
275
        expected = [1, 2, 3, 'a', 'b', 'c']
276
        result = list(concatenate_streams(streams))
277
278
        self.assertEqual(expected, result)
279