diff --git a/Lib/test/test_csv.py b/Lib/test/test_csv.py index 13b68cc2255..3e86af0f8c5 100644 --- a/Lib/test/test_csv.py +++ b/Lib/test/test_csv.py @@ -276,7 +276,6 @@ def test_write_lineterminator(self): f'1,2{lineterminator}' f'"\r","\n"{lineterminator}') - @unittest.expectedFailure # TODO: RUSTPYTHON def test_write_iterable(self): self._write_test(iter(['a', 1, 'p,q']), 'a,1,"p,q"') self._write_test(iter(['a', 1, None]), 'a,1,') @@ -319,7 +318,6 @@ def test_writerows_with_none(self): self.assertEqual(fileobj.read(), 'a\r\n""\r\n') - @unittest.expectedFailure # TODO: RUSTPYTHON def test_write_empty_fields(self): self._write_test((), '') self._write_test([''], '""') diff --git a/crates/stdlib/src/csv.rs b/crates/stdlib/src/csv.rs index 3fbafab8dca..cb7cecff416 100644 --- a/crates/stdlib/src/csv.rs +++ b/crates/stdlib/src/csv.rs @@ -1380,6 +1380,55 @@ mod _csv { self.write.call((s,), vm) } + fn writerow_minimal(&self, row: PyObjectRef, vm: &VirtualMachine) -> PyResult { + let _state = self.state.lock(); + + let row: ArgIterable = ArgIterable::try_from_object(vm, row.clone()).map_err(|_e| { + new_csv_error( + vm, + format!("'{}' object is not iterable", row.class().name()), + ) + })?; + + let fields = row.iter(vm)?.collect::>>()?; + let single_field = fields.len() == 1; + let mut output = Vec::new(); + + for (index, field) in fields.into_iter().enumerate() { + if index > 0 { + output.push(self.dialect.delimiter); + } + + let stringified; + let data: &[u8] = match_class!(match field { + ref s @ PyStr => s.as_bytes(), + crate::builtins::PyNone => b"", + ref obj => { + stringified = obj.str(vm)?; + stringified.as_bytes() + } + }); + + // CPython quotes a QUOTE_MINIMAL field if it contains the + // delimiter, the quote character, '\r', '\n', or the line + // terminator, regardless of which line terminator is + // configured. A row with a single empty field is also quoted + // so that it is not read back as an empty line. + if field_needs_quotes(data, self.dialect) || (single_field && data.is_empty()) { + write_quoted_field(&mut output, data, self.dialect, vm)?; + } else { + output.extend_from_slice(data); + } + } + + write_lineterminator(&mut output, self.dialect.lineterminator); + + let s = core::str::from_utf8(&output) + .map_err(|_| vm.new_unicode_decode_error("csv not utf8"))?; + + self.write.call((s,), vm) + } + #[pymethod] fn writerow(&self, row: PyObjectRef, vm: &VirtualMachine) -> PyResult { match self.dialect.quoting { @@ -1387,6 +1436,7 @@ mod _csv { QuoteStyle::Strings | QuoteStyle::Notnull => { return self.writerow_quoted_strings(row, vm); } + QuoteStyle::Minimal => return self.writerow_minimal(row, vm), _ => {} } diff --git a/extra_tests/snippets/stdlib_csv.py b/extra_tests/snippets/stdlib_csv.py index dc2186d17ac..0664bfd8d92 100644 --- a/extra_tests/snippets/stdlib_csv.py +++ b/extra_tests/snippets/stdlib_csv.py @@ -147,3 +147,48 @@ def test_quote_none_reader_skipinitialspace_escapechar(): test_quote_none_reader_skipinitialspace_escapechar() + + +def test_quote_minimal_writer_lineterminator(): + # https://github.com/RustPython/RustPython/issues/8302 + # QUOTE_MINIMAL must quote '\r' and '\n' regardless of the line terminator. + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="!") + writer.writerow(["a", "b"]) + writer.writerow([1, 2]) + writer.writerow(["\r", "\n"]) + assert buf.getvalue() == 'a,b!1,2!"\r","\n"!' + + nul = io.StringIO() + csv.writer(nul, lineterminator="\0").writerow(["\r", "\n"]) + assert nul.getvalue() == '"\r","\n"\0' + + crlf = io.StringIO() + csv.writer(crlf, lineterminator="!").writerow(["\r\n"]) + assert crlf.getvalue() == '"\r\n"!' + + # the terminator character itself still triggers quoting + term = io.StringIO() + csv.writer(term, lineterminator="!").writerow(["a!b", "c"]) + assert term.getvalue() == '"a!b",c!' + + # default terminator behavior is unchanged + default = io.StringIO() + csv.writer(default).writerow(["\r", "\n"]) + assert default.getvalue() == '"\r","\n"\r\n' + + +test_quote_minimal_writer_lineterminator() + + +def test_quote_minimal_writer_empty_fields(): + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow([""]) + writer.writerow([None]) + writer.writerow([]) + writer.writerow(["", ""]) + assert buf.getvalue() == '""\r\n""\r\n\r\n,\r\n' + + +test_quote_minimal_writer_empty_fields()