diff --git a/Lib/test/test_structseq.py b/Lib/test/test_structseq.py index a9fe19302..8ef6dd2fe 100644 --- a/Lib/test/test_structseq.py +++ b/Lib/test/test_structseq.py @@ -1,6 +1,12 @@ +import copy +import gc import os +import pickle +import re +import textwrap import time import unittest +from test.support import script_helper class StructSeqTest(unittest.TestCase): @@ -37,7 +43,7 @@ class StructSeqTest(unittest.TestCase): # os.stat() gives a complicated struct sequence. st = os.stat(__file__) rep = repr(st) - self.assertTrue(rep.startswith("os.stat_result")) + self.assertStartsWith(rep, "os.stat_result") self.assertIn("st_mode=", rep) self.assertIn("st_ino=", rep) self.assertIn("st_dev=", rep) @@ -81,6 +87,7 @@ class StructSeqTest(unittest.TestCase): self.assertEqual(t.n_unnamed_fields, 0) self.assertEqual(t.n_fields, time._STRUCT_TM_ITEMS) + @unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: Unexpected keyword argument dict def test_constructor(self): t = time.struct_time @@ -89,10 +96,72 @@ class StructSeqTest(unittest.TestCase): self.assertRaises(TypeError, t, "123") self.assertRaises(TypeError, t, "123", dict={}) self.assertRaises(TypeError, t, "123456789", dict=None) + self.assertRaises(TypeError, t, seq="123456789", dict={}) + + self.assertEqual(t("123456789"), tuple("123456789")) + self.assertEqual(t("123456789", {}), tuple("123456789")) + self.assertEqual(t("123456789", dict={}), tuple("123456789")) + self.assertEqual(t(sequence="123456789", dict={}), tuple("123456789")) + + self.assertEqual(t("1234567890"), tuple("123456789")) + self.assertEqual(t("1234567890").tm_zone, "0") + self.assertEqual(t("123456789", {"tm_zone": "some zone"}), tuple("123456789")) + self.assertEqual(t("123456789", {"tm_zone": "some zone"}).tm_zone, "some zone") s = "123456789" self.assertEqual("".join(t(s)), s) + @unittest.expectedFailure # TODO: RUSTPYTHON; Wrong error message + def test_constructor_with_duplicate_fields(self): + t = time.struct_time + + error_message = re.escape("got duplicate or unexpected field name(s)") + with self.assertRaisesRegex(TypeError, error_message): + t("1234567890", dict={"tm_zone": "some zone"}) + with self.assertRaisesRegex(TypeError, error_message): + t("1234567890", dict={"tm_zone": "some zone", "tm_mon": 1}) + with self.assertRaisesRegex(TypeError, error_message): + t("1234567890", dict={"error": 0, "tm_zone": "some zone"}) + with self.assertRaisesRegex(TypeError, error_message): + t("1234567890", dict={"error": 0, "tm_zone": "some zone", "tm_mon": 1}) + + @unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: expected at most 1 arguments, got 2 + def test_constructor_with_duplicate_unnamed_fields(self): + assert os.stat_result.n_unnamed_fields > 0 + n_visible_fields = os.stat_result.n_sequence_fields + + r = os.stat_result(range(n_visible_fields), {'st_atime': -1.0}) + self.assertEqual(r.st_atime, -1.0) + self.assertEqual(r, tuple(range(n_visible_fields))) + + r = os.stat_result((*range(n_visible_fields), -1.0)) + self.assertEqual(r.st_atime, -1.0) + self.assertEqual(r, tuple(range(n_visible_fields))) + + with self.assertRaisesRegex(TypeError, + re.escape("got duplicate or unexpected field name(s)")): + os.stat_result((*range(n_visible_fields), -1.0), {'st_atime': -1.0}) + + @unittest.expectedFailure # TODO: RUSTPYTHON; Wrong error message + def test_constructor_with_unknown_fields(self): + t = time.struct_time + + error_message = re.escape("got duplicate or unexpected field name(s)") + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"tm_year": 0}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"tm_year": 0, "tm_mon": 1}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"tm_zone": "some zone", "tm_mon": 1}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"tm_zone": "some zone", "error": 0}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"error": 0, "tm_zone": "some zone", "tm_mon": 1}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"error": 0}) + with self.assertRaisesRegex(TypeError, error_message): + t("123456789", dict={"tm_zone": "some zone", "error": 0}) + def test_eviltuple(self): class Exc(Exception): pass @@ -106,9 +175,80 @@ class StructSeqTest(unittest.TestCase): self.assertRaises(Exc, time.struct_time, C()) - def test_reduce(self): + def test_pickling(self): t = time.gmtime() - x = t.__reduce__() + for proto in range(pickle.HIGHEST_PROTOCOL + 1): + p = pickle.dumps(t, proto) + t2 = pickle.loads(p) + self.assertEqual(t2.__class__, t.__class__) + self.assertEqual(t2, t) + self.assertEqual(t2.tm_year, t.tm_year) + self.assertEqual(t2.tm_zone, t.tm_zone) + + @unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: expected at most 1 arguments, got 2 + def test_pickling_with_unnamed_fields(self): + assert os.stat_result.n_unnamed_fields > 0 + + r = os.stat_result(range(os.stat_result.n_sequence_fields), + {'st_atime': 1.0, 'st_atime_ns': 2.0}) + for proto in range(pickle.HIGHEST_PROTOCOL + 1): + p = pickle.dumps(r, proto) + r2 = pickle.loads(p) + self.assertEqual(r2.__class__, r.__class__) + self.assertEqual(r2, r) + self.assertEqual(r2.st_mode, r.st_mode) + self.assertEqual(r2.st_atime, r.st_atime) + self.assertEqual(r2.st_atime_ns, r.st_atime_ns) + + def test_copying(self): + n_fields = time.struct_time.n_fields + t = time.struct_time([[i] for i in range(n_fields)]) + + t2 = copy.copy(t) + self.assertEqual(t2.__class__, t.__class__) + self.assertEqual(t2, t) + self.assertEqual(t2.tm_year, t.tm_year) + self.assertEqual(t2.tm_zone, t.tm_zone) + self.assertIs(t2[0], t[0]) + self.assertIs(t2.tm_year, t.tm_year) + + t3 = copy.deepcopy(t) + self.assertEqual(t3.__class__, t.__class__) + self.assertEqual(t3, t) + self.assertEqual(t3.tm_year, t.tm_year) + self.assertEqual(t3.tm_zone, t.tm_zone) + self.assertIsNot(t3[0], t[0]) + self.assertIsNot(t3.tm_year, t.tm_year) + + @unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: expected at most 1 arguments, got 2 + def test_copying_with_unnamed_fields(self): + assert os.stat_result.n_unnamed_fields > 0 + + n_sequence_fields = os.stat_result.n_sequence_fields + r = os.stat_result([[i] for i in range(n_sequence_fields)], + {'st_atime': [1.0], 'st_atime_ns': [2.0]}) + + r2 = copy.copy(r) + self.assertEqual(r2.__class__, r.__class__) + self.assertEqual(r2, r) + self.assertEqual(r2.st_mode, r.st_mode) + self.assertEqual(r2.st_atime, r.st_atime) + self.assertEqual(r2.st_atime_ns, r.st_atime_ns) + self.assertIs(r2[0], r[0]) + self.assertIs(r2.st_mode, r.st_mode) + self.assertIs(r2.st_atime, r.st_atime) + self.assertIs(r2.st_atime_ns, r.st_atime_ns) + + r3 = copy.deepcopy(r) + self.assertEqual(r3.__class__, r.__class__) + self.assertEqual(r3, r) + self.assertEqual(r3.st_mode, r.st_mode) + self.assertEqual(r3.st_atime, r.st_atime) + self.assertEqual(r3.st_atime_ns, r.st_atime_ns) + self.assertIsNot(r3[0], r[0]) + self.assertIsNot(r3.st_mode, r.st_mode) + self.assertIsNot(r3.st_atime, r.st_atime) + self.assertIsNot(r3.st_atime_ns, r.st_atime_ns) def test_extended_getslice(self): # Test extended slicing by comparing with list slicing. @@ -133,6 +273,104 @@ class StructSeqTest(unittest.TestCase): self.assertEqual(os.stat_result.n_unnamed_fields, 3) self.assertEqual(os.stat_result.__match_args__, expected_args) + def test_copy_replace_all_fields_visible(self): + assert os.times_result.n_unnamed_fields == 0 + assert os.times_result.n_sequence_fields == os.times_result.n_fields + + t = os.times() + + # visible fields + self.assertEqual(copy.replace(t), t) + self.assertIsInstance(copy.replace(t), os.times_result) + self.assertEqual(copy.replace(t, user=1.5), (1.5, *t[1:])) + self.assertEqual(copy.replace(t, system=2.5), (t[0], 2.5, *t[2:])) + self.assertEqual(copy.replace(t, user=1.5, system=2.5), (1.5, 2.5, *t[2:])) + + # unknown fields + with self.assertRaisesRegex(TypeError, 'unexpected field name'): + copy.replace(t, error=-1) + with self.assertRaisesRegex(TypeError, 'unexpected field name'): + copy.replace(t, user=1, error=-1) + + @unittest.expectedFailure # TODO: RUSTPYTHON; Wrong error message + def test_copy_replace_with_invisible_fields(self): + assert time.struct_time.n_unnamed_fields == 0 + assert time.struct_time.n_sequence_fields < time.struct_time.n_fields + + t = time.gmtime(0) + + # visible fields + t2 = copy.replace(t) + self.assertEqual(t2, (1970, 1, 1, 0, 0, 0, 3, 1, 0)) + self.assertIsInstance(t2, time.struct_time) + t3 = copy.replace(t, tm_year=2000) + self.assertEqual(t3, (2000, 1, 1, 0, 0, 0, 3, 1, 0)) + self.assertEqual(t3.tm_year, 2000) + t4 = copy.replace(t, tm_mon=2) + self.assertEqual(t4, (1970, 2, 1, 0, 0, 0, 3, 1, 0)) + self.assertEqual(t4.tm_mon, 2) + t5 = copy.replace(t, tm_year=2000, tm_mon=2) + self.assertEqual(t5, (2000, 2, 1, 0, 0, 0, 3, 1, 0)) + self.assertEqual(t5.tm_year, 2000) + self.assertEqual(t5.tm_mon, 2) + + # named invisible fields + self.assertHasAttr(t, 'tm_zone') + with self.assertRaisesRegex(AttributeError, 'readonly attribute'): + t.tm_zone = 'some other zone' + self.assertEqual(t2.tm_zone, t.tm_zone) + self.assertEqual(t3.tm_zone, t.tm_zone) + self.assertEqual(t4.tm_zone, t.tm_zone) + t6 = copy.replace(t, tm_zone='some other zone') + self.assertEqual(t, t6) + self.assertEqual(t6.tm_zone, 'some other zone') + t7 = copy.replace(t, tm_year=2000, tm_zone='some other zone') + self.assertEqual(t7, (2000, 1, 1, 0, 0, 0, 3, 1, 0)) + self.assertEqual(t7.tm_year, 2000) + self.assertEqual(t7.tm_zone, 'some other zone') + + # unknown fields + with self.assertRaisesRegex(TypeError, 'unexpected field name'): + copy.replace(t, error=2) + with self.assertRaisesRegex(TypeError, 'unexpected field name'): + copy.replace(t, tm_year=2000, error=2) + with self.assertRaisesRegex(TypeError, 'unexpected field name'): + copy.replace(t, tm_zone='some other zone', error=2) + + def test_copy_replace_with_unnamed_fields(self): + assert os.stat_result.n_unnamed_fields > 0 + + r = os.stat_result(range(os.stat_result.n_sequence_fields)) + + error_message = re.escape('__replace__() is not supported') + with self.assertRaisesRegex(TypeError, error_message): + copy.replace(r) + with self.assertRaisesRegex(TypeError, error_message): + copy.replace(r, st_mode=1) + with self.assertRaisesRegex(TypeError, error_message): + copy.replace(r, error=2) + with self.assertRaisesRegex(TypeError, error_message): + copy.replace(r, st_mode=1, error=2) + + def test_reference_cycle(self): + # gh-122527: Check that a structseq that's part of a reference cycle + # with its own type doesn't crash. Previously, if the type's dictionary + # was cleared first, the structseq instance would crash in the + # destructor. + script_helper.assert_python_ok("-c", textwrap.dedent(r""" + import time + t = time.gmtime() + type(t).refcyle = t + """)) + + def test_replace_gc_tracked(self): + # Verify that __replace__ results are properly GC-tracked + time_struct = time.gmtime(0) + lst = [] + replaced_struct = time_struct.__replace__(tm_year=lst) + lst.append(replaced_struct) + + self.assertTrue(gc.is_tracked(replaced_struct)) if __name__ == "__main__": unittest.main() diff --git a/crates/derive-impl/src/pystructseq.rs b/crates/derive-impl/src/pystructseq.rs index 4059aba63..874f85741 100644 --- a/crates/derive-impl/src/pystructseq.rs +++ b/crates/derive-impl/src/pystructseq.rs @@ -102,12 +102,12 @@ fn parse_fields(input: &mut DeriveInput) -> Result { bail_span!(input, "Only #[pystruct_sequence(...)] form is allowed"); }; - let idents: Vec<_> = l + let idents = l .nested .iter() .filter_map(|n| n.get_ident()) .cloned() - .collect(); + .collect::>(); for ident in idents { match ident.to_string().as_str() { @@ -205,7 +205,7 @@ pub(crate) fn impl_pystruct_sequence_data( let n_unnamed_fields = field_info.n_unnamed_fields(); // Generate field index constants for visible fields (with cfg guards) - let field_indices: Vec<_> = visible_fields + let field_indices = visible_fields .iter() .enumerate() .map(|(i, field)| { @@ -216,78 +216,58 @@ pub(crate) fn impl_pystruct_sequence_data( pub const #const_name: usize = #i; } }) - .collect(); + .collect::>(); // Generate field name entries with cfg guards for named fields - let named_field_names: Vec<_> = named_fields + let named_field_names = named_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { stringify!(#ident), } - } else { - quote! { - #(#cfg_attrs)* - { stringify!(#ident) }, - } + quote! { + #(#cfg_attrs)* + { stringify!(#ident) }, } }) - .collect(); + .collect::>(); // Generate field name entries with cfg guards for skipped fields - let skipped_field_names: Vec<_> = skipped_fields + let skipped_field_names = skipped_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { stringify!(#ident), } - } else { - quote! { - #(#cfg_attrs)* - { stringify!(#ident) }, - } + quote! { + #(#cfg_attrs)* + { stringify!(#ident) }, } }) - .collect(); + .collect::>(); // Generate into_tuple items with cfg guards - let visible_tuple_items: Vec<_> = visible_fields + let visible_tuple_items = visible_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { - ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm), - } - } else { - quote! { - #(#cfg_attrs)* - { ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm) }, - } + quote! { + #(#cfg_attrs)* + { ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm) }, } }) - .collect(); + .collect::>(); - let skipped_tuple_items: Vec<_> = skipped_fields + let skipped_tuple_items = skipped_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { - ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm), - } - } else { - quote! { - #(#cfg_attrs)* - { ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm) }, - } + quote! { + #(#cfg_attrs)* + { ::rustpython_vm::convert::ToPyObject::to_pyobject(self.#ident, vm) }, } }) - .collect(); + .collect::>(); // Generate TryFromObject impl only when try_from_object=true let try_from_object_impl = if try_from_object { @@ -317,44 +297,33 @@ pub(crate) fn impl_pystruct_sequence_data( // Generate try_from_elements trait override only when try_from_object=true let try_from_elements_trait_override = if try_from_object { - let visible_field_inits: Vec<_> = visible_fields + let visible_field_inits = visible_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { #ident: iter.next().unwrap().clone().try_into_value(vm)?, } - } else { - quote! { - #(#cfg_attrs)* - #ident: iter.next().unwrap().clone().try_into_value(vm)?, - } + quote! { + #(#cfg_attrs)* + #ident: iter.next().unwrap().clone().try_into_value(vm)?, } }) - .collect(); - let skipped_field_inits: Vec<_> = skipped_fields + .collect::>(); + + let skipped_field_inits = skipped_fields .iter() .map(|f| { let ident = &f.ident; let cfg_attrs = &f.cfg_attrs; - if cfg_attrs.is_empty() { - quote! { - #ident: match iter.next() { - Some(v) => v.clone().try_into_value(vm)?, - None => vm.ctx.none(), - }, - } - } else { - quote! { - #(#cfg_attrs)* - #ident: match iter.next() { - Some(v) => v.clone().try_into_value(vm)?, - None => vm.ctx.none(), - }, - } + quote! { + #(#cfg_attrs)* + #ident: match iter.next() { + Some(v) => v.clone().try_into_value(vm)?, + None => vm.ctx.none(), + }, } }) - .collect(); + .collect::>(); + quote! { fn try_from_elements( elements: Vec<::rustpython_vm::PyObjectRef>, @@ -426,6 +395,7 @@ impl ItemMeta for PyStructSequenceMeta { fn from_inner(inner: ItemMetaInner) -> Self { Self { inner } } + fn inner(&self) -> &ItemMetaInner { &self.inner }