sources_test.py

class TestCsvFileSource(unittest.TestCase):

    CSV_LINES_FILE = """header1,header2,header3
value1,value2,value3
"""


    def test_gzipped_csv(self):
        out = StringIO()
        with tempfile.NamedTemporaryFile() as f:
            with gzip.GzipFile(filename=f.name, mode='wb') as gz:
                gz.write(self.CSV_LINES_FILE)

            csv = CsvFileSource(
                    f.name,
                    compression_type=CompressionTypes.GZIP)

            # first record as a dict
            rec = csv.read_records(f.name, None).next()

            # make a dict from the csv lines
            lines = self.CSV_LINES_FILE.split('\n')
            ltup = zip(lines[0].split(','), lines[1].split(','))
            ldict = {header: value for (header, value) in ltup}

            # assert both dicts are equal
            for (header, value) in rec.iteritems():
                self.assertEqual(value, ldict[header])