Add native _datetime module (#8872)

* Add native _datetime module

Assisted-by: Claude Code:claude-opus-5-5
Assisted-by: Codex:GPT-6

* Use pymember and safe timezone storage in datetime

Assisted-by: Codex:GPT-6
This commit is contained in:
1ndahous3 authored and GitHub committed 2026-09-29 11:45:11 +09:00
1 parent 04eb0ad94c
commit f909d2a30d
12 files changed
+4896 -16

No files matched your search

+1
View File
@@ -7326,6 +7326,7 @@ class ExtensionModuleTests(unittest.TestCase):
""")
script_helper.assert_python_ok('-c', script)
@unittest.expectedFailure # TODO: RUSTPYTHON; no concurrent.futures.InterpreterPoolExecutor
def test_concurrent_initialization_subinterpreter(self):
# gh-136421: Concurrent initialization of _datetime across multiple
# interpreters wasn't thread-safe due to its static types.
+2 -8
View File
@@ -21,14 +21,8 @@ def load_tests(loader, tests, pattern):
for modname in ['datetime', '_datetime', '_pydatetime', '_strptime']:
sys.modules.pop(modname, None)
test_modules = [
pure_tests,
# fast_tests # XXX: RUSTPYTHON; not supported yet
]
test_suffixes = [
"_Pure",
# "_Fast" # XXX: RUSTPYTHON; not supported yet
]
test_modules = [pure_tests, fast_tests]
test_suffixes = ["_Pure", "_Fast"]
# XXX(gb) First run all the _Pure tests, then all the _Fast tests. You might
# not believe this, but in spite of all the sys.modules trickery running a _Pure
-1
View File
@@ -175,7 +175,6 @@ class custom_descriptor:
class TestPredicates(IsTestBase):
@unittest.expectedFailure # TODO: RUSTPYTHON; datetime.timedelta.days is a property until _datetime members exist
def test_excluding_predicates(self):
global tb
self.istest(inspect.isbuiltin, 'sys.exit')
-1
View File
@@ -1134,7 +1134,6 @@ class TestBinaryPlistlib(unittest.TestCase):
with self.assertRaisesRegex(TypeError, msg):
plistlib.dumps(dt, fmt=plistlib.FMT_BINARY, aware_datetime=False)
@unittest.expectedFailure # TODO: RUSTPYTHON; Wrong error message
def test_dump_utc_aware_datetime_without_aware_datetime_option(self):
dt = datetime.datetime(2345, 6, 7, 8, tzinfo=datetime.UTC)
msg = "can't subtract offset-naive and offset-aware datetimes"
-1
View File
@@ -1614,7 +1614,6 @@ class TestDescriptions(unittest.TestCase):
self.assertEqual(self._get_summary_line(set().__contains__),
"__contains__(object, /) method of builtins.set instance")
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_unbound_builtin_classmethod_noargs(self):
self.assertEqual(self._get_summary_line(datetime.datetime.__dict__['utcnow']),
"utcnow(type, /) unbound datetime.datetime method")
+4 -3
View File
@@ -218,9 +218,9 @@ pub fn derive_from_args(input: TokenStream) -> TokenStream {
/// Declares an offset member on a payload field. The struct `#[pyclass]` builds
/// a `PyClassDef::MEMBERS` table and `extend_class` registers one
/// `member_descriptor` per entry. The member kind is inferred from the field
/// type: `bool` / `AtomicBool`, `i32` / `AtomicI32`, `u32` / `AtomicU32`,
/// type: `bool` / `AtomicBool`, `u8` / `AtomicU8`, `i32` / `AtomicI32`, `u32` / `AtomicU32`,
/// `isize` / `AtomicIsize`, `f64` / `AtomicF64`, or an object pointer
/// (`PyObjectRef`, `PyRef<T>`, `Option` of those, `PyAtomicRef<PyObject>`,
/// (`PyObjectRef`, `PyRef<T>`, `Option` of those, `PyObjectCell`, `PyAtomicRef<PyObject>`,
/// `PyAtomicRef<Option<PyObject>>`, `PyAtomicRef<Option<T>>`,
/// `&'static Py<T>`, `&'static PyStrInterned`).
///
@@ -230,7 +230,8 @@ pub fn derive_from_args(input: TokenStream) -> TokenStream {
/// - `writable`: accept stores. Members are readonly without it. A writable
/// object member must be `PyAtomicRef<PyObject>` (never null) or
/// `PyAtomicRef<Option<PyObject>>` (nullable), a writable bool must be
/// `AtomicBool`, a writable int must be `AtomicI32`, a writable uint must be
/// `AtomicBool`, a writable unsigned byte must be `AtomicU8`, a writable int
/// must be `AtomicI32`, a writable uint must be
/// `AtomicU32`, a writable double must be `AtomicF64`, and a writable
/// py_ssize_t must be `AtomicIsize`.
/// - `audit_read`: audit `object.__getattr__` before the load.
File diff suppressed because it is too large. Load diff
+2
View File
@@ -54,6 +54,7 @@ mod math;
#[cfg(all(feature = "host_env", any(unix, windows)))]
mod mmap;
mod _datetime;
mod _heapq;
mod _queue;
mod pickle;
@@ -257,6 +258,7 @@ pub fn stdlib_module_defs(ctx: &Context) -> Vec<&'static builtins::PyModuleDef>
pickle::module_def(ctx),
pyexpat::module_def(ctx),
pystruct::module_def(ctx),
_datetime::module_def(ctx),
_heapq::module_def(ctx),
_queue::module_def(ctx),
random::module_def(ctx),
+209
View File
@@ -324,6 +324,8 @@ pub enum MemberKind {
/// as `None`. Assignment is rejected.
String = 5,
Object = 6,
/// `unsigned char`. Writable cells are `AtomicU8`.
UByte = 9,
/// `unsigned short`. Writable cells are `AtomicU16`.
UShort = 10,
Uint = 11,
@@ -365,6 +367,7 @@ impl MemberKind {
4 => Some(Self::Double),
5 => Some(Self::String),
6 => Some(Self::Object),
9 => Some(Self::UByte),
10 => Some(Self::UShort),
11 => Some(Self::Uint),
12 => Some(Self::ULong),
@@ -443,6 +446,15 @@ impl MemberLayout for core::sync::atomic::AtomicU16 {
}
impl MemberCell for core::sync::atomic::AtomicU16 {}
impl MemberLayout for u8 {
const KIND: MemberKind = MemberKind::UByte;
}
impl MemberLayout for core::sync::atomic::AtomicU8 {
const KIND: MemberKind = MemberKind::UByte;
const ATOMIC: bool = true;
}
impl MemberCell for core::sync::atomic::AtomicU8 {}
// `i64` is `long` where `c_long` is 8 bytes, and `long long` otherwise.
impl MemberLayout for i64 {
const KIND: MemberKind = MemberKind::from_i64();
@@ -534,6 +546,10 @@ impl MemberLayout for PyObjectRef {
impl MemberLayout for Option<PyObjectRef> {
const KIND: MemberKind = MemberKind::Object;
}
impl MemberLayout for crate::object::PyObjectCell {
const KIND: MemberKind = MemberKind::Object;
const ATOMIC: bool = true;
}
impl<T> MemberLayout for PyRef<T> {
const KIND: MemberKind = MemberKind::Object;
}
@@ -746,6 +762,7 @@ impl PyMemberDescriptor {
| MemberKind::LongLong
| MemberKind::Short
| MemberKind::String
| MemberKind::UByte
| MemberKind::Uint
| MemberKind::ULong
| MemberKind::ULongLong
@@ -929,6 +946,21 @@ fn load_u16(obj: &PyObject, offset: isize, atomic: bool) -> u16 {
}
}
fn load_u8(obj: &PyObject, offset: isize, atomic: bool) -> u8 {
let addr = member_addr(obj, offset);
if atomic {
// SAFETY: `PyMemberFlags::ATOMIC` or a writable member addresses an `AtomicU8`.
unsafe {
(*addr.cast::<core::sync::atomic::AtomicU8>())
.load(core::sync::atomic::Ordering::Relaxed)
}
} else {
// SAFETY: a readonly unsigned char member addresses a `u8` that is not
// written after publication.
unsafe { addr.cast::<u8>().read() }
}
}
fn load_i64(obj: &PyObject, offset: isize, atomic: bool) -> i64 {
let addr = member_addr(obj, offset);
if atomic {
@@ -1172,6 +1204,10 @@ fn member_get_one(
.ctx
.new_int(load_u16(obj, offset, member.atomic_storage()))
.into(),
MemberKind::UByte => vm
.ctx
.new_int(load_u8(obj, offset, member.atomic_storage()))
.into(),
MemberKind::Long => vm
.ctx
.new_int(load_c_long(obj, offset, member.atomic_storage()))
@@ -1356,6 +1392,21 @@ fn member_set_one(
warn_member(vm, "Truncation of value to unsigned short")?;
}
}
MemberKind::UByte => {
let PySetterValue::Assign(value) = value else {
return Err(vm.new_type_error("can't delete numeric/char attribute"));
};
let long_val = member_as_c_long(&value, vm)?;
let stored = long_val as u8;
// SAFETY: a writable unsigned char member addresses an `AtomicU8`.
unsafe {
(*member_addr(obj, offset).cast::<core::sync::atomic::AtomicU8>())
.store(stored, core::sync::atomic::Ordering::Relaxed);
}
if long_val > u8::MAX as core::ffi::c_long || long_val < 0 {
warn_member(vm, "Truncation of value to unsigned char")?;
}
}
MemberKind::Long => {
let PySetterValue::Assign(value) = value else {
return Err(vm.new_type_error("can't delete numeric/char attribute"));
@@ -2130,3 +2181,161 @@ impl Comparable for PyMethodWrapper {
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{protocol::PyNumberMethods, types::AsNumber};
use core::sync::atomic::{AtomicU8, Ordering};
#[pyclass(name = "UByteMembers", module = false)]
#[derive(Debug, PyPayload)]
#[repr(C)]
struct UByteMembers {
prefix: u8,
#[pymember]
readonly: u8,
#[pymember(writable)]
writable: AtomicU8,
suffix: u8,
}
#[pyclass(with(AsNumber))]
impl UByteMembers {}
impl AsNumber for UByteMembers {
fn as_number() -> &'static PyNumberMethods {
static METHODS: PyNumberMethods = PyNumberMethods {
index: Some(|_, vm| Ok(vm.ctx.new_int(42).into())),
..PyNumberMethods::NOT_IMPLEMENTED
};
&METHODS
}
}
#[pyclass(name = "ReadonlyAtomicMembers", module = false)]
#[derive(Debug, PyPayload)]
struct ReadonlyAtomicMembers {
#[pymember]
byte: AtomicU8,
#[pymember]
object: crate::object::PyObjectCell,
}
#[pyclass]
impl ReadonlyAtomicMembers {}
#[test]
fn readonly_atomic_members_use_atomic_storage() {
crate::Interpreter::without_stdlib(Default::default()).enter(|vm| {
let _class = ReadonlyAtomicMembers::make_static_type();
let obj = ReadonlyAtomicMembers {
byte: AtomicU8::new(128),
object: None.into(),
}
.into_ref(&vm.ctx);
for name in ["byte", "object"] {
let descriptor = obj.class().as_object().get_attr(name, vm).unwrap();
let descriptor = descriptor.downcast_ref::<PyMemberDescriptor>().unwrap();
assert!(descriptor.member.readonly());
assert!(descriptor.member.atomic_storage());
let err = obj
.as_object()
.set_attr(name, vm.ctx.none(), vm)
.unwrap_err();
assert!(err.fast_isinstance(vm.ctx.exceptions.attribute_error));
}
assert!(vm.is_none(&obj.as_object().get_attr("object", vm).unwrap()));
obj.byte.store(255, Ordering::Relaxed);
let value: PyObjectRef = vm.ctx.new_int(42).into();
obj.object.store(Some(value.clone()));
assert!(obj.as_object().get_attr("object", vm).unwrap().is(&value));
let byte = obj.as_object().get_attr("byte", vm).unwrap();
assert_eq!(
u8::try_from(
byte.downcast_ref::<crate::builtins::PyInt>()
.unwrap()
.as_bigint()
)
.unwrap(),
255
);
obj.object.store(None);
assert!(vm.is_none(&obj.as_object().get_attr("object", vm).unwrap()));
});
}
#[test]
fn unsigned_byte_members_preserve_adjacent_fields() {
crate::Interpreter::without_stdlib(Default::default()).enter(|vm| {
let _class = UByteMembers::make_static_type();
let obj = UByteMembers {
prefix: 17,
readonly: 255,
writable: AtomicU8::new(128),
suffix: 29,
}
.into_ref(&vm.ctx);
assert_eq!(MemberKind::from_i32(9), Some(MemberKind::UByte));
for (name, expected) in [("readonly", 255), ("writable", 128)] {
let value = obj.as_object().get_attr(name, vm).unwrap();
let value = value.downcast_ref::<crate::builtins::PyInt>().unwrap();
assert_eq!(i32::try_from(value.as_bigint()).unwrap(), expected);
}
obj.as_object()
.set_attr("writable", vm.ctx.new_int(255), vm)
.unwrap();
assert_eq!(obj.writable.load(Ordering::Relaxed), 255);
assert_eq!((obj.prefix, obj.readonly, obj.suffix), (17, 255, 29));
let err = obj
.as_object()
.set_attr("readonly", vm.ctx.new_int(0), vm)
.unwrap_err();
assert!(err.fast_isinstance(vm.ctx.exceptions.attribute_error));
let err = obj.as_object().del_attr("writable", vm).unwrap_err();
assert!(err.fast_isinstance(vm.ctx.exceptions.type_error));
assert_eq!(obj.writable.load(Ordering::Relaxed), 255);
let err = obj
.as_object()
.set_attr("writable", vm.ctx.new_int(i128::MAX), vm)
.unwrap_err();
assert!(err.fast_isinstance(vm.ctx.exceptions.overflow_error));
assert_eq!(obj.writable.load(Ordering::Relaxed), 255);
obj.as_object()
.set_attr("writable", obj.to_owned(), vm)
.unwrap();
assert_eq!(obj.writable.load(Ordering::Relaxed), 42);
let filters = vm.state.warnings.filters.to_owned();
filters.borrow_vec_mut().insert(
0,
vm.ctx
.new_tuple(vec![
vm.ctx.new_str("error").into(),
vm.ctx.none(),
vm.ctx.exceptions.runtime_warning.to_owned().into(),
vm.ctx.none(),
vm.ctx.new_int(0).into(),
])
.into(),
);
vm.state.warnings.filters_mutated();
for (input, expected) in [(256, 0), (-1, 255)] {
let err = obj
.as_object()
.set_attr("writable", vm.ctx.new_int(input), vm)
.unwrap_err();
assert!(err.fast_isinstance(vm.ctx.exceptions.runtime_warning));
// As in CPython, truncation is stored before the warning is raised.
assert_eq!(obj.writable.load(Ordering::Relaxed), expected);
assert_eq!((obj.prefix, obj.readonly, obj.suffix), (17, 255, 29));
}
});
}
}
+2 -2
View File
@@ -110,8 +110,8 @@ pub mod windows;
pub use self::convert::{TryFromBorrowedObject, TryFromObject};
pub use self::object::{
AsObject, Py, PyAtomicRef, PyExact, PyObject, PyObjectRef, PyPayload, PyRef, PyRefExact,
PyResult, PyStackRef, PyWeakRef,
AsObject, Py, PyAtomicRef, PyExact, PyObject, PyObjectCell, PyObjectRef, PyPayload, PyRef,
PyRefExact, PyResult, PyStackRef, PyWeakRef,
};
pub use self::vm::runtime;
pub use self::vm::{
+111
View File
@@ -1,4 +1,5 @@
use super::{
Traverse, TraverseFn,
core::{Py, PyObject, PyObjectRef, PyRef},
payload::PyPayload,
};
@@ -724,6 +725,57 @@ impl PyAtomicRef<Option<PyObject>> {
}
}
/// A nullable object reference that can be replaced without invalidating readers.
///
/// Unlike [`PyAtomicRef`], this cell only exposes owned reads. Concurrent stores
/// use the same QSBR reclamation as object slots without retaining old values
/// for the lifetime of a Python frame.
#[repr(transparent)]
pub struct PyObjectCell(PyAtomicRef<Option<PyObject>>);
impl From<Option<PyObjectRef>> for PyObjectCell {
fn from(value: Option<PyObjectRef>) -> Self {
Self(value.into())
}
}
impl PyObjectCell {
/// Return an owned reference to the current value, or `None` for an empty cell.
#[inline]
pub fn load_owned(&self) -> Option<PyObjectRef> {
self.0.load_owned()
}
/// Replace the stored value and return the previous reference, still owned.
#[inline]
pub fn store(&self, value: Option<PyObjectRef>) -> Option<PyObjectRef> {
self.0.store(value)
}
}
impl fmt::Debug for PyObjectCell {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_tuple("PyObjectCell")
.field(&self.load_owned())
.finish()
}
}
// SAFETY: the underlying cell visits its single owned reference while mutating
// threads are stopped. It does not clone the reference during traversal.
unsafe impl Traverse for PyObjectCell {
#[inline]
fn traverse(&self, traverse_fn: &mut TraverseFn<'_>) {
self.0.traverse(traverse_fn);
}
}
// Object members address a single pointer-sized cell at the field's offset.
const _: () = assert!(
core::mem::size_of::<PyObjectCell>() == core::mem::size_of::<*mut PyObject>()
&& core::mem::align_of::<PyObjectCell>() == core::mem::align_of::<*mut PyObject>()
);
/// Atomic borrowed (non-ref-counted) optional reference to a Python object.
/// Unlike `PyAtomicRef`, this does NOT own the reference.
/// The pointed-to object must outlive this reference.
@@ -899,3 +951,62 @@ impl IntoPyException for PyBaseExceptionRef {
self
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn object_cell_snapshots_survive_replacement_and_clear() {
crate::Interpreter::without_stdlib(Default::default()).enter(|vm| {
let cell = PyObjectCell::from(Some(vm.ctx.new_bytes(vec![1, 2, 3]).into()));
let first = cell.load_owned().unwrap();
let previous = cell.store(Some(vm.ctx.new_bytes(vec![4, 5, 6]).into()));
assert!(previous.as_ref().unwrap().is(&first));
drop(previous);
assert_eq!(first.strong_count(), 1);
assert_eq!(
first
.downcast_ref::<crate::builtins::PyBytes>()
.unwrap()
.as_bytes(),
&[1, 2, 3]
);
let second = cell.load_owned().unwrap();
let previous = cell.store(None);
assert!(previous.as_ref().unwrap().is(&second));
drop(previous);
assert!(cell.load_owned().is_none());
assert!(cell.store(None).is_none());
assert_eq!(second.strong_count(), 1);
assert_eq!(
second
.downcast_ref::<crate::builtins::PyBytes>()
.unwrap()
.as_bytes(),
&[4, 5, 6]
);
});
}
#[test]
fn object_cell_traverses_current_reference_once_without_cloning() {
crate::Interpreter::without_stdlib(Default::default()).enter(|vm| {
let cell = PyObjectCell::from(None);
cell.traverse(&mut |_| panic!("empty cell owns no edge"));
let value: PyObjectRef = vm.ctx.new_bytes(vec![1, 2, 3]).into();
assert!(cell.store(Some(value.clone())).is_none());
let references = value.strong_count();
let mut edges = 0;
cell.traverse(&mut |child| {
assert!(child.is(&value));
assert_eq!(value.strong_count(), references);
edges += 1;
});
assert_eq!(edges, 1);
drop(cell.store(None));
cell.traverse(&mut |_| panic!("cleared cell owns no edge"));
});
}
}
+239
View File
@@ -5,9 +5,12 @@ See http://www.zope.org/Members/fdrake/DateTimeWiki/TestCases
# import copy
import datetime as datetime_module
import errno
import gc
import random
import sys
import time as _time
import weakref
from datetime import MAXYEAR, MINYEAR, date, datetime, time, timedelta, timezone, tzinfo
from operator import eq, floordiv, ge, gt, le, lt, mod, ne, truediv
@@ -3792,3 +3795,239 @@ assert_equal(as_datetime, datetime_sc)
assert_equal(datetime_sc, as_datetime)
'''
for base, value, fields in (
(date, date(2024, 2, 29), ("year", "month", "day")),
(time, time(1, 2, 3), ("hour", "minute", "second", "microsecond", "fold")),
(
datetime,
datetime(2024, 2, 29, 1, 2, 3),
("year", "month", "day", "hour", "minute", "second", "microsecond", "fold"),
),
):
assert value.replace() == value
for field in fields:
assert_raises(TypeError, value.replace, **{field: None})
class InvalidIsoformat(base):
def isoformat(self, *args):
return 123
invalid = InvalidIsoformat(1) if base is time else InvalidIsoformat(2024, 2, 29)
assert_raises(TypeError, str, invalid)
for value in (time(1, 2, 3), datetime(2024, 2, 29)):
assert_raises(TypeError, value.isoformat, timespec=None)
assert_raises(TypeError, datetime(2024, 2, 29).isoformat, None)
# Malformed but accepted pickle state must not panic during formatting.
if sys.implementation.name == "rustpython":
assert isinstance(date(b"\x00\x00\x01\x01").ctime(), str)
class DeltaIntOverrides(int):
def __mul__(self, other):
return 5
def __rmul__(self, other):
return 7
def __rfloordiv__(self, other):
return 13
assert_equal(timedelta(seconds=DeltaIntOverrides(1)), timedelta(microseconds=5))
assert_equal(timedelta(seconds=1) * DeltaIntOverrides(2), timedelta(microseconds=5))
assert_equal(timedelta(seconds=1) // DeltaIntOverrides(2), timedelta(microseconds=13))
assert_equal(
timedelta(seconds=1) / DeltaIntOverrides(2), timedelta(microseconds=500000)
)
class DeltaFloatRatio(float):
def as_integer_ratio(self):
return DeltaIntOverrides(2), DeltaIntOverrides(3)
assert_equal(timedelta(seconds=1) * DeltaFloatRatio(), timedelta(microseconds=2))
assert_equal(timedelta(seconds=1) / DeltaFloatRatio(), timedelta(microseconds=4))
class DeltaProduct(int):
remainder = 7
def __radd__(self, other):
return self
def __divmod__(self, other):
return 0, self.remainder
class DeltaInput(int):
def __mul__(self, other):
return DeltaProduct(0)
assert_equal(timedelta(seconds=DeltaInput(1)), timedelta(microseconds=7))
DeltaProduct.remainder = -1
with assert_raises(TypeError):
timedelta(seconds=DeltaInput(1))
assert_equal(str(timedelta() / timedelta(microseconds=-1)), "-0.0")
# A date's Unicode pickle month must not be truncated before validation.
assert_raises(TypeError, date, "\0\0\u0101\1")
# datetime/time apply their fold mask before checking the month/hour; they
# must still reject a non-latin-1 state with the dedicated unpickle error.
assert_raises(ValueError, datetime, "\0\0\u0101\1\0\0\0\0\0\0")
assert_raises(ValueError, time, "\u0101\0\0\0\0\0")
def test_astimezone_subclass_state():
constructed = []
class TaggedDateTime(datetime_module.datetime):
__slots__ = ("marker",)
def __new__(cls, *args, **kwargs):
value = super().__new__(cls, *args, **kwargs)
value.marker = object()
constructed.append(value)
return value
class TargetZone(datetime_module.tzinfo):
def fromutc(self, value):
assert value.tzinfo is self
assert value.fold == 0
assert value is constructed[-1]
return value
value = TaggedDateTime(2020, 1, 2, 3, tzinfo=datetime_module.timezone.utc, fold=1)
zone = TargetZone()
converted = value.astimezone(zone)
assert len(constructed) == 2
assert converted is constructed[-1]
assert converted.fold == 0
assert converted.tzinfo is zone
assert value.fold == 1
assert value.tzinfo is datetime_module.timezone.utc
def test_timezone_cycles():
# CPython's fixed-offset timezone is not GC-tracked. RustPython must keep
# collecting these cycles when replacing its Python implementation.
if sys.implementation.name != "rustpython":
return
class Offset(datetime_module.timedelta):
pass
class Name(str):
pass
for kind in ("offset", "name"):
value = Offset(seconds=1) if kind == "offset" else Name("local")
zone = (
datetime_module.timezone(value)
if kind == "offset"
else datetime_module.timezone(datetime_module.timedelta(seconds=1), value)
)
value.back = zone
reference = weakref.ref(value)
del value, zone
gc.collect()
assert reference() is None
def test_timestamp_errno():
if sys.platform == "win32":
try:
datetime_module.datetime.fromtimestamp(-1)
except OSError as error:
assert error.errno == errno.EINVAL
else:
raise AssertionError("Windows localtime accepted a negative timestamp")
test_astimezone_subclass_state()
test_timezone_cycles()
test_timestamp_errno()
if sys.implementation.name == "rustpython":
# A constructor may return any object; native arithmetic must check it.
class NonDateResult(datetime):
return_other = False
def __new__(cls, *args, **kwargs):
if cls.return_other:
return object()
return super().__new__(cls, *args, **kwargs)
class NonzeroDst(tzinfo):
def utcoffset(self, value):
return timedelta(hours=2)
def dst(self, value):
return timedelta(hours=1)
zone = NonzeroDst()
value = NonDateResult(2024, 2, 29, tzinfo=zone)
NonDateResult.return_other = True
assert_raises(TypeError, value.utctimetuple)
assert_raises(TypeError, zone.fromutc, value)
assert_raises(TypeError, value.astimezone, timezone.utc)
def test_readonly_datetime_descriptors():
cases = (
(timedelta, (-2, 12345, 654321), {}, ("days", "seconds", "microseconds")),
(date, (2024, 2, 29), {}, ("year", "month", "day")),
(
time,
(12, 34, 56, 123456),
{"fold": 1, "tzinfo": timezone.utc},
("hour", "minute", "second", "microsecond", "fold", "tzinfo"),
),
(
datetime,
(2024, 2, 29, 12, 34, 56, 123456),
{"fold": 1, "tzinfo": timezone.utc},
(
"year",
"month",
"day",
"hour",
"minute",
"second",
"microsecond",
"fold",
"tzinfo",
),
),
)
for base, args, kwargs, names in cases:
class Subclass(base):
pass
reference = base(*args, **kwargs)
value = Subclass(*args, **kwargs)
for name in names:
descriptor = getattr(base, name)
expected = getattr(reference, name)
# Instance dictionaries must not shadow inherited readonly fields.
value.__dict__[name] = object()
assert getattr(value, name) == expected
assert descriptor.__get__(value) == expected
assert_raises(AttributeError, setattr, value, name, expected)
assert_raises(AttributeError, delattr, value, name)
assert_raises(TypeError, descriptor.__get__, object())
if "tzinfo" in names:
assert base.tzinfo.__get__(value) is timezone.utc
assert base.tzinfo.__get__(base(*args)) is None
test_readonly_datetime_descriptors()