monty 0.0.21

A sandboxed, snapshotable Python interpreter written in Rust.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227

//! Coverage for buffered file I/O error paths that exercise host-side failures.
//!
//! These paths cannot easily be triggered through `mount-fs` test_cases because
//! the in-process `MountTable` does not raise mid-read; instead we drive the
//! VM directly so we can resume an OS call with an exception (or with an
//! interned-string return value that hits the `InternBytes`/empty-string
//! materialisation branch of [`apply_buffer_store`]).

use monty::MontyRun;
use monty_types::{
    CompileOptions, ExcType, ExtFunctionResult, FileMode, MontyException, MontyFileHandle, MontyObject, PrintWriter,
    ResourceTracker,
};

/// Drives an `open()` followed by a single read/write OS call, then resumes
/// the second call with the caller-provided result. Returns the final
/// `Complete` value. Both OS calls are matched against the expected function
/// names (compared by stable [`OsFunctionCall::name`] string) so a regression
/// in dispatch fails the test loudly.
fn run_with_open_then_io(
    code: &str,
    expected_io_fn_name: &str,
    file_handle: MontyFileHandle,
    io_result: ExtFunctionResult,
) -> Result<MontyObject, MontyException> {
    let runner = MontyRun::new(code.to_owned(), "test.py", vec![], CompileOptions::default()).unwrap();
    let progress = runner
        .start(vec![], ResourceTracker::default(), PrintWriter::Stdout)
        .unwrap();
    let open_call = progress.into_os_call().expect("expected Open OsCall");
    assert_eq!(open_call.function_call.name(), "open");
    let progress = open_call
        .resume(MontyObject::FileHandle(file_handle), PrintWriter::Stdout)
        .unwrap();
    let io_call = progress.into_os_call().expect("expected follow-up OsCall");
    assert_eq!(io_call.function_call.name(), expected_io_fn_name);
    let progress = io_call.resume(io_result, PrintWriter::Stdout)?;
    Ok(progress.into_complete().expect("expected Complete after resume"))
}

fn file_handle(path: &str, mode: &str) -> MontyFileHandle {
    MontyFileHandle {
        path: path.to_owned(),
        mode: mode.parse::<FileMode>().unwrap(),
        position: 0,
    }
}

// ---------------------------------------------------------------------------
// Host-side read failure: VM must clear `pending_read` so a retry sees a
// fresh slot, and the user-visible exception must surface unchanged.
// ---------------------------------------------------------------------------

#[test]
fn read_host_error_clears_pending_read_and_propagates() {
    let code = r"
f = open('/x.txt')
try:
    f.read(5)
    result = 'no-error'
except OSError as exc:
    result = str(exc)
result
";
    let host_exc = MontyException::new(ExcType::OSError, Some("disk on fire".to_owned()));
    let result = run_with_open_then_io(code, "Path.read_text", file_handle("/x.txt", "r"), host_exc.into())
        .expect("script should complete after catching the host error");
    assert_eq!(result, MontyObject::String("disk on fire".to_owned()));
}

#[test]
fn readline_host_error_then_retry() {
    // After a failed buffered read the file's `pending_read` slot must be
    // cleared so a subsequent successful read replays the OS call. The
    // second OS call is observed here as a new `OsCall` rather than a
    // misrouted resume.
    let code = r"
f = open('/x.txt')
try:
    f.readline()
except OSError:
    pass
f.readline()
";
    let runner = MontyRun::new(code.to_owned(), "test.py", vec![], CompileOptions::default()).unwrap();
    let progress = runner
        .start(vec![], ResourceTracker::default(), PrintWriter::Stdout)
        .unwrap();
    let open_call = progress.into_os_call().expect("expected Open OsCall");
    let progress = open_call
        .resume(MontyObject::FileHandle(file_handle("/x.txt", "r")), PrintWriter::Stdout)
        .unwrap();
    let read_call = progress.into_os_call().expect("expected ReadText OsCall");
    assert_eq!(read_call.function_call.name(), "Path.read_text");
    // Fail the first ReadText.
    let host_exc = MontyException::new(ExcType::OSError, Some("transient".to_owned()));
    let progress = read_call
        .resume(ExtFunctionResult::Error(host_exc), PrintWriter::Stdout)
        .unwrap();
    // Retry should issue a second ReadText now that pending_read was cleared.
    let retry_call = progress.into_os_call().expect("expected retry OsCall");
    assert_eq!(retry_call.function_call.name(), "Path.read_text");
    let final_progress = retry_call
        .resume(MontyObject::String("alpha\nbeta\n".to_owned()), PrintWriter::Stdout)
        .unwrap();
    let result = final_progress.into_complete().expect("expected Complete");
    assert_eq!(result, MontyObject::String("alpha\n".to_owned()));
}

