(self, items, rows, settings=None)
| 63 | return content |
| 64 | |
| 65 | async def assertExportedJsonLines(self, items, rows, settings=None): |
| 66 | settings = settings or {} |
| 67 | settings.update( |
| 68 | { |
| 69 | "FEEDS": { |
| 70 | self._random_temp_filename() / "jl" / self._file_mark: { |
| 71 | "format": "jl" |
| 72 | }, |
| 73 | }, |
| 74 | } |
| 75 | ) |
| 76 | batch_size = Settings(settings).getint("FEED_EXPORT_BATCH_ITEM_COUNT") |
| 77 | rows = [{k: v for k, v in row.items() if v} for row in rows] |
| 78 | data = await self.exported_data(items, settings) |
| 79 | for batch in data["jl"]: |
| 80 | got_batch = [ |
| 81 | json.loads(to_unicode(batch_item)) for batch_item in batch.splitlines() |
| 82 | ] |
| 83 | expected_batch, rows = rows[:batch_size], rows[batch_size:] |
| 84 | assert got_batch == expected_batch |
| 85 | |
| 86 | async def assertExportedCsv(self, items, header, rows, settings=None): |
| 87 | settings = settings or {} |
nothing calls this directly
no test coverage detected