(obj: &PyObject, size: Option<usize>, vm: &VirtualMachine)
| 62 | } |
| 63 | |
| 64 | pub fn file_readline(obj: &PyObject, size: Option<usize>, vm: &VirtualMachine) -> PyResult { |
| 65 | let args = size.map_or_else(Vec::new, |size| vec![vm.ctx.new_int(size).into()]); |
| 66 | let ret = vm.call_method(obj, "readline", args)?; |
| 67 | let eof_err = || { |
| 68 | vm.new_exception( |
| 69 | vm.ctx.exceptions.eof_error.to_owned(), |
| 70 | vec![vm.ctx.new_str(ascii!("EOF when reading a line")).into()], |
| 71 | ) |
| 72 | }; |
| 73 | let ret = match_class!(match ret { |
| 74 | s @ PyStr => { |
| 75 | // Use as_wtf8() to handle strings with surrogates (e.g., surrogateescape) |
| 76 | let s_wtf8 = s.as_wtf8(); |
| 77 | if s_wtf8.is_empty() { |
| 78 | return Err(eof_err()); |
| 79 | } |
| 80 | // '\n' is ASCII, so we can check bytes directly |
| 81 | if s_wtf8.as_bytes().last() == Some(&b'\n') { |
| 82 | let no_nl = &s_wtf8[..s_wtf8.len() - 1]; |
| 83 | vm.ctx.new_str(no_nl).into() |
| 84 | } else { |
| 85 | s.into() |
| 86 | } |
| 87 | } |
| 88 | b @ PyBytes => { |
| 89 | let buf = b.as_bytes(); |
| 90 | if buf.is_empty() { |
| 91 | return Err(eof_err()); |
| 92 | } |
| 93 | if buf.last() == Some(&b'\n') { |
| 94 | vm.ctx.new_bytes(buf[..buf.len() - 1].to_owned()).into() |
| 95 | } else { |
| 96 | b.into() |
| 97 | } |
| 98 | } |
| 99 | _ => return Err(vm.new_type_error("object.readline() returned non-string".to_owned())), |
| 100 | }); |
| 101 | Ok(ret) |
| 102 | } |
no test coverage detected