// ---------------------------------------------------------------------------
// Host-side write failure: VM must roll back `position` / `file_length`,
// leaving the file's logical `tell()` exactly where it was before the call.
// ---------------------------------------------------------------------------

#[test]
fn write_host_error_rolls_back_position() {
    let code = r"
f = open('/x.txt', 'w')
try:
    f.write('abc')
except OSError as exc:
    after = (f.tell(), str(exc))
after
";
    let host_exc = MontyException::new(ExcType::OSError, Some("device full".to_owned()));
    // `w` mode pre-truncates at open time, so the first user write actually
    // emits AppendText (see `OpenFile::with_state` setting `first_write_done`).
    let result = run_with_open_then_io(code, "Path.append_text", file_handle("/x.txt", "w"), host_exc.into())
        .expect("script should complete after catching the host error");
    // tell() must read 0 — the failed write was rolled back.
    assert_eq!(
        result,
        MontyObject::Tuple(vec![MontyObject::Int(0), MontyObject::String("device full".to_owned()),])
    );
}

#[test]
fn write_host_error_binary_rolls_back_position() {
    let code = r"
f = open('/x.bin', 'wb')
try:
    f.write(b'\x00\x01\x02')
except OSError:
    pass
f.tell()
";
    let host_exc = MontyException::new(ExcType::OSError, Some("io".to_owned()));
    let result = run_with_open_then_io(code, "Path.append_bytes", file_handle("/x.bin", "wb"), host_exc.into())
        .expect("script should complete");
    assert_eq!(result, MontyObject::Int(0));
}

// ---------------------------------------------------------------------------
// apply_buffer_store covers two materialisation branches: `Value::Ref`
// (heap-allocated content) and `Value::InternString` / `Value::InternBytes`
// (small/empty strings that get interned at the host boundary). The empty
// `""` text case and the small `b""` binary case both go through the
// intern branch.
// ---------------------------------------------------------------------------

#[test]
fn buffer_store_handles_empty_text_via_intern() {
    let code = r"
f = open('/empty.txt')
f.read(5)
";
    let result = run_with_open_then_io(
        code,
        "Path.read_text",
        file_handle("/empty.txt", "r"),
        MontyObject::String(String::new()).into(),
    )
    .expect("script should complete");
    assert_eq!(result, MontyObject::String(String::new()));
}

#[test]
fn buffer_store_handles_empty_bytes_via_intern() {
    let code = r"
f = open('/empty.bin', 'rb')
f.read(5)
";
    let result = run_with_open_then_io(
        code,
        "Path.read_bytes",
        file_handle("/empty.bin", "rb"),
        MontyObject::Bytes(Vec::new()).into(),
    )
    .expect("script should complete");
    assert_eq!(result, MontyObject::Bytes(Vec::new()));
}

// ---------------------------------------------------------------------------
// apply_write_position: a host that returns a non-int (or negative) byte
// count must surface as a clean Python error rather than panic. The
// surfaced error from `as_int(...)?` is a TypeError; the negative-int
// branch raises an internal RuntimeError. We assert on both.
// ---------------------------------------------------------------------------

#[test]
fn write_position_non_int_result_raises_type_error() {
    let code = r"
f = open('/x.txt', 'w')
try:
    f.write('abc')
    result = 'no-error'
except TypeError as exc:
    result = ('type', str(exc))
result
";
    // Host returns a string for the byte count — should raise TypeError.
    let result = run_with_open_then_io(
        code,
        "Path.append_text",
        file_handle("/x.txt", "w"),
        MontyObject::String("oops".to_owned()).into(),
    )
    .expect("script should complete");
    // The exact message comes from `as_int`; we just verify the tag.
    match result {
        MontyObject::Tuple(items) => {
            assert_eq!(items.len(), 2);
            assert_eq!(items[0], MontyObject::String("type".to_owned()));
        }
        other => panic!("expected ('type', msg) tuple, got {other:?}"),
    }
}