Support frozen ElementTree attribute mappings (#9124)

* Support frozen ElementTree attribute mappings

Preserve assigned mapping identity while copying constructor attributes, and use checked union ownership through Element operations.

Assisted-by: OpenAI Codex:gpt-6.1-sol

* Sync final ElementTree runtime and companion tests

Import ElementTree.py from CPython d220409a13b77de7a6c30abe16ea90db01528edc through update_lib migrate. Copy six complete canonical tests with update_lib copy-test, preserving all surrounding test markers. This aligns pure-Python and native frozendict diagnostic expectations and covers the runtime import companion changes.

Assisted-by: OpenAI Codex:model version unavailable
This commit is contained in:
youknowdot authored and GitHub committed 2026-10-15 10:27:39 +09:00
1 parent 13d0c83753
commit 2ddba03c39
5 files changed
+237 -67

No files matched your search

+50 -4
View File
@@ -370,8 +370,7 @@ class ElementTreeTest(unittest.TestCase):
self.serialize_check(element, '<tag key="value"><subtag /></tag>') # 4
element.remove(subelement)
self.serialize_check(element, '<tag key="value" />') # 5
with self.assertRaisesRegex(ValueError,
r'Element\.remove\(.+\): element not found'):
with self.assertRaises(ValueError):
element.remove(subelement)
self.serialize_check(element, '<tag key="value" />') # 6
element[0:0] = [subelement, subelement, subelement]
@@ -1255,6 +1254,9 @@ class ElementTreeTest(unittest.TestCase):
comm = ET.Comment('<spam> & ham')
# comments are not escaped
self.assertEqual(ET.tostring(comm), b'<!--<spam> & ham-->')
self.assertEqual(ET.tostring(comm, method='html'), b'<!--<spam> & ham-->')
# no comments in text serialization
self.assertEqual(ET.tostring(comm, method='text'), b'')
def test_processinginstruction_serialization(self):
# Test ProcessingInstruction directly
@@ -1271,6 +1273,11 @@ class ElementTreeTest(unittest.TestCase):
self.assertEqual(ET.tostring(ET.PI('test', '<testing&>\xe3'), 'latin-1'),
b"<?xml version='1.0' encoding='latin-1'?>\n"
b"<?test <testing&>\xe3?>")
pi = ET.PI('test', 'ham & eggs < spam')
self.assertEqual(ET.tostring(pi), b'<?test ham & eggs < spam?>')
self.assertEqual(ET.tostring(pi, method='html'), b'<?test ham & eggs < spam?>')
# no processing instructions in text serialization
self.assertEqual(ET.tostring(pi, method='text'), b'')
@support.subTests('tag', ("script", "style", "xmp", "iframe", "noembed", "noframes"))
def test_html_cdata_elems_serialization(self, tag):
@@ -1469,17 +1476,39 @@ class IterparseTest(unittest.TestCase):
def test_resource_warnings_not_exhausted(self):
# Not exhausting the iterator still closes the underlying file (bpo-43292)
# Not closing before del should emit ResourceWarning
it = ET.iterparse(SIMPLE_XMLFILE)
with warnings_helper.check_no_resource_warning(self):
it.close()
del it
gc_collect()
it = ET.iterparse(SIMPLE_XMLFILE)
with self.assertWarns(ResourceWarning) as wm:
del it
gc_collect()
# Not 'unclosed file'.
self.assertIn('unclosed iterparse iterator', str(wm.warning))
self.assertIn(repr(SIMPLE_XMLFILE), str(wm.warning))
self.assertEqual(wm.filename, __file__)
it = ET.iterparse(SIMPLE_XMLFILE)
with warnings_helper.check_no_resource_warning(self):
action, elem = next(it)
it.close()
self.assertEqual((action, elem.tag), ('end', 'element'))
del it, elem
gc_collect()
it = ET.iterparse(SIMPLE_XMLFILE)
with self.assertWarns(ResourceWarning) as wm:
action, elem = next(it)
self.assertEqual((action, elem.tag), ('end', 'element'))
del it, elem
gc_collect()
self.assertIn('unclosed iterparse iterator', str(wm.warning))
self.assertIn(repr(SIMPLE_XMLFILE), str(wm.warning))
self.assertEqual(wm.filename, __file__)
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_resource_warnings_failed_iteration(self):
@@ -1502,9 +1531,19 @@ class IterparseTest(unittest.TestCase):
it = ET.iterparse(SIMPLE_XMLFILE)
with warnings_helper.check_no_resource_warning(self):
list(it)
it.close()
del it
gc_collect()
it = ET.iterparse(SIMPLE_XMLFILE)
with self.assertWarns(ResourceWarning) as wm:
list(it)
del it
gc_collect()
self.assertIn('unclosed iterparse iterator', str(wm.warning))
self.assertIn(repr(SIMPLE_XMLFILE), str(wm.warning))
self.assertEqual(wm.filename, __file__)
def test_close_not_exhausted(self):
iterparse = ET.iterparse
@@ -4514,6 +4553,9 @@ class KeywordArgsTest(unittest.TestCase):
ET.Element('a', dict(href="#"), id="foo"),
ET.Element('a', href="#", id="foo"),
ET.Element('a', dict(href="#", id="foo"), href="#", id="foo"),
ET.Element('a', frozendict(href="#", id="foo")),
ET.Element('a', frozendict(href="#"), id="foo"),
ET.Element('a', attrib=frozendict(href="#", id="foo")),
]
for e in elements:
self.assertEqual(e.tag, 'a')
@@ -4521,10 +4563,14 @@ class KeywordArgsTest(unittest.TestCase):
e2 = ET.SubElement(elements[0], 'foobar', attrib={'key1': 'value1'})
self.assertEqual(e2.attrib['key1'], 'value1')
e3 = ET.SubElement(elements[0], 'foobar',
attrib=frozendict({'key1': 'value1'}))
self.assertEqual(e3.attrib['key1'], 'value1')
with self.assertRaisesRegex(TypeError, 'must be dict, not str'):
errmsg = 'must be dict or frozendict, not str'
with self.assertRaisesRegex(TypeError, errmsg):
ET.Element('a', "I'm not a dict")
with self.assertRaisesRegex(TypeError, 'must be dict, not str'):
with self.assertRaisesRegex(TypeError, errmsg):
ET.Element('a', attrib="I'm not a dict")
# --------------------------------------------------------------------
+38 -20
View File
@@ -83,15 +83,12 @@ __all__ = [
"SubElement",
"tostring", "tostringlist",
"TreeBuilder",
"VERSION",
"XML", "XMLID",
"XMLParser", "XMLPullParser",
"register_namespace",
"canonicalize", "C14NWriterTarget",
]
VERSION = "1.3.0"
import sys
import re
import warnings
@@ -167,9 +164,9 @@ class Element:
"""
def __init__(self, tag, attrib={}, **extra):
if not isinstance(attrib, dict):
raise TypeError("attrib must be dict, not %s" % (
def __init__(self, tag, /, attrib={}, **extra):
if not isinstance(attrib, (dict, frozendict)):
raise TypeError("attrib must be dict or frozendict, not %s" % (
attrib.__class__.__name__,))
self.tag = tag
self.attrib = {**attrib, **extra}
@@ -266,12 +263,15 @@ class Element:
ValueError is raised if a matching element could not be found.
"""
# assert iselement(element)
try:
self._children.remove(subelement)
except ValueError:
# to align the error type with the C implementation
if isinstance(subelement, type) or not iselement(subelement):
raise TypeError('expected an Element, not %s' %
type(subelement).__name__) from None
# to align the error message with the C implementation
raise ValueError("Element.remove(x): element not found") from None
raise ValueError(f"{subelement!r} not in {self!r}") from None
def find(self, path, namespaces=None):
"""Find first matching element by tag name or path.
@@ -420,7 +420,7 @@ class Element:
yield t
def SubElement(parent, tag, attrib={}, **extra):
def SubElement(parent, tag, /, attrib={}, **extra):
"""Subelement factory which creates an element instance, and appends it
to an existing parent.
@@ -928,9 +928,9 @@ def _serialize_html(write, elem, qnames, namespaces, **kwargs):
tag = elem.tag
text = elem.text
if tag is Comment:
write("<!--%s-->" % _escape_cdata(text))
write("<!--%s-->" % text)
elif tag is ProcessingInstruction:
write("<?%s?>" % _escape_cdata(text))
write("<?%s?>" % text)
else:
tag = qnames[tag]
if tag is None:
@@ -954,12 +954,15 @@ def _serialize_html(write, elem, qnames, namespaces, **kwargs):
for k, v in items:
if isinstance(k, QName):
k = k.text
if isinstance(v, QName):
v = qnames[v.text]
k = qnames[k]
if v is None:
write(" %s" % k) # empty attr
else:
v = _escape_attrib_html(v)
# FIXME: handle boolean attributes
write(" %s=\"%s\"" % (qnames[k], v))
if isinstance(v, QName):
v = qnames[v.text]
else:
v = _escape_attrib_html(v)
write(" %s=\"%s\"" % (k, v))
write(">")
ltag = tag.lower()
if text:
@@ -1275,16 +1278,20 @@ def iterparse(source, events=None, parser=None):
gen = iterator(source)
class IterParseIterator(collections.abc.Iterator):
__next__ = gen.__next__
def close(self):
nonlocal close_source
if close_source:
source.close()
close_source = False
gen.close()
def __del__(self):
# TODO: Emit a ResourceWarning if it was not explicitly closed.
# (When the close() method will be supported in all maintained Python versions.)
def __del__(self, _warn=warnings.warn):
if close_source:
source.close()
try:
_warn(f"unclosed iterparse iterator {source.name!r}", ResourceWarning, stacklevel=2)
finally:
source.close()
it = IterParseIterator()
it.root = None
@@ -2118,3 +2125,14 @@ except ImportError:
pass
else:
_set_factories(Comment, ProcessingInstruction)
# --------------------------------------------------------------------
def __getattr__(name):
if name == "VERSION":
from warnings import _deprecated
_deprecated("VERSION", remove=(3, 20))
return "1.3.0" # Do not change
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+67 -43
View File
@@ -26,7 +26,7 @@ pub(crate) mod _elementtree {
use crate::vm::{
AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject,
VirtualMachine, atomic_func,
builtins::{PyDict, PyDictRef, PyList, PyModule, PyStr, PyType, PyTypeRef},
builtins::{PyAnyDict, PyDict, PyDictRef, PyList, PyModule, PyStr, PyType, PyTypeRef},
function::{FuncArgs, PySetterValue},
protocol::{PyMappingMethods, PyNumberMethods, PySequenceMethods},
sliceable::{NameElement, SequenceIndex, SliceableSequenceOp},
@@ -174,7 +174,7 @@ pub(crate) mod _elementtree {
tail: Joined,
/// `None` until an attribute is actually needed, mirroring the C
/// implementation's lazily allocated `extra->attrib`.
attrib: Option<PyDictRef>,
attrib: Option<PyRef<PyAnyDict>>,
children: Vec<PyObjectRef>,
}
@@ -222,8 +222,9 @@ pub(crate) mod _elementtree {
}
impl PyElement {
fn new(tag: PyObjectRef, attrib: Option<PyDictRef>, vm: &VirtualMachine) -> Self {
let attrib = attrib.filter(|d| !d.is_empty());
fn new(tag: PyObjectRef, attrib: Option<PyRef<PyAnyDict>>, vm: &VirtualMachine) -> Self {
// CPython elides only empty exact dicts, preserving frozen/subclass identity.
let attrib = attrib.filter(|d| !d.class().is(vm.ctx.types.dict_type) || !d.is_empty());
Self {
inner: PyRwLock::new(ElementInner {
tag,
@@ -287,15 +288,15 @@ pub(crate) mod _elementtree {
Ok(joined)
}
fn attrib_or_new(&self, vm: &VirtualMachine) -> PyDictRef {
fn attrib_or_new(&self, vm: &VirtualMachine) -> PyRef<PyAnyDict> {
let mut inner = self.inner.write();
inner
.attrib
.get_or_insert_with(|| vm.ctx.new_dict())
.get_or_insert_with(|| vm.ctx.new_dict().into())
.clone()
}
fn attrib_opt(&self) -> Option<PyDictRef> {
fn attrib_opt(&self) -> Option<PyRef<PyAnyDict>> {
self.inner.read().attrib.clone()
}
}
@@ -356,14 +357,21 @@ pub(crate) mod _elementtree {
let positional_attrib = args.args.get(extra_positional + 1).cloned();
let attrib = match positional_attrib {
Some(obj) => {
let dict = obj.downcast::<PyDict>().map_err(|obj| {
vm.new_type_error(format!(
"{func_name}() argument {} must be dict, not {}",
extra_positional + 2,
obj.class().name()
))
})?;
let copied = dict.copy().into_ref(&vm.ctx);
let expected = if extra_positional == 0 {
"dict or frozendict"
} else {
"dict"
};
let dict = PyAnyDict::from_object(&obj)
.filter(|_| extra_positional == 0 || obj.downcastable::<PyDict>())
.ok_or_else(|| {
vm.new_type_error(format!(
"{func_name}() argument {} must be {expected}, not {}",
extra_positional + 2,
obj.class().name()
))
})?;
let copied = dict.copy_as_dict(vm)?.into_ref(&vm.ctx);
for (key, value) in args.kwargs.drain(..) {
copied.set_item(&key, value, vm)?;
}
@@ -373,13 +381,13 @@ pub(crate) mod _elementtree {
let popped = args.kwargs.shift_remove("attrib");
let copied = match popped {
Some(obj) => {
let dict = obj.downcast::<PyDict>().map_err(|obj| {
let dict = PyAnyDict::from_object(&obj).ok_or_else(|| {
vm.new_type_error(format!(
"attrib must be dict, not {}",
"attrib must be dict or frozendict, not {}",
obj.class().name()
))
})?;
dict.copy().into_ref(&vm.ctx)
dict.copy_as_dict(vm)?.into_ref(&vm.ctx)
}
None => vm.ctx.new_dict(),
};
@@ -454,7 +462,7 @@ pub(crate) mod _elementtree {
fn init(zelf: &Py<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {
let (tag, attrib) = parse_attrib_args(args, "Element", 0, vm)?;
let attrib = attrib.filter(|d| !d.is_empty());
let attrib = attrib.filter(|d| !d.is_empty()).map(Into::into);
let _recycle = {
let mut inner = zelf.inner.write();
let old_attrib = if attrib.is_some() {
@@ -525,16 +533,21 @@ pub(crate) mod _elementtree {
}
#[pygetset]
fn attrib(zelf: &Py<Self>, vm: &VirtualMachine) -> PyDictRef {
fn attrib(zelf: &Py<Self>, vm: &VirtualMachine) -> PyRef<PyAnyDict> {
zelf.attrib_or_new(vm)
}
#[pygetset(setter)]
fn set_attrib(zelf: &Py<Self>, value: PySetterValue, vm: &VirtualMachine) -> PyResult<()> {
let value = setter_value(value, vm)?;
let dict = value.downcast::<PyDict>().map_err(|obj| {
vm.new_type_error(format!("attrib must be dict, not {}", obj.class().name()))
})?;
let dict = PyAnyDict::from_object(&value)
.ok_or_else(|| {
vm.new_type_error(format!(
"attrib must be dict or frozendict, not {}",
value.class().name()
))
})?
.to_owned();
let _recycle = zelf.inner.write().attrib.replace(dict);
Ok(())
}
@@ -645,7 +658,7 @@ pub(crate) mod _elementtree {
let Some(attrib) = zelf.attrib_opt() else {
return Ok(default);
};
Ok(attrib.get_item_opt(&*args.key, vm)?.unwrap_or(default))
Ok(attrib.inner_getitem_opt(&*args.key, vm)?.unwrap_or(default))
}
#[pymethod]
@@ -664,7 +677,7 @@ pub(crate) mod _elementtree {
let Some(attrib) = zelf.attrib_opt() else {
return vec![];
};
attrib.into_iter().map(|(k, _)| k).collect()
attrib.keys_vec()
}
#[pymethod]
@@ -673,6 +686,7 @@ pub(crate) mod _elementtree {
return vec![];
};
attrib
.items_vec()
.into_iter()
.map(|(k, v)| vm.ctx.new_tuple(vec![k, v]).into())
.collect()
@@ -692,7 +706,7 @@ pub(crate) mod _elementtree {
))
})?;
let attrib = attrib.copy().into_ref(&vm.ctx);
create_element(zelf.class().to_owned(), tag, Some(attrib), vm)
create_element(zelf.class().to_owned(), tag, Some(attrib.into()), vm)
}
#[pymethod]
@@ -847,7 +861,10 @@ pub(crate) mod _elementtree {
fn __getstate__(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<PyDictRef> {
let inner = zelf.inner.read();
let children = vm.ctx.new_list(inner.children.clone());
let attrib = inner.attrib.clone().unwrap_or_else(|| vm.ctx.new_dict());
let attrib = inner
.attrib
.clone()
.unwrap_or_else(|| vm.ctx.new_dict().into());
let text = inner.text.obj.clone();
let tail = inner.tail.obj.clone();
let tag = inner.tag.clone();
@@ -891,14 +908,16 @@ pub(crate) mod _elementtree {
None => None,
};
let attrib = match attrib {
Some(attrib) if !vm.is_none(&attrib) => {
Some(attrib.downcast::<PyDict>().map_err(|obj| {
vm.new_type_error(format!(
"attrib must be dict, not {}",
obj.class().name()
))
})?)
}
Some(attrib) if !vm.is_none(&attrib) => Some(
PyAnyDict::from_object(&attrib)
.ok_or_else(|| {
vm.new_type_error(format!(
"attrib must be dict or frozendict, not {}",
attrib.class().name()
))
})?
.to_owned(),
),
_ => None,
};
let new_text = match text {
@@ -1051,7 +1070,7 @@ pub(crate) mod _elementtree {
fn create_element(
cls: PyTypeRef,
tag: PyObjectRef,
attrib: Option<PyDictRef>,
attrib: Option<PyRef<PyAnyDict>>,
vm: &VirtualMachine,
) -> PyResult {
let element = PyElement::new(tag, attrib, vm);
@@ -1101,11 +1120,14 @@ pub(crate) mod _elementtree {
};
let tag = deepcopy_obj(&tag, memo, vm)?;
let attrib = match attrib {
Some(attrib) => Some(
deepcopy_obj(attrib.as_object(), memo, vm)?
.downcast::<PyDict>()
.map_err(|_| vm.new_type_error("attrib must be dict"))?,
),
Some(attrib) => {
let copied = deepcopy_obj(attrib.as_object(), memo, vm)?;
Some(
PyAnyDict::from_object(&copied)
.ok_or_else(|| vm.new_type_error("attrib must be dict or frozendict"))?
.to_owned(),
)
}
None => None,
};
let new = create_element(zelf.class().to_owned(), tag, attrib, vm)?;
@@ -1279,7 +1301,7 @@ pub(crate) mod _elementtree {
))
})?;
let (tag, attrib) = parse_attrib_args(args, "SubElement", 1, vm)?;
let element = create_element(parent.class().to_owned(), tag, attrib, vm)?;
let element = create_element(parent.class().to_owned(), tag, attrib.map(Into::into), vm)?;
parent.push(element.clone());
Ok(element)
}
@@ -1555,7 +1577,9 @@ pub(crate) mod _elementtree {
let factory = self.state.read().element_factory.clone();
let node = match factory {
// CPython adopts the caller's dict rather than copying it.
None => PyElement::new(tag, attrib, vm).into_ref(&vm.ctx).into(),
None => PyElement::new(tag, attrib.map(Into::into), vm)
.into_ref(&vm.ctx)
.into(),
Some(factory) => {
let attrib = attrib.unwrap_or_else(|| vm.ctx.new_dict());
factory.call((tag, attrib), vm)?
+9
View File
@@ -205,6 +205,15 @@ impl PyAnyDict {
pub fn is_empty(&self) -> bool {
self.as_dict().is_empty()
}
pub fn copy_as_dict(&self, vm: &VirtualMachine) -> PyResult<PyDict> {
if self.is_empty() || self.uses_builtin_iter(vm) {
return Ok(self.as_dict().copy());
}
let copied = PyDict::default();
copied.merge_object(self.as_object().to_owned(), vm)?;
Ok(copied)
}
#[must_use]
pub fn keys_vec(&self) -> Vec<PyObjectRef> {
self.as_dict().keys_vec()
+73
View File
@@ -83,3 +83,76 @@ for value, type_name in (
assert str(caught.exception) == (
"remove() argument must be xml.etree.ElementTree.Element, not " + type_name
)
def test_element_frozen_attrib_identity_bounded():
class Frozen(frozendict):
def __getitem__(self, key):
raise AssertionError("Element.get must use the stored entries")
def keys(self):
raise AssertionError("Element.keys must use the stored entries")
def items(self):
raise AssertionError("Element.items must use the stored entries")
for mapping in (frozendict(), frozendict(a=1), Frozen(a=1)):
element = Element("root")
element.attrib = mapping
assert element.attrib is mapping
assert element.get("a", 7) == (1 if mapping else 7)
assert element.get("missing", 7) == 7
assert element.keys() == (["a"] if mapping else [])
assert element.items() == ([("a", 1)] if mapping else [])
with assert_raises(AttributeError):
del element.attrib
assert element.attrib is mapping
element.clear()
assert type(element.attrib) is dict and element.attrib == {}
element.set("new", 2)
assert element.attrib == {"new": 2}
assert dict(mapping) == ({"a": 1} if mapping else {})
def test_element_frozen_attrib_mutation_bounded():
class Key:
def __hash__(self):
raise AssertionError("frozen assignment must fail before hashing")
element = Element("root")
mapping = frozendict(a=1)
element.attrib = mapping
for key in ("a", "new", Key()):
with assert_raises(TypeError) as caught:
element.set(key, 2)
assert str(caught.exception) == (
"frozendict object does not support item assignment"
)
assert element.attrib is mapping and mapping == {"a": 1}
def test_element_frozen_attrib_copy_state_bounded():
import copy
for mapping in (frozendict(), frozendict(a=[1])):
element = Element("root")
element.attrib = mapping
shallow = copy.copy(element)
assert shallow is not element and shallow.attrib is mapping
deep = copy.deepcopy(element)
assert deep is not element
assert type(deep.attrib) is frozendict and deep.attrib == mapping
if mapping:
assert deep.attrib["a"] is not mapping["a"]
deep.attrib["a"].append(2)
assert mapping["a"] == [1]
state = element.__getstate__()
assert state["attrib"] is mapping
restored = Element("restored")
restored.__setstate__(state)
assert restored.attrib is mapping
test_element_frozen_attrib_identity_bounded()
test_element_frozen_attrib_mutation_bounded()
test_element_frozen_attrib_copy_state_bounded()