(path: crate::PyObjectRef, vm: &VirtualMachine)
| 1465 | |
| 1466 | #[pyfunction] |
| 1467 | fn _path_splitroot_ex(path: crate::PyObjectRef, vm: &VirtualMachine) -> PyResult<PyTupleRef> { |
| 1468 | // Handle path-like objects via os.fspath, but without null check (non_strict=True) |
| 1469 | let path = if let Some(fspath) = vm.get_method(path.clone(), identifier!(vm, __fspath__)) { |
| 1470 | fspath?.call((), vm)? |
| 1471 | } else { |
| 1472 | path |
| 1473 | }; |
| 1474 | |
| 1475 | // Convert to wide string, validating UTF-8 for bytes input |
| 1476 | let (wide, is_bytes): (Vec<u16>, bool) = if let Some(s) = path.downcast_ref::<PyStr>() { |
| 1477 | // Use encode_wide which handles WTF-8 (including surrogates) |
| 1478 | let wide: Vec<u16> = s.as_wtf8().encode_wide().collect(); |
| 1479 | (wide, false) |
| 1480 | } else if let Some(b) = path.downcast_ref::<PyBytes>() { |
| 1481 | // On Windows, bytes must be valid UTF-8 - this raises UnicodeDecodeError if not |
| 1482 | let s = core::str::from_utf8(b.as_bytes()).map_err(|e| { |
| 1483 | vm.new_exception_msg( |
| 1484 | vm.ctx.exceptions.unicode_decode_error.to_owned(), |
| 1485 | format!( |
| 1486 | "'utf-8' codec can't decode byte {:#x} in position {}: invalid start byte", |
| 1487 | b.as_bytes().get(e.valid_up_to()).copied().unwrap_or(0), |
| 1488 | e.valid_up_to() |
| 1489 | ) |
| 1490 | .into(), |
| 1491 | ) |
| 1492 | })?; |
| 1493 | let wide: Vec<u16> = s.encode_utf16().collect(); |
| 1494 | (wide, true) |
| 1495 | } else { |
| 1496 | return Err(vm.new_type_error(format!( |
| 1497 | "expected str or bytes, not {}", |
| 1498 | path.class().name() |
| 1499 | ))); |
| 1500 | }; |
| 1501 | |
| 1502 | // Normalize slashes for parsing |
| 1503 | let normalized: Vec<u16> = wide |
| 1504 | .iter() |
| 1505 | .map(|&c| if c == b'/' as u16 { b'\\' as u16 } else { c }) |
| 1506 | .collect(); |
| 1507 | |
| 1508 | let (drv_size, root_size) = skiproot(&normalized); |
| 1509 | |
| 1510 | // Return as bytes if input was bytes, preserving the original content |
| 1511 | if is_bytes { |
| 1512 | // Convert UTF-16 back to UTF-8 for bytes output |
| 1513 | let drv = String::from_utf16(&wide[..drv_size]) |
| 1514 | .map_err(|e| vm.new_unicode_decode_error(e.to_string()))?; |
| 1515 | let root = String::from_utf16(&wide[drv_size..drv_size + root_size]) |
| 1516 | .map_err(|e| vm.new_unicode_decode_error(e.to_string()))?; |
| 1517 | let tail = String::from_utf16(&wide[drv_size + root_size..]) |
| 1518 | .map_err(|e| vm.new_unicode_decode_error(e.to_string()))?; |
| 1519 | Ok(vm.ctx.new_tuple(vec![ |
| 1520 | vm.ctx.new_bytes(drv.into_bytes()).into(), |
| 1521 | vm.ctx.new_bytes(root.into_bytes()).into(), |
| 1522 | vm.ctx.new_bytes(tail.into_bytes()).into(), |
| 1523 | ])) |
| 1524 | } else { |
nothing calls this directly
no test coverage detected