csv: support multi-character lineterminator in writer (#8328)

* csv: support multi-character lineterminator in writer

Store the dialect line terminator as an owned String instead of the
single-byte csv_core::Terminator, dropping PyDialect's Copy derive. The
manual writer paths emit the full terminator; the csv-core-backed
QUOTE_ALL/QUOTE_NONNUMERIC paths emit a sentinel byte (preserving
csv-core's quote/empty-record bookkeeping) and append the real
terminator. field_needs_quotes/escape now quote a field containing any
terminator byte. The reader ignores lineterminator and always uses CRLF,
matching CPython and avoiding mid-UTF-8 record splits.

* csv: unmark now-passing test_write_lineterminator

* csv: address review feedback

- Remove the redundant per-branch sentinel terminator setup in to_writer;
  the unconditional terminator call after the match is the single source.
- Reword the empty-lineterminator errors to "must not be empty" (the
  constraint is non-empty, not single-character) on both entry points.
- Promote the sentinel invariant check in writerow from debug_assert_eq!
  to assert_eq! so it also guards release builds.
- Add a snippet case for a field containing a line-break byte to ensure
  the csv-core path drops only the trailing terminator.

* csv: reject non-ASCII lineterminator

The writer decides what to quote and escape by comparing raw bytes, so a
non-ASCII terminator quoted a field that merely shared a UTF-8 lead byte,
and QUOTE_NONE escaped individual bytes of the terminator and then failed to
decode the record back to a string.

Reject non-ASCII terminators, including lone surrogates, as csv.Error when
the dialect is parsed, and leave code-point-wise handling to a follow-up
(#8310). The non-string error message now matches CPython as well.

Lib/test/test_csv.py

@@ -262,7 +262,6 @@ class Test_Csv(unittest.TestCase):
         self._write_test(['C\\', '6', '7', 'X"'], 'C\\\\,6,7,"X"""',
                          escapechar='\\', quoting=csv.QUOTE_MINIMAL)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_write_lineterminator(self):
         for lineterminator in '\r\n', '\n', '\r', '!@#', '\0':
             with self.subTest(lineterminator=lineterminator):

crates/stdlib/src/csv.rs

@@ -77,18 +77,27 @@ mod _csv {
 
     #[pyattr]
     #[pyclass(module = "csv", name = "Dialect")]
-    #[derive(Debug, PyPayload, Clone, Copy)]
+    #[derive(Debug, PyPayload, Clone)]
     struct PyDialect {
         delimiter: u8,
         quotechar: Option<u8>,
         escapechar: Option<u8>,
         doublequote: bool,
         skipinitialspace: bool,
-        lineterminator: csv_core::Terminator,
+        lineterminator: String,
         quoting: QuoteStyle,
         strict: bool,
     }
 
+    /// Placeholder single-byte terminator for the csv-core writer paths
+    /// (`QUOTE_ALL` / `QUOTE_NONNUMERIC`). csv-core can only emit a single byte
+    /// for the record terminator, but its `terminator()` call also performs
+    /// essential bookkeeping — closing the final quote and emitting `""` for an
+    /// empty record — that must not be bypassed. So the writer emits this
+    /// sentinel byte, and `writerow` strips it and appends the real (possibly
+    /// multi-character) line terminator afterwards.
+    const CSV_CORE_TERMINATOR_SENTINEL: u8 = b'\n';
+
     impl Constructor for PyDialect {
         type Args = PyObjectRef;
 
@@ -121,11 +130,7 @@ mod _csv {
 
         #[pygetset]
         fn lineterminator(&self, vm: &VirtualMachine) -> PyRef<PyStr> {
-            match self.lineterminator {
-                Terminator::CRLF => vm.ctx.new_str("\r\n".to_string()),
-                Terminator::Any(t) => vm.ctx.new_str(format!("{}", t as char)),
-                _ => unreachable!(),
-            }
+            vm.ctx.new_str(self.lineterminator.clone())
         }
 
         #[pygetset]
@@ -230,19 +235,42 @@ mod _csv {
         })
     }
 
-    fn prase_lineterminator_from_obj(vm: &VirtualMachine, obj: &PyObject) -> PyResult<Terminator> {
+    /// Validate that a line terminator is ASCII and return it as a `str`.
+    ///
+    /// The writer's quoting and escaping predicates compare raw bytes, so a
+    /// non-ASCII terminator would either quote a field that merely shares a
+    /// UTF-8 lead byte or splice an escape character into the middle of a
+    /// multi-byte sequence. Reject those here.
+    ///
+    /// The ASCII check must come before any UTF-8 conversion so that lone
+    /// surrogates are reported as this `csv.Error` too.
+    ///
+    /// TODO: RUSTPYTHON; handle non-ASCII terminators code-point-wise as part
+    /// of full Unicode dialect support.
+    fn ascii_lineterminator<'a>(vm: &VirtualMachine, s: &'a PyStr) -> PyResult<&'a str> {
+        if !s.as_wtf8().is_ascii() {
+            return Err(new_csv_error(
+                vm,
+                r#""lineterminator" must be an ASCII string"#,
+            ));
+        }
+        // An ASCII string is always valid UTF-8.
+        s.to_str()
+            .ok_or_else(|| new_csv_error(vm, r#""lineterminator" must be a string"#))
+    }
+
+    fn prase_lineterminator_from_obj(vm: &VirtualMachine, obj: &PyObject) -> PyResult<String> {
         match_class!(match obj.get_attr("lineterminator", vm)? {
             s @ PyStr => {
-                Ok(if s.as_bytes().eq(b"\r\n") {
-                    csv_core::Terminator::CRLF
-                } else if let Some(t) = s.as_bytes().first() {
-                    // Due to limitations in the current implementation within csv_core
-                    // the support for multiple characters in lineterminator is not complete.
-                    // only capture the first character
-                    csv_core::Terminator::Any(*t)
-                } else {
-                    return Err(new_csv_error(vm, r#""lineterminator" must be a string"#));
-                })
+                // Store the full line terminator string. CPython accepts an
+                // arbitrary-length terminator; the manual writer paths emit it
+                // verbatim and the csv-core writer path appends it after a
+                // sentinel terminator (see `writerow`).
+                let value = ascii_lineterminator(vm, &s)?;
+                if value.is_empty() {
+                    return Err(new_csv_error(vm, r#""lineterminator" must not be empty"#));
+                }
+                Ok(value.to_owned())
             }
             attr => {
                 Err(vm.new_type_error(format!(
@@ -344,7 +372,7 @@ mod _csv {
         let g = GLOBAL_HASHMAP.lock();
 
         if let Some(dialect) = g.get(name.as_str()) {
-            return Ok(*dialect);
+            return Ok(dialect.clone());
         }
 
         Err(new_csv_error(vm, "unknown dialect"))
@@ -540,7 +568,7 @@ mod _csv {
         escapechar: Option<u8>,
         doublequote: Option<bool>,
         skipinitialspace: Option<bool>,
-        lineterminator: Option<csv_core::Terminator>,
+        lineterminator: Option<String>,
         quoting: Option<QuoteStyle>,
         strict: Option<bool>,
     }
@@ -629,15 +657,22 @@ mod _csv {
             };
 
             if let Some(lineterminator) = args.kwargs.swap_remove("lineterminator") {
-                res.lineterminator = Some(csv_core::Terminator::Any(
-                    lineterminator
-                        .try_to_value::<&str>(vm)?
-                        .bytes()
-                        .exactly_one()
-                        .map_err(|_| {
-                            vm.new_type_error(r#""lineterminator" must be a 1-character string"#)
-                        })?,
-                ))
+                let s = lineterminator.downcast_ref::<PyStr>().ok_or_else(|| {
+                    vm.new_type_error(format!(
+                        r#""lineterminator" must be a string, not {}"#,
+                        lineterminator.class().name()
+                    ))
+                })?;
+                let value = ascii_lineterminator(vm, s)?;
+                // Preserve the previous behavior of rejecting an empty terminator
+                // (full validation parity is deferred to a follow-up). Any
+                // non-empty string, including multi-character ones, is stored.
+                if value.is_empty() {
+                    return Err(vm
+                        .new_type_error(r#""lineterminator" must not be empty"#)
+                        .into());
+                }
+                res.lineterminator = Some(value.to_owned());
             };
 
             if let Some(doublequote) = args.kwargs.swap_remove("doublequote") {
@@ -717,7 +752,7 @@ mod _csv {
     }
 
     impl FormatOptions {
-        const fn update_py_dialect(&self, mut res: PyDialect) -> PyDialect {
+        fn update_py_dialect(&self, mut res: PyDialect) -> PyDialect {
             macro_rules! check_and_fill {
                 ($res:ident, $e:ident) => {{
                     if let Some(t) = self.$e {
@@ -741,7 +776,9 @@ mod _csv {
             };
 
             check_and_fill!(res, quoting);
-            check_and_fill!(res, lineterminator);
+            if let Some(t) = &self.lineterminator {
+                res.lineterminator.clone_from(t);
+            };
             check_and_fill!(res, strict);
             res
         }
@@ -751,16 +788,16 @@ mod _csv {
                 DialectItem::Str(name) => {
                     let g = GLOBAL_HASHMAP.lock();
                     if let Some(dialect) = g.get(name) {
-                        Ok(self.update_py_dialect(*dialect))
+                        Ok(self.update_py_dialect(dialect.clone()))
                     } else {
                         Err(new_csv_error(vm, format!("{name} is not registered.")))
                     }
                     // TODO: Maybe need to update the obj from HashMap
                 }
-                DialectItem::Obj(o) => Ok(self.update_py_dialect(*o)),
+                DialectItem::Obj(o) => Ok(self.update_py_dialect(o.clone())),
                 DialectItem::None => {
                     let g = GLOBAL_HASHMAP.lock();
-                    let res = *g.get("excel").unwrap();
+                    let res = g.get("excel").unwrap().clone();
                     Ok(self.update_py_dialect(res))
                 }
             }
@@ -788,27 +825,6 @@ mod _csv {
             skipinitialspace
         }
 
-        fn get_lineterminator(&self) -> csv_core::Terminator {
-            let mut lineterminator = match &self.dialect {
-                DialectItem::Str(name) => {
-                    let g = GLOBAL_HASHMAP.lock();
-                    if let Some(dialect) = g.get(name) {
-                        dialect.lineterminator
-                    } else {
-                        Terminator::CRLF
-                    }
-                }
-                DialectItem::Obj(obj) => obj.lineterminator,
-                _ => Terminator::CRLF,
-            };
-
-            if let Some(attr) = self.lineterminator {
-                lineterminator = attr
-            }
-
-            lineterminator
-        }
-
         fn get_quoting(&self) -> QuoteStyle {
             let mut quoting = match &self.dialect {
                 DialectItem::Str(name) => {
@@ -832,11 +848,11 @@ mod _csv {
 
         fn to_reader(&self) -> csv_core::Reader {
             let dialect = match &self.dialect {
-                DialectItem::Str(name) => GLOBAL_HASHMAP.lock().get(name).copied(),
-                DialectItem::Obj(obj) => Some(*obj),
+                DialectItem::Str(name) => GLOBAL_HASHMAP.lock().get(name).cloned(),
+                DialectItem::Obj(obj) => Some(obj.clone()),
                 DialectItem::None => {
                     let g = GLOBAL_HASHMAP.lock();
-                    Some(*g.get("excel").unwrap())
+                    Some(g.get("excel").unwrap().clone())
                 }
             };
 
@@ -868,10 +884,6 @@ mod _csv {
                 reader = reader.quoting(self.quoting != Some(QuoteStyle::None));
             }
 
-            if let Some(t) = self.lineterminator {
-                reader = reader.terminator(t);
-            }
-
             if let Some(t) = self.doublequote {
                 reader = reader.double_quote(t);
             }
@@ -880,7 +892,12 @@ mod _csv {
                 reader = reader.escape(self.escapechar);
             }
 
-            reader = reader.terminator(self.lineterminator.unwrap_or(Terminator::CRLF));
+            // CPython's reader ignores the dialect's `lineterminator` entirely and
+            // only recognizes `\r`, `\n`, and `\r\n` as record separators. Match
+            // that: always use CRLF mode. Feeding a multi-byte terminator's first
+            // byte here would otherwise split records mid-UTF-8 and raise a
+            // UnicodeDecodeError.
+            reader = reader.terminator(Terminator::CRLF);
 
             reader.build()
         }
@@ -893,8 +910,7 @@ mod _csv {
                     if let Some(dialect) = g.get(name) {
                         let mut builder = builder
                             .delimiter(dialect.delimiter)
-                            .double_quote(dialect.doublequote)
-                            .terminator(dialect.lineterminator);
+                            .double_quote(dialect.doublequote);
 
                         if let Some(t) = dialect.quotechar {
                             builder = builder.quote(t);
@@ -910,8 +926,7 @@ mod _csv {
                 DialectItem::Obj(obj) => {
                     let mut builder = builder
                         .delimiter(obj.delimiter)
-                        .double_quote(obj.doublequote)
-                        .terminator(obj.lineterminator);
+                        .double_quote(obj.doublequote);
 
                     if let Some(t) = obj.quotechar {
                         builder = builder.quote(t);
@@ -934,7 +949,7 @@ mod _csv {
                 writer = writer.double_quote(t);
             }
 
-            writer = writer.terminator(self.get_lineterminator());
+            writer = writer.terminator(Terminator::Any(CSV_CORE_TERMINATOR_SENTINEL));
 
             if let Some(e) = self.escapechar {
                 writer = writer.escape(e);
@@ -979,8 +994,8 @@ mod _csv {
         }
 
         #[pygetset]
-        const fn dialect(&self, _vm: &VirtualMachine) -> PyDialect {
-            self.dialect
+        fn dialect(&self, _vm: &VirtualMachine) -> PyDialect {
+            self.dialect.clone()
         }
     }
 
@@ -1014,7 +1029,7 @@ mod _csv {
             &mut self,
             input: &[u8],
             index: usize,
-            dialect: PyDialect,
+            dialect: &PyDialect,
             unquoted_escape: bool,
         ) -> (QuoteScanEvent, usize) {
             let byte = input[index];
@@ -1067,7 +1082,7 @@ mod _csv {
 
     fn read_quote_record(
         input: &[u8],
-        dialect: PyDialect,
+        dialect: &PyDialect,
         field_limit: isize,
         vm: &VirtualMachine,
     ) -> PyResult<Vec<PyObjectRef>> {
@@ -1179,13 +1194,13 @@ mod _csv {
             ) || (zelf.dialect.quoting == QuoteStyle::None
                 && zelf.dialect.escapechar.is_some());
             if use_quote_record {
-                let out = read_quote_record(input, zelf.dialect, field_limit, vm)?;
+                let out = read_quote_record(input, &zelf.dialect, field_limit, vm)?;
                 *line_num += 1;
                 return Ok(PyIterReturn::Return(vm.ctx.new_list(out).into()));
             }
 
             #[inline]
-            fn trim_initial_spaces(input: &[u8], dialect: PyDialect) -> Vec<u8> {
+            fn trim_initial_spaces(input: &[u8], dialect: &PyDialect) -> Vec<u8> {
                 let mut trimmed = Vec::with_capacity(input.len());
                 let mut scan_state = QuoteScanState::new();
                 let mut index = 0;
@@ -1215,7 +1230,7 @@ mod _csv {
             }
 
             let input = if *skipinitialspace {
-                String::from_utf8(trim_initial_spaces(input, zelf.dialect)).unwrap()
+                String::from_utf8(trim_initial_spaces(input, &zelf.dialect)).unwrap()
             } else {
                 String::from_utf8(input.to_vec()).unwrap()
             };
@@ -1314,7 +1329,7 @@ mod _csv {
     fn write_quoted_field(
         output: &mut Vec<u8>,
         data: &[u8],
-        dialect: PyDialect,
+        dialect: &PyDialect,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
         let quotechar = dialect
@@ -1346,7 +1361,7 @@ mod _csv {
     fn write_unquoted_field(
         output: &mut Vec<u8>,
         data: &[u8],
-        dialect: PyDialect,
+        dialect: &PyDialect,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
         for &byte in data {
@@ -1361,36 +1376,38 @@ mod _csv {
         Ok(())
     }
 
-    fn field_needs_quotes(data: &[u8], dialect: PyDialect) -> bool {
+    fn field_needs_quotes(data: &[u8], dialect: &PyDialect) -> bool {
         data.iter().any(|&byte| {
             byte == dialect.delimiter
                 || dialect.quotechar == Some(byte)
                 || matches!(byte, b'\r' | b'\n')
-                || matches!(dialect.lineterminator, Terminator::Any(t) if byte == t)
+                // CPython quotes a field containing any character of the line
+                // terminator. The terminator is ASCII-validated at parse time, so
+                // comparing raw bytes cannot match part of a multi-byte character.
+                // TODO: RUSTPYTHON; supporting non-ASCII terminators needs
+                //       code-point-wise quoting and escaping as part of full
+                //       Unicode dialect support.
+                || dialect.lineterminator.as_bytes().contains(&byte)
         })
     }
 
-    fn field_needs_escape(byte: u8, dialect: PyDialect) -> bool {
+    fn field_needs_escape(byte: u8, dialect: &PyDialect) -> bool {
         byte == dialect.delimiter
             || dialect.quotechar == Some(byte)
             || dialect.escapechar == Some(byte)
             || matches!(byte, b'\r' | b'\n')
-            || matches!(dialect.lineterminator, Terminator::Any(t) if byte == t)
+            || dialect.lineterminator.as_bytes().contains(&byte)
     }
 
-    fn write_lineterminator(output: &mut Vec<u8>, terminator: Terminator) {
-        match terminator {
-            Terminator::CRLF => output.extend_from_slice(b"\r\n"),
-            Terminator::Any(byte) => output.push(byte),
-            _ => unreachable!(),
-        }
+    fn write_lineterminator(output: &mut Vec<u8>, terminator: &str) {
+        output.extend_from_slice(terminator.as_bytes());
     }
 
     #[pyclass(flags(DISALLOW_INSTANTIATION))]
     impl Writer {
         #[pygetset(name = "dialect")]
-        const fn get_dialect(&self, _vm: &VirtualMachine) -> PyDialect {
-            self.dialect
+        fn get_dialect(&self, _vm: &VirtualMachine) -> PyDialect {
+            self.dialect.clone()
         }
 
         fn writerow_quoted_strings(&self, row: PyObjectRef, vm: &VirtualMachine) -> PyResult {
@@ -1421,12 +1438,12 @@ mod _csv {
                 });
 
                 let should_quote = match self.dialect.quoting {
-                    QuoteStyle::Strings => is_str || field_needs_quotes(data, self.dialect),
+                    QuoteStyle::Strings => is_str || field_needs_quotes(data, &self.dialect),
                     QuoteStyle::Notnull => !is_none,
                     _ => unreachable!(),
                 };
                 if should_quote {
-                    write_quoted_field(&mut output, data, self.dialect, vm)?;
+                    write_quoted_field(&mut output, data, &self.dialect, vm)?;
                 } else if single_field && data.is_empty() {
                     return Err(new_csv_error(
                         vm,
@@ -1437,7 +1454,7 @@ mod _csv {
                 }
             }
 
-            write_lineterminator(&mut output, self.dialect.lineterminator);
+            write_lineterminator(&mut output, &self.dialect.lineterminator);
             let s =
                 core::str::from_utf8(&output).map_err(|e| new_not_utf8_error(vm, &output, e))?;
             self.write.call((s,), vm)
@@ -1479,10 +1496,10 @@ mod _csv {
                     ));
                 }
 
-                write_unquoted_field(&mut output, data, self.dialect, vm)?;
+                write_unquoted_field(&mut output, data, &self.dialect, vm)?;
             }
 
-            write_lineterminator(&mut output, self.dialect.lineterminator);
+            write_lineterminator(&mut output, &self.dialect.lineterminator);
 
             let s =
                 core::str::from_utf8(&output).map_err(|e| new_not_utf8_error(vm, &output, e))?;
@@ -1524,14 +1541,14 @@ mod _csv {
                 // 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)?;
+                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);
+            write_lineterminator(&mut output, &self.dialect.lineterminator);
 
             let s =
                 core::str::from_utf8(&output).map_err(|e| new_not_utf8_error(vm, &output, e))?;
@@ -1607,8 +1624,16 @@ mod _csv {
                 handle_res!(writer.terminator(&mut buffer[buffer_offset..]));
             }
 
-            let s = core::str::from_utf8(&buffer[..buffer_offset])
-                .map_err(|e| new_not_utf8_error(vm, &buffer[..buffer_offset], e))?;
+            // csv-core just emitted the single-byte sentinel terminator (after
+            // closing the final quote / emitting an empty record as needed).
+            // Drop that sentinel byte and append the real, possibly
+            // multi-character, line terminator.
+            assert_eq!(buffer[buffer_offset - 1], CSV_CORE_TERMINATOR_SENTINEL);
+            let mut output = buffer[..buffer_offset - 1].to_vec();
+            output.extend_from_slice(self.dialect.lineterminator.as_bytes());
+
+            let s =
+                core::str::from_utf8(&output).map_err(|e| new_not_utf8_error(vm, &output, e))?;
 
             self.write.call((s,), vm)
         }

extra_tests/snippets/stdlib_csv.py

@@ -1,5 +1,6 @@
 import csv
 import io
+import sys
 
 from testutils import assert_raises
 
@@ -181,6 +182,119 @@ def test_quote_minimal_writer_lineterminator():
 test_quote_minimal_writer_lineterminator()
 
 
+def test_multichar_lineterminator():
+    # https://github.com/RustPython/RustPython/issues/8322
+    # The writer must store and emit a full multi-character line terminator.
+    for lineterminator in "\r\n", "\n", "\r", "!@#", "\0":
+        buf = io.StringIO()
+        writer = csv.writer(buf, lineterminator=lineterminator)
+        writer.writerow(["a", "b"])
+        writer.writerow([1, 2])
+        writer.writerow(["\r", "\n"])
+        assert buf.getvalue() == (
+            f'a,b{lineterminator}1,2{lineterminator}"\r","\n"{lineterminator}'
+        ), (lineterminator, buf.getvalue())
+
+    # A field is quoted when it contains any byte of the terminator (QUOTE_MINIMAL).
+    for field, expected in [
+        ("a@b", '"a@b",x!@#'),
+        ("a!b", '"a!b",x!@#'),
+        ("a#b", '"a#b",x!@#'),
+        ("abc", "abc,x!@#"),
+    ]:
+        buf = io.StringIO()
+        csv.writer(buf, lineterminator="!@#").writerow([field, "x"])
+        assert buf.getvalue() == expected, (field, buf.getvalue())
+
+    # The csv-core-backed QUOTE_ALL / QUOTE_NONNUMERIC paths emit the full
+    # terminator too, and keep the state machine correct across rows.
+    allq = io.StringIO()
+    writer = csv.writer(allq, lineterminator="!@#", quoting=csv.QUOTE_ALL)
+    writer.writerow(["a", "b"])
+    writer.writerow(["c", "d"])
+    assert allq.getvalue() == '"a","b"!@#"c","d"!@#', allq.getvalue()
+
+    nonnum = io.StringIO()
+    csv.writer(nonnum, lineterminator="!@#", quoting=csv.QUOTE_NONNUMERIC).writerow(
+        ["a", 1]
+    )
+    assert nonnum.getvalue() == '"a",1!@#', nonnum.getvalue()
+
+    # A field that itself contains a line-break byte must be kept intact: the
+    # csv-core path drops only the trailing record terminator, not a byte from
+    # the field data.
+    embedded = io.StringIO()
+    csv.writer(embedded, lineterminator="!@#", quoting=csv.QUOTE_ALL).writerow(
+        ["x\ny", "z"]
+    )
+    assert embedded.getvalue() == '"x\ny","z"!@#', embedded.getvalue()
+
+    # QUOTE_NONE escapes any byte of the terminator.
+    none = io.StringIO()
+    csv.writer(
+        none, lineterminator="!@#", quoting=csv.QUOTE_NONE, escapechar="\\"
+    ).writerow(["a!b", "x"])
+    assert none.getvalue() == "a\\!b,x!@#", none.getvalue()
+
+    # register_dialect round-trips a multi-character terminator.
+    csv.register_dialect("multichar_lt", delimiter=",", lineterminator="!@#")
+    try:
+        reg = io.StringIO()
+        csv.writer(reg, dialect="multichar_lt").writerow(["a", "b"])
+        assert reg.getvalue() == "a,b!@#", reg.getvalue()
+    finally:
+        csv.unregister_dialect("multichar_lt")
+
+    # The dialect attribute reflects the full terminator.
+    assert (
+        csv.writer(io.StringIO(), lineterminator="!@#").dialect.lineterminator == "!@#"
+    )
+
+    # The reader ignores lineterminator (like CPython) and only splits on \r\n.
+    assert list(csv.reader(io.StringIO("a,b!@#c,d!@#"), lineterminator="!@#")) == [
+        ["a", "b!@#c", "d!@#"]
+    ]
+
+
+test_multichar_lineterminator()
+
+
+def test_reject_non_ascii_lineterminator():
+    # CPython accepts non-ASCII line terminators; RustPython rejects them
+    # because the writer quotes and escapes byte by byte. Supporting them
+    # requires code-point-wise handling as part of full Unicode dialect support.
+    with assert_raises(csv.Error):
+        csv.writer(io.StringIO(), lineterminator="é")
+
+    with assert_raises(csv.Error):
+        csv.writer(io.StringIO(), lineterminator="\x85")
+
+    with assert_raises(csv.Error):
+        csv.writer(io.StringIO(), lineterminator="\ud800")
+
+    with assert_raises(csv.Error):
+        csv.writer(
+            io.StringIO(), lineterminator="é", quoting=csv.QUOTE_NONE, escapechar="\\"
+        )
+
+    with assert_raises(csv.Error):
+        csv.register_dialect("non_ascii_lt", lineterminator="é")
+
+    class NonAsciiDialect(csv.excel):
+        lineterminator = "é"
+
+    with assert_raises(csv.Error):
+        NonAsciiDialect()
+
+    buf = io.StringIO()
+    csv.writer(buf, lineterminator="!@#").writerow(["a", "b"])
+    assert buf.getvalue() == "a,b!@#"
+
+
+if sys.implementation.name == "rustpython":
+    test_reject_non_ascii_lineterminator()
+
+
 def test_quote_minimal_writer_empty_fields():
     buf = io.StringIO()
     writer = csv.writer(buf)