use std::{
error,
fmt::{self, Write},
mem, str,
sync::Arc,
};
use serde::{Deserialize, Serialize};
use strum::{Display, EnumString, IntoStaticStr};
use crate::format::StringRepr;
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct MontyException {
exc_type: ExcType,
message: Option<String>,
traceback: Vec<StackFrame>,
#[serde(default)]
data: ExcData,
}
const REPEAT_FRAMES_SHOWN: usize = 3;
impl fmt::Display for MontyException {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if !self.traceback.is_empty() {
writeln!(f, "Traceback (most recent call last):")?;
}
let mut i = 0;
while i < self.traceback.len() {
let frame = &self.traceback[i];
let mut repeat_count = 1;
while i + repeat_count < self.traceback.len()
&& frames_are_identical(frame, &self.traceback[i + repeat_count])
{
repeat_count += 1;
}
if repeat_count > REPEAT_FRAMES_SHOWN {
for j in 0..REPEAT_FRAMES_SHOWN {
write!(f, "{}", self.traceback[i + j])?;
}
let collapsed = repeat_count - REPEAT_FRAMES_SHOWN;
writeln!(f, " [Previous line repeated {collapsed} more times]")?;
i += repeat_count;
} else {
for j in 0..repeat_count {
write!(f, "{}", self.traceback[i + j])?;
}
i += repeat_count;
}
}
if let Some(msg) = &self.message {
write!(f, "{}: {}", self.exc_type, msg)
} else {
write!(f, "{}", self.exc_type)
}
}
}
impl error::Error for MontyException {}
impl MontyException {
#[must_use]
pub fn new(exc_type: ExcType, message: Option<String>) -> Self {
Self {
exc_type,
message,
traceback: vec![],
data: ExcData::None,
}
}
#[must_use]
pub fn with_traceback(exc_type: ExcType, message: Option<String>, traceback: Vec<StackFrame>) -> Self {
Self {
exc_type,
message,
traceback,
data: ExcData::None,
}
}
#[must_use]
pub fn with_data(mut self, data: ExcData) -> Self {
self.data = data;
self
}
#[must_use]
pub fn data(&self) -> &ExcData {
&self.data
}
#[must_use]
pub fn unicode_data(&self) -> Option<&UnicodeErrorData> {
self.data.unicode()
}
#[must_use]
pub fn json_data(&self) -> Option<&JsonErrorData> {
self.data.json()
}
#[must_use]
pub fn take_data(&mut self) -> ExcData {
mem::take(&mut self.data)
}
pub fn add_traceback(&mut self, traceback: impl IntoIterator<Item = StackFrame>) {
self.traceback.extend(traceback);
}
#[must_use]
pub fn runtime_error(err: impl fmt::Display) -> Self {
Self {
exc_type: ExcType::RuntimeError,
message: Some(err.to_string()),
traceback: vec![],
data: ExcData::None,
}
}
#[must_use]
pub fn exc_type(&self) -> ExcType {
self.exc_type
}
#[must_use]
pub fn message(&self) -> Option<&str> {
self.message.as_deref()
}
#[must_use]
pub fn into_message(self) -> Option<String> {
self.message
}
#[must_use]
pub fn traceback(&self) -> &[StackFrame] {
&self.traceback
}
#[must_use]
pub fn summary(&self) -> String {
if let Some(msg) = &self.message {
format!("{}: {}", self.exc_type, msg)
} else {
self.exc_type.to_string()
}
}
#[must_use]
pub fn py_repr(&self) -> String {
let type_str: &'static str = self.exc_type.into();
if let Some(msg) = &self.message {
format!("{}({})", type_str, StringRepr(msg))
} else {
format!("{type_str}()")
}
}
}
fn frames_are_identical(a: &StackFrame, b: &StackFrame) -> bool {
a.filename == b.filename && a.start.line == b.start.line && a.frame_name == b.frame_name
}
#[derive(
Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Display, EnumString, IntoStaticStr, Serialize, Deserialize,
)]
pub enum ExcType {
#[default]
Exception,
BaseException,
SystemExit,
KeyboardInterrupt,
ArithmeticError,
OverflowError,
ZeroDivisionError,
LookupError,
IndexError,
KeyError,
RuntimeError,
NotImplementedError,
RecursionError,
AttributeError,
FrozenInstanceError,
NameError,
UnboundLocalError,
ValueError,
UnicodeDecodeError,
UnicodeEncodeError,
#[strum(serialize = "json.JSONDecodeError")]
JsonDecodeError,
ImportError,
ModuleNotFoundError,
OSError,
FileNotFoundError,
FileExistsError,
IsADirectoryError,
NotADirectoryError,
PermissionError,
#[strum(serialize = "io.UnsupportedOperation")]
UnsupportedOperation,
TimeoutError,
AssertionError,
MemoryError,
StopIteration,
SyntaxError,
TypeError,
#[strum(serialize = "re.PatternError")]
RePatternError,
}
impl ExcType {
#[must_use]
pub fn is_subclass_of(self, handler_type: Self) -> bool {
if self == handler_type {
return true;
}
match handler_type {
Self::BaseException => true,
Self::Exception => !matches!(self, Self::BaseException | Self::KeyboardInterrupt | Self::SystemExit),
Self::LookupError => matches!(self, Self::KeyError | Self::IndexError),
Self::ArithmeticError => matches!(self, Self::ZeroDivisionError | Self::OverflowError),
Self::RuntimeError => matches!(self, Self::RecursionError | Self::NotImplementedError),
Self::AttributeError => matches!(self, Self::FrozenInstanceError),
Self::NameError => matches!(self, Self::UnboundLocalError),
Self::ValueError => matches!(
self,
Self::UnicodeDecodeError
| Self::UnicodeEncodeError
| Self::JsonDecodeError
| Self::UnsupportedOperation
),
Self::ImportError => matches!(self, Self::ModuleNotFoundError),
Self::OSError => matches!(
self,
Self::FileNotFoundError
| Self::FileExistsError
| Self::IsADirectoryError
| Self::NotADirectoryError
| Self::PermissionError
| Self::UnsupportedOperation
| Self::TimeoutError
),
_ => false,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub enum ExcData {
#[default]
None,
Unicode(Box<UnicodeErrorData>),
Json(Box<JsonErrorData>),
}
impl ExcData {
#[must_use]
pub fn unicode(&self) -> Option<&UnicodeErrorData> {
match self {
Self::Unicode(data) => Some(data),
_ => None,
}
}
#[must_use]
pub fn json(&self) -> Option<&JsonErrorData> {
match self {
Self::Json(data) => Some(data),
_ => None,
}
}
#[must_use]
pub fn estimate_size(&self) -> usize {
match self {
Self::None => 0,
Self::Unicode(data) => data.estimate_size(),
Self::Json(data) => data.estimate_size(),
}
}
}
#[derive(Debug, Clone, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub struct UnicodeErrorData {
pub encoding: String,
pub object: UnicodeErrorObject,
pub start: usize,
pub end: usize,
pub reason: String,
}
#[derive(Debug, Clone, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub enum UnicodeErrorObject {
Bytes(Vec<u8>),
Str(String),
}
impl UnicodeErrorData {
pub const MAX_OBJECT_LEN: usize = 64 * 1024;
#[must_use]
pub fn encode(encoding: &str, object: &str, start: usize, end: usize, reason: &str) -> ExcData {
if object.len() <= Self::MAX_OBJECT_LEN {
ExcData::Unicode(Box::new(Self {
encoding: encoding.to_owned(),
object: UnicodeErrorObject::Str(object.to_owned()),
start,
end,
reason: reason.to_owned(),
}))
} else {
ExcData::None
}
}
#[must_use]
pub fn decode(encoding: &str, object: &[u8], start: usize, end: usize, reason: &str) -> ExcData {
if object.len() <= Self::MAX_OBJECT_LEN {
ExcData::Unicode(Box::new(Self {
encoding: encoding.to_owned(),
object: UnicodeErrorObject::Bytes(object.to_vec()),
start,
end,
reason: reason.to_owned(),
}))
} else {
ExcData::None
}
}
#[must_use]
pub fn estimate_size(&self) -> usize {
let object_len = match &self.object {
UnicodeErrorObject::Bytes(b) => b.len(),
UnicodeErrorObject::Str(s) => s.len(),
};
mem::size_of::<Self>() + self.encoding.len() + object_len + self.reason.len()
}
}
#[derive(Debug, Clone, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub struct JsonErrorData {
pub msg: String,
pub doc: Option<String>,
pub pos: usize,
pub lineno: usize,
pub colno: usize,
}
impl JsonErrorData {
pub const MAX_DOC_LEN: usize = 64 * 1024;
#[must_use]
pub fn build(msg: &str, doc: &[u8], pos: usize, lineno: usize, colno: usize) -> ExcData {
let doc = if doc.len() <= Self::MAX_DOC_LEN {
str::from_utf8(doc).ok().map(ToOwned::to_owned)
} else {
None
};
ExcData::Json(Box::new(Self {
msg: msg.to_owned(),
doc,
pos,
lineno,
colno,
}))
}
#[must_use]
pub fn estimate_size(&self) -> usize {
mem::size_of::<Self>() + self.msg.len() + self.doc.as_ref().map_or(0, String::len)
}
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct StackFrame {
pub filename: String,
pub start: CodeLoc,
pub end: CodeLoc,
pub frame_name: Option<String>,
pub preview_line: Option<Arc<str>>,
pub hide_caret: bool,
pub hide_frame_name: bool,
}
impl fmt::Display for StackFrame {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if self.hide_frame_name {
write!(f, r#" File "{}", line {}"#, self.filename, self.start.line)?;
} else {
write!(f, r#" File "{}", line {}, in "#, self.filename, self.start.line)?;
if let Some(frame_name) = &self.frame_name {
f.write_str(frame_name)?;
} else {
f.write_str("<module>")?;
}
}
if let Some(line) = &self.preview_line {
if self.start.line != self.end.line {
f.write_char('\n')?;
for block_line in line.lines() {
writeln!(f, " {block_line}")?;
}
return Ok(());
}
let trimmed = line.trim_start();
writeln!(f, "\n {trimmed}")?;
if !self.hide_caret {
let leading_spaces = line.len() - trimmed.len();
let caret_start = if self.start.column as usize > leading_spaces {
4 + self.start.column as usize - leading_spaces - 1
} else {
4
};
f.write_str(&" ".repeat(caret_start))?;
let caret_len = (self.end.column - self.start.column).max(1) as usize;
writeln!(f, "{}", "~".repeat(caret_len))?;
}
} else {
f.write_char('\n')?;
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]
pub struct CodeLoc {
pub line: u32,
pub column: u32,
}
impl Default for CodeLoc {
fn default() -> Self {
Self { line: 1, column: 1 }
}
}
impl CodeLoc {
#[must_use]
pub fn new(line: u32, column: u32) -> Self {
Self {
line: line.saturating_add(1),
column: column.saturating_add(1),
}
}
}
#[must_use]
pub fn unicode_decode_error_msg(codec: &str, first_byte: u8, start: usize, end: usize, reason: &str) -> String {
debug_assert!(
end > start,
"unicode_decode_error_msg: end ({end}) must be > start ({start})"
);
if end - start == 1 {
format!("'{codec}' codec can't decode byte 0x{first_byte:02x} in position {start}: {reason}")
} else {
let last = end - 1;
format!("'{codec}' codec can't decode bytes in position {start}-{last}: {reason}")
}
}