use std::{fmt, ops::Deref};
use crate::{
ExcType, MontyException, MontyObject,
args::{ArgValues, FromArgs, LaxBool, ToArgs, ToMontyObject},
bytecode::VM,
exception_private::RunResult,
heap::{ContainsHeap, DropWithHeap, Heap, HeapData},
intern::{Interns, StaticStrings},
resource::ResourceTracker,
types::{file::FileMode, str::StringRepr},
value::Value,
};
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum OsFunctionCall {
Exists(MontyPath),
IsFile(MontyPath),
IsDir(MontyPath),
IsSymlink(MontyPath),
ReadText(MontyPath),
ReadBytes(MontyPath),
Stat(MontyPath),
Iterdir(MontyPath),
Resolve(MontyPath),
Absolute(MontyPath),
WriteText(PathStringDataArgs),
AppendText(PathStringDataArgs),
WriteBytes(PathBytesDataArgs),
AppendBytes(PathBytesDataArgs),
Open(OpenCallArgs),
Mkdir(MkdirCallArgs),
Unlink(MontyPath),
Rmdir(MontyPath),
Rename(RenameCallArgs),
Getenv(GetenvArgs),
GetEnviron,
DateToday,
DateTimeNow(MontyObject),
Used,
}
impl OsFunctionCall {
#[must_use]
pub fn name(&self) -> &'static str {
match self {
Self::Exists(_) => "Path.exists",
Self::IsFile(_) => "Path.is_file",
Self::IsDir(_) => "Path.is_dir",
Self::IsSymlink(_) => "Path.is_symlink",
Self::Open(_) => "Open",
Self::ReadText(_) => "Path.read_text",
Self::ReadBytes(_) => "Path.read_bytes",
Self::WriteText(_) => "Path.write_text",
Self::WriteBytes(_) => "Path.write_bytes",
Self::AppendText(_) => "Path.append_text",
Self::AppendBytes(_) => "Path.append_bytes",
Self::Mkdir(_) => "Path.mkdir",
Self::Unlink(_) => "Path.unlink",
Self::Rmdir(_) => "Path.rmdir",
Self::Iterdir(_) => "Path.iterdir",
Self::Stat(_) => "Path.stat",
Self::Rename(_) => "Path.rename",
Self::Resolve(_) => "Path.resolve",
Self::Absolute(_) => "Path.absolute",
Self::Getenv(_) => "os.getenv",
Self::GetEnviron => "os.environ",
Self::DateToday => "date.today",
Self::DateTimeNow(_) => "datetime.now",
Self::Used => unreachable!("OsFunctionCall::Used inspected after take_function_call"),
}
}
#[must_use]
pub fn to_args(self) -> (Vec<MontyObject>, Vec<(MontyObject, MontyObject)>) {
match self {
Self::Exists(p)
| Self::IsFile(p)
| Self::IsDir(p)
| Self::IsSymlink(p)
| Self::ReadText(p)
| Self::ReadBytes(p)
| Self::Stat(p)
| Self::Iterdir(p)
| Self::Resolve(p)
| Self::Absolute(p)
| Self::Unlink(p)
| Self::Rmdir(p) => (vec![p.into_monty_object()], vec![]),
Self::WriteText(a) | Self::AppendText(a) => a.to_args(),
Self::WriteBytes(a) | Self::AppendBytes(a) => a.to_args(),
Self::Open(a) => a.to_args(),
Self::Mkdir(a) => a.to_args(),
Self::Rename(a) => a.to_args(),
Self::Getenv(a) => a.to_args(),
Self::GetEnviron | Self::DateToday => (vec![], vec![]),
Self::DateTimeNow(tz) => (vec![tz], vec![]),
Self::Used => unreachable!("OsFunctionCall::Used dispatched after take_function_call"),
}
}
#[must_use]
pub fn is_filesystem(&self) -> bool {
!matches!(
self,
Self::Getenv(_) | Self::GetEnviron | Self::DateToday | Self::DateTimeNow(_)
)
}
#[must_use]
pub fn is_write(&self) -> bool {
match self {
Self::WriteText(_)
| Self::WriteBytes(_)
| Self::AppendText(_)
| Self::AppendBytes(_)
| Self::Mkdir(_)
| Self::Unlink(_)
| Self::Rmdir(_)
| Self::Rename(_) => true,
Self::Open(args) => args.mode.create(),
_ => false,
}
}
#[must_use]
pub fn is_existence_check(&self) -> bool {
matches!(
self,
Self::Exists(_) | Self::IsFile(_) | Self::IsDir(_) | Self::IsSymlink(_)
)
}
#[must_use]
pub fn primary_path(&self) -> Option<&str> {
match self {
Self::Exists(p)
| Self::IsFile(p)
| Self::IsDir(p)
| Self::IsSymlink(p)
| Self::ReadText(p)
| Self::ReadBytes(p)
| Self::Stat(p)
| Self::Iterdir(p)
| Self::Resolve(p)
| Self::Absolute(p)
| Self::Unlink(p)
| Self::Rmdir(p) => Some(p.as_str()),
Self::WriteText(a) | Self::AppendText(a) => Some(a.path.as_str()),
Self::WriteBytes(a) | Self::AppendBytes(a) => Some(a.path.as_str()),
Self::Open(a) => Some(a.path.as_str()),
Self::Mkdir(a) => Some(a.path.as_str()),
Self::Rename(a) => Some(a.src.as_str()),
Self::Getenv(_) | Self::GetEnviron | Self::DateToday | Self::DateTimeNow(_) => None,
Self::Used => unreachable!("OsFunctionCall::Used inspected after take_function_call"),
}
}
#[must_use]
pub fn on_no_handler(&self) -> MontyException {
if self.is_filesystem() {
let path = self.primary_path().unwrap_or("<unknown>");
MontyException::new(
ExcType::PermissionError,
Some(format!("Permission denied: {}", StringRepr(path))),
)
} else {
MontyException::new(
ExcType::RuntimeError,
Some(format!("'{}' is not supported in this environment", self.name())),
)
}
}
}
impl fmt::Display for OsFunctionCall {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.name())
}
}
impl DropWithHeap for OsFunctionCall {
fn drop_with_heap<H: ContainsHeap>(self, _heap: &mut H) {
drop(self);
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct PathStringDataArgs {
pub path: MontyPath,
pub data: String,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct PathBytesDataArgs {
pub path: MontyPath,
pub data: Vec<u8>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct OpenCallArgs {
pub path: MontyPath,
pub mode: FileMode,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct MkdirCallArgs {
pub path: MontyPath,
#[from_args(kw_only)]
pub parents: bool,
#[from_args(kw_only)]
pub exist_ok: bool,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct RenameCallArgs {
pub src: MontyPath,
pub dst: MontyPath,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, monty_macros::ToArgs)]
pub struct GetenvArgs {
pub key: String,
pub default: MontyObject,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct MontyPath(String);
impl MontyPath {
#[must_use]
pub fn new(path: String) -> Self {
Self(path)
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
#[must_use]
pub fn into_string(self) -> String {
self.0
}
}
impl Deref for MontyPath {
type Target = str;
fn deref(&self) -> &str {
&self.0
}
}
impl From<String> for MontyPath {
fn from(s: String) -> Self {
Self(s)
}
}
impl From<&str> for MontyPath {
fn from(s: &str) -> Self {
Self(s.to_owned())
}
}
impl ToMontyObject for MontyPath {
fn into_monty_object(self) -> MontyObject {
MontyObject::Path(self.0)
}
}
#[must_use]
pub(crate) fn is_path_os_method(method: StaticStrings) -> bool {
matches!(
method,
StaticStrings::Exists
| StaticStrings::IsFile
| StaticStrings::IsDir
| StaticStrings::IsSymlink
| StaticStrings::ReadText
| StaticStrings::ReadBytes
| StaticStrings::StatMethod
| StaticStrings::Iterdir
| StaticStrings::Resolve
| StaticStrings::Absolute
| StaticStrings::Unlink
| StaticStrings::Rmdir
| StaticStrings::WriteText
| StaticStrings::AppendText
| StaticStrings::WriteBytes
| StaticStrings::AppendBytes
| StaticStrings::Mkdir
| StaticStrings::Rename
)
}
pub(crate) fn build_path_os_call(
method: StaticStrings,
path: MontyPath,
args: ArgValues,
vm: &mut VM<'_, impl ResourceTracker>,
) -> RunResult<Option<OsFunctionCall>> {
macro_rules! path_only {
($name:literal, $variant:ident) => {{
args.check_zero_args($name, vm.heap)?;
OsFunctionCall::$variant(path)
}};
}
let call = match method {
StaticStrings::Exists => path_only!("exists", Exists),
StaticStrings::IsFile => path_only!("is_file", IsFile),
StaticStrings::IsDir => path_only!("is_dir", IsDir),
StaticStrings::IsSymlink => path_only!("is_symlink", IsSymlink),
StaticStrings::ReadText => path_only!("read_text", ReadText),
StaticStrings::ReadBytes => path_only!("read_bytes", ReadBytes),
StaticStrings::StatMethod => path_only!("stat", Stat),
StaticStrings::Iterdir => path_only!("iterdir", Iterdir),
StaticStrings::Resolve => path_only!("resolve", Resolve),
StaticStrings::Absolute => path_only!("absolute", Absolute),
StaticStrings::Unlink => path_only!("unlink", Unlink),
StaticStrings::Rmdir => path_only!("rmdir", Rmdir),
StaticStrings::WriteText => {
OsFunctionCall::WriteText(extract_str_data("write_text", path, args, vm.heap, vm.interns)?)
}
StaticStrings::AppendText => {
OsFunctionCall::AppendText(extract_str_data("append_text", path, args, vm.heap, vm.interns)?)
}
StaticStrings::WriteBytes => {
OsFunctionCall::WriteBytes(extract_bytes_data("write_bytes", path, args, vm.heap, vm.interns)?)
}
StaticStrings::AppendBytes => {
OsFunctionCall::AppendBytes(extract_bytes_data("append_bytes", path, args, vm.heap, vm.interns)?)
}
StaticStrings::Mkdir => OsFunctionCall::Mkdir(extract_mkdir_args(path, args, vm)?),
StaticStrings::Rename => OsFunctionCall::Rename(extract_rename_args(path, args, vm.heap, vm.interns)?),
_ => {
let _ = path;
args.drop_with_heap(vm.heap);
return Ok(None);
}
};
Ok(Some(call))
}
fn extract_str_data(
method: &'static str,
path: MontyPath,
args: ArgValues,
heap: &mut Heap<impl ResourceTracker>,
interns: &Interns,
) -> RunResult<PathStringDataArgs> {
let data = arg_or_missing_data(method, args, heap)?;
let data_str = value_to_owned_string(&data, heap, interns);
let py_type = data.py_type_name_heap(heap, interns);
data.drop_with_heap(heap);
match data_str {
Some(data) => Ok(PathStringDataArgs { path, data }),
None => Err(ExcType::type_error(format!("data must be str, not {py_type}"))),
}
}
fn extract_bytes_data(
method: &'static str,
path: MontyPath,
args: ArgValues,
heap: &mut Heap<impl ResourceTracker>,
interns: &Interns,
) -> RunResult<PathBytesDataArgs> {
let data = arg_or_missing_data(method, args, heap)?;
let bytes = value_to_owned_bytes(&data, heap, interns);
let py_type = data.py_type_name_heap(heap, interns);
data.drop_with_heap(heap);
match bytes {
Some(data) => Ok(PathBytesDataArgs { path, data }),
None => Err(ExcType::type_error(format!(
"memoryview: a bytes-like object is required, not '{py_type}'"
))),
}
}
#[derive(FromArgs)]
#[from_args(name = "Path.mkdir", style = def)]
struct PathMkdirArgs {
#[from_args(default = 0o777_i64)]
mode: i64,
#[from_args(default = LaxBool::new(false))]
parents: LaxBool,
#[from_args(default = LaxBool::new(false))]
exist_ok: LaxBool,
}
fn extract_mkdir_args(
path: MontyPath,
args: ArgValues,
vm: &mut VM<'_, impl ResourceTracker>,
) -> RunResult<MkdirCallArgs> {
let PathMkdirArgs {
mode,
parents,
exist_ok,
} = PathMkdirArgs::from_args(args, vm)?;
let _ = mode;
Ok(MkdirCallArgs {
path,
parents: parents.bool(),
exist_ok: exist_ok.bool(),
})
}
fn extract_rename_args(
src: MontyPath,
args: ArgValues,
heap: &mut Heap<impl ResourceTracker>,
interns: &Interns,
) -> RunResult<RenameCallArgs> {
let target = args.get_one_arg("rename", heap)?;
let dst_str = value_to_owned_string(&target, heap, interns);
target.drop_with_heap(heap);
match dst_str {
Some(dst) => Ok(RenameCallArgs {
src,
dst: MontyPath::new(dst),
}),
None => Err(ExcType::type_error(
"Path.rename() argument 'target' must be str or Path".to_owned(),
)),
}
}
fn arg_or_missing_data(
method: &'static str,
args: ArgValues,
heap: &mut Heap<impl ResourceTracker>,
) -> RunResult<Value> {
if matches!(args, ArgValues::Empty) {
return Err(ExcType::type_error(format!(
"Path.{method}() missing 1 required positional argument: 'data'"
)));
}
args.get_one_arg(method, heap)
}
fn value_to_owned_string(value: &Value, heap: &Heap<impl ResourceTracker>, interns: &Interns) -> Option<String> {
match value {
Value::InternString(id) => Some(interns.get_str(*id).to_owned()),
Value::Ref(id) => match heap.get(*id) {
HeapData::Str(s) => Some(s.as_str().to_owned()),
HeapData::Path(p) => Some(p.as_str().to_owned()),
_ => None,
},
_ => None,
}
}
fn value_to_owned_bytes(value: &Value, heap: &Heap<impl ResourceTracker>, interns: &Interns) -> Option<Vec<u8>> {
match value {
Value::InternBytes(id) => Some(interns.get_bytes(*id).to_owned()),
Value::Ref(id) => match heap.get(*id) {
HeapData::Bytes(b) => Some(b.as_slice().to_owned()),
_ => None,
},
_ => None,
}
}
#[must_use]
pub fn file_stat(mode: i64, size: i64, mtime: f64) -> MontyObject {
let mode = if mode < 0o1000 { mode | 0o100_000 } else { mode };
stat_result(mode, 0, 0, 1, 0, 0, size, mtime, mtime, mtime)
}
#[must_use]
pub fn dir_stat(mode: i64, mtime: f64) -> MontyObject {
let mode = if mode < 0o1000 { mode | 0o040_000 } else { mode };
stat_result(mode, 0, 0, 2, 0, 0, 4096, mtime, mtime, mtime)
}
#[must_use]
pub fn symlink_stat(mode: i64, mtime: f64) -> MontyObject {
let mode = if mode < 0o1000 { mode | 0o120_000 } else { mode };
stat_result(mode, 0, 0, 1, 0, 0, 0, mtime, mtime, mtime)
}
#[must_use]
#[expect(clippy::too_many_arguments)]
pub fn stat_result(
st_mode: i64,
st_ino: i64,
st_dev: i64,
st_nlink: i64,
st_uid: i64,
st_gid: i64,
st_size: i64,
st_atime: f64,
st_mtime: f64,
st_ctime: f64,
) -> MontyObject {
MontyObject::NamedTuple {
type_name: STAT_RESULT_TYPE_NAME.to_owned(),
field_names: STAT_RESULT_FIELDS.iter().map(|s| (*s).to_owned()).collect(),
values: vec![
MontyObject::Int(st_mode),
MontyObject::Int(st_ino),
MontyObject::Int(st_dev),
MontyObject::Int(st_nlink),
MontyObject::Int(st_uid),
MontyObject::Int(st_gid),
MontyObject::Int(st_size),
MontyObject::Float(st_atime),
MontyObject::Float(st_mtime),
MontyObject::Float(st_ctime),
],
}
}
const STAT_RESULT_TYPE_NAME: &str = "StatResult";
const STAT_RESULT_FIELDS: &[&str] = &[
"st_mode", "st_ino", "st_dev", "st_nlink", "st_uid", "st_gid", "st_size", "st_atime", "st_mtime", "st_ctime",
];