rustpython_vm/
warn.rs

1use crate::{
2    AsObject, Context, Py, PyAtomicRef, PyObject, PyObjectRef, PyResult, VirtualMachine,
3    anystr::SplitLinesArgs,
4    builtins::{
5        PyBaseExceptionRef, PyDict, PyDictRef, PyList, PyListRef, PyStr, PyStrInterned, PyStrRef,
6        PyTuple, PyTupleRef, PyType, PyTypeRef,
7    },
8    convert::TryFromObject,
9};
10use core::sync::atomic::{AtomicUsize, Ordering};
11use rustpython_common::lock::OnceCell;
12
13pub struct WarningsState {
14    pub filters: PyAtomicRef<PyList>,
15    pub once_registry: PyAtomicRef<PyDict>,
16    pub default_action: PyAtomicRef<PyStr>,
17    pub filters_version: AtomicUsize,
18    pub context_var: OnceCell<PyObjectRef>,
19    lock_count: AtomicUsize,
20}
21
22impl WarningsState {
23    fn create_default_filters(ctx: &Context) -> PyListRef {
24        // init_filters(): non-debug default filter set.
25        ctx.new_list(vec![
26            ctx.new_tuple(vec![
27                ctx.new_str("default").into(),
28                ctx.none(),
29                ctx.exceptions.deprecation_warning.as_object().to_owned(),
30                ctx.new_str("__main__").into(),
31                ctx.new_int(0).into(),
32            ])
33            .into(),
34            ctx.new_tuple(vec![
35                ctx.new_str("ignore").into(),
36                ctx.none(),
37                ctx.exceptions.deprecation_warning.as_object().to_owned(),
38                ctx.none(),
39                ctx.new_int(0).into(),
40            ])
41            .into(),
42            ctx.new_tuple(vec![
43                ctx.new_str("ignore").into(),
44                ctx.none(),
45                ctx.exceptions
46                    .pending_deprecation_warning
47                    .as_object()
48                    .to_owned(),
49                ctx.none(),
50                ctx.new_int(0).into(),
51            ])
52            .into(),
53            ctx.new_tuple(vec![
54                ctx.new_str("ignore").into(),
55                ctx.none(),
56                ctx.exceptions.import_warning.as_object().to_owned(),
57                ctx.none(),
58                ctx.new_int(0).into(),
59            ])
60            .into(),
61            ctx.new_tuple(vec![
62                ctx.new_str("ignore").into(),
63                ctx.none(),
64                ctx.exceptions.resource_warning.as_object().to_owned(),
65                ctx.none(),
66                ctx.new_int(0).into(),
67            ])
68            .into(),
69        ])
70    }
71
72    pub fn init_state(ctx: &Context) -> Self {
73        Self {
74            filters: Self::create_default_filters(ctx).into(),
75            once_registry: ctx.new_dict().into(),
76            default_action: ctx.new_str("default").into(),
77            filters_version: AtomicUsize::new(0),
78            context_var: OnceCell::new(),
79            lock_count: AtomicUsize::new(0),
80        }
81    }
82
83    pub fn acquire_lock(&self) {
84        self.lock_count.fetch_add(1, Ordering::SeqCst);
85    }
86
87    pub fn release_lock(&self) -> bool {
88        let prev = self.lock_count.load(Ordering::SeqCst);
89        if prev == 0 {
90            return false;
91        }
92        self.lock_count.fetch_sub(1, Ordering::SeqCst);
93        true
94    }
95
96    pub fn filters_mutated(&self) {
97        self.filters_version.fetch_add(1, Ordering::SeqCst);
98    }
99}
100
101/// None matches everything; plain strings do exact comparison;
102/// regex objects use .match().
103fn check_matched(obj: &PyObject, arg: &PyObject, vm: &VirtualMachine) -> PyResult<bool> {
104    if vm.is_none(obj) {
105        return Ok(true);
106    }
107    if obj.class().is(vm.ctx.types.str_type) {
108        return obj.rich_compare_bool(arg, crate::types::PyComparisonOp::Eq, vm);
109    }
110    let result = vm.call_method(obj, "match", (arg.to_owned(),))?;
111    result.is_true(vm)
112}
113
114fn get_warnings_attr(
115    vm: &VirtualMachine,
116    attr_name: &'static PyStrInterned,
117    try_import: bool,
118) -> PyResult<Option<PyObjectRef>> {
119    let module = if try_import
120        && !vm
121            .state
122            .finalizing
123            .load(core::sync::atomic::Ordering::SeqCst)
124    {
125        match vm.import("warnings", 0) {
126            Ok(module) => module,
127            Err(e) if e.fast_isinstance(vm.ctx.exceptions.import_error) => return Ok(None),
128            Err(e) => return Err(e),
129        }
130    } else {
131        let Ok(modules) = vm.sys_module.get_attr(identifier!(vm, modules), vm) else {
132            return Ok(None);
133        };
134        match modules.get_item(vm.ctx.intern_str("warnings"), vm) {
135            Ok(module) => module,
136            Err(_) => return Ok(None),
137        }
138    };
139
140    vm.get_attribute_opt(&module, attr_name)
141}
142
143/// Get the warnings filters list from `sys.modules['warnings'].filters`,
144/// falling back to the interpreter cache (`st->filters`).
145fn get_warnings_filters(vm: &VirtualMachine) -> PyResult<PyListRef> {
146    if let Some(filters_obj) = get_warnings_attr(vm, identifier!(vm, filters), false)? {
147        let filters = PyListRef::try_from_object(vm, filters_obj)
148            .map_err(|_| vm.new_value_error("_warnings.filters must be a list"))?;
149        vm.state
150            .warnings
151            .filters
152            .swap_to_temporary_refs(filters, vm);
153    }
154    Ok(vm.state.warnings.filters.to_owned())
155}
156
157fn get_warnings_context_filters(vm: &VirtualMachine) -> PyResult<Option<PyListRef>> {
158    let Some(ctx_var) = vm.state.warnings.context_var.get() else {
159        return Ok(None);
160    };
161    if vm.is_none(ctx_var) {
162        return Ok(None);
163    }
164    let ctx = vm.call_method(ctx_var, "get", (vm.ctx.none(),))?;
165    if vm.is_none(&ctx) {
166        return Ok(None);
167    }
168    let filters = ctx.get_attr("_filters", vm)?;
169    PyListRef::try_from_object(vm, filters)
170        .map(Some)
171        .map_err(|_| vm.new_value_error("_filters of warnings._warnings_context must be a list"))
172}
173
174fn bless_my_loader(
175    module_globals: &PyObject,
176    vm: &VirtualMachine,
177) -> PyResult<Option<PyObjectRef>> {
178    let external = vm
179        .importlib
180        .get_attr("_bootstrap_external", vm)
181        .or_else(|_| vm.import("importlib._bootstrap_external", 0))?;
182    if vm.is_none(&external) {
183        return Ok(None);
184    }
185    let bless = external.get_attr("_bless_my_loader", vm)?;
186    let loader = bless.call((module_globals.to_owned(),), vm)?;
187    if vm.is_none(&loader) {
188        Ok(None)
189    } else {
190        Ok(Some(loader))
191    }
192}
193
194pub(crate) fn get_source_line(
195    module_globals: &PyObject,
196    lineno: usize,
197    vm: &VirtualMachine,
198) -> PyResult<Option<PyObjectRef>> {
199    let Some(loader) = bless_my_loader(module_globals, vm)? else {
200        return Ok(None);
201    };
202
203    let module_name = if let Some(dict) = module_globals.downcast_ref::<PyDict>() {
204        match dict.get_item_opt(identifier!(vm, __name__), vm)? {
205            Some(name) => name,
206            None => return Ok(None),
207        }
208    } else {
209        match module_globals.get_item(identifier!(vm, __name__), vm) {
210            Ok(name) => name,
211            Err(_) => return Ok(None),
212        }
213    };
214
215    let Some(get_source) = vm.get_attribute_opt(&loader, vm.ctx.intern_str("get_source"))? else {
216        return Ok(None);
217    };
218    let source = get_source.call((module_name,), vm)?;
219    if vm.is_none(&source) {
220        return Ok(None);
221    }
222
223    let Some(source_str) = source.downcast_ref::<PyStr>() else {
224        return Err(vm.new_type_error(format!("expected str, not {}", source.class().name())));
225    };
226    let lines = source_str.splitlines(SplitLinesArgs { keepends: false }, vm);
227    if lineno == 0 {
228        return Err(vm.new_index_error("list index out of range"));
229    }
230    lines
231        .get(lineno - 1)
232        .cloned()
233        .ok_or_else(|| vm.new_index_error("list index out of range"))
234        .map(Some)
235}
236
237/// Get the default action from `sys.modules['warnings'].defaultaction`,
238/// falling back to vm.state.warnings.default_action.
239fn get_default_action(vm: &VirtualMachine) -> PyResult<PyObjectRef> {
240    if let Some(action) = get_warnings_attr(vm, identifier!(vm, defaultaction), false)? {
241        if !action.class().is(vm.ctx.types.str_type) {
242            return Err(vm.new_type_error(format!(
243                "_warnings.defaultaction must be a string, not '{}'",
244                action.class().name()
245            )));
246        }
247        let action = PyStrRef::try_from_object(vm, action)?;
248        vm.state
249            .warnings
250            .default_action
251            .swap_to_temporary_refs(action, vm);
252    }
253    Ok(vm.state.warnings.default_action.to_owned().into())
254}
255
256/// Get the once registry from `sys.modules['warnings'].onceregistry`,
257/// falling back to vm.state.warnings.once_registry.
258fn get_once_registry(vm: &VirtualMachine) -> PyResult<PyObjectRef> {
259    if let Some(registry) = get_warnings_attr(vm, identifier!(vm, onceregistry), false)? {
260        if !registry.class().is(vm.ctx.types.dict_type) {
261            return Err(vm.new_type_error(format!(
262                "_warnings.onceregistry must be a dict, not '{}'",
263                registry.class().name()
264            )));
265        }
266        let registry = PyDictRef::try_from_object(vm, registry)?;
267        vm.state
268            .warnings
269            .once_registry
270            .swap_to_temporary_refs(registry, vm);
271    }
272    Ok(vm.state.warnings.once_registry.to_owned().into())
273}
274
275fn already_warned(
276    registry: &PyObject,
277    key: &PyObject,
278    should_set: bool,
279    vm: &VirtualMachine,
280) -> PyResult<bool> {
281    if vm.is_none(registry) {
282        return Ok(false);
283    }
284
285    let current_version = vm.state.warnings.filters_version.load(Ordering::SeqCst);
286    let version_obj = registry.get_item(identifier!(&vm.ctx, version), vm).ok();
287
288    let version_matches = version_obj.as_ref().is_some_and(|v| {
289        v.try_int(vm)
290            .is_ok_and(|i| i.as_u32_mask() as usize == current_version)
291    });
292
293    if version_matches {
294        if let Ok(val) = registry.get_item(key, vm)
295            && val.is_true(vm)?
296        {
297            return Ok(true);
298        }
299    } else if let Ok(dict) = PyDictRef::try_from_object(vm, registry.to_owned()) {
300        dict.clear();
301        dict.set_item(
302            identifier!(&vm.ctx, version),
303            vm.ctx.new_int(current_version).into(),
304            vm,
305        )?;
306    }
307
308    if should_set {
309        registry.set_item(key, vm.ctx.true_value.clone().into(), vm)?;
310    }
311    Ok(false)
312}
313
314/// Create a `(text, category)` or `(text, category, 0)` key and record
315/// it in the registry via `already_warned`.
316fn update_registry(
317    registry: &PyObject,
318    text: &PyObject,
319    category: &PyObject,
320    add_zero: bool,
321    vm: &VirtualMachine,
322) -> PyResult<bool> {
323    let altkey: PyObjectRef = if add_zero {
324        PyTuple::new_ref(
325            vec![
326                text.to_owned(),
327                category.to_owned(),
328                vm.ctx.new_int(0).into(),
329            ],
330            &vm.ctx,
331        )
332        .into()
333    } else {
334        PyTuple::new_ref(vec![text.to_owned(), category.to_owned()], &vm.ctx).into()
335    };
336    already_warned(registry, &altkey, true, vm)
337}
338
339fn normalize_module(filename: &Py<PyStr>, vm: &VirtualMachine) -> PyObjectRef {
340    match filename.byte_len() {
341        0 => vm.new_pyobj("<unknown>"),
342        len if len >= 3 && filename.as_bytes().ends_with(b".py") => {
343            vm.new_pyobj(&filename.as_wtf8()[..len - 3])
344        }
345        _ => filename.as_object().to_owned(),
346    }
347}
348
349/// Search a filters list for a matching action.
350fn filter_search(
351    category: &PyObject,
352    text: &PyObject,
353    lineno: usize,
354    module: &PyObject,
355    filters: &PyListRef,
356    list_name: &str,
357    vm: &VirtualMachine,
358) -> PyResult<Option<PyObjectRef>> {
359    // filters could change while we are iterating over it.
360    // Re-check list length each iteration.
361    let mut i = 0;
362    while i < filters.borrow_vec().len() {
363        let Some(tmp_item) = filters.borrow_vec().get(i).cloned() else {
364            break;
365        };
366        let tmp_item = PyTupleRef::try_from_object(vm, tmp_item)
367            .ok()
368            .filter(|t| t.as_slice().len() == 5)
369            .ok_or_else(|| {
370                vm.new_value_error(format!("warnings.{list_name} item {i} isn't a 5-tuple"))
371            })?;
372
373        /* action, msg, cat, mod, ln = item */
374        let action = &tmp_item.as_slice()[0];
375        if !action.class().is(vm.ctx.types.str_type) {
376            return Err(vm.new_type_error(format!(
377                "action must be a string, not '{}'",
378                action.class().name()
379            )));
380        }
381        let good_msg = check_matched(&tmp_item.as_slice()[1], text, vm)?;
382        let is_subclass = category.is_subclass(&tmp_item.as_slice()[2], vm)?;
383        let good_mod = check_matched(&tmp_item.as_slice()[3], module, vm)?;
384        let ln: usize = tmp_item.as_slice()[4]
385            .try_int(vm)
386            .map_or(0, |v| v.as_u32_mask() as _);
387
388        if good_msg && is_subclass && good_mod && (ln == 0 || lineno == ln) {
389            return Ok(Some(action.to_owned()));
390        }
391        i += 1;
392    }
393    Ok(None)
394}
395
396fn get_filter(
397    category: &PyObject,
398    text: &PyObject,
399    lineno: usize,
400    module: &PyObject,
401    vm: &VirtualMachine,
402) -> PyResult {
403    if let Some(context_filters) = get_warnings_context_filters(vm)? {
404        if let Some(action) = filter_search(
405            category,
406            text,
407            lineno,
408            module,
409            &context_filters,
410            "_warnings_context _filters",
411            vm,
412        )? {
413            return Ok(action);
414        }
415        return get_default_action(vm);
416    }
417
418    let filters = get_warnings_filters(vm)?;
419    if let Some(action) = filter_search(category, text, lineno, module, &filters, "filters", vm)? {
420        return Ok(action);
421    }
422    get_default_action(vm)
423}
424
425pub fn warn(
426    message: PyObjectRef,
427    category: Option<PyTypeRef>,
428    stack_level: isize,
429    source: Option<PyObjectRef>,
430    vm: &VirtualMachine,
431) -> PyResult<()> {
432    warn_with_skip(message, category, stack_level, source, None, vm)
433}
434
435/// Warn that `coro` was never awaited.
436///
437/// Prefer `warnings._warn_unawaited_coroutine` so origin tracking can
438/// format the creation traceback. If that helper is missing, broken, or
439/// not a RuntimeWarning-as-error, fall back to a plain RuntimeWarning.
440/// Never raises; leftover exceptions become unraisable.
441pub fn warn_unawaited_coroutine(coro: &PyObject, qualname: &Py<PyStr>, vm: &VirtualMachine) {
442    let unraisable_msg = format!(
443        "Exception ignored while finalizing coroutine {}",
444        coro.repr(vm)
445            .map_or_else(|_| String::new(), |s| s.to_string())
446    );
447    let mut warned = false;
448    if let Some(helper) =
449        get_warnings_attr(vm, vm.ctx.intern_str("_warn_unawaited_coroutine"), true)
450            .ok()
451            .flatten()
452    {
453        match helper.call((coro.to_owned(),), vm) {
454            Ok(_) => warned = true,
455            Err(e) => {
456                if e.fast_isinstance(vm.ctx.exceptions.runtime_warning) {
457                    warned = true;
458                }
459                vm.run_unraisable(e, Some(unraisable_msg.clone()), coro.to_owned());
460            }
461        }
462    }
463    if !warned {
464        let msg = format!("coroutine '{qualname}' was never awaited");
465        if let Err(e) =
466            crate::stdlib::_warnings::warn(vm.ctx.exceptions.runtime_warning, msg, 1, vm)
467        {
468            vm.run_unraisable(e, Some(unraisable_msg), coro.to_owned());
469        }
470    }
471}
472
473/// do_warn: resolve context via setup_context, then call warn_explicit.
474pub fn warn_with_skip(
475    message: PyObjectRef,
476    category: Option<PyTypeRef>,
477    mut stack_level: isize,
478    source: Option<PyObjectRef>,
479    skip_file_prefixes: Option<&Py<PyTuple>>,
480    vm: &VirtualMachine,
481) -> PyResult<()> {
482    if let Some(prefixes) = skip_file_prefixes
483        && !prefixes.as_slice().is_empty()
484        && stack_level < 2
485    {
486        stack_level = 2;
487    }
488    let (filename, lineno, module, registry) = setup_context(stack_level, skip_file_prefixes, vm)?;
489    warn_explicit(
490        category, message, filename, lineno, module, registry, None, source, vm,
491    )
492}
493
494/// Core warning logic matching `warn_explicit()` in `_warnings.c`.
495#[allow(clippy::too_many_arguments)]
496pub fn warn_explicit(
497    category: Option<PyTypeRef>,
498    message: PyObjectRef,
499    filename: PyStrRef,
500    lineno: usize,
501    module: Option<PyObjectRef>,
502    registry: PyObjectRef,
503    source_line: Option<PyObjectRef>,
504    source: Option<PyObjectRef>,
505    vm: &VirtualMachine,
506) -> PyResult<()> {
507    // Normalize module. None → silent return (late-shutdown safety).
508    let module = module.unwrap_or_else(|| normalize_module(&filename, vm));
509    if vm.is_none(&module) {
510        return Ok(());
511    }
512
513    // Normalize message.
514    let is_warning = message.fast_isinstance(vm.ctx.exceptions.warning);
515    let (text, category, message) = if is_warning {
516        let text = message.str(vm)?;
517        let cat = message.class().to_owned();
518        (text, cat, message)
519    } else {
520        // For non-Warning messages, convert to string via str()
521        let text = message.str(vm)?;
522        let cat = category.unwrap_or_else(|| vm.ctx.exceptions.user_warning.to_owned());
523        let instance = cat.as_object().call((text.clone(),), vm)?;
524        (text, cat, instance)
525    };
526
527    let lineno_obj: PyObjectRef = vm.ctx.new_int(lineno).into();
528
529    // key = (text, category, lineno)
530    let key: PyObjectRef = PyTuple::new_ref(
531        vec![
532            text.clone().into(),
533            category.as_object().to_owned(),
534            lineno_obj.clone(),
535        ],
536        &vm.ctx,
537    )
538    .into();
539
540    // Check if already warned
541    if !vm.is_none(&registry) && already_warned(&registry, &key, false, vm)? {
542        return Ok(());
543    }
544
545    // Get filter action
546    let action = get_filter(category.as_object(), text.as_object(), lineno, &module, vm)?;
547    let action_str = PyStrRef::try_from_object(vm, action)
548        .map_err(|_| vm.new_type_error("action must be a string"))?;
549
550    if action_str.as_bytes() == b"error" {
551        let exc = PyBaseExceptionRef::try_from_object(vm, message)?;
552        return Err(exc);
553    }
554    if action_str.as_bytes() == b"ignore" {
555        return Ok(());
556    }
557
558    // For everything except "always"/"all", record in registry then
559    // check per-action registries.
560    let already = if action_str.as_wtf8() != "always" && action_str.as_wtf8() != "all" {
561        if !vm.is_none(&registry) {
562            registry.set_item(&*key, vm.ctx.true_value.clone().into(), vm)?;
563        }
564
565        let action_s = action_str.to_str();
566        match action_s {
567            Some("once") => {
568                let reg = if vm.is_none(&registry) {
569                    get_once_registry(vm)?
570                } else {
571                    registry
572                };
573                update_registry(&reg, text.as_ref(), category.as_object(), false, vm)?
574            }
575            Some("module") => {
576                if !vm.is_none(&registry) {
577                    update_registry(&registry, text.as_ref(), category.as_object(), false, vm)?
578                } else {
579                    false
580                }
581            }
582            Some("default") => false,
583            _ => {
584                return Err(vm.new_runtime_error(format!(
585                    "Unrecognized action ({action_str}) in warnings.filters:\n {action_str}"
586                )));
587            }
588        }
589    } else {
590        false
591    };
592
593    if already {
594        return Ok(());
595    }
596
597    call_show_warning(
598        category,
599        &text,
600        message,
601        filename,
602        lineno,
603        lineno_obj,
604        source_line,
605        source,
606        vm,
607    )
608}
609
610#[allow(clippy::too_many_arguments)]
611fn call_show_warning(
612    category: PyTypeRef,
613    text: &Py<PyStr>,
614    message: PyObjectRef,
615    filename: PyStrRef,
616    lineno: usize,
617    lineno_obj: PyObjectRef,
618    source_line: Option<PyObjectRef>,
619    source: Option<PyObjectRef>,
620    vm: &VirtualMachine,
621) -> PyResult<()> {
622    let Some(show_fn) = get_warnings_attr(vm, identifier!(vm, _showwarnmsg), source.is_some())?
623    else {
624        show_warning(&filename, lineno, text, &category, source_line, vm);
625        return Ok(());
626    };
627
628    if !show_fn.is_callable() {
629        return Err(vm.new_type_error("warnings._showwarnmsg() must be set to a callable"));
630    }
631
632    let Some(warnmsg_cls) = get_warnings_attr(vm, identifier!(vm, WarningMessage), false)? else {
633        return Err(vm.new_runtime_error("unable to get warnings.WarningMessage"));
634    };
635
636    let msg = warnmsg_cls.call(
637        vec![
638            message,
639            category.into(),
640            filename.into(),
641            lineno_obj,
642            vm.ctx.none(),
643            vm.ctx.none(),
644            vm.unwrap_or_none(source),
645        ],
646        vm,
647    )?;
648
649    show_fn.call((msg,), vm)?;
650    Ok(())
651}
652
653fn show_warning(
654    filename: &Py<PyStr>,
655    lineno: usize,
656    text: &Py<PyStr>,
657    category: &Py<PyType>,
658    source_line: Option<PyObjectRef>,
659    vm: &VirtualMachine,
660) {
661    let stderr = crate::stdlib::sys::PyStderr(vm);
662    writeln!(
663        stderr,
664        "{}:{}: {}: {}",
665        filename,
666        lineno,
667        category.name(),
668        text
669    );
670    if let Some(source_line) = source_line
671        && let Ok(line) = source_line.str(vm)
672    {
673        writeln!(stderr, "{line}");
674    }
675}
676
677/// Check if a frame's filename starts with any of the given prefixes.
678fn is_filename_to_skip(frame: &crate::frame::FrameObject, prefixes: &Py<PyTuple>) -> bool {
679    let filename = frame.f_code().co_filename();
680    let filename_bytes = filename.as_bytes();
681    prefixes.as_slice().iter().any(|prefix| {
682        prefix
683            .downcast_ref::<PyStr>()
684            .is_some_and(|s| filename_bytes.starts_with(s.as_bytes()))
685    })
686}
687
688/// Like FrameObject::next_external_frame but also skips frames matching prefixes.
689fn next_external_frame_with_skip(
690    frame: &crate::frame::FrameObjectRef,
691    skip_file_prefixes: Option<&Py<PyTuple>>,
692    vm: &VirtualMachine,
693) -> Option<crate::frame::FrameObjectRef> {
694    let mut f = frame.f_back(vm);
695    loop {
696        let current: crate::frame::FrameObjectRef = f.take()?;
697        if current.is_internal_frame()
698            || skip_file_prefixes.is_some_and(|p| is_filename_to_skip(&current, p))
699        {
700            f = current.f_back(vm);
701        } else {
702            return Some(current);
703        }
704    }
705}
706
707/// filename, module, and registry are new refs, globals is borrowed
708/// Returns `Ok` on success, or `Err` on error (no new refs)
709fn setup_context(
710    mut stack_level: isize,
711    skip_file_prefixes: Option<&Py<PyTuple>>,
712    vm: &VirtualMachine,
713) -> PyResult<(PyStrRef, usize, Option<PyObjectRef>, PyObjectRef)> {
714    // Materialize the topmost frame (including light frames) so stack
715    // level counting is correct across the full Python frame chain.
716    let mut f = crate::frame::current_thread_frame_materialize(vm);
717
718    // Stack level comparisons to Python code is off by one as there is no
719    // warnings-related stack level to avoid.
720    if stack_level <= 0 || f.as_ref().is_some_and(|frame| frame.is_internal_frame()) {
721        while {
722            stack_level -= 1;
723            stack_level > 0
724        } {
725            match f {
726                Some(tmp) => f = tmp.f_back(vm),
727                None => break,
728            }
729        }
730    } else {
731        while {
732            stack_level -= 1;
733            stack_level > 0
734        } {
735            match f {
736                Some(tmp) => f = next_external_frame_with_skip(&tmp, skip_file_prefixes, vm),
737                None => break,
738            }
739        }
740    }
741
742    let (globals, filename, lineno) = if let Some(f) = f {
743        (
744            f.iframe().globals().to_owned(),
745            f.iframe().code().source_path(),
746            f.lineno().max(0) as usize,
747        )
748    } else if let Some(frame) = vm.current_frame() {
749        // We have a frame but it wasn't found during stack walking
750        (
751            frame.iframe().globals().to_owned(),
752            vm.ctx.intern_str("<sys>"),
753            1,
754        )
755    } else {
756        // No frames on the stack - use sys.__dict__ (interp->sysdict)
757        let globals = vm
758            .sys_module
759            .as_object()
760            .get_attr(identifier!(vm, __dict__), vm)
761            .and_then(|d| {
762                d.downcast::<crate::builtins::PyDict>()
763                    .map_err(|_| vm.new_type_error("sys.__dict__ is not a dictionary"))
764            })?;
765        (globals, vm.ctx.intern_str("<sys>"), 0)
766    };
767
768    let registry = match globals.get_item("__warningregistry__", vm) {
769        Ok(r) => r,
770        Err(_) => {
771            let r = vm.ctx.new_dict();
772            globals.set_item("__warningregistry__", r.clone().into(), vm)?;
773            r.into()
774        }
775    };
776
777    // Setup module.
778    let module = globals
779        .get_item("__name__", vm)
780        .unwrap_or_else(|_| vm.new_pyobj("<string>"));
781    Ok((filename.to_owned(), lineno, Some(module), registry))
782}
rustpython_vm/warn.rs

rustpython_vm/
warn.rs
