Rust-related/RustPython
1
0
Fork 1
mirror of https://github.com/RustPython/RustPython.git synced 2026-06-02 19:39:49 +09:00
Code Issues Packages Projects Releases Wiki Activity

impl more msvcrt

Browse Source
This commit is contained in:
Jeong, YunWon
2026-01-24 12:30:06 +09:00
committed by Jeong YunWon
parent 1251fbf0ba
commit 26d64b9fda
3 changed files with 158 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
Lib/test/test_msvcrt.py vendored
View File

@@ -15,7 +15,6 @@ import msvcrt
class TestFileOperations(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'msvcrt' has no attribute 'locking'
def test_locking(self):
with open(TESTFN, "w") as f:
self.addCleanup(os_helper.unlink, TESTFN)
@@ -23,7 +22,6 @@ class TestFileOperations(unittest.TestCase):
msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
self.assertRaises(OSError, msvcrt.locking, f.fileno(), msvcrt.LK_NBLCK, 1)
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'msvcrt' has no attribute 'locking'
def test_unlockfile(self):
with open(TESTFN, "w") as f:
self.addCleanup(os_helper.unlink, TESTFN)
@@ -39,7 +37,6 @@ class TestFileOperations(unittest.TestCase):
msvcrt.setmode(f.fileno(), os.O_BINARY)
msvcrt.setmode(f.fileno(), os.O_TEXT)
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module '_winapi' has no attribute 'CreateFile'. Did you mean: 'CreatePipe'?
def test_open_osfhandle(self):
h = _winapi.CreateFile(TESTFN_ASCII, _winapi.GENERIC_WRITE, 0, 0, 1, 128, 0)
self.addCleanup(os_helper.unlink, TESTFN_ASCII)
@@ -80,7 +77,6 @@ class TestConsoleIO(unittest.TestCase):
''')
self.run_in_separated_process(code)
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'msvcrt' has no attribute 'ungetch'. Did you mean: 'getch'?
def test_getch(self):
msvcrt.ungetch(b'c')
self.assertEqual(msvcrt.getch(), b'c')
@@ -98,7 +94,6 @@ class TestConsoleIO(unittest.TestCase):
def test_getwch(self):
self.check_getwch('getwch')
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'msvcrt' has no attribute 'ungetch'. Did you mean: 'getch'?
def test_getche(self):
msvcrt.ungetch(b'c')
self.assertEqual(msvcrt.getche(), b'c')
@@ -114,7 +109,6 @@ class TestConsoleIO(unittest.TestCase):
class TestOther(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: module 'msvcrt' has no attribute 'heapmin'
def test_heap_min(self):
try:
msvcrt.heapmin()

71
crates/vm/src/stdlib/msvcrt.rs
View File

@@ -31,8 +31,25 @@ mod msvcrt {
fn _getwche() -> u32;
fn _putch(c: u32) -> i32;
fn _putwch(c: u16) -> u32;
fn _ungetch(c: i32) -> i32;
fn _ungetwch(c: u32) -> u32;
fn _locking(fd: i32, mode: i32, nbytes: i64) -> i32;
fn _heapmin() -> i32;
fn _kbhit() -> i32;
}
// Locking mode constants
#[pyattr]
const LK_UNLCK: i32 = 0; // Unlock
#[pyattr]
const LK_LOCK: i32 = 1; // Lock (blocking)
#[pyattr]
const LK_NBLCK: i32 = 2; // Non-blocking lock
#[pyattr]
const LK_RLCK: i32 = 3; // Lock for reading (same as LK_LOCK)
#[pyattr]
const LK_NBRLCK: i32 = 4; // Non-blocking lock for reading (same as LK_NBLCK)
#[pyfunction]
fn getch() -> Vec<u8> {
let c = unsafe { _getch() };
@@ -73,6 +90,60 @@ mod msvcrt {
Ok(())
}
#[pyfunction]
fn ungetch(b: PyRef<PyBytes>, vm: &VirtualMachine) -> PyResult<()> {
let &c = b.as_bytes().iter().exactly_one().map_err(|_| {
vm.new_type_error("ungetch() argument must be a byte string of length 1")
})?;
let ret = unsafe { suppress_iph!(_ungetch(c as i32)) };
if ret == -1 {
// EOF returned means the buffer is full
Err(vm.new_os_error(libc::ENOSPC))
} else {
Ok(())
}
}
#[pyfunction]
fn ungetwch(s: PyStrRef, vm: &VirtualMachine) -> PyResult<()> {
let c =
s.as_str().chars().exactly_one().map_err(|_| {
vm.new_type_error("ungetwch() argument must be a string of length 1")
})?;
let ret = unsafe { suppress_iph!(_ungetwch(c as u32)) };
if ret == 0xFFFF {
// WEOF returned means the buffer is full
Err(vm.new_os_error(libc::ENOSPC))
} else {
Ok(())
}
}
#[pyfunction]
fn kbhit() -> i32 {
unsafe { _kbhit() }
}
#[pyfunction]
fn locking(fd: i32, mode: i32, nbytes: i64, vm: &VirtualMachine) -> PyResult<()> {
let ret = unsafe { suppress_iph!(_locking(fd, mode, nbytes)) };
if ret == -1 {
Err(vm.new_last_errno_error())
} else {
Ok(())
}
}
#[pyfunction]
fn heapmin(vm: &VirtualMachine) -> PyResult<()> {
let ret = unsafe { suppress_iph!(_heapmin()) };
if ret == -1 {
Err(vm.new_last_errno_error())
} else {
Ok(())
}
}
unsafe extern "C" {
fn _setmode(fd: crt_fd::Borrowed<'_>, flags: i32) -> i32;
}

100
crates/vm/src/stdlib/winapi.rs
View File

@@ -31,19 +31,57 @@ mod _winapi {
LCMAP_TRADITIONAL_CHINESE, LCMAP_UPPERCASE,
},
Storage::FileSystem::{
COPY_FILE_ALLOW_DECRYPTED_DESTINATION, COPY_FILE_COPY_SYMLINK,
COPY_FILE_FAIL_IF_EXISTS, COPY_FILE_NO_BUFFERING, COPY_FILE_NO_OFFLOAD,
COPY_FILE_OPEN_SOURCE_FOR_WRITE, COPY_FILE_REQUEST_COMPRESSED_TRAFFIC,
COPY_FILE_REQUEST_SECURITY_PRIVILEGES, COPY_FILE_RESTARTABLE,
COPY_FILE_RESUME_FROM_PAUSE, COPYFILE2_CALLBACK_CHUNK_FINISHED,
COPYFILE2_CALLBACK_CHUNK_STARTED, COPYFILE2_CALLBACK_ERROR,
COPYFILE2_CALLBACK_POLL_CONTINUE, COPYFILE2_CALLBACK_STREAM_FINISHED,
COPYFILE2_CALLBACK_STREAM_STARTED, COPYFILE2_PROGRESS_CANCEL,
COPYFILE2_PROGRESS_CONTINUE, COPYFILE2_PROGRESS_PAUSE, COPYFILE2_PROGRESS_QUIET,
COPYFILE2_PROGRESS_STOP, FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED,
FILE_GENERIC_READ, FILE_GENERIC_WRITE, FILE_TYPE_CHAR, FILE_TYPE_DISK, FILE_TYPE_PIPE,
FILE_TYPE_REMOTE, FILE_TYPE_UNKNOWN, OPEN_EXISTING, PIPE_ACCESS_DUPLEX,
PIPE_ACCESS_INBOUND, SYNCHRONIZE,
COPY_FILE_ALLOW_DECRYPTED_DESTINATION,
COPY_FILE_COPY_SYMLINK,
COPY_FILE_FAIL_IF_EXISTS,
COPY_FILE_NO_BUFFERING,
COPY_FILE_NO_OFFLOAD,
COPY_FILE_OPEN_SOURCE_FOR_WRITE,
COPY_FILE_REQUEST_COMPRESSED_TRAFFIC,
COPY_FILE_REQUEST_SECURITY_PRIVILEGES,
COPY_FILE_RESTARTABLE,
COPY_FILE_RESUME_FROM_PAUSE,
COPYFILE2_CALLBACK_CHUNK_FINISHED,
COPYFILE2_CALLBACK_CHUNK_STARTED,
COPYFILE2_CALLBACK_ERROR,
COPYFILE2_CALLBACK_POLL_CONTINUE,
COPYFILE2_CALLBACK_STREAM_FINISHED,
COPYFILE2_CALLBACK_STREAM_STARTED,
COPYFILE2_PROGRESS_CANCEL,
COPYFILE2_PROGRESS_CONTINUE,
COPYFILE2_PROGRESS_PAUSE,
COPYFILE2_PROGRESS_QUIET,
COPYFILE2_PROGRESS_STOP,
CREATE_ALWAYS,
// CreateFile constants
CREATE_NEW,
FILE_ATTRIBUTE_NORMAL,
FILE_FLAG_BACKUP_SEMANTICS,
FILE_FLAG_DELETE_ON_CLOSE,
FILE_FLAG_FIRST_PIPE_INSTANCE,
FILE_FLAG_NO_BUFFERING,
FILE_FLAG_OPEN_REPARSE_POINT,
FILE_FLAG_OVERLAPPED,
FILE_FLAG_POSIX_SEMANTICS,
FILE_FLAG_RANDOM_ACCESS,
FILE_FLAG_SEQUENTIAL_SCAN,
FILE_FLAG_WRITE_THROUGH,
FILE_GENERIC_READ,
FILE_GENERIC_WRITE,
FILE_SHARE_DELETE,
FILE_SHARE_READ,
FILE_SHARE_WRITE,
FILE_TYPE_CHAR,
FILE_TYPE_DISK,
FILE_TYPE_PIPE,
FILE_TYPE_REMOTE,
FILE_TYPE_UNKNOWN,
OPEN_ALWAYS,
OPEN_EXISTING,
PIPE_ACCESS_DUPLEX,
PIPE_ACCESS_INBOUND,
SYNCHRONIZE,
TRUNCATE_EXISTING,
},
System::{
Console::{STD_ERROR_HANDLE, STD_INPUT_HANDLE, STD_OUTPUT_HANDLE},
@@ -78,6 +116,42 @@ mod _winapi {
WindowsSysResult(unsafe { windows_sys::Win32::Foundation::CloseHandle(handle.0) })
}
/// CreateFile - Create or open a file or I/O device.
#[pyfunction]
#[allow(clippy::too_many_arguments)]
fn CreateFile(
file_name: PyStrRef,
desired_access: u32,
share_mode: u32,
_security_attributes: PyObjectRef, // Always NULL (0)
creation_disposition: u32,
flags_and_attributes: u32,
_template_file: PyObjectRef, // Always NULL (0)
vm: &VirtualMachine,
) -> PyResult<WinHandle> {
use windows_sys::Win32::Storage::FileSystem::CreateFileW;
let file_name_wide = file_name.as_wtf8().to_wide_with_nul();
let handle = unsafe {
CreateFileW(
file_name_wide.as_ptr(),
desired_access,
share_mode,
null(),
creation_disposition,
flags_and_attributes,
null_mut(),
)
};
if handle == INVALID_HANDLE_VALUE {
return Err(vm.new_last_os_error());
}
Ok(WinHandle(handle))
}
#[pyfunction]
fn GetStdHandle(
std_handle: windows_sys::Win32::System::Console::STD_HANDLE,