(self, items, header, rows, settings=None)
| 84 | assert got_batch == expected_batch |
| 85 | |
| 86 | async def assertExportedCsv(self, items, header, rows, settings=None): |
| 87 | settings = settings or {} |
| 88 | settings.update( |
| 89 | { |
| 90 | "FEEDS": { |
| 91 | self._random_temp_filename() / "csv" / self._file_mark: { |
| 92 | "format": "csv" |
| 93 | }, |
| 94 | }, |
| 95 | } |
| 96 | ) |
| 97 | batch_size = Settings(settings).getint("FEED_EXPORT_BATCH_ITEM_COUNT") |
| 98 | data = await self.exported_data(items, settings) |
| 99 | for batch in data["csv"]: |
| 100 | got_batch = csv.DictReader(to_unicode(batch).splitlines()) |
| 101 | assert list(header) == got_batch.fieldnames |
| 102 | expected_batch, rows = rows[:batch_size], rows[batch_size:] |
| 103 | assert list(got_batch) == expected_batch |
| 104 | |
| 105 | async def assertExportedXml(self, items, rows, settings=None): |
| 106 | settings = settings or {} |
nothing calls this directly
no test coverage detected