Mailman 3 February 2024 - Python-checkins

gh-116030: test_unparse: Add ``ctx`` argument to ``ast.Name`` calls (#116031)
by JelleZijlstra Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/3b63d0769f49171f53e9cecc686fa01a38… commit: 3b63d0769f49171f53e9cecc686fa01a383bd4b1 branch: main author: Kirill Podoprigora <kirill.bast9(a)mail.ru> committer: JelleZijlstra <jelle.zijlstra(a)gmail.com> date: 2024-02-28T03:04:23-08:00 summary: gh-116030: test_unparse: Add ``ctx`` argument to ``ast.Name`` calls (#116031) files: M Lib/test/test_unparse.py diff --git a/Lib/test/test_unparse.py b/Lib/test/test_unparse.py index 106704ba8c9c2d..bb15f64c59dbd1 100644 --- a/Lib/test/test_unparse.py +++ b/Lib/test/test_unparse.py @@ -370,13 +370,13 @@ def test_slices(self): self.check_ast_roundtrip("a[i:j, k]") def test_invalid_raise(self): - self.check_invalid(ast.Raise(exc=None, cause=ast.Name(id="X"))) + self.check_invalid(ast.Raise(exc=None, cause=ast.Name(id="X", ctx=ast.Load()))) def test_invalid_fstring_value(self): self.check_invalid( ast.JoinedStr( values=[ - ast.Name(id="test"), + ast.Name(id="test", ctx=ast.Load()), ast.Constant(value="test") ] ) @@ -718,7 +718,7 @@ def test_function_with_type_params_and_bound(self): body=[ast.Pass()], decorator_list=[], returns=None, - type_params=[ast.TypeVar("T", bound=ast.Name("int"))], + type_params=[ast.TypeVar("T", bound=ast.Name("int", ctx=ast.Load()))], ) ast.fix_missing_locations(node) self.assertEqual(ast.unparse(node), "def f[T: int]():\n pass")

1 0

gh-115773: Add tests to exercise the _Py_DebugOffsets structure (#115774)
by pablogsal Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/1752b51012269eaa35f7a28f162d18479a… commit: 1752b51012269eaa35f7a28f162d18479a4f72aa branch: main author: Pablo Galindo Salgado <Pablogsal(a)gmail.com> committer: pablogsal <Pablogsal(a)gmail.com> date: 2024-02-28T10:17:34Z summary: gh-115773: Add tests to exercise the _Py_DebugOffsets structure (#115774) files: A Lib/test/test_external_inspection.py A Modules/_testexternalinspection.c M Include/internal/pycore_runtime.h M Include/internal/pycore_runtime_init.h M Modules/Setup M Modules/Setup.stdlib.in M Tools/build/generate_stdlib_module_names.py M configure M configure.ac M pyconfig.h.in diff --git a/Include/internal/pycore_runtime.h b/Include/internal/pycore_runtime.h index 0c9c59e85b2fcf..dc6f6f100f7a92 100644 --- a/Include/internal/pycore_runtime.h +++ b/Include/internal/pycore_runtime.h @@ -55,74 +55,81 @@ typedef struct _Py_DebugOffsets { uint64_t version; // Runtime state offset; struct _runtime_state { - off_t finalizing; - off_t interpreters_head; + uint64_t finalizing; + uint64_t interpreters_head; } runtime_state; // Interpreter state offset; struct _interpreter_state { - off_t next; - off_t threads_head; - off_t gc; - off_t imports_modules; - off_t sysdict; - off_t builtins; - off_t ceval_gil; - off_t gil_runtime_state_locked; - off_t gil_runtime_state_holder; + uint64_t next; + uint64_t threads_head; + uint64_t gc; + uint64_t imports_modules; + uint64_t sysdict; + uint64_t builtins; + uint64_t ceval_gil; + uint64_t gil_runtime_state_locked; + uint64_t gil_runtime_state_holder; } interpreter_state; // Thread state offset; struct _thread_state{ - off_t prev; - off_t next; - off_t interp; - off_t current_frame; - off_t thread_id; - off_t native_thread_id; + uint64_t prev; + uint64_t next; + uint64_t interp; + uint64_t current_frame; + uint64_t thread_id; + uint64_t native_thread_id; } thread_state; // InterpreterFrame offset; struct _interpreter_frame { - off_t previous; - off_t executable; - off_t instr_ptr; - off_t localsplus; - off_t owner; + uint64_t previous; + uint64_t executable; + uint64_t instr_ptr; + uint64_t localsplus; + uint64_t owner; } interpreter_frame; // CFrame offset; struct _cframe { - off_t current_frame; - off_t previous; + uint64_t current_frame; + uint64_t previous; } cframe; // Code object offset; struct _code_object { - off_t filename; - off_t name; - off_t linetable; - off_t firstlineno; - off_t argcount; - off_t localsplusnames; - off_t localspluskinds; - off_t co_code_adaptive; + uint64_t filename; + uint64_t name; + uint64_t linetable; + uint64_t firstlineno; + uint64_t argcount; + uint64_t localsplusnames; + uint64_t localspluskinds; + uint64_t co_code_adaptive; } code_object; // PyObject offset; struct _pyobject { - off_t ob_type; + uint64_t ob_type; } pyobject; // PyTypeObject object offset; struct _type_object { - off_t tp_name; + uint64_t tp_name; } type_object; // PyTuple object offset; struct _tuple_object { - off_t ob_item; + uint64_t ob_item; } tuple_object; + + // Unicode object offset; + struct _unicode_object { + uint64_t state; + uint64_t length; + size_t asciiobject_size; + } unicode_object; } _Py_DebugOffsets; /* Full Python runtime state */ diff --git a/Include/internal/pycore_runtime_init.h b/Include/internal/pycore_runtime_init.h index d093047d4bc09d..cc47b9a82e2879 100644 --- a/Include/internal/pycore_runtime_init.h +++ b/Include/internal/pycore_runtime_init.h @@ -83,6 +83,11 @@ extern PyTypeObject _PyExc_MemoryError; .tuple_object = { \ .ob_item = offsetof(PyTupleObject, ob_item), \ }, \ + .unicode_object = { \ + .state = offsetof(PyUnicodeObject, _base._base.state), \ + .length = offsetof(PyUnicodeObject, _base._base.length), \ + .asciiobject_size = sizeof(PyASCIIObject), \ + }, \ }, \ .allocators = { \ .standard = _pymem_allocators_standard_INIT(runtime), \ diff --git a/Lib/test/test_external_inspection.py b/Lib/test/test_external_inspection.py new file mode 100644 index 00000000000000..86c07de507e39c --- /dev/null +++ b/Lib/test/test_external_inspection.py @@ -0,0 +1,84 @@ +import unittest +import os +import textwrap +import importlib +import sys +from test.support import os_helper, SHORT_TIMEOUT +from test.support.script_helper import make_script + +import subprocess + +PROCESS_VM_READV_SUPPORTED = False + +try: + from _testexternalinspection import PROCESS_VM_READV_SUPPORTED + from _testexternalinspection import get_stack_trace +except ImportError: + unittest.skip("Test only runs when _testexternalinspection is available") + +def _make_test_script(script_dir, script_basename, source): + to_return = make_script(script_dir, script_basename, source) + importlib.invalidate_caches() + return to_return + +class TestGetStackTrace(unittest.TestCase): + + @unittest.skipIf(sys.platform != "darwin" and sys.platform != "linux", "Test only runs on Linux and MacOS") + @unittest.skipIf(sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED, "Test only runs on Linux with process_vm_readv support") + def test_remote_stack_trace(self): + # Spawn a process with some realistic Python code + script = textwrap.dedent("""\ + import time, sys, os + def bar(): + for x in range(100): + if x == 50: + baz() + def baz(): + foo() + + def foo(): + fifo = sys.argv[1] + with open(sys.argv[1], "w") as fifo: + fifo.write("ready") + time.sleep(1000) + + bar() + """) + stack_trace = None + with os_helper.temp_dir() as work_dir: + script_dir = os.path.join(work_dir, "script_pkg") + os.mkdir(script_dir) + fifo = f"{work_dir}/the_fifo" + os.mkfifo(fifo) + script_name = _make_test_script(script_dir, 'script', script) + try: + p = subprocess.Popen([sys.executable, script_name, str(fifo)]) + with open(fifo, "r") as fifo_file: + response = fifo_file.read() + self.assertEqual(response, "ready") + stack_trace = get_stack_trace(p.pid) + except PermissionError: + self.skipTest("Insufficient permissions to read the stack trace") + finally: + os.remove(fifo) + p.kill() + p.terminate() + p.wait(timeout=SHORT_TIMEOUT) + + + expected_stack_trace = [ + 'foo', + 'baz', + 'bar', + '<module>' + ] + self.assertEqual(stack_trace, expected_stack_trace) + + @unittest.skipIf(sys.platform != "darwin" and sys.platform != "linux", "Test only runs on Linux and MacOS") + @unittest.skipIf(sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED, "Test only runs on Linux with process_vm_readv support") + def test_self_trace(self): + stack_trace = get_stack_trace(os.getpid()) + self.assertEqual(stack_trace[0], "test_self_trace") + +if __name__ == "__main__": + unittest.main() diff --git a/Modules/Setup b/Modules/Setup index 8ad9a5aebbfcaa..cd1cf24c25d406 100644 --- a/Modules/Setup +++ b/Modules/Setup @@ -285,6 +285,7 @@ PYTHONPATH=$(COREPYTHONPATH) #_testcapi _testcapimodule.c #_testimportmultiple _testimportmultiple.c #_testmultiphase _testmultiphase.c +#_testexternalinspection _testexternalinspection.c #_testsinglephase _testsinglephase.c # --- diff --git a/Modules/Setup.stdlib.in b/Modules/Setup.stdlib.in index e98775a4808765..73b082691a3fd4 100644 --- a/Modules/Setup.stdlib.in +++ b/Modules/Setup.stdlib.in @@ -171,6 +171,7 @@ @MODULE__TESTIMPORTMULTIPLE_TRUE@_testimportmultiple _testimportmultiple.c @MODULE__TESTMULTIPHASE_TRUE@_testmultiphase _testmultiphase.c @MODULE__TESTMULTIPHASE_TRUE@_testsinglephase _testsinglephase.c +@MODULE__TESTEXTERNALINSPECTION_TRUE@_testexternalinspection _testexternalinspection.c @MODULE__CTYPES_TEST_TRUE@_ctypes_test _ctypes/_ctypes_test.c # Limited API template modules; must be built as shared modules. diff --git a/Modules/_testexternalinspection.c b/Modules/_testexternalinspection.c new file mode 100644 index 00000000000000..19ee3b8dd1428d --- /dev/null +++ b/Modules/_testexternalinspection.c @@ -0,0 +1,629 @@ +#define _GNU_SOURCE + +#ifdef __linux__ +# include <elf.h> +# include <sys/uio.h> +# if INTPTR_MAX == INT64_MAX +# define Elf_Ehdr Elf64_Ehdr +# define Elf_Shdr Elf64_Shdr +# define Elf_Phdr Elf64_Phdr +# else +# define Elf_Ehdr Elf32_Ehdr +# define Elf_Shdr Elf32_Shdr +# define Elf_Phdr Elf32_Phdr +# endif +# include <sys/mman.h> +#endif + +#ifdef __APPLE__ +# include <libproc.h> +# include <mach-o/fat.h> +# include <mach-o/loader.h> +# include <mach-o/nlist.h> +# include <mach/mach.h> +# include <mach/mach_vm.h> +# include <mach/machine.h> +# include <sys/mman.h> +# include <sys/proc.h> +# include <sys/sysctl.h> +#endif + +#include <errno.h> +#include <fcntl.h> +#include <stddef.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/param.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <unistd.h> + +#ifndef Py_BUILD_CORE_BUILTIN +# define Py_BUILD_CORE_MODULE 1 +#endif +#include "Python.h" +#include <internal/pycore_runtime.h> + +#ifndef HAVE_PROCESS_VM_READV +# define HAVE_PROCESS_VM_READV 0 +#endif + +#ifdef __APPLE__ +static void* +analyze_macho64(mach_port_t proc_ref, void* base, void* map) +{ + struct mach_header_64* hdr = (struct mach_header_64*)map; + int ncmds = hdr->ncmds; + + int cmd_cnt = 0; + struct segment_command_64* cmd = map + sizeof(struct mach_header_64); + + mach_vm_size_t size = 0; + mach_msg_type_number_t count = sizeof(vm_region_basic_info_data_64_t); + mach_vm_address_t address = (mach_vm_address_t)base; + vm_region_basic_info_data_64_t region_info; + mach_port_t object_name; + + for (int i = 0; cmd_cnt < 2 && i < ncmds; i++) { + if (cmd->cmd == LC_SEGMENT_64 && strcmp(cmd->segname, "__DATA") == 0) { + while (cmd->filesize != size) { + address += size; + if (mach_vm_region( + proc_ref, + &address, + &size, + VM_REGION_BASIC_INFO_64, + (vm_region_info_t)&region_info, // cppcheck-suppress [uninitvar] + &count, + &object_name) + != KERN_SUCCESS) + { + PyErr_SetString(PyExc_RuntimeError, "Cannot get any more VM maps.\n"); + return NULL; + } + } + base = (void*)address - cmd->vmaddr; + + int nsects = cmd->nsects; + struct section_64* sec = + (struct section_64*)((void*)cmd + sizeof(struct segment_command_64)); + for (int j = 0; j < nsects; j++) { + if (strcmp(sec[j].sectname, "PyRuntime") == 0) { + return base + sec[j].addr; + } + } + cmd_cnt++; + } + + cmd = (struct segment_command_64*)((void*)cmd + cmd->cmdsize); + } + return NULL; +} + +static void* +analyze_macho(char* path, void* base, mach_vm_size_t size, mach_port_t proc_ref) +{ + int fd = open(path, O_RDONLY); + if (fd == -1) { + PyErr_Format(PyExc_RuntimeError, "Cannot open binary %s\n", path); + return NULL; + } + + struct stat fs; + if (fstat(fd, &fs) == -1) { + PyErr_Format(PyExc_RuntimeError, "Cannot get size of binary %s\n", path); + close(fd); + return NULL; + } + + void* map = mmap(0, fs.st_size, PROT_READ, MAP_SHARED, fd, 0); + if (map == MAP_FAILED) { + PyErr_Format(PyExc_RuntimeError, "Cannot map binary %s\n", path); + close(fd); + return NULL; + } + + void* result = NULL; + + struct mach_header_64* hdr = (struct mach_header_64*)map; + switch (hdr->magic) { + case MH_MAGIC: + case MH_CIGAM: + case FAT_MAGIC: + case FAT_CIGAM: + PyErr_SetString(PyExc_RuntimeError, "32-bit Mach-O binaries are not supported"); + break; + case MH_MAGIC_64: + case MH_CIGAM_64: + result = analyze_macho64(proc_ref, base, map); + break; + default: + PyErr_SetString(PyExc_RuntimeError, "Unknown Mach-O magic"); + break; + } + + munmap(map, fs.st_size); + if (close(fd) != 0) { + PyErr_SetFromErrno(PyExc_OSError); + } + return result; +} + +static mach_port_t +pid_to_task(pid_t pid) +{ + mach_port_t task; + kern_return_t result; + + result = task_for_pid(mach_task_self(), pid, &task); + if (result != KERN_SUCCESS) { + PyErr_Format(PyExc_PermissionError, "Cannot get task for PID %d", pid); + return 0; + } + return task; +} + +static void* +get_py_runtime_macos(pid_t pid) +{ + mach_vm_address_t address = 0; + mach_vm_size_t size = 0; + mach_msg_type_number_t count = sizeof(vm_region_basic_info_data_64_t); + vm_region_basic_info_data_64_t region_info; + mach_port_t object_name; + + mach_port_t proc_ref = pid_to_task(pid); + if (proc_ref == 0) { + PyErr_SetString(PyExc_PermissionError, "Cannot get task for PID"); + return NULL; + } + + int match_found = 0; + char map_filename[MAXPATHLEN + 1]; + void* result_address = NULL; + while (mach_vm_region( + proc_ref, + &address, + &size, + VM_REGION_BASIC_INFO_64, + (vm_region_info_t)&region_info, + &count, + &object_name) + == KERN_SUCCESS) + { + int path_len = proc_regionfilename(pid, address, map_filename, MAXPATHLEN); + if (path_len == 0) { + address += size; + continue; + } + + char* filename = strrchr(map_filename, '/'); + if (filename != NULL) { + filename++; // Move past the '/' + } else { + filename = map_filename; // No path, use the whole string + } + + // Check if the filename starts with "python" or "libpython" + if (!match_found && strncmp(filename, "python", 6) == 0) { + match_found = 1; + result_address = analyze_macho(map_filename, (void*)address, size, proc_ref); + } + if (strncmp(filename, "libpython", 9) == 0) { + match_found = 1; + result_address = analyze_macho(map_filename, (void*)address, size, proc_ref); + break; + } + + address += size; + } + return result_address; +} +#endif + +#ifdef __linux__ +void* +find_python_map_start_address(pid_t pid, char* result_filename) +{ + char maps_file_path[64]; + sprintf(maps_file_path, "/proc/%d/maps", pid); + + FILE* maps_file = fopen(maps_file_path, "r"); + if (maps_file == NULL) { + PyErr_SetFromErrno(PyExc_OSError); + return NULL; + } + + int match_found = 0; + + char line[256]; + char map_filename[PATH_MAX]; + void* result_address = 0; + while (fgets(line, sizeof(line), maps_file) != NULL) { + unsigned long start_address = 0; + sscanf(line, "%lx-%*x %*s %*s %*s %*s %s", &start_address, map_filename); + char* filename = strrchr(map_filename, '/'); + if (filename != NULL) { + filename++; // Move past the '/' + } else { + filename = map_filename; // No path, use the whole string + } + + // Check if the filename starts with "python" or "libpython" + if (!match_found && strncmp(filename, "python", 6) == 0) { + match_found = 1; + result_address = (void*)start_address; + strcpy(result_filename, map_filename); + } + if (strncmp(filename, "libpython", 9) == 0) { + match_found = 1; + result_address = (void*)start_address; + strcpy(result_filename, map_filename); + break; + } + } + + fclose(maps_file); + + if (!match_found) { + map_filename[0] = '\0'; + } + + return result_address; +} + +void* +get_py_runtime_linux(pid_t pid) +{ + char elf_file[256]; + void* start_address = (void*)find_python_map_start_address(pid, elf_file); + + if (start_address == 0) { + PyErr_SetString(PyExc_RuntimeError, "No memory map associated with python or libpython found"); + return NULL; + } + + void* result = NULL; + void* file_memory = NULL; + + int fd = open(elf_file, O_RDONLY); + if (fd < 0) { + PyErr_SetFromErrno(PyExc_OSError); + goto exit; + } + + struct stat file_stats; + if (fstat(fd, &file_stats) != 0) { + PyErr_SetFromErrno(PyExc_OSError); + goto exit; + } + + file_memory = mmap(NULL, file_stats.st_size, PROT_READ, MAP_PRIVATE, fd, 0); + if (file_memory == MAP_FAILED) { + PyErr_SetFromErrno(PyExc_OSError); + goto exit; + } + + Elf_Ehdr* elf_header = (Elf_Ehdr*)file_memory; + + Elf_Shdr* section_header_table = (Elf_Shdr*)(file_memory + elf_header->e_shoff); + + Elf_Shdr* shstrtab_section = &section_header_table[elf_header->e_shstrndx]; + char* shstrtab = (char*)(file_memory + shstrtab_section->sh_offset); + + Elf_Shdr* py_runtime_section = NULL; + for (int i = 0; i < elf_header->e_shnum; i++) { + if (strcmp(".PyRuntime", shstrtab + section_header_table[i].sh_name) == 0) { + py_runtime_section = &section_header_table[i]; + break; + } + } + + Elf_Phdr* program_header_table = (Elf_Phdr*)(file_memory + elf_header->e_phoff); + // Find the first PT_LOAD segment + Elf_Phdr* first_load_segment = NULL; + for (int i = 0; i < elf_header->e_phnum; i++) { + if (program_header_table[i].p_type == PT_LOAD) { + first_load_segment = &program_header_table[i]; + break; + } + } + + if (py_runtime_section != NULL && first_load_segment != NULL) { + uintptr_t elf_load_addr = first_load_segment->p_vaddr + - (first_load_segment->p_vaddr % first_load_segment->p_align); + result = start_address + py_runtime_section->sh_addr - elf_load_addr; + } + +exit: + if (close(fd) != 0) { + PyErr_SetFromErrno(PyExc_OSError); + } + if (file_memory != NULL) { + munmap(file_memory, file_stats.st_size); + } + return result; +} +#endif + +ssize_t +read_memory(pid_t pid, void* remote_address, size_t len, void* dst) +{ + ssize_t total_bytes_read = 0; +#ifdef __linux__ + struct iovec local[1]; + struct iovec remote[1]; + ssize_t result = 0; + ssize_t read = 0; + + do { + local[0].iov_base = dst + result; + local[0].iov_len = len - result; + remote[0].iov_base = (void*)(remote_address + result); + remote[0].iov_len = len - result; + + read = process_vm_readv(pid, local, 1, remote, 1, 0); + if (read < 0) { + PyErr_SetFromErrno(PyExc_OSError); + return -1; + } + + result += read; + } while ((size_t)read != local[0].iov_len); + total_bytes_read = result; +#elif defined(__APPLE__) + ssize_t result = -1; + kern_return_t kr = mach_vm_read_overwrite( + pid_to_task(pid), + (mach_vm_address_t)remote_address, + len, + (mach_vm_address_t)dst, + (mach_vm_size_t*)&result); + + if (kr != KERN_SUCCESS) { + switch (kr) { + case KERN_PROTECTION_FAILURE: + PyErr_SetString(PyExc_PermissionError, "Not enough permissions to read memory"); + break; + case KERN_INVALID_ARGUMENT: + PyErr_SetString(PyExc_PermissionError, "Invalid argument to mach_vm_read_overwrite"); + break; + default: + PyErr_SetString(PyExc_RuntimeError, "Unknown error reading memory"); + } + return -1; + } + total_bytes_read = len; +#else + return -1; +#endif + return total_bytes_read; +} + +int +read_string(pid_t pid, _Py_DebugOffsets* debug_offsets, void* address, char* buffer, Py_ssize_t size) +{ + Py_ssize_t len; + ssize_t bytes_read = + read_memory(pid, address + debug_offsets->unicode_object.length, sizeof(Py_ssize_t), &len); + if (bytes_read == -1) { + return -1; + } + if (len >= size) { + PyErr_SetString(PyExc_RuntimeError, "Buffer too small"); + return -1; + } + size_t offset = debug_offsets->unicode_object.asciiobject_size; + bytes_read = read_memory(pid, address + offset, len, buffer); + if (bytes_read == -1) { + return -1; + } + buffer[len] = '\0'; + return 0; +} + +void* +get_py_runtime(pid_t pid) +{ +#if defined(__linux__) + return get_py_runtime_linux(pid); +#elif defined(__APPLE__) + return get_py_runtime_macos(pid); +#else + return NULL; +#endif +} + +static int +parse_code_object( + int pid, + PyObject* result, + struct _Py_DebugOffsets* offsets, + void* address, + void** previous_frame) +{ + void* address_of_function_name; + read_memory( + pid, + (void*)(address + offsets->code_object.name), + sizeof(void*), + &address_of_function_name); + + if (address_of_function_name == NULL) { + PyErr_SetString(PyExc_RuntimeError, "No function name found"); + return -1; + } + + char function_name[256]; + if (read_string(pid, offsets, address_of_function_name, function_name, sizeof(function_name)) != 0) { + return -1; + } + + PyObject* py_function_name = PyUnicode_FromString(function_name); + if (py_function_name == NULL) { + return -1; + } + + if (PyList_Append(result, py_function_name) == -1) { + Py_DECREF(py_function_name); + return -1; + } + Py_DECREF(py_function_name); + + return 0; +} + +static int +parse_frame_object( + int pid, + PyObject* result, + struct _Py_DebugOffsets* offsets, + void* address, + void** previous_frame) +{ + ssize_t bytes_read = read_memory( + pid, + (void*)(address + offsets->interpreter_frame.previous), + sizeof(void*), + previous_frame); + if (bytes_read == -1) { + return -1; + } + + char owner; + bytes_read = + read_memory(pid, (void*)(address + offsets->interpreter_frame.owner), sizeof(char), &owner); + if (bytes_read < 0) { + return -1; + } + + if (owner == FRAME_OWNED_BY_CSTACK) { + return 0; + } + + void* address_of_code_object; + bytes_read = read_memory( + pid, + (void*)(address + offsets->interpreter_frame.executable), + sizeof(void*), + &address_of_code_object); + if (bytes_read == -1) { + return -1; + } + + if (address_of_code_object == NULL) { + return 0; + } + return parse_code_object(pid, result, offsets, address_of_code_object, previous_frame); +} + +static PyObject* +get_stack_trace(PyObject* self, PyObject* args) +{ +#if (!defined(__linux__) && !defined(__APPLE__)) || (defined(__linux__) && !HAVE_PROCESS_VM_READV) + PyErr_SetString(PyExc_RuntimeError, "get_stack_trace is not supported on this platform"); + return NULL; +#endif + int pid; + + if (!PyArg_ParseTuple(args, "i", &pid)) { + return NULL; + } + + void* runtime_start_address = get_py_runtime(pid); + if (runtime_start_address == NULL) { + if (!PyErr_Occurred()) { + PyErr_SetString(PyExc_RuntimeError, "Failed to get .PyRuntime address"); + } + return NULL; + } + size_t size = sizeof(struct _Py_DebugOffsets); + struct _Py_DebugOffsets local_debug_offsets; + + ssize_t bytes_read = read_memory(pid, runtime_start_address, size, &local_debug_offsets); + if (bytes_read == -1) { + return NULL; + } + off_t thread_state_list_head = local_debug_offsets.runtime_state.interpreters_head; + + void* address_of_interpreter_state; + bytes_read = read_memory( + pid, + (void*)(runtime_start_address + thread_state_list_head), + sizeof(void*), + &address_of_interpreter_state); + if (bytes_read == -1) { + return NULL; + } + + if (address_of_interpreter_state == NULL) { + PyErr_SetString(PyExc_RuntimeError, "No interpreter state found"); + return NULL; + } + + void* address_of_thread; + bytes_read = read_memory( + pid, + (void*)(address_of_interpreter_state + local_debug_offsets.interpreter_state.threads_head), + sizeof(void*), + &address_of_thread); + if (bytes_read == -1) { + return NULL; + } + + PyObject* result = PyList_New(0); + if (result == NULL) { + return NULL; + } + + // No Python frames are available for us (can happen at tear-down). + if (address_of_thread != NULL) { + void* address_of_current_frame; + (void)read_memory( + pid, + (void*)(address_of_thread + local_debug_offsets.thread_state.current_frame), + sizeof(void*), + &address_of_current_frame); + while (address_of_current_frame != NULL) { + if (parse_frame_object( + pid, + result, + &local_debug_offsets, + address_of_current_frame, + &address_of_current_frame) + < 0) + { + Py_DECREF(result); + return NULL; + } + } + } + + return result; +} + +static PyMethodDef methods[] = { + {"get_stack_trace", get_stack_trace, METH_VARARGS, "Get the Python stack from a given PID"}, + {NULL, NULL, 0, NULL}, +}; + +static struct PyModuleDef module = { + .m_base = PyModuleDef_HEAD_INIT, + .m_name = "_testexternalinspection", + .m_size = -1, + .m_methods = methods, +}; + +PyMODINIT_FUNC +PyInit__testexternalinspection(void) +{ + PyObject* mod = PyModule_Create(&module); + int rc = PyModule_AddIntConstant(mod, "PROCESS_VM_READV_SUPPORTED", HAVE_PROCESS_VM_READV); + if (rc < 0) { + Py_DECREF(mod); + return NULL; + } + return mod; +} diff --git a/Tools/build/generate_stdlib_module_names.py b/Tools/build/generate_stdlib_module_names.py index 5dce4e042d1eb4..588dfda50658be 100644 --- a/Tools/build/generate_stdlib_module_names.py +++ b/Tools/build/generate_stdlib_module_names.py @@ -34,6 +34,7 @@ '_testinternalcapi', '_testmultiphase', '_testsinglephase', + '_testexternalinspection', '_xxsubinterpreters', '_xxinterpchannels', '_xxinterpqueues', diff --git a/configure b/configure index b565cdbfae1327..c204c9eb499559 100755 --- a/configure +++ b/configure @@ -661,6 +661,8 @@ MODULE__XXTESTFUZZ_FALSE MODULE__XXTESTFUZZ_TRUE MODULE_XXSUBTYPE_FALSE MODULE_XXSUBTYPE_TRUE +MODULE__TESTEXTERNALINSPECTION_FALSE +MODULE__TESTEXTERNALINSPECTION_TRUE MODULE__TESTMULTIPHASE_FALSE MODULE__TESTMULTIPHASE_TRUE MODULE__TESTIMPORTMULTIPLE_FALSE @@ -17997,6 +17999,12 @@ if test "x$ac_cv_func_preadv2" = xyes then : printf "%s\n" "#define HAVE_PREADV2 1" >>confdefs.h +fi +ac_fn_c_check_func "$LINENO" "process_vm_readv" "ac_cv_func_process_vm_readv" +if test "x$ac_cv_func_process_vm_readv" = xyes +then : + printf "%s\n" "#define HAVE_PROCESS_VM_READV 1" >>confdefs.h + fi ac_fn_c_check_func "$LINENO" "pthread_cond_timedwait_relative_np" "ac_cv_func_pthread_cond_timedwait_relative_np" if test "x$ac_cv_func_pthread_cond_timedwait_relative_np" = xyes @@ -30688,6 +30696,44 @@ fi printf "%s\n" "$py_cv_module__testmultiphase" >&6; } + { printf "%s\n" "$as_me:${as_lineno-$LINENO}: checking for stdlib extension module _testexternalinspection" >&5 +printf %s "checking for stdlib extension module _testexternalinspection... " >&6; } + if test "$py_cv_module__testexternalinspection" != "n/a" +then : + + if test "$TEST_MODULES" = yes +then : + if true +then : + py_cv_module__testexternalinspection=yes +else $as_nop + py_cv_module__testexternalinspection=missing +fi +else $as_nop + py_cv_module__testexternalinspection=disabled +fi + +fi + as_fn_append MODULE_BLOCK "MODULE__TESTEXTERNALINSPECTION_STATE=$py_cv_module__testexternalinspection$as_nl" + if test "x$py_cv_module__testexternalinspection" = xyes +then : + + + + +fi + if test "$py_cv_module__testexternalinspection" = yes; then + MODULE__TESTEXTERNALINSPECTION_TRUE= + MODULE__TESTEXTERNALINSPECTION_FALSE='#' +else + MODULE__TESTEXTERNALINSPECTION_TRUE='#' + MODULE__TESTEXTERNALINSPECTION_FALSE= +fi + + { printf "%s\n" "$as_me:${as_lineno-$LINENO}: result: $py_cv_module__testexternalinspection" >&5 +printf "%s\n" "$py_cv_module__testexternalinspection" >&6; } + + { printf "%s\n" "$as_me:${as_lineno-$LINENO}: checking for stdlib extension module xxsubtype" >&5 printf %s "checking for stdlib extension module xxsubtype... " >&6; } if test "$py_cv_module_xxsubtype" != "n/a" @@ -31300,6 +31346,10 @@ if test -z "${MODULE__TESTMULTIPHASE_TRUE}" && test -z "${MODULE__TESTMULTIPHASE as_fn_error $? "conditional \"MODULE__TESTMULTIPHASE\" was never defined. Usually this means the macro was only invoked conditionally." "$LINENO" 5 fi +if test -z "${MODULE__TESTEXTERNALINSPECTION_TRUE}" && test -z "${MODULE__TESTEXTERNALINSPECTION_FALSE}"; then + as_fn_error $? "conditional \"MODULE__TESTEXTERNALINSPECTION\" was never defined. +Usually this means the macro was only invoked conditionally." "$LINENO" 5 +fi if test -z "${MODULE_XXSUBTYPE_TRUE}" && test -z "${MODULE_XXSUBTYPE_FALSE}"; then as_fn_error $? "conditional \"MODULE_XXSUBTYPE\" was never defined. Usually this means the macro was only invoked conditionally." "$LINENO" 5 diff --git a/configure.ac b/configure.ac index 0694ef06364f82..702e588357789c 100644 --- a/configure.ac +++ b/configure.ac @@ -4920,7 +4920,7 @@ AC_CHECK_FUNCS([ \ mknod mknodat mktime mmap mremap nice openat opendir pathconf pause pipe \ pipe2 plock poll posix_fadvise posix_fallocate posix_openpt posix_spawn posix_spawnp \ posix_spawn_file_actions_addclosefrom_np \ - pread preadv preadv2 pthread_cond_timedwait_relative_np pthread_condattr_setclock pthread_init \ + pread preadv preadv2 process_vm_readv pthread_cond_timedwait_relative_np pthread_condattr_setclock pthread_init \ pthread_kill ptsname ptsname_r pwrite pwritev pwritev2 readlink readlinkat readv realpath renameat \ rtpSpawn sched_get_priority_max sched_rr_get_interval sched_setaffinity \ sched_setparam sched_setscheduler sem_clockwait sem_getvalue sem_open \ @@ -7581,6 +7581,7 @@ PY_STDLIB_MOD([_testinternalcapi], [test "$TEST_MODULES" = yes]) PY_STDLIB_MOD([_testbuffer], [test "$TEST_MODULES" = yes]) PY_STDLIB_MOD([_testimportmultiple], [test "$TEST_MODULES" = yes], [test "$ac_cv_func_dlopen" = yes]) PY_STDLIB_MOD([_testmultiphase], [test "$TEST_MODULES" = yes], [test "$ac_cv_func_dlopen" = yes]) +PY_STDLIB_MOD([_testexternalinspection], [test "$TEST_MODULES" = yes]) PY_STDLIB_MOD([xxsubtype], [test "$TEST_MODULES" = yes]) PY_STDLIB_MOD([_xxtestfuzz], [test "$TEST_MODULES" = yes]) PY_STDLIB_MOD([_ctypes_test], diff --git a/pyconfig.h.in b/pyconfig.h.in index 63a3437ebb3eac..b93cac9d23c796 100644 --- a/pyconfig.h.in +++ b/pyconfig.h.in @@ -933,6 +933,9 @@ /* Define to 1 if you have the <process.h> header file. */ #undef HAVE_PROCESS_H +/* Define to 1 if you have the `process_vm_readv' function. */ +#undef HAVE_PROCESS_VM_READV + /* Define if your compiler supports function prototype */ #undef HAVE_PROTOTYPES

1 0

gh-105858: Expose some union-related objects as internal APIs (GH-116025)
by encukou Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/d53560deb2c9ae12147201003fe63b2666… commit: d53560deb2c9ae12147201003fe63b266654ee21 branch: main author: Jelle Zijlstra <jelle.zijlstra(a)gmail.com> committer: encukou <encukou(a)gmail.com> date: 2024-02-28T09:56:40Z summary: gh-105858: Expose some union-related objects as internal APIs (GH-116025) We now use these in the AST parsing code after gh-105880. A few comparable types (e.g., NoneType) are already exposed as internal APIs. files: M Include/internal/pycore_unionobject.h diff --git a/Include/internal/pycore_unionobject.h b/Include/internal/pycore_unionobject.h index 87264635b6e1cf..6ece7134cdeca0 100644 --- a/Include/internal/pycore_unionobject.h +++ b/Include/internal/pycore_unionobject.h @@ -8,9 +8,11 @@ extern "C" { # error "this header requires Py_BUILD_CORE define" #endif -extern PyTypeObject _PyUnion_Type; +// For extensions created by test_peg_generator +PyAPI_DATA(PyTypeObject) _PyUnion_Type; +PyAPI_FUNC(PyObject *) _Py_union_type_or(PyObject *, PyObject *); + #define _PyUnion_Check(op) Py_IS_TYPE((op), &_PyUnion_Type) -extern PyObject *_Py_union_type_or(PyObject *, PyObject *); #define _PyGenericAlias_Check(op) PyObject_TypeCheck((op), &Py_GenericAliasType) extern PyObject *_Py_subs_parameters(PyObject *, PyObject *, PyObject *, PyObject *);

1 0

[3.11] gh-115961: Improve tests for compressed file-like objects (GH-115963) (GH-116039)
by serhiy-storchaka Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/becc61596f79a3bb2bfeb7134581ae4663… commit: becc61596f79a3bb2bfeb7134581ae46638d34dc branch: 3.11 author: Serhiy Storchaka <storchaka(a)gmail.com> committer: serhiy-storchaka <storchaka(a)gmail.com> date: 2024-02-28T09:38:39Z summary: [3.11] gh-115961: Improve tests for compressed file-like objects (GH-115963) (GH-116039) * Increase coverage for compressed file-like objects initialized with a file name, an open file object, a file object opened by file descriptor, and a file-like object without name and mode attributes (io.BytesIO) * Increase coverage for name, fileno(), mode, readable(), writable(), seekable() in different modes and states * No longer skip tests with bytes names * Test objects implementing the path protocol, not just pathlib.Path. (cherry picked from commit e72576c48b8be1e4f22c2f387f9769efa073c5be) files: M Lib/test/test_bz2.py M Lib/test/test_gzip.py M Lib/test/test_lzma.py M Lib/test/test_tarfile.py M Lib/test/test_zipfile.py diff --git a/Lib/test/test_bz2.py b/Lib/test/test_bz2.py index ba1c02c833847c..16dee59d1c6b24 100644 --- a/Lib/test/test_bz2.py +++ b/Lib/test/test_bz2.py @@ -3,19 +3,19 @@ import array import unittest +import io from io import BytesIO, DEFAULT_BUFFER_SIZE import os import pickle import glob import tempfile -import pathlib import random import shutil import subprocess import threading from test.support import import_helper from test.support import threading_helper -from test.support.os_helper import unlink +from test.support.os_helper import unlink, FakePath import _compression import sys @@ -537,12 +537,136 @@ def testMultiStreamOrdering(self): with BZ2File(self.filename) as bz2f: self.assertEqual(bz2f.read(), data1 + data2) + def testOpenFilename(self): + with BZ2File(self.filename, "wb") as f: + f.write(b'content') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, "ab") as f: + f.write(b'appendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithName(self): + with open(self.filename, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithoutName(self): + bio = BytesIO() + with BZ2File(bio, 'wb') as f: + f.write(b'content') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + with BZ2File(bio, 'ab') as f: + f.write(b'appendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + bio.seek(0) + with BZ2File(bio, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + with self.assertRaises(ValueError): + f.fileno() + + def testOpenFileWithIntName(self): + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with open(fd, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with open(fd, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + with self.assertRaises(ValueError): + f.fileno() + def testOpenBytesFilename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with BZ2File(bytes_filename, "wb") as f: f.write(self.DATA) with BZ2File(bytes_filename, "rb") as f: @@ -552,7 +676,7 @@ def testOpenBytesFilename(self): self.assertEqual(f.read(), self.DATA) def testOpenPathLikeFilename(self): - filename = pathlib.Path(self.filename) + filename = FakePath(self.filename) with BZ2File(filename, "wb") as f: f.write(self.DATA) with BZ2File(filename, "rb") as f: diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py index 6de413e5056ef0..eec27f89bb8276 100644 --- a/Lib/test/test_gzip.py +++ b/Lib/test/test_gzip.py @@ -5,7 +5,6 @@ import functools import io import os -import pathlib import struct import sys import unittest @@ -78,16 +77,18 @@ def test_write(self): f.close() def test_write_read_with_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.GzipFile(filename, 'w') as f: f.write(data1 * 50) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) with gzip.GzipFile(filename, 'a') as f: f.write(data1) with gzip.GzipFile(filename) as f: d = f.read() self.assertEqual(d, data1 * 51) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) # The following test_write_xy methods test that write accepts # the corresponding bytes-like object type as input @@ -471,13 +472,118 @@ def test_textio_readlines(self): with io.TextIOWrapper(f, encoding="ascii") as t: self.assertEqual(t.readlines(), lines) + def test_fileobj_with_name(self): + with open(self.filename, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + def test_fileobj_from_fdopen(self): # Issue #13781: Opening a GzipFile for writing fails when using a # fileobj created with os.fdopen(). - fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT) - with os.fdopen(fd, "wb") as f: - with gzip.GzipFile(fileobj=f, mode="w") as g: - pass + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_EXCL) + with os.fdopen(fd, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with os.fdopen(fd, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with os.fdopen(fd, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with os.fdopen(fd, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) def test_fileobj_mode(self): gzip.GzipFile(self.filename, "wb").close() @@ -507,17 +613,69 @@ def test_fileobj_mode(self): def test_bytes_filename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with gzip.GzipFile(bytes_filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, bytes_filename) with gzip.GzipFile(bytes_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, bytes_filename) # Sanity check that we are actually operating on the right file. with gzip.GzipFile(str_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, str_filename) + + def test_fileobj_without_name(self): + bio = io.BytesIO() + with gzip.GzipFile(fileobj=bio, mode='wb') as f: + f.write(data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + bio.seek(0) + with gzip.GzipFile(fileobj=bio, mode='rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + + def test_fileobj_and_filename(self): + filename2 = self.filename + 'new' + with (open(self.filename, 'wb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='wb') as f): + f.write(data1 * 50) + self.assertEqual(f.name, filename2) + with (open(self.filename, 'rb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='rb') as f): + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, filename2) + # Sanity check that we are actually operating on the right file. + with gzip.GzipFile(self.filename, 'rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, self.filename) def test_decompress_limited(self): """Decompressed data buffering should be limited""" @@ -646,13 +804,16 @@ def test_binary_modes(self): self.assertEqual(file_data, uncompressed) def test_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.open(filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, self.filename) with gzip.open(filename, "ab") as f: f.write(data1) + self.assertEqual(f.name, self.filename) with gzip.open(filename) as f: self.assertEqual(f.read(), data1 * 51) + self.assertEqual(f.name, self.filename) def test_implicit_binary_modes(self): # Test implicit binary modes (no "b" or "t" in mode string). diff --git a/Lib/test/test_lzma.py b/Lib/test/test_lzma.py index d2ae133a41ff40..0d0c34a910d87a 100644 --- a/Lib/test/test_lzma.py +++ b/Lib/test/test_lzma.py @@ -2,7 +2,6 @@ import array from io import BytesIO, UnsupportedOperation, DEFAULT_BUFFER_SIZE import os -import pathlib import pickle import random import sys @@ -12,7 +11,7 @@ from test.support import _4G, bigmemtest from test.support.import_helper import import_module from test.support.os_helper import ( - TESTFN, unlink + TESTFN, unlink, FakePath ) lzma = import_module("lzma") @@ -544,7 +543,7 @@ def test_init(self): pass def test_init_with_PathLike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename, COMPRESSED_XZ): with LZMAFile(filename) as f: self.assertEqual(f.read(), INPUT) @@ -581,11 +580,10 @@ def test_init_with_x_mode(self): self.addCleanup(unlink, TESTFN) for mode in ("x", "xb"): unlink(TESTFN) - with LZMAFile(TESTFN, mode): + with LZMAFile(TESTFN, mode) as f: pass with self.assertRaises(FileExistsError): - with LZMAFile(TESTFN, mode): - pass + LZMAFile(TESTFN, mode) def test_init_bad_mode(self): with self.assertRaises(ValueError): @@ -866,17 +864,59 @@ def test_read_from_file(self): with LZMAFile(TESTFN) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) def test_read_from_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) with TempFile(TESTFN, COMPRESSED_XZ): with LZMAFile(bytes_filename) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + def test_read_from_fileobj(self): + with TempFile(TESTFN, COMPRESSED_XZ): + with open(TESTFN, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def test_read_from_fileobj_with_int_name(self): + with TempFile(TESTFN, COMPRESSED_XZ): + fd = os.open(TESTFN, os.O_RDONLY) + with open(fd, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + def test_read_incomplete(self): with LZMAFile(BytesIO(COMPRESSED_XZ[:128])) as f: self.assertRaises(EOFError, f.read) @@ -1059,6 +1099,17 @@ def test_write_to_file(self): try: with LZMAFile(TESTFN, "w") as f: f.write(INPUT) + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + expected = lzma.compress(INPUT) with open(TESTFN, "rb") as f: self.assertEqual(f.read(), expected) @@ -1066,10 +1117,7 @@ def test_write_to_file(self): unlink(TESTFN) def test_write_to_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) try: with LZMAFile(bytes_filename, "w") as f: f.write(INPUT) @@ -1079,6 +1127,51 @@ def test_write_to_file_with_bytes_filename(self): finally: unlink(TESTFN) + def test_write_to_fileobj(self): + try: + with open(TESTFN, "wb") as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + + def test_write_to_fileobj_with_int_name(self): + try: + fd = os.open(TESTFN, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + def test_write_append_to_file(self): part1 = INPUT[:1024] part2 = INPUT[1024:1536] @@ -1284,7 +1377,7 @@ def test_filename(self): self.assertEqual(f.read(), INPUT * 2) def test_with_pathlike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename): with lzma.open(filename, "wb") as f: f.write(INPUT) diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index e960f5616b308b..389da7be3a3247 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -487,7 +487,30 @@ def test_length_zero_header(self): with tarfile.open(support.findfile('recursion.tar')) as tar: pass + def test_extractfile_attrs(self): + # gh-74468: TarFile.name must name a file, not a parent archive. + file = self.tar.getmember('ustar/regtype') + with self.tar.extractfile(file) as fobj: + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) + self.assertIs(fobj.closed, False) + self.assertIs(fobj.closed, True) + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) + + class MiscReadTestBase(CommonReadTest): + is_stream = False def requires_name_attribute(self): pass @@ -780,6 +803,7 @@ def requires_name_attribute(self): class StreamReadTest(CommonReadTest, unittest.TestCase): prefix="r|" + is_stream = True def test_read_through(self): # Issue #11224: A poorly designed _FileInFile.read() method diff --git a/Lib/test/test_zipfile.py b/Lib/test/test_zipfile.py index b4ce055bb57fd2..bb5dad2d17bbd4 100644 --- a/Lib/test/test_zipfile.py +++ b/Lib/test/test_zipfile.py @@ -28,7 +28,7 @@ captured_stdout, captured_stderr, requires_subprocess ) from test.support.os_helper import ( - TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count + TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath ) @@ -162,7 +162,7 @@ def test_open(self): self.zip_open_test(f, self.compression) def test_open_with_pathlike(self): - path = pathlib.Path(TESTFN2) + path = FakePath(TESTFN2) self.zip_open_test(path, self.compression) with zipfile.ZipFile(path, "r", self.compression) as zipfp: self.assertIsInstance(zipfp.filename, str) @@ -449,6 +449,27 @@ def write(self, data): self.assertEqual(zipfp.read('file1'), b'data1') self.assertEqual(zipfp.read('file2'), b'data2') + def test_zipextfile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + zipfp.writestr(fname, "bogus") + + with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: + with zipfp.open(fname) as fid: + self.assertEqual(fid.name, fname) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertEqual(fid.mode, 'r') + self.assertIs(fid.readable(), True) + self.assertIs(fid.writable(), False) + self.assertIs(fid.seekable(), True) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertEqual(fid.name, fname) + self.assertEqual(fid.mode, 'r') + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertRaises(ValueError, fid.readable) + self.assertIs(fid.writable(), False) + self.assertRaises(ValueError, fid.seekable) def tearDown(self): unlink(TESTFN) @@ -580,17 +601,16 @@ def test_write_default_name(self): def test_io_on_closed_zipextfile(self): fname = "somefile.txt" - with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: zipfp.writestr(fname, "bogus") with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: with zipfp.open(fname) as fid: fid.close() + self.assertIs(fid.closed, True) self.assertRaises(ValueError, fid.read) self.assertRaises(ValueError, fid.seek, 0) self.assertRaises(ValueError, fid.tell) - self.assertRaises(ValueError, fid.readable) - self.assertRaises(ValueError, fid.seekable) def test_write_to_readonly(self): """Check that trying to call write() on a readonly ZipFile object @@ -1287,6 +1307,21 @@ def test_issue44439(self): self.assertEqual(data.write(q), LENGTH) self.assertEqual(zip.getinfo('data').file_size, LENGTH) + def test_zipwritefile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: + with zipfp.open(fname, 'w') as fid: + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + class StoredWriterTests(AbstractWriterTests, unittest.TestCase): compression = zipfile.ZIP_STORED @@ -1489,7 +1524,7 @@ def test_write_pathlike(self): fp.write("print(42)\n") with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: - zipfp.writepy(pathlib.Path(TESTFN2) / "mod1.py") + zipfp.writepy(FakePath(os.path.join(TESTFN2, "mod1.py"))) names = zipfp.namelist() self.assertCompiledIn('mod1.py', names) finally: @@ -1547,7 +1582,7 @@ def test_extract_with_target(self): def test_extract_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_with_target(pathlib.Path(extdir)) + self._test_extract_with_target(FakePath(extdir)) def test_extract_all(self): with temp_cwd(): @@ -1582,7 +1617,7 @@ def test_extract_all_with_target(self): def test_extract_all_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_all_with_target(pathlib.Path(extdir)) + self._test_extract_all_with_target(FakePath(extdir)) def check_file(self, filename, content): self.assertTrue(os.path.isfile(filename)) @@ -1855,7 +1890,7 @@ def test_is_zip_erroneous_file(self): fp.write("this is not a legal zip file\n") self.assertFalse(zipfile.is_zipfile(TESTFN)) # - passing a path-like object - self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) + self.assertFalse(zipfile.is_zipfile(FakePath(TESTFN))) # - passing a file object with open(TESTFN, "rb") as fp: self.assertFalse(zipfile.is_zipfile(fp)) @@ -2957,7 +2992,7 @@ def test_from_file(self): self.assertEqual(zi.file_size, os.path.getsize(__file__)) def test_from_file_pathlike(self): - zi = zipfile.ZipInfo.from_file(pathlib.Path(__file__)) + zi = zipfile.ZipInfo.from_file(FakePath(__file__)) self.assertEqual(posixpath.basename(zi.filename), 'test_zipfile.py') self.assertFalse(zi.is_dir()) self.assertEqual(zi.file_size, os.path.getsize(__file__))

1 0

[3.12] gh-115961: Improve tests for compressed file-like objects (GH-115963) (GH-116032)
by serhiy-storchaka Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/e5e98626a15dd62eccbb6a281ae0263f00… commit: e5e98626a15dd62eccbb6a281ae0263f005f4942 branch: 3.12 author: Miss Islington (bot) <31488909+miss-islington(a)users.noreply.github.com> committer: serhiy-storchaka <storchaka(a)gmail.com> date: 2024-02-28T11:00:50+02:00 summary: [3.12] gh-115961: Improve tests for compressed file-like objects (GH-115963) (GH-116032) * Increase coverage for compressed file-like objects initialized with a file name, an open file object, a file object opened by file descriptor, and a file-like object without name and mode attributes (io.BytesIO) * Increase coverage for name, fileno(), mode, readable(), writable(), seekable() in different modes and states * No longer skip tests with bytes names * Test objects implementing the path protocol, not just pathlib.Path. (cherry picked from commit e72576c48b8be1e4f22c2f387f9769efa073c5be) Co-authored-by: Serhiy Storchaka <storchaka(a)gmail.com> files: M Lib/test/test_bz2.py M Lib/test/test_gzip.py M Lib/test/test_lzma.py M Lib/test/test_tarfile.py M Lib/test/test_zipfile/test_core.py diff --git a/Lib/test/test_bz2.py b/Lib/test/test_bz2.py index 1f0b9adc3698b4..772f0eacce28f5 100644 --- a/Lib/test/test_bz2.py +++ b/Lib/test/test_bz2.py @@ -3,19 +3,19 @@ import array import unittest +import io from io import BytesIO, DEFAULT_BUFFER_SIZE import os import pickle import glob import tempfile -import pathlib import random import shutil import subprocess import threading from test.support import import_helper from test.support import threading_helper -from test.support.os_helper import unlink +from test.support.os_helper import unlink, FakePath import _compression import sys @@ -537,12 +537,136 @@ def testMultiStreamOrdering(self): with BZ2File(self.filename) as bz2f: self.assertEqual(bz2f.read(), data1 + data2) + def testOpenFilename(self): + with BZ2File(self.filename, "wb") as f: + f.write(b'content') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, "ab") as f: + f.write(b'appendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithName(self): + with open(self.filename, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithoutName(self): + bio = BytesIO() + with BZ2File(bio, 'wb') as f: + f.write(b'content') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + with BZ2File(bio, 'ab') as f: + f.write(b'appendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + bio.seek(0) + with BZ2File(bio, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + with self.assertRaises(ValueError): + f.fileno() + + def testOpenFileWithIntName(self): + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with open(fd, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with open(fd, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + with self.assertRaises(ValueError): + f.fileno() + def testOpenBytesFilename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with BZ2File(bytes_filename, "wb") as f: f.write(self.DATA) with BZ2File(bytes_filename, "rb") as f: @@ -552,7 +676,7 @@ def testOpenBytesFilename(self): self.assertEqual(f.read(), self.DATA) def testOpenPathLikeFilename(self): - filename = pathlib.Path(self.filename) + filename = FakePath(self.filename) with BZ2File(filename, "wb") as f: f.write(self.DATA) with BZ2File(filename, "rb") as f: diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py index 128f933787a3f6..d220c7d06e50c9 100644 --- a/Lib/test/test_gzip.py +++ b/Lib/test/test_gzip.py @@ -5,7 +5,6 @@ import functools import io import os -import pathlib import struct import sys import unittest @@ -79,16 +78,18 @@ def test_write(self): f.close() def test_write_read_with_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.GzipFile(filename, 'w') as f: f.write(data1 * 50) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) with gzip.GzipFile(filename, 'a') as f: f.write(data1) with gzip.GzipFile(filename) as f: d = f.read() self.assertEqual(d, data1 * 51) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) # The following test_write_xy methods test that write accepts # the corresponding bytes-like object type as input @@ -472,13 +473,118 @@ def test_textio_readlines(self): with io.TextIOWrapper(f, encoding="ascii") as t: self.assertEqual(t.readlines(), lines) + def test_fileobj_with_name(self): + with open(self.filename, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + def test_fileobj_from_fdopen(self): # Issue #13781: Opening a GzipFile for writing fails when using a # fileobj created with os.fdopen(). - fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT) - with os.fdopen(fd, "wb") as f: - with gzip.GzipFile(fileobj=f, mode="w") as g: - pass + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_EXCL) + with os.fdopen(fd, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with os.fdopen(fd, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with os.fdopen(fd, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with os.fdopen(fd, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) def test_fileobj_mode(self): gzip.GzipFile(self.filename, "wb").close() @@ -508,17 +614,69 @@ def test_fileobj_mode(self): def test_bytes_filename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with gzip.GzipFile(bytes_filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, bytes_filename) with gzip.GzipFile(bytes_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, bytes_filename) # Sanity check that we are actually operating on the right file. with gzip.GzipFile(str_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, str_filename) + + def test_fileobj_without_name(self): + bio = io.BytesIO() + with gzip.GzipFile(fileobj=bio, mode='wb') as f: + f.write(data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + bio.seek(0) + with gzip.GzipFile(fileobj=bio, mode='rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + + def test_fileobj_and_filename(self): + filename2 = self.filename + 'new' + with (open(self.filename, 'wb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='wb') as f): + f.write(data1 * 50) + self.assertEqual(f.name, filename2) + with (open(self.filename, 'rb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='rb') as f): + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, filename2) + # Sanity check that we are actually operating on the right file. + with gzip.GzipFile(self.filename, 'rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, self.filename) def test_decompress_limited(self): """Decompressed data buffering should be limited""" @@ -707,13 +865,16 @@ def test_binary_modes(self): self.assertEqual(file_data, uncompressed) def test_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.open(filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, self.filename) with gzip.open(filename, "ab") as f: f.write(data1) + self.assertEqual(f.name, self.filename) with gzip.open(filename) as f: self.assertEqual(f.read(), data1 * 51) + self.assertEqual(f.name, self.filename) def test_implicit_binary_modes(self): # Test implicit binary modes (no "b" or "t" in mode string). diff --git a/Lib/test/test_lzma.py b/Lib/test/test_lzma.py index 65e6488c5d7b10..db290e139327e0 100644 --- a/Lib/test/test_lzma.py +++ b/Lib/test/test_lzma.py @@ -2,7 +2,6 @@ import array from io import BytesIO, UnsupportedOperation, DEFAULT_BUFFER_SIZE import os -import pathlib import pickle import random import sys @@ -12,7 +11,7 @@ from test.support import _4G, bigmemtest from test.support.import_helper import import_module from test.support.os_helper import ( - TESTFN, unlink + TESTFN, unlink, FakePath ) lzma = import_module("lzma") @@ -548,7 +547,7 @@ def test_init(self): pass def test_init_with_PathLike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename, COMPRESSED_XZ): with LZMAFile(filename) as f: self.assertEqual(f.read(), INPUT) @@ -585,11 +584,10 @@ def test_init_with_x_mode(self): self.addCleanup(unlink, TESTFN) for mode in ("x", "xb"): unlink(TESTFN) - with LZMAFile(TESTFN, mode): + with LZMAFile(TESTFN, mode) as f: pass with self.assertRaises(FileExistsError): - with LZMAFile(TESTFN, mode): - pass + LZMAFile(TESTFN, mode) def test_init_bad_mode(self): with self.assertRaises(ValueError): @@ -867,17 +865,59 @@ def test_read_from_file(self): with LZMAFile(TESTFN) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) def test_read_from_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) with TempFile(TESTFN, COMPRESSED_XZ): with LZMAFile(bytes_filename) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + def test_read_from_fileobj(self): + with TempFile(TESTFN, COMPRESSED_XZ): + with open(TESTFN, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def test_read_from_fileobj_with_int_name(self): + with TempFile(TESTFN, COMPRESSED_XZ): + fd = os.open(TESTFN, os.O_RDONLY) + with open(fd, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + def test_read_incomplete(self): with LZMAFile(BytesIO(COMPRESSED_XZ[:128])) as f: self.assertRaises(EOFError, f.read) @@ -1051,6 +1091,17 @@ def test_write_to_file(self): try: with LZMAFile(TESTFN, "w") as f: f.write(INPUT) + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + expected = lzma.compress(INPUT) with open(TESTFN, "rb") as f: self.assertEqual(f.read(), expected) @@ -1058,10 +1109,7 @@ def test_write_to_file(self): unlink(TESTFN) def test_write_to_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) try: with LZMAFile(bytes_filename, "w") as f: f.write(INPUT) @@ -1071,6 +1119,51 @@ def test_write_to_file_with_bytes_filename(self): finally: unlink(TESTFN) + def test_write_to_fileobj(self): + try: + with open(TESTFN, "wb") as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + + def test_write_to_fileobj_with_int_name(self): + try: + fd = os.open(TESTFN, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + def test_write_append_to_file(self): part1 = INPUT[:1024] part2 = INPUT[1024:1536] @@ -1276,7 +1369,7 @@ def test_filename(self): self.assertEqual(f.read(), INPUT * 2) def test_with_pathlike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename): with lzma.open(filename, "wb") as f: f.write(INPUT) diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index 71489ea493f94a..9aa17267490384 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -487,14 +487,32 @@ def test_length_zero_header(self): with tarfile.open(support.findfile('recursion.tar')) as tar: pass - def test_extractfile_name(self): + def test_extractfile_attrs(self): # gh-74468: TarFile.name must name a file, not a parent archive. file = self.tar.getmember('ustar/regtype') with self.tar.extractfile(file) as fobj: self.assertEqual(fobj.name, 'ustar/regtype') + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) + self.assertIs(fobj.closed, False) + self.assertIs(fobj.closed, True) + self.assertEqual(fobj.name, 'ustar/regtype') + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) class MiscReadTestBase(CommonReadTest): + is_stream = False def requires_name_attribute(self): pass @@ -787,6 +805,7 @@ def requires_name_attribute(self): class StreamReadTest(CommonReadTest, unittest.TestCase): prefix="r|" + is_stream = True def test_read_through(self): # Issue #11224: A poorly designed _FileInFile.read() method diff --git a/Lib/test/test_zipfile/test_core.py b/Lib/test/test_zipfile/test_core.py index ecc3dd6d0b5aa5..b63eb0c9339864 100644 --- a/Lib/test/test_zipfile/test_core.py +++ b/Lib/test/test_zipfile/test_core.py @@ -4,7 +4,6 @@ import io import itertools import os -import pathlib import posixpath import struct import subprocess @@ -25,7 +24,7 @@ captured_stdout, captured_stderr, requires_subprocess ) from test.support.os_helper import ( - TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count + TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath ) @@ -160,7 +159,7 @@ def test_open(self): self.zip_open_test(f, self.compression) def test_open_with_pathlike(self): - path = pathlib.Path(TESTFN2) + path = FakePath(TESTFN2) self.zip_open_test(path, self.compression) with zipfile.ZipFile(path, "r", self.compression) as zipfp: self.assertIsInstance(zipfp.filename, str) @@ -447,6 +446,27 @@ def write(self, data): self.assertEqual(zipfp.read('file1'), b'data1') self.assertEqual(zipfp.read('file2'), b'data2') + def test_zipextfile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + zipfp.writestr(fname, "bogus") + + with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: + with zipfp.open(fname) as fid: + self.assertEqual(fid.name, fname) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertEqual(fid.mode, 'r') + self.assertIs(fid.readable(), True) + self.assertIs(fid.writable(), False) + self.assertIs(fid.seekable(), True) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertEqual(fid.name, fname) + self.assertEqual(fid.mode, 'r') + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertRaises(ValueError, fid.readable) + self.assertIs(fid.writable(), False) + self.assertRaises(ValueError, fid.seekable) def tearDown(self): unlink(TESTFN) @@ -578,17 +598,16 @@ def test_write_default_name(self): def test_io_on_closed_zipextfile(self): fname = "somefile.txt" - with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: zipfp.writestr(fname, "bogus") with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: with zipfp.open(fname) as fid: fid.close() + self.assertIs(fid.closed, True) self.assertRaises(ValueError, fid.read) self.assertRaises(ValueError, fid.seek, 0) self.assertRaises(ValueError, fid.tell) - self.assertRaises(ValueError, fid.readable) - self.assertRaises(ValueError, fid.seekable) def test_write_to_readonly(self): """Check that trying to call write() on a readonly ZipFile object @@ -1285,6 +1304,21 @@ def test_issue44439(self): self.assertEqual(data.write(q), LENGTH) self.assertEqual(zip.getinfo('data').file_size, LENGTH) + def test_zipwritefile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: + with zipfp.open(fname, 'w') as fid: + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + class StoredWriterTests(AbstractWriterTests, unittest.TestCase): compression = zipfile.ZIP_STORED @@ -1487,7 +1521,7 @@ def test_write_pathlike(self): fp.write("print(42)\n") with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: - zipfp.writepy(pathlib.Path(TESTFN2) / "mod1.py") + zipfp.writepy(FakePath(os.path.join(TESTFN2, "mod1.py"))) names = zipfp.namelist() self.assertCompiledIn('mod1.py', names) finally: @@ -1545,7 +1579,7 @@ def test_extract_with_target(self): def test_extract_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_with_target(pathlib.Path(extdir)) + self._test_extract_with_target(FakePath(extdir)) def test_extract_all(self): with temp_cwd(): @@ -1580,7 +1614,7 @@ def test_extract_all_with_target(self): def test_extract_all_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_all_with_target(pathlib.Path(extdir)) + self._test_extract_all_with_target(FakePath(extdir)) def check_file(self, filename, content): self.assertTrue(os.path.isfile(filename)) @@ -1893,7 +1927,7 @@ def test_is_zip_erroneous_file(self): fp.write("this is not a legal zip file\n") self.assertFalse(zipfile.is_zipfile(TESTFN)) # - passing a path-like object - self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) + self.assertFalse(zipfile.is_zipfile(FakePath(TESTFN))) # - passing a file object with open(TESTFN, "rb") as fp: self.assertFalse(zipfile.is_zipfile(fp)) @@ -3013,7 +3047,7 @@ def test_from_file(self): self.assertEqual(zi.file_size, os.path.getsize(__file__)) def test_from_file_pathlike(self): - zi = zipfile.ZipInfo.from_file(pathlib.Path(__file__)) + zi = zipfile.ZipInfo.from_file(FakePath(__file__)) self.assertEqual(posixpath.basename(zi.filename), 'test_core.py') self.assertFalse(zi.is_dir()) self.assertEqual(zi.file_size, os.path.getsize(__file__))

1 0

gh-115961: Improve tests for compressed file-like objects (GH-115963)
by serhiy-storchaka Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/e72576c48b8be1e4f22c2f387f9769efa0… commit: e72576c48b8be1e4f22c2f387f9769efa073c5be branch: main author: Serhiy Storchaka <storchaka(a)gmail.com> committer: serhiy-storchaka <storchaka(a)gmail.com> date: 2024-02-28T07:51:08Z summary: gh-115961: Improve tests for compressed file-like objects (GH-115963) * Increase coverage for compressed file-like objects initialized with a file name, an open file object, a file object opened by file descriptor, and a file-like object without name and mode attributes (io.BytesIO) * Increase coverage for name, fileno(), mode, readable(), writable(), seekable() in different modes and states * No longer skip tests with bytes names * Test objects implementing the path protocol, not just pathlib.Path. files: M Lib/test/test_bz2.py M Lib/test/test_gzip.py M Lib/test/test_lzma.py M Lib/test/test_tarfile.py M Lib/test/test_zipfile/test_core.py diff --git a/Lib/test/test_bz2.py b/Lib/test/test_bz2.py index 1f0b9adc3698b4..772f0eacce28f5 100644 --- a/Lib/test/test_bz2.py +++ b/Lib/test/test_bz2.py @@ -3,19 +3,19 @@ import array import unittest +import io from io import BytesIO, DEFAULT_BUFFER_SIZE import os import pickle import glob import tempfile -import pathlib import random import shutil import subprocess import threading from test.support import import_helper from test.support import threading_helper -from test.support.os_helper import unlink +from test.support.os_helper import unlink, FakePath import _compression import sys @@ -537,12 +537,136 @@ def testMultiStreamOrdering(self): with BZ2File(self.filename) as bz2f: self.assertEqual(bz2f.read(), data1 + data2) + def testOpenFilename(self): + with BZ2File(self.filename, "wb") as f: + f.write(b'content') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, "ab") as f: + f.write(b'appendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with BZ2File(self.filename, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithName(self): + with open(self.filename, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + with open(self.filename, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + with self.assertRaises(ValueError): + f.fileno() + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def testOpenFileWithoutName(self): + bio = BytesIO() + with BZ2File(bio, 'wb') as f: + f.write(b'content') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + with BZ2File(bio, 'ab') as f: + f.write(b'appendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertRaises(ValueError, f.fileno) + + bio.seek(0) + with BZ2File(bio, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertRaises(io.UnsupportedOperation, f.fileno) + with self.assertRaises(ValueError): + f.fileno() + + def testOpenFileWithIntName(self): + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with BZ2File(raw, 'wb') as f: + f.write(b'content') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with open(fd, 'ab') as raw: + with BZ2File(raw, 'ab') as f: + f.write(b'appendix') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertRaises(ValueError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with open(fd, 'rb') as raw: + with BZ2File(raw, 'rb') as f: + self.assertEqual(f.read(), b'contentappendix') + self.assertEqual(f.fileno(), raw.fileno()) + with self.assertRaises(ValueError): + f.fileno() + def testOpenBytesFilename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with BZ2File(bytes_filename, "wb") as f: f.write(self.DATA) with BZ2File(bytes_filename, "rb") as f: @@ -552,7 +676,7 @@ def testOpenBytesFilename(self): self.assertEqual(f.read(), self.DATA) def testOpenPathLikeFilename(self): - filename = pathlib.Path(self.filename) + filename = FakePath(self.filename) with BZ2File(filename, "wb") as f: f.write(self.DATA) with BZ2File(filename, "rb") as f: diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py index 128f933787a3f6..d220c7d06e50c9 100644 --- a/Lib/test/test_gzip.py +++ b/Lib/test/test_gzip.py @@ -5,7 +5,6 @@ import functools import io import os -import pathlib import struct import sys import unittest @@ -79,16 +78,18 @@ def test_write(self): f.close() def test_write_read_with_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.GzipFile(filename, 'w') as f: f.write(data1 * 50) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) with gzip.GzipFile(filename, 'a') as f: f.write(data1) with gzip.GzipFile(filename) as f: d = f.read() self.assertEqual(d, data1 * 51) self.assertIsInstance(f.name, str) + self.assertEqual(f.name, self.filename) # The following test_write_xy methods test that write accepts # the corresponding bytes-like object type as input @@ -472,13 +473,118 @@ def test_textio_readlines(self): with io.TextIOWrapper(f, encoding="ascii") as t: self.assertEqual(t.readlines(), lines) + def test_fileobj_with_name(self): + with open(self.filename, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + with open(self.filename, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, raw.name) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, raw.name) + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + def test_fileobj_from_fdopen(self): # Issue #13781: Opening a GzipFile for writing fails when using a # fileobj created with os.fdopen(). - fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT) - with os.fdopen(fd, "wb") as f: - with gzip.GzipFile(fileobj=f, mode="w") as g: - pass + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_EXCL) + with os.fdopen(fd, "xb") as raw: + with gzip.GzipFile(fileobj=raw, mode="x") as f: + f.write(b'one') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with os.fdopen(fd, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="w") as f: + f.write(b'two') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + with os.fdopen(fd, "ab") as raw: + with gzip.GzipFile(fileobj=raw, mode="a") as f: + f.write(b'three') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + + fd = os.open(self.filename, os.O_RDONLY) + with os.fdopen(fd, "rb") as raw: + with gzip.GzipFile(fileobj=raw, mode="r") as f: + self.assertEqual(f.read(), b'twothree') + self.assertEqual(f.name, '') + self.assertEqual(f.fileno(), raw.fileno()) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) def test_fileobj_mode(self): gzip.GzipFile(self.filename, "wb").close() @@ -508,17 +614,69 @@ def test_fileobj_mode(self): def test_bytes_filename(self): str_filename = self.filename - try: - bytes_filename = str_filename.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(str_filename) with gzip.GzipFile(bytes_filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, bytes_filename) with gzip.GzipFile(bytes_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, bytes_filename) # Sanity check that we are actually operating on the right file. with gzip.GzipFile(str_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, str_filename) + + def test_fileobj_without_name(self): + bio = io.BytesIO() + with gzip.GzipFile(fileobj=bio, mode='wb') as f: + f.write(data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.WRITE) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), True) + + bio.seek(0) + with gzip.GzipFile(fileobj=bio, mode='rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, '') + self.assertRaises(io.UnsupportedOperation, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertEqual(f.name, '') + self.assertRaises(AttributeError, f.fileno) + self.assertEqual(f.mode, gzip.READ) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + + def test_fileobj_and_filename(self): + filename2 = self.filename + 'new' + with (open(self.filename, 'wb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='wb') as f): + f.write(data1 * 50) + self.assertEqual(f.name, filename2) + with (open(self.filename, 'rb') as fileobj, + gzip.GzipFile(fileobj=fileobj, filename=filename2, mode='rb') as f): + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, filename2) + # Sanity check that we are actually operating on the right file. + with gzip.GzipFile(self.filename, 'rb') as f: + self.assertEqual(f.read(), data1 * 50) + self.assertEqual(f.name, self.filename) def test_decompress_limited(self): """Decompressed data buffering should be limited""" @@ -707,13 +865,16 @@ def test_binary_modes(self): self.assertEqual(file_data, uncompressed) def test_pathlike_file(self): - filename = pathlib.Path(self.filename) + filename = os_helper.FakePath(self.filename) with gzip.open(filename, "wb") as f: f.write(data1 * 50) + self.assertEqual(f.name, self.filename) with gzip.open(filename, "ab") as f: f.write(data1) + self.assertEqual(f.name, self.filename) with gzip.open(filename) as f: self.assertEqual(f.read(), data1 * 51) + self.assertEqual(f.name, self.filename) def test_implicit_binary_modes(self): # Test implicit binary modes (no "b" or "t" in mode string). diff --git a/Lib/test/test_lzma.py b/Lib/test/test_lzma.py index 65e6488c5d7b10..db290e139327e0 100644 --- a/Lib/test/test_lzma.py +++ b/Lib/test/test_lzma.py @@ -2,7 +2,6 @@ import array from io import BytesIO, UnsupportedOperation, DEFAULT_BUFFER_SIZE import os -import pathlib import pickle import random import sys @@ -12,7 +11,7 @@ from test.support import _4G, bigmemtest from test.support.import_helper import import_module from test.support.os_helper import ( - TESTFN, unlink + TESTFN, unlink, FakePath ) lzma = import_module("lzma") @@ -548,7 +547,7 @@ def test_init(self): pass def test_init_with_PathLike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename, COMPRESSED_XZ): with LZMAFile(filename) as f: self.assertEqual(f.read(), INPUT) @@ -585,11 +584,10 @@ def test_init_with_x_mode(self): self.addCleanup(unlink, TESTFN) for mode in ("x", "xb"): unlink(TESTFN) - with LZMAFile(TESTFN, mode): + with LZMAFile(TESTFN, mode) as f: pass with self.assertRaises(FileExistsError): - with LZMAFile(TESTFN, mode): - pass + LZMAFile(TESTFN, mode) def test_init_bad_mode(self): with self.assertRaises(ValueError): @@ -867,17 +865,59 @@ def test_read_from_file(self): with LZMAFile(TESTFN) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) def test_read_from_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) with TempFile(TESTFN, COMPRESSED_XZ): with LZMAFile(bytes_filename) as f: self.assertEqual(f.read(), INPUT) self.assertEqual(f.read(), b"") + def test_read_from_fileobj(self): + with TempFile(TESTFN, COMPRESSED_XZ): + with open(TESTFN, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + def test_read_from_fileobj_with_int_name(self): + with TempFile(TESTFN, COMPRESSED_XZ): + fd = os.open(TESTFN, os.O_RDONLY) + with open(fd, 'rb') as raw: + with LZMAFile(raw) as f: + self.assertEqual(f.read(), INPUT) + self.assertEqual(f.read(), b"") + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), True) + self.assertIs(f.writable(), False) + self.assertIs(f.seekable(), True) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + def test_read_incomplete(self): with LZMAFile(BytesIO(COMPRESSED_XZ[:128])) as f: self.assertRaises(EOFError, f.read) @@ -1051,6 +1091,17 @@ def test_write_to_file(self): try: with LZMAFile(TESTFN, "w") as f: f.write(INPUT) + self.assertIsInstance(f.fileno(), int) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + expected = lzma.compress(INPUT) with open(TESTFN, "rb") as f: self.assertEqual(f.read(), expected) @@ -1058,10 +1109,7 @@ def test_write_to_file(self): unlink(TESTFN) def test_write_to_file_with_bytes_filename(self): - try: - bytes_filename = TESTFN.encode("ascii") - except UnicodeEncodeError: - self.skipTest("Temporary file name needs to be ASCII") + bytes_filename = os.fsencode(TESTFN) try: with LZMAFile(bytes_filename, "w") as f: f.write(INPUT) @@ -1071,6 +1119,51 @@ def test_write_to_file_with_bytes_filename(self): finally: unlink(TESTFN) + def test_write_to_fileobj(self): + try: + with open(TESTFN, "wb") as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + + def test_write_to_fileobj_with_int_name(self): + try: + fd = os.open(TESTFN, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + with open(fd, 'wb') as raw: + with LZMAFile(raw, "w") as f: + f.write(INPUT) + self.assertEqual(f.fileno(), raw.fileno()) + self.assertIs(f.readable(), False) + self.assertIs(f.writable(), True) + self.assertIs(f.seekable(), False) + self.assertIs(f.closed, False) + self.assertIs(f.closed, True) + self.assertRaises(ValueError, f.fileno) + self.assertRaises(ValueError, f.readable) + self.assertRaises(ValueError, f.writable) + self.assertRaises(ValueError, f.seekable) + + expected = lzma.compress(INPUT) + with open(TESTFN, "rb") as f: + self.assertEqual(f.read(), expected) + finally: + unlink(TESTFN) + def test_write_append_to_file(self): part1 = INPUT[:1024] part2 = INPUT[1024:1536] @@ -1276,7 +1369,7 @@ def test_filename(self): self.assertEqual(f.read(), INPUT * 2) def test_with_pathlike_filename(self): - filename = pathlib.Path(TESTFN) + filename = FakePath(TESTFN) with TempFile(filename): with lzma.open(filename, "wb") as f: f.write(INPUT) diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index 51f070e96047a6..a047780fdd7e17 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -507,14 +507,32 @@ def test_length_zero_header(self): with tarfile.open(support.findfile('recursion.tar', subdir='archivetestdata')): pass - def test_extractfile_name(self): + def test_extractfile_attrs(self): # gh-74468: TarFile.name must name a file, not a parent archive. file = self.tar.getmember('ustar/regtype') with self.tar.extractfile(file) as fobj: self.assertEqual(fobj.name, 'ustar/regtype') + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) + self.assertIs(fobj.closed, False) + self.assertIs(fobj.closed, True) + self.assertEqual(fobj.name, 'ustar/regtype') + self.assertRaises(AttributeError, fobj.fileno) + self.assertIs(fobj.readable(), True) + self.assertIs(fobj.writable(), False) + if self.is_stream: + self.assertRaises(AttributeError, fobj.seekable) + else: + self.assertIs(fobj.seekable(), True) class MiscReadTestBase(CommonReadTest): + is_stream = False def requires_name_attribute(self): pass @@ -807,6 +825,7 @@ def requires_name_attribute(self): class StreamReadTest(CommonReadTest, unittest.TestCase): prefix="r|" + is_stream = True def test_read_through(self): # Issue #11224: A poorly designed _FileInFile.read() method diff --git a/Lib/test/test_zipfile/test_core.py b/Lib/test/test_zipfile/test_core.py index 087fa8d65cc336..7d89f753448dd7 100644 --- a/Lib/test/test_zipfile/test_core.py +++ b/Lib/test/test_zipfile/test_core.py @@ -4,7 +4,6 @@ import io import itertools import os -import pathlib import posixpath import struct import subprocess @@ -25,7 +24,7 @@ captured_stdout, captured_stderr, requires_subprocess ) from test.support.os_helper import ( - TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count + TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count, FakePath ) @@ -160,7 +159,7 @@ def test_open(self): self.zip_open_test(f, self.compression) def test_open_with_pathlike(self): - path = pathlib.Path(TESTFN2) + path = FakePath(TESTFN2) self.zip_open_test(path, self.compression) with zipfile.ZipFile(path, "r", self.compression) as zipfp: self.assertIsInstance(zipfp.filename, str) @@ -447,6 +446,27 @@ def write(self, data): self.assertEqual(zipfp.read('file1'), b'data1') self.assertEqual(zipfp.read('file2'), b'data2') + def test_zipextfile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + zipfp.writestr(fname, "bogus") + + with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: + with zipfp.open(fname) as fid: + self.assertEqual(fid.name, fname) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertEqual(fid.mode, 'r') + self.assertIs(fid.readable(), True) + self.assertIs(fid.writable(), False) + self.assertIs(fid.seekable(), True) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertEqual(fid.name, fname) + self.assertEqual(fid.mode, 'r') + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertRaises(ValueError, fid.readable) + self.assertIs(fid.writable(), False) + self.assertRaises(ValueError, fid.seekable) def tearDown(self): unlink(TESTFN) @@ -578,17 +598,16 @@ def test_write_default_name(self): def test_io_on_closed_zipextfile(self): fname = "somefile.txt" - with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: zipfp.writestr(fname, "bogus") with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: with zipfp.open(fname) as fid: fid.close() + self.assertIs(fid.closed, True) self.assertRaises(ValueError, fid.read) self.assertRaises(ValueError, fid.seek, 0) self.assertRaises(ValueError, fid.tell) - self.assertRaises(ValueError, fid.readable) - self.assertRaises(ValueError, fid.seekable) def test_write_to_readonly(self): """Check that trying to call write() on a readonly ZipFile object @@ -1285,6 +1304,21 @@ def test_issue44439(self): self.assertEqual(data.write(q), LENGTH) self.assertEqual(zip.getinfo('data').file_size, LENGTH) + def test_zipwritefile_attrs(self): + fname = "somefile.txt" + with zipfile.ZipFile(TESTFN2, mode="w", compression=self.compression) as zipfp: + with zipfp.open(fname, 'w') as fid: + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + self.assertIs(fid.closed, False) + self.assertIs(fid.closed, True) + self.assertRaises(io.UnsupportedOperation, fid.fileno) + self.assertIs(fid.readable(), False) + self.assertIs(fid.writable(), True) + self.assertIs(fid.seekable(), False) + class StoredWriterTests(AbstractWriterTests, unittest.TestCase): compression = zipfile.ZIP_STORED @@ -1487,7 +1521,7 @@ def test_write_pathlike(self): fp.write("print(42)\n") with TemporaryFile() as t, zipfile.PyZipFile(t, "w") as zipfp: - zipfp.writepy(pathlib.Path(TESTFN2) / "mod1.py") + zipfp.writepy(FakePath(os.path.join(TESTFN2, "mod1.py"))) names = zipfp.namelist() self.assertCompiledIn('mod1.py', names) finally: @@ -1545,7 +1579,7 @@ def test_extract_with_target(self): def test_extract_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_with_target(pathlib.Path(extdir)) + self._test_extract_with_target(FakePath(extdir)) def test_extract_all(self): with temp_cwd(): @@ -1580,7 +1614,7 @@ def test_extract_all_with_target(self): def test_extract_all_with_target_pathlike(self): with temp_dir() as extdir: - self._test_extract_all_with_target(pathlib.Path(extdir)) + self._test_extract_all_with_target(FakePath(extdir)) def check_file(self, filename, content): self.assertTrue(os.path.isfile(filename)) @@ -1893,7 +1927,7 @@ def test_is_zip_erroneous_file(self): fp.write("this is not a legal zip file\n") self.assertFalse(zipfile.is_zipfile(TESTFN)) # - passing a path-like object - self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) + self.assertFalse(zipfile.is_zipfile(FakePath(TESTFN))) # - passing a file object with open(TESTFN, "rb") as fp: self.assertFalse(zipfile.is_zipfile(fp)) @@ -3013,7 +3047,7 @@ def test_from_file(self): self.assertEqual(zi.file_size, os.path.getsize(__file__)) def test_from_file_pathlike(self): - zi = zipfile.ZipInfo.from_file(pathlib.Path(__file__)) + zi = zipfile.ZipInfo.from_file(FakePath(__file__)) self.assertEqual(posixpath.basename(zi.filename), 'test_core.py') self.assertFalse(zi.is_dir()) self.assertEqual(zi.file_size, os.path.getsize(__file__))

1 0

gh-116026: Try disabling rebuilds of dependents in Homebrew (#116027)
by JelleZijlstra Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/02beb9f0208d22fd8bd893e6e6ec813f7e… commit: 02beb9f0208d22fd8bd893e6e6ec813f7e51b235 branch: main author: Jelle Zijlstra <jelle.zijlstra(a)gmail.com> committer: JelleZijlstra <jelle.zijlstra(a)gmail.com> date: 2024-02-27T21:35:25-08:00 summary: gh-116026: Try disabling rebuilds of dependents in Homebrew (#116027) files: M .github/workflows/reusable-macos.yml diff --git a/.github/workflows/reusable-macos.yml b/.github/workflows/reusable-macos.yml index ba62d9568c6b80..65b73cd791c75c 100644 --- a/.github/workflows/reusable-macos.yml +++ b/.github/workflows/reusable-macos.yml @@ -17,6 +17,7 @@ jobs: HOMEBREW_NO_ANALYTICS: 1 HOMEBREW_NO_AUTO_UPDATE: 1 HOMEBREW_NO_INSTALL_CLEANUP: 1 + HOMEBREW_NO_INSTALLED_DEPENDENTS_CHECK: 1 PYTHONSTRICTEXTENSIONBUILD: 1 strategy: fail-fast: false

1 0

gh-105858: Improve AST node constructors (#105880)
by JelleZijlstra Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/ed4dfd8825b49e16a0fcb9e67baf1b58bb… commit: ed4dfd8825b49e16a0fcb9e67baf1b58bb8d438f branch: main author: Jelle Zijlstra <jelle.zijlstra(a)gmail.com> committer: JelleZijlstra <jelle.zijlstra(a)gmail.com> date: 2024-02-27T18:13:03-08:00 summary: gh-105858: Improve AST node constructors (#105880) Demonstration: >>> ast.FunctionDef.__annotations__ {'name': <class 'str'>, 'args': <class 'ast.arguments'>, 'body': list[ast.stmt], 'decorator_list': list[ast.expr], 'returns': ast.expr | None, 'type_comment': str | None, 'type_params': list[ast.type_param]} >>> ast.FunctionDef() <stdin>:1: DeprecationWarning: FunctionDef.__init__ missing 1 required positional argument: 'name'. This will become an error in Python 3.15. <stdin>:1: DeprecationWarning: FunctionDef.__init__ missing 1 required positional argument: 'args'. This will become an error in Python 3.15. <ast.FunctionDef object at 0x101959460> >>> node = ast.FunctionDef(name="foo", args=ast.arguments()) >>> node.decorator_list [] >>> ast.FunctionDef(whatever="you want", name="x", args=ast.arguments()) <stdin>:1: DeprecationWarning: FunctionDef.__init__ got an unexpected keyword argument 'whatever'. Support for arbitrary keyword arguments is deprecated and will be removed in Python 3.15. <ast.FunctionDef object at 0x1019581f0> files: A Misc/NEWS.d/next/Core and Builtins/2023-06-16-21-29-06.gh-issue-105858.Q7h0EV.rst M Doc/library/ast.rst M Doc/whatsnew/3.13.rst M Include/internal/pycore_global_objects_fini_generated.h M Include/internal/pycore_global_strings.h M Include/internal/pycore_runtime_init_generated.h M Include/internal/pycore_unicodeobject_generated.h M Lib/test/test_ast.py M Parser/asdl_c.py M Python/Python-ast.c diff --git a/Doc/library/ast.rst b/Doc/library/ast.rst index c943c2f498173e..d629393f023c8a 100644 --- a/Doc/library/ast.rst +++ b/Doc/library/ast.rst @@ -103,20 +103,15 @@ Node classes For example, to create and populate an :class:`ast.UnaryOp` node, you could use :: - node = ast.UnaryOp() - node.op = ast.USub() - node.operand = ast.Constant() - node.operand.value = 5 - node.operand.lineno = 0 - node.operand.col_offset = 0 - node.lineno = 0 - node.col_offset = 0 - - or the more compact :: - node = ast.UnaryOp(ast.USub(), ast.Constant(5, lineno=0, col_offset=0), lineno=0, col_offset=0) + If a field that is optional in the grammar is omitted from the constructor, + it defaults to ``None``. If a list field is omitted, it defaults to the empty + list. If any other field is omitted, a :exc:`DeprecationWarning` is raised + and the AST node will not have this field. In Python 3.15, this condition will + raise an error. + .. versionchanged:: 3.8 Class :class:`ast.Constant` is now used for all constants. @@ -140,6 +135,14 @@ Node classes In the meantime, instantiating them will return an instance of a different class. +.. deprecated-removed:: 3.13 3.15 + + Previous versions of Python allowed the creation of AST nodes that were missing + required fields. Similarly, AST node constructors allowed arbitrary keyword + arguments that were set as attributes of the AST node, even if they did not + match any of the fields of the AST node. This behavior is deprecated and will + be removed in Python 3.15. + .. note:: The descriptions of the specific node classes displayed here were initially adapted from the fantastic `Green Tree diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst index ca5a0e7515994f..3a277d7ce1585f 100644 --- a/Doc/whatsnew/3.13.rst +++ b/Doc/whatsnew/3.13.rst @@ -206,6 +206,21 @@ array ast --- +* The constructors of node types in the :mod:`ast` module are now stricter + in the arguments they accept, and have more intuitive behaviour when + arguments are omitted. + + If an optional field on an AST node is not included as an argument when + constructing an instance, the field will now be set to ``None``. Similarly, + if a list field is omitted, that field will now be set to an empty list. + (Previously, in both cases, the attribute would be missing on the newly + constructed AST node instance.) + + If other arguments are omitted, a :exc:`DeprecationWarning` is emitted. + This will cause an exception in Python 3.15. Similarly, passing a keyword + argument that does not map to a field on the AST node is now deprecated, + and will raise an exception in Python 3.15. + * :func:`ast.parse` now accepts an optional argument ``optimize`` which is passed on to the :func:`compile` built-in. This makes it possible to obtain an optimized ``AST``. diff --git a/Include/internal/pycore_global_objects_fini_generated.h b/Include/internal/pycore_global_objects_fini_generated.h index 3253b5271a9b7c..e8b62bd0dcd369 100644 --- a/Include/internal/pycore_global_objects_fini_generated.h +++ b/Include/internal/pycore_global_objects_fini_generated.h @@ -753,6 +753,7 @@ _PyStaticObjects_CheckRefcnt(PyInterpreterState *interp) { _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_check_retval_)); _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_dealloc_warn)); _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_feature_version)); + _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_field_types)); _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_fields_)); _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_finalizing)); _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(_find_and_load)); diff --git a/Include/internal/pycore_global_strings.h b/Include/internal/pycore_global_strings.h index 8780f7ef949165..b41f5c8952021e 100644 --- a/Include/internal/pycore_global_strings.h +++ b/Include/internal/pycore_global_strings.h @@ -242,6 +242,7 @@ struct _Py_global_strings { STRUCT_FOR_ID(_check_retval_) STRUCT_FOR_ID(_dealloc_warn) STRUCT_FOR_ID(_feature_version) + STRUCT_FOR_ID(_field_types) STRUCT_FOR_ID(_fields_) STRUCT_FOR_ID(_finalizing) STRUCT_FOR_ID(_find_and_load) diff --git a/Include/internal/pycore_runtime_init_generated.h b/Include/internal/pycore_runtime_init_generated.h index a9d514856dab1f..016eae02a2103d 100644 --- a/Include/internal/pycore_runtime_init_generated.h +++ b/Include/internal/pycore_runtime_init_generated.h @@ -751,6 +751,7 @@ extern "C" { INIT_ID(_check_retval_), \ INIT_ID(_dealloc_warn), \ INIT_ID(_feature_version), \ + INIT_ID(_field_types), \ INIT_ID(_fields_), \ INIT_ID(_finalizing), \ INIT_ID(_find_and_load), \ diff --git a/Include/internal/pycore_unicodeobject_generated.h b/Include/internal/pycore_unicodeobject_generated.h index f3b064e2a2cb25..64c4cf8c077056 100644 --- a/Include/internal/pycore_unicodeobject_generated.h +++ b/Include/internal/pycore_unicodeobject_generated.h @@ -567,6 +567,9 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) { string = &_Py_ID(_feature_version); assert(_PyUnicode_CheckConsistency(string, 1)); _PyUnicode_InternInPlace(interp, &string); + string = &_Py_ID(_field_types); + assert(_PyUnicode_CheckConsistency(string, 1)); + _PyUnicode_InternInPlace(interp, &string); string = &_Py_ID(_fields_); assert(_PyUnicode_CheckConsistency(string, 1)); _PyUnicode_InternInPlace(interp, &string); diff --git a/Lib/test/test_ast.py b/Lib/test/test_ast.py index d49c149da42592..7cecf319e3638f 100644 --- a/Lib/test/test_ast.py +++ b/Lib/test/test_ast.py @@ -525,17 +525,38 @@ def test_field_attr_existence(self): if name == 'Index': continue if self._is_ast_node(name, item): - x = item() + x = self._construct_ast_class(item) if isinstance(x, ast.AST): self.assertIs(type(x._fields), tuple) + def _construct_ast_class(self, cls): + kwargs = {} + for name, typ in cls.__annotations__.items(): + if typ is str: + kwargs[name] = 'capybara' + elif typ is int: + kwargs[name] = 42 + elif typ is object: + kwargs[name] = b'capybara' + elif isinstance(typ, type) and issubclass(typ, ast.AST): + kwargs[name] = self._construct_ast_class(typ) + return cls(**kwargs) + def test_arguments(self): x = ast.arguments() self.assertEqual(x._fields, ('posonlyargs', 'args', 'vararg', 'kwonlyargs', 'kw_defaults', 'kwarg', 'defaults')) - - with self.assertRaises(AttributeError): - x.args + self.assertEqual(x.__annotations__, { + 'posonlyargs': list[ast.arg], + 'args': list[ast.arg], + 'vararg': ast.arg | None, + 'kwonlyargs': list[ast.arg], + 'kw_defaults': list[ast.expr], + 'kwarg': ast.arg | None, + 'defaults': list[ast.expr], + }) + + self.assertEqual(x.args, []) self.assertIsNone(x.vararg) x = ast.arguments(*range(1, 8)) @@ -551,7 +572,7 @@ def test_field_attr_writable_deprecated(self): self.assertEqual(x._fields, 666) def test_field_attr_writable(self): - x = ast.Constant() + x = ast.Constant(1) # We can assign to _fields x._fields = 666 self.assertEqual(x._fields, 666) @@ -611,15 +632,22 @@ def test_classattrs_deprecated(self): self.assertEqual([str(w.message) for w in wlog], [ 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', + "Constant.__init__ missing 1 required positional argument: 'value'. This will become " + 'an error in Python 3.15.', 'Attribute n is deprecated and will be removed in Python 3.14; use value instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', 'Attribute n is deprecated and will be removed in Python 3.14; use value instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', + "Constant.__init__ missing 1 required positional argument: 'value'. This will become " + 'an error in Python 3.15.', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', 'Attribute n is deprecated and will be removed in Python 3.14; use value instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', + "Constant.__init__ got an unexpected keyword argument 'foo'. Support for " + 'arbitrary keyword arguments is deprecated and will be removed in Python ' + '3.15.', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', 'Attribute n is deprecated and will be removed in Python 3.14; use value instead', 'ast.Num is deprecated and will be removed in Python 3.14; use ast.Constant instead', @@ -636,7 +664,8 @@ def test_classattrs_deprecated(self): ]) def test_classattrs(self): - x = ast.Constant() + with self.assertWarns(DeprecationWarning): + x = ast.Constant() self.assertEqual(x._fields, ('value', 'kind')) with self.assertRaises(AttributeError): @@ -651,7 +680,7 @@ def test_classattrs(self): with self.assertRaises(AttributeError): x.foobar - x = ast.Constant(lineno=2) + x = ast.Constant(lineno=2, value=3) self.assertEqual(x.lineno, 2) x = ast.Constant(42, lineno=0) @@ -662,8 +691,9 @@ def test_classattrs(self): self.assertRaises(TypeError, ast.Constant, 1, None, 2) self.assertRaises(TypeError, ast.Constant, 1, None, 2, lineno=0) - # Arbitrary keyword arguments are supported - self.assertEqual(ast.Constant(1, foo='bar').foo, 'bar') + # Arbitrary keyword arguments are supported (but deprecated) + with self.assertWarns(DeprecationWarning): + self.assertEqual(ast.Constant(1, foo='bar').foo, 'bar') with self.assertRaisesRegex(TypeError, "Constant got multiple values for argument 'value'"): ast.Constant(1, value=2) @@ -815,11 +845,11 @@ def test_isinstance(self): assertBytesDeprecated(self.assertNotIsInstance, Constant('42'), Bytes) assertNameConstantDeprecated(self.assertNotIsInstance, Constant(42), NameConstant) assertEllipsisDeprecated(self.assertNotIsInstance, Constant(42), Ellipsis) - assertNumDeprecated(self.assertNotIsInstance, Constant(), Num) - assertStrDeprecated(self.assertNotIsInstance, Constant(), Str) - assertBytesDeprecated(self.assertNotIsInstance, Constant(), Bytes) - assertNameConstantDeprecated(self.assertNotIsInstance, Constant(), NameConstant) - assertEllipsisDeprecated(self.assertNotIsInstance, Constant(), Ellipsis) + assertNumDeprecated(self.assertNotIsInstance, Constant(None), Num) + assertStrDeprecated(self.assertNotIsInstance, Constant(None), Str) + assertBytesDeprecated(self.assertNotIsInstance, Constant(None), Bytes) + assertNameConstantDeprecated(self.assertNotIsInstance, Constant(1), NameConstant) + assertEllipsisDeprecated(self.assertNotIsInstance, Constant(None), Ellipsis) class S(str): pass with assertStrDeprecated(): @@ -888,8 +918,9 @@ def test_module(self): self.assertEqual(x.body, body) def test_nodeclasses(self): - # Zero arguments constructor explicitly allowed - x = ast.BinOp() + # Zero arguments constructor explicitly allowed (but deprecated) + with self.assertWarns(DeprecationWarning): + x = ast.BinOp() self.assertEqual(x._fields, ('left', 'op', 'right')) # Random attribute allowed too @@ -927,8 +958,9 @@ def test_nodeclasses(self): self.assertEqual(x.right, 3) self.assertEqual(x.lineno, 0) - # Random kwargs also allowed - x = ast.BinOp(1, 2, 3, foobarbaz=42) + # Random kwargs also allowed (but deprecated) + with self.assertWarns(DeprecationWarning): + x = ast.BinOp(1, 2, 3, foobarbaz=42) self.assertEqual(x.foobarbaz, 42) def test_no_fields(self): @@ -941,8 +973,9 @@ def test_pickling(self): for protocol in range(pickle.HIGHEST_PROTOCOL + 1): for ast in (compile(i, "?", "exec", 0x400) for i in exec_tests): - ast2 = pickle.loads(pickle.dumps(ast, protocol)) - self.assertEqual(to_tuple(ast2), to_tuple(ast)) + with self.subTest(ast=ast, protocol=protocol): + ast2 = pickle.loads(pickle.dumps(ast, protocol)) + self.assertEqual(to_tuple(ast2), to_tuple(ast)) def test_invalid_sum(self): pos = dict(lineno=2, col_offset=3) @@ -1310,8 +1343,9 @@ def test_copy_location(self): 'lineno=1, col_offset=4, end_lineno=1, end_col_offset=5), lineno=1, ' 'col_offset=0, end_lineno=1, end_col_offset=5))' ) - src = ast.Call(col_offset=1, lineno=1, end_lineno=1, end_col_offset=1) - new = ast.copy_location(src, ast.Call(col_offset=None, lineno=None)) + func = ast.Name('spam', ast.Load()) + src = ast.Call(col_offset=1, lineno=1, end_lineno=1, end_col_offset=1, func=func) + new = ast.copy_location(src, ast.Call(col_offset=None, lineno=None, func=func)) self.assertIsNone(new.end_lineno) self.assertIsNone(new.end_col_offset) self.assertEqual(new.lineno, 1) @@ -1570,15 +1604,15 @@ def test_level_as_none(self): self.assertIn('sleep', ns) def test_recursion_direct(self): - e = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0) + e = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0, operand=ast.Constant(1)) e.operand = e with self.assertRaises(RecursionError): with support.infinite_recursion(): compile(ast.Expression(e), "<test>", "eval") def test_recursion_indirect(self): - e = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0) - f = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0) + e = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0, operand=ast.Constant(1)) + f = ast.UnaryOp(op=ast.Not(), lineno=0, col_offset=0, operand=ast.Constant(1)) e.operand = f f.operand = e with self.assertRaises(RecursionError): @@ -2866,6 +2900,23 @@ def visit_Call(self, node: ast.Call): self.assertASTTransformation(PrintToLog, code, expected) +class ASTConstructorTests(unittest.TestCase): + """Test the autogenerated constructors for AST nodes.""" + + def test_FunctionDef(self): + args = ast.arguments() + self.assertEqual(args.args, []) + self.assertEqual(args.posonlyargs, []) + with self.assertWarnsRegex(DeprecationWarning, + r"FunctionDef\.__init__ missing 1 required positional argument: 'name'"): + node = ast.FunctionDef(args=args) + self.assertFalse(hasattr(node, "name")) + self.assertEqual(node.decorator_list, []) + node = ast.FunctionDef(name='foo', args=args) + self.assertEqual(node.name, 'foo') + self.assertEqual(node.decorator_list, []) + + @support.cpython_only class ModuleStateTests(unittest.TestCase): # bpo-41194, bpo-41261, bpo-41631: The _ast module uses a global state. diff --git a/Misc/NEWS.d/next/Core and Builtins/2023-06-16-21-29-06.gh-issue-105858.Q7h0EV.rst b/Misc/NEWS.d/next/Core and Builtins/2023-06-16-21-29-06.gh-issue-105858.Q7h0EV.rst new file mode 100644 index 00000000000000..9338f662374c9a --- /dev/null +++ b/Misc/NEWS.d/next/Core and Builtins/2023-06-16-21-29-06.gh-issue-105858.Q7h0EV.rst @@ -0,0 +1,8 @@ +Improve the constructors for :mod:`ast` nodes. Arguments of list types now +default to an empty list if omitted, and optional fields default to ``None``. +AST nodes now have an +``__annotations__`` attribute with the expected types of their attributes. +Passing unrecognized extra arguments to AST nodes is deprecated and will +become an error in Python 3.15. Omitting a required argument to an AST node +is deprecated and will become an error in Python 3.15. Patch by Jelle +Zijlstra. diff --git a/Parser/asdl_c.py b/Parser/asdl_c.py index ce92672bf00776..865fd76acf697d 100755 --- a/Parser/asdl_c.py +++ b/Parser/asdl_c.py @@ -15,6 +15,13 @@ MAX_COL = 80 AUTOGEN_MESSAGE = "// File automatically generated by {}.\n\n" +builtin_type_to_c_type = { + "identifier": "PyUnicode_Type", + "string": "PyUnicode_Type", + "int": "PyLong_Type", + "constant": "PyBaseObject_Type", +} + def get_c_type(name): """Return a string for the C name of the type. @@ -764,6 +771,67 @@ def visitConstructor(self, cons, name): self.emit("};",0) +class AnnotationsVisitor(PickleVisitor): + def visitModule(self, mod): + self.file.write(textwrap.dedent(''' + static int + add_ast_annotations(struct ast_state *state) + { + bool cond; + ''')) + for dfn in mod.dfns: + self.visit(dfn) + self.file.write(textwrap.dedent(''' + return 1; + } + ''')) + + def visitProduct(self, prod, name): + self.emit_annotations(name, prod.fields) + + def visitSum(self, sum, name): + for t in sum.types: + self.visitConstructor(t, name) + + def visitConstructor(self, cons, name): + self.emit_annotations(cons.name, cons.fields) + + def emit_annotations(self, name, fields): + self.emit(f"PyObject *{name}_annotations = PyDict_New();", 1) + self.emit(f"if (!{name}_annotations) return 0;", 1) + for field in fields: + self.emit("{", 1) + if field.type in builtin_type_to_c_type: + self.emit(f"PyObject *type = (PyObject *)&{builtin_type_to_c_type[field.type]};", 2) + else: + self.emit(f"PyObject *type = state->{field.type}_type;", 2) + if field.opt: + self.emit("type = _Py_union_type_or(type, Py_None);", 2) + self.emit("cond = type != NULL;", 2) + self.emit_annotations_error(name, 2) + elif field.seq: + self.emit("type = Py_GenericAlias((PyObject *)&PyList_Type, type);", 2) + self.emit("cond = type != NULL;", 2) + self.emit_annotations_error(name, 2) + else: + self.emit("Py_INCREF(type);", 2) + self.emit(f"cond = PyDict_SetItemString({name}_annotations, \"{field.name}\", type) == 0;", 2) + self.emit("Py_DECREF(type);", 2) + self.emit_annotations_error(name, 2) + self.emit("}", 1) + self.emit(f'cond = PyObject_SetAttrString(state->{name}_type, "_field_types", {name}_annotations) == 0;', 1) + self.emit_annotations_error(name, 1) + self.emit(f'cond = PyObject_SetAttrString(state->{name}_type, "__annotations__", {name}_annotations) == 0;', 1) + self.emit_annotations_error(name, 1) + self.emit(f"Py_DECREF({name}_annotations);", 1) + + def emit_annotations_error(self, name, depth): + self.emit("if (!cond) {", depth) + self.emit(f"Py_DECREF({name}_annotations);", depth + 1) + self.emit("return 0;", depth + 1) + self.emit("}", depth) + + class PyTypesVisitor(PickleVisitor): def visitModule(self, mod): @@ -812,7 +880,7 @@ def visitModule(self, mod): Py_ssize_t i, numfields = 0; int res = -1; - PyObject *key, *value, *fields; + PyObject *key, *value, *fields, *remaining_fields = NULL; if (PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), state->_fields, &fields) < 0) { goto cleanup; } @@ -821,6 +889,13 @@ def visitModule(self, mod): if (numfields == -1) { goto cleanup; } + remaining_fields = PySet_New(fields); + } + else { + remaining_fields = PySet_New(NULL); + } + if (remaining_fields == NULL) { + goto cleanup; } res = 0; /* if no error occurs, this stays 0 to the end */ @@ -840,6 +915,11 @@ def visitModule(self, mod): goto cleanup; } res = PyObject_SetAttr(self, name, PyTuple_GET_ITEM(args, i)); + if (PySet_Discard(remaining_fields, name) < 0) { + res = -1; + Py_DECREF(name); + goto cleanup; + } Py_DECREF(name); if (res < 0) { goto cleanup; @@ -852,13 +932,14 @@ def visitModule(self, mod): if (contains == -1) { res = -1; goto cleanup; - } else if (contains == 1) { - Py_ssize_t p = PySequence_Index(fields, key); + } + else if (contains == 1) { + int p = PySet_Discard(remaining_fields, key); if (p == -1) { res = -1; goto cleanup; } - if (p < PyTuple_GET_SIZE(args)) { + if (p == 0) { PyErr_Format(PyExc_TypeError, "%.400s got multiple values for argument '%U'", Py_TYPE(self)->tp_name, key); @@ -866,15 +947,91 @@ def visitModule(self, mod): goto cleanup; } } + else if ( + PyUnicode_CompareWithASCIIString(key, "lineno") != 0 && + PyUnicode_CompareWithASCIIString(key, "col_offset") != 0 && + PyUnicode_CompareWithASCIIString(key, "end_lineno") != 0 && + PyUnicode_CompareWithASCIIString(key, "end_col_offset") != 0 + ) { + if (PyErr_WarnFormat( + PyExc_DeprecationWarning, 1, + "%.400s.__init__ got an unexpected keyword argument '%U'. " + "Support for arbitrary keyword arguments is deprecated " + "and will be removed in Python 3.15.", + Py_TYPE(self)->tp_name, key + ) < 0) { + res = -1; + goto cleanup; + } + } res = PyObject_SetAttr(self, key, value); if (res < 0) { goto cleanup; } } } + Py_ssize_t size = PySet_Size(remaining_fields); + PyObject *field_types = NULL, *remaining_list = NULL; + if (size > 0) { + if (!PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), &_Py_ID(_field_types), + &field_types)) { + res = -1; + goto cleanup; + } + remaining_list = PySequence_List(remaining_fields); + if (!remaining_list) { + goto set_remaining_cleanup; + } + for (Py_ssize_t i = 0; i < size; i++) { + PyObject *name = PyList_GET_ITEM(remaining_list, i); + PyObject *type = PyDict_GetItemWithError(field_types, name); + if (!type) { + if (!PyErr_Occurred()) { + PyErr_SetObject(PyExc_KeyError, name); + } + goto set_remaining_cleanup; + } + if (_PyUnion_Check(type)) { + // optional field + // do nothing, we'll have set a None default on the class + } + else if (Py_IS_TYPE(type, &Py_GenericAliasType)) { + // list field + PyObject *empty = PyList_New(0); + if (!empty) { + goto set_remaining_cleanup; + } + res = PyObject_SetAttr(self, name, empty); + Py_DECREF(empty); + if (res < 0) { + goto set_remaining_cleanup; + } + } + else { + // simple field (e.g., identifier) + if (PyErr_WarnFormat( + PyExc_DeprecationWarning, 1, + "%.400s.__init__ missing 1 required positional argument: '%U'. " + "This will become an error in Python 3.15.", + Py_TYPE(self)->tp_name, name + ) < 0) { + res = -1; + goto cleanup; + } + } + } + Py_DECREF(remaining_list); + Py_DECREF(field_types); + } cleanup: Py_XDECREF(fields); + Py_XDECREF(remaining_fields); return res; + set_remaining_cleanup: + Py_XDECREF(remaining_list); + Py_XDECREF(field_types); + res = -1; + goto cleanup; } /* Pickling support */ @@ -886,14 +1043,75 @@ def visitModule(self, mod): return NULL; } - PyObject *dict; + PyObject *dict = NULL, *fields = NULL, *remaining_fields = NULL, + *remaining_dict = NULL, *positional_args = NULL; if (PyObject_GetOptionalAttr(self, state->__dict__, &dict) < 0) { return NULL; } + PyObject *result = NULL; if (dict) { - return Py_BuildValue("O()N", Py_TYPE(self), dict); + // Serialize the fields as positional args if possible, because if we + // serialize them as a dict, during unpickling they are set only *after* + // the object is constructed, which will now trigger a DeprecationWarning + // if the AST type has required fields. + if (PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), state->_fields, &fields) < 0) { + goto cleanup; + } + if (fields) { + Py_ssize_t numfields = PySequence_Size(fields); + if (numfields == -1) { + Py_DECREF(dict); + goto cleanup; + } + remaining_dict = PyDict_Copy(dict); + Py_DECREF(dict); + if (!remaining_dict) { + goto cleanup; + } + positional_args = PyList_New(0); + if (!positional_args) { + goto cleanup; + } + for (Py_ssize_t i = 0; i < numfields; i++) { + PyObject *name = PySequence_GetItem(fields, i); + if (!name) { + goto cleanup; + } + PyObject *value = PyDict_GetItemWithError(remaining_dict, name); + if (!value) { + if (PyErr_Occurred()) { + goto cleanup; + } + break; + } + if (PyList_Append(positional_args, value) < 0) { + goto cleanup; + } + if (PyDict_DelItem(remaining_dict, name) < 0) { + goto cleanup; + } + Py_DECREF(name); + } + PyObject *args_tuple = PyList_AsTuple(positional_args); + if (!args_tuple) { + goto cleanup; + } + result = Py_BuildValue("ONO", Py_TYPE(self), args_tuple, + remaining_dict); + } + else { + result = Py_BuildValue("O()N", Py_TYPE(self), dict); + } + } + else { + result = Py_BuildValue("O()", Py_TYPE(self)); } - return Py_BuildValue("O()", Py_TYPE(self)); +cleanup: + Py_XDECREF(fields); + Py_XDECREF(remaining_fields); + Py_XDECREF(remaining_dict); + Py_XDECREF(positional_args); + return result; } static PyMemberDef ast_type_members[] = { @@ -1117,6 +1335,9 @@ def visitModule(self, mod): for dfn in mod.dfns: self.visit(dfn) self.file.write(textwrap.dedent(''' + if (!add_ast_annotations(state)) { + return -1; + } return 0; } ''')) @@ -1534,6 +1755,8 @@ def generate_module_def(mod, metadata, f, internal_h): #include "pycore_lock.h" // _PyOnceFlag #include "pycore_interp.h" // _PyInterpreterState.ast #include "pycore_pystate.h" // _PyInterpreterState_GET() + #include "pycore_unionobject.h" // _Py_union_type_or + #include "structmember.h" #include <stddef.h> struct validator { @@ -1651,6 +1874,7 @@ def write_source(mod, metadata, f, internal_h_file): v = ChainOfVisitors( SequenceConstructorVisitor(f), PyTypesDeclareVisitor(f), + AnnotationsVisitor(f), PyTypesVisitor(f), Obj2ModPrototypeVisitor(f), FunctionVisitor(f), diff --git a/Python/Python-ast.c b/Python/Python-ast.c index d77e986ba067a3..46387493214829 100644 --- a/Python/Python-ast.c +++ b/Python/Python-ast.c @@ -7,6 +7,8 @@ #include "pycore_lock.h" // _PyOnceFlag #include "pycore_interp.h" // _PyInterpreterState.ast #include "pycore_pystate.h" // _PyInterpreterState_GET() +#include "pycore_unionobject.h" // _Py_union_type_or +#include "structmember.h" #include <stddef.h> struct validator { @@ -816,6 +818,4170 @@ static const char * const TypeVarTuple_fields[]={ }; +static int +add_ast_annotations(struct ast_state *state) +{ + bool cond; + PyObject *Module_annotations = PyDict_New(); + if (!Module_annotations) return 0; + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + cond = PyDict_SetItemString(Module_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + } + { + PyObject *type = state->type_ignore_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + cond = PyDict_SetItemString(Module_annotations, "type_ignores", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Module_type, "_field_types", + Module_annotations) == 0; + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Module_type, "__annotations__", + Module_annotations) == 0; + if (!cond) { + Py_DECREF(Module_annotations); + return 0; + } + Py_DECREF(Module_annotations); + PyObject *Interactive_annotations = PyDict_New(); + if (!Interactive_annotations) return 0; + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Interactive_annotations); + return 0; + } + cond = PyDict_SetItemString(Interactive_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Interactive_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Interactive_type, "_field_types", + Interactive_annotations) == 0; + if (!cond) { + Py_DECREF(Interactive_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Interactive_type, "__annotations__", + Interactive_annotations) == 0; + if (!cond) { + Py_DECREF(Interactive_annotations); + return 0; + } + Py_DECREF(Interactive_annotations); + PyObject *Expression_annotations = PyDict_New(); + if (!Expression_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Expression_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Expression_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Expression_type, "_field_types", + Expression_annotations) == 0; + if (!cond) { + Py_DECREF(Expression_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Expression_type, "__annotations__", + Expression_annotations) == 0; + if (!cond) { + Py_DECREF(Expression_annotations); + return 0; + } + Py_DECREF(Expression_annotations); + PyObject *FunctionType_annotations = PyDict_New(); + if (!FunctionType_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionType_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionType_annotations, "argtypes", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionType_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(FunctionType_annotations, "returns", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionType_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->FunctionType_type, "_field_types", + FunctionType_annotations) == 0; + if (!cond) { + Py_DECREF(FunctionType_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->FunctionType_type, "__annotations__", + FunctionType_annotations) == 0; + if (!cond) { + Py_DECREF(FunctionType_annotations); + return 0; + } + Py_DECREF(FunctionType_annotations); + PyObject *FunctionDef_annotations = PyDict_New(); + if (!FunctionDef_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(FunctionDef_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->arguments_type; + Py_INCREF(type); + cond = PyDict_SetItemString(FunctionDef_annotations, "args", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionDef_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionDef_annotations, "decorator_list", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionDef_annotations, "returns", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionDef_annotations, "type_comment", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->type_param_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(FunctionDef_annotations, "type_params", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->FunctionDef_type, "_field_types", + FunctionDef_annotations) == 0; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->FunctionDef_type, "__annotations__", + FunctionDef_annotations) == 0; + if (!cond) { + Py_DECREF(FunctionDef_annotations); + return 0; + } + Py_DECREF(FunctionDef_annotations); + PyObject *AsyncFunctionDef_annotations = PyDict_New(); + if (!AsyncFunctionDef_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, "name", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->arguments_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, "args", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, "body", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, + "decorator_list", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, "returns", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, + "type_comment", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + { + PyObject *type = state->type_param_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFunctionDef_annotations, + "type_params", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->AsyncFunctionDef_type, "_field_types", + AsyncFunctionDef_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->AsyncFunctionDef_type, + "__annotations__", + AsyncFunctionDef_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncFunctionDef_annotations); + return 0; + } + Py_DECREF(AsyncFunctionDef_annotations); + PyObject *ClassDef_annotations = PyDict_New(); + if (!ClassDef_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(ClassDef_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyDict_SetItemString(ClassDef_annotations, "bases", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + { + PyObject *type = state->keyword_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyDict_SetItemString(ClassDef_annotations, "keywords", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyDict_SetItemString(ClassDef_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyDict_SetItemString(ClassDef_annotations, "decorator_list", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + { + PyObject *type = state->type_param_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyDict_SetItemString(ClassDef_annotations, "type_params", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->ClassDef_type, "_field_types", + ClassDef_annotations) == 0; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->ClassDef_type, "__annotations__", + ClassDef_annotations) == 0; + if (!cond) { + Py_DECREF(ClassDef_annotations); + return 0; + } + Py_DECREF(ClassDef_annotations); + PyObject *Return_annotations = PyDict_New(); + if (!Return_annotations) return 0; + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Return_annotations); + return 0; + } + cond = PyDict_SetItemString(Return_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Return_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Return_type, "_field_types", + Return_annotations) == 0; + if (!cond) { + Py_DECREF(Return_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Return_type, "__annotations__", + Return_annotations) == 0; + if (!cond) { + Py_DECREF(Return_annotations); + return 0; + } + Py_DECREF(Return_annotations); + PyObject *Delete_annotations = PyDict_New(); + if (!Delete_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Delete_annotations); + return 0; + } + cond = PyDict_SetItemString(Delete_annotations, "targets", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Delete_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Delete_type, "_field_types", + Delete_annotations) == 0; + if (!cond) { + Py_DECREF(Delete_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Delete_type, "__annotations__", + Delete_annotations) == 0; + if (!cond) { + Py_DECREF(Delete_annotations); + return 0; + } + Py_DECREF(Delete_annotations); + PyObject *Assign_annotations = PyDict_New(); + if (!Assign_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + cond = PyDict_SetItemString(Assign_annotations, "targets", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Assign_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + cond = PyDict_SetItemString(Assign_annotations, "type_comment", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Assign_type, "_field_types", + Assign_annotations) == 0; + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Assign_type, "__annotations__", + Assign_annotations) == 0; + if (!cond) { + Py_DECREF(Assign_annotations); + return 0; + } + Py_DECREF(Assign_annotations); + PyObject *TypeAlias_annotations = PyDict_New(); + if (!TypeAlias_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeAlias_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + } + { + PyObject *type = state->type_param_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + cond = PyDict_SetItemString(TypeAlias_annotations, "type_params", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeAlias_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->TypeAlias_type, "_field_types", + TypeAlias_annotations) == 0; + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->TypeAlias_type, "__annotations__", + TypeAlias_annotations) == 0; + if (!cond) { + Py_DECREF(TypeAlias_annotations); + return 0; + } + Py_DECREF(TypeAlias_annotations); + PyObject *AugAssign_annotations = PyDict_New(); + if (!AugAssign_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AugAssign_annotations, "target", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AugAssign_annotations); + return 0; + } + } + { + PyObject *type = state->operator_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AugAssign_annotations, "op", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AugAssign_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AugAssign_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AugAssign_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->AugAssign_type, "_field_types", + AugAssign_annotations) == 0; + if (!cond) { + Py_DECREF(AugAssign_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->AugAssign_type, "__annotations__", + AugAssign_annotations) == 0; + if (!cond) { + Py_DECREF(AugAssign_annotations); + return 0; + } + Py_DECREF(AugAssign_annotations); + PyObject *AnnAssign_annotations = PyDict_New(); + if (!AnnAssign_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AnnAssign_annotations, "target", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AnnAssign_annotations, "annotation", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + cond = PyDict_SetItemString(AnnAssign_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyLong_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(AnnAssign_annotations, "simple", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->AnnAssign_type, "_field_types", + AnnAssign_annotations) == 0; + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->AnnAssign_type, "__annotations__", + AnnAssign_annotations) == 0; + if (!cond) { + Py_DECREF(AnnAssign_annotations); + return 0; + } + Py_DECREF(AnnAssign_annotations); + PyObject *For_annotations = PyDict_New(); + if (!For_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(For_annotations, "target", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(For_annotations, "iter", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + cond = PyDict_SetItemString(For_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + cond = PyDict_SetItemString(For_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + cond = PyDict_SetItemString(For_annotations, "type_comment", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->For_type, "_field_types", + For_annotations) == 0; + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->For_type, "__annotations__", + For_annotations) == 0; + if (!cond) { + Py_DECREF(For_annotations); + return 0; + } + Py_DECREF(For_annotations); + PyObject *AsyncFor_annotations = PyDict_New(); + if (!AsyncFor_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AsyncFor_annotations, "target", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(AsyncFor_annotations, "iter", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFor_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFor_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncFor_annotations, "type_comment", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->AsyncFor_type, "_field_types", + AsyncFor_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->AsyncFor_type, "__annotations__", + AsyncFor_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncFor_annotations); + return 0; + } + Py_DECREF(AsyncFor_annotations); + PyObject *While_annotations = PyDict_New(); + if (!While_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(While_annotations, "test", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + cond = PyDict_SetItemString(While_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + cond = PyDict_SetItemString(While_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->While_type, "_field_types", + While_annotations) == 0; + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->While_type, "__annotations__", + While_annotations) == 0; + if (!cond) { + Py_DECREF(While_annotations); + return 0; + } + Py_DECREF(While_annotations); + PyObject *If_annotations = PyDict_New(); + if (!If_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(If_annotations, "test", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + cond = PyDict_SetItemString(If_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + cond = PyDict_SetItemString(If_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->If_type, "_field_types", + If_annotations) == 0; + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->If_type, "__annotations__", + If_annotations) == 0; + if (!cond) { + Py_DECREF(If_annotations); + return 0; + } + Py_DECREF(If_annotations); + PyObject *With_annotations = PyDict_New(); + if (!With_annotations) return 0; + { + PyObject *type = state->withitem_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + cond = PyDict_SetItemString(With_annotations, "items", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + cond = PyDict_SetItemString(With_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + cond = PyDict_SetItemString(With_annotations, "type_comment", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->With_type, "_field_types", + With_annotations) == 0; + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->With_type, "__annotations__", + With_annotations) == 0; + if (!cond) { + Py_DECREF(With_annotations); + return 0; + } + Py_DECREF(With_annotations); + PyObject *AsyncWith_annotations = PyDict_New(); + if (!AsyncWith_annotations) return 0; + { + PyObject *type = state->withitem_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncWith_annotations, "items", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncWith_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + cond = PyDict_SetItemString(AsyncWith_annotations, "type_comment", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->AsyncWith_type, "_field_types", + AsyncWith_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->AsyncWith_type, "__annotations__", + AsyncWith_annotations) == 0; + if (!cond) { + Py_DECREF(AsyncWith_annotations); + return 0; + } + Py_DECREF(AsyncWith_annotations); + PyObject *Match_annotations = PyDict_New(); + if (!Match_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Match_annotations, "subject", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Match_annotations); + return 0; + } + } + { + PyObject *type = state->match_case_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Match_annotations); + return 0; + } + cond = PyDict_SetItemString(Match_annotations, "cases", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Match_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Match_type, "_field_types", + Match_annotations) == 0; + if (!cond) { + Py_DECREF(Match_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Match_type, "__annotations__", + Match_annotations) == 0; + if (!cond) { + Py_DECREF(Match_annotations); + return 0; + } + Py_DECREF(Match_annotations); + PyObject *Raise_annotations = PyDict_New(); + if (!Raise_annotations) return 0; + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + cond = PyDict_SetItemString(Raise_annotations, "exc", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + cond = PyDict_SetItemString(Raise_annotations, "cause", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Raise_type, "_field_types", + Raise_annotations) == 0; + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Raise_type, "__annotations__", + Raise_annotations) == 0; + if (!cond) { + Py_DECREF(Raise_annotations); + return 0; + } + Py_DECREF(Raise_annotations); + PyObject *Try_annotations = PyDict_New(); + if (!Try_annotations) return 0; + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + cond = PyDict_SetItemString(Try_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + } + { + PyObject *type = state->excepthandler_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + cond = PyDict_SetItemString(Try_annotations, "handlers", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + cond = PyDict_SetItemString(Try_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + cond = PyDict_SetItemString(Try_annotations, "finalbody", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Try_type, "_field_types", + Try_annotations) == 0; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Try_type, "__annotations__", + Try_annotations) == 0; + if (!cond) { + Py_DECREF(Try_annotations); + return 0; + } + Py_DECREF(Try_annotations); + PyObject *TryStar_annotations = PyDict_New(); + if (!TryStar_annotations) return 0; + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + cond = PyDict_SetItemString(TryStar_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + } + { + PyObject *type = state->excepthandler_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + cond = PyDict_SetItemString(TryStar_annotations, "handlers", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + cond = PyDict_SetItemString(TryStar_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + cond = PyDict_SetItemString(TryStar_annotations, "finalbody", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->TryStar_type, "_field_types", + TryStar_annotations) == 0; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->TryStar_type, "__annotations__", + TryStar_annotations) == 0; + if (!cond) { + Py_DECREF(TryStar_annotations); + return 0; + } + Py_DECREF(TryStar_annotations); + PyObject *Assert_annotations = PyDict_New(); + if (!Assert_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Assert_annotations, "test", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Assert_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Assert_annotations); + return 0; + } + cond = PyDict_SetItemString(Assert_annotations, "msg", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Assert_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Assert_type, "_field_types", + Assert_annotations) == 0; + if (!cond) { + Py_DECREF(Assert_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Assert_type, "__annotations__", + Assert_annotations) == 0; + if (!cond) { + Py_DECREF(Assert_annotations); + return 0; + } + Py_DECREF(Assert_annotations); + PyObject *Import_annotations = PyDict_New(); + if (!Import_annotations) return 0; + { + PyObject *type = state->alias_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Import_annotations); + return 0; + } + cond = PyDict_SetItemString(Import_annotations, "names", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Import_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Import_type, "_field_types", + Import_annotations) == 0; + if (!cond) { + Py_DECREF(Import_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Import_type, "__annotations__", + Import_annotations) == 0; + if (!cond) { + Py_DECREF(Import_annotations); + return 0; + } + Py_DECREF(Import_annotations); + PyObject *ImportFrom_annotations = PyDict_New(); + if (!ImportFrom_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + cond = PyDict_SetItemString(ImportFrom_annotations, "module", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + } + { + PyObject *type = state->alias_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + cond = PyDict_SetItemString(ImportFrom_annotations, "names", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyLong_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + cond = PyDict_SetItemString(ImportFrom_annotations, "level", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->ImportFrom_type, "_field_types", + ImportFrom_annotations) == 0; + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->ImportFrom_type, "__annotations__", + ImportFrom_annotations) == 0; + if (!cond) { + Py_DECREF(ImportFrom_annotations); + return 0; + } + Py_DECREF(ImportFrom_annotations); + PyObject *Global_annotations = PyDict_New(); + if (!Global_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Global_annotations); + return 0; + } + cond = PyDict_SetItemString(Global_annotations, "names", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Global_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Global_type, "_field_types", + Global_annotations) == 0; + if (!cond) { + Py_DECREF(Global_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Global_type, "__annotations__", + Global_annotations) == 0; + if (!cond) { + Py_DECREF(Global_annotations); + return 0; + } + Py_DECREF(Global_annotations); + PyObject *Nonlocal_annotations = PyDict_New(); + if (!Nonlocal_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Nonlocal_annotations); + return 0; + } + cond = PyDict_SetItemString(Nonlocal_annotations, "names", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Nonlocal_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Nonlocal_type, "_field_types", + Nonlocal_annotations) == 0; + if (!cond) { + Py_DECREF(Nonlocal_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Nonlocal_type, "__annotations__", + Nonlocal_annotations) == 0; + if (!cond) { + Py_DECREF(Nonlocal_annotations); + return 0; + } + Py_DECREF(Nonlocal_annotations); + PyObject *Expr_annotations = PyDict_New(); + if (!Expr_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Expr_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Expr_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Expr_type, "_field_types", + Expr_annotations) == 0; + if (!cond) { + Py_DECREF(Expr_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Expr_type, "__annotations__", + Expr_annotations) == 0; + if (!cond) { + Py_DECREF(Expr_annotations); + return 0; + } + Py_DECREF(Expr_annotations); + PyObject *Pass_annotations = PyDict_New(); + if (!Pass_annotations) return 0; + cond = PyObject_SetAttrString(state->Pass_type, "_field_types", + Pass_annotations) == 0; + if (!cond) { + Py_DECREF(Pass_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Pass_type, "__annotations__", + Pass_annotations) == 0; + if (!cond) { + Py_DECREF(Pass_annotations); + return 0; + } + Py_DECREF(Pass_annotations); + PyObject *Break_annotations = PyDict_New(); + if (!Break_annotations) return 0; + cond = PyObject_SetAttrString(state->Break_type, "_field_types", + Break_annotations) == 0; + if (!cond) { + Py_DECREF(Break_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Break_type, "__annotations__", + Break_annotations) == 0; + if (!cond) { + Py_DECREF(Break_annotations); + return 0; + } + Py_DECREF(Break_annotations); + PyObject *Continue_annotations = PyDict_New(); + if (!Continue_annotations) return 0; + cond = PyObject_SetAttrString(state->Continue_type, "_field_types", + Continue_annotations) == 0; + if (!cond) { + Py_DECREF(Continue_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Continue_type, "__annotations__", + Continue_annotations) == 0; + if (!cond) { + Py_DECREF(Continue_annotations); + return 0; + } + Py_DECREF(Continue_annotations); + PyObject *BoolOp_annotations = PyDict_New(); + if (!BoolOp_annotations) return 0; + { + PyObject *type = state->boolop_type; + Py_INCREF(type); + cond = PyDict_SetItemString(BoolOp_annotations, "op", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(BoolOp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(BoolOp_annotations); + return 0; + } + cond = PyDict_SetItemString(BoolOp_annotations, "values", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(BoolOp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->BoolOp_type, "_field_types", + BoolOp_annotations) == 0; + if (!cond) { + Py_DECREF(BoolOp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->BoolOp_type, "__annotations__", + BoolOp_annotations) == 0; + if (!cond) { + Py_DECREF(BoolOp_annotations); + return 0; + } + Py_DECREF(BoolOp_annotations); + PyObject *NamedExpr_annotations = PyDict_New(); + if (!NamedExpr_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(NamedExpr_annotations, "target", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(NamedExpr_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(NamedExpr_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(NamedExpr_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->NamedExpr_type, "_field_types", + NamedExpr_annotations) == 0; + if (!cond) { + Py_DECREF(NamedExpr_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->NamedExpr_type, "__annotations__", + NamedExpr_annotations) == 0; + if (!cond) { + Py_DECREF(NamedExpr_annotations); + return 0; + } + Py_DECREF(NamedExpr_annotations); + PyObject *BinOp_annotations = PyDict_New(); + if (!BinOp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(BinOp_annotations, "left", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(BinOp_annotations); + return 0; + } + } + { + PyObject *type = state->operator_type; + Py_INCREF(type); + cond = PyDict_SetItemString(BinOp_annotations, "op", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(BinOp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(BinOp_annotations, "right", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(BinOp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->BinOp_type, "_field_types", + BinOp_annotations) == 0; + if (!cond) { + Py_DECREF(BinOp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->BinOp_type, "__annotations__", + BinOp_annotations) == 0; + if (!cond) { + Py_DECREF(BinOp_annotations); + return 0; + } + Py_DECREF(BinOp_annotations); + PyObject *UnaryOp_annotations = PyDict_New(); + if (!UnaryOp_annotations) return 0; + { + PyObject *type = state->unaryop_type; + Py_INCREF(type); + cond = PyDict_SetItemString(UnaryOp_annotations, "op", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(UnaryOp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(UnaryOp_annotations, "operand", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(UnaryOp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->UnaryOp_type, "_field_types", + UnaryOp_annotations) == 0; + if (!cond) { + Py_DECREF(UnaryOp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->UnaryOp_type, "__annotations__", + UnaryOp_annotations) == 0; + if (!cond) { + Py_DECREF(UnaryOp_annotations); + return 0; + } + Py_DECREF(UnaryOp_annotations); + PyObject *Lambda_annotations = PyDict_New(); + if (!Lambda_annotations) return 0; + { + PyObject *type = state->arguments_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Lambda_annotations, "args", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Lambda_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Lambda_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Lambda_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Lambda_type, "_field_types", + Lambda_annotations) == 0; + if (!cond) { + Py_DECREF(Lambda_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Lambda_type, "__annotations__", + Lambda_annotations) == 0; + if (!cond) { + Py_DECREF(Lambda_annotations); + return 0; + } + Py_DECREF(Lambda_annotations); + PyObject *IfExp_annotations = PyDict_New(); + if (!IfExp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(IfExp_annotations, "test", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(IfExp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(IfExp_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(IfExp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(IfExp_annotations, "orelse", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(IfExp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->IfExp_type, "_field_types", + IfExp_annotations) == 0; + if (!cond) { + Py_DECREF(IfExp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->IfExp_type, "__annotations__", + IfExp_annotations) == 0; + if (!cond) { + Py_DECREF(IfExp_annotations); + return 0; + } + Py_DECREF(IfExp_annotations); + PyObject *Dict_annotations = PyDict_New(); + if (!Dict_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + cond = PyDict_SetItemString(Dict_annotations, "keys", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + cond = PyDict_SetItemString(Dict_annotations, "values", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Dict_type, "_field_types", + Dict_annotations) == 0; + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Dict_type, "__annotations__", + Dict_annotations) == 0; + if (!cond) { + Py_DECREF(Dict_annotations); + return 0; + } + Py_DECREF(Dict_annotations); + PyObject *Set_annotations = PyDict_New(); + if (!Set_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Set_annotations); + return 0; + } + cond = PyDict_SetItemString(Set_annotations, "elts", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Set_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Set_type, "_field_types", + Set_annotations) == 0; + if (!cond) { + Py_DECREF(Set_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Set_type, "__annotations__", + Set_annotations) == 0; + if (!cond) { + Py_DECREF(Set_annotations); + return 0; + } + Py_DECREF(Set_annotations); + PyObject *ListComp_annotations = PyDict_New(); + if (!ListComp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(ListComp_annotations, "elt", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ListComp_annotations); + return 0; + } + } + { + PyObject *type = state->comprehension_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ListComp_annotations); + return 0; + } + cond = PyDict_SetItemString(ListComp_annotations, "generators", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ListComp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->ListComp_type, "_field_types", + ListComp_annotations) == 0; + if (!cond) { + Py_DECREF(ListComp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->ListComp_type, "__annotations__", + ListComp_annotations) == 0; + if (!cond) { + Py_DECREF(ListComp_annotations); + return 0; + } + Py_DECREF(ListComp_annotations); + PyObject *SetComp_annotations = PyDict_New(); + if (!SetComp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(SetComp_annotations, "elt", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(SetComp_annotations); + return 0; + } + } + { + PyObject *type = state->comprehension_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(SetComp_annotations); + return 0; + } + cond = PyDict_SetItemString(SetComp_annotations, "generators", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(SetComp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->SetComp_type, "_field_types", + SetComp_annotations) == 0; + if (!cond) { + Py_DECREF(SetComp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->SetComp_type, "__annotations__", + SetComp_annotations) == 0; + if (!cond) { + Py_DECREF(SetComp_annotations); + return 0; + } + Py_DECREF(SetComp_annotations); + PyObject *DictComp_annotations = PyDict_New(); + if (!DictComp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(DictComp_annotations, "key", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(DictComp_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + } + { + PyObject *type = state->comprehension_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + cond = PyDict_SetItemString(DictComp_annotations, "generators", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->DictComp_type, "_field_types", + DictComp_annotations) == 0; + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->DictComp_type, "__annotations__", + DictComp_annotations) == 0; + if (!cond) { + Py_DECREF(DictComp_annotations); + return 0; + } + Py_DECREF(DictComp_annotations); + PyObject *GeneratorExp_annotations = PyDict_New(); + if (!GeneratorExp_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(GeneratorExp_annotations, "elt", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(GeneratorExp_annotations); + return 0; + } + } + { + PyObject *type = state->comprehension_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(GeneratorExp_annotations); + return 0; + } + cond = PyDict_SetItemString(GeneratorExp_annotations, "generators", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(GeneratorExp_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->GeneratorExp_type, "_field_types", + GeneratorExp_annotations) == 0; + if (!cond) { + Py_DECREF(GeneratorExp_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->GeneratorExp_type, "__annotations__", + GeneratorExp_annotations) == 0; + if (!cond) { + Py_DECREF(GeneratorExp_annotations); + return 0; + } + Py_DECREF(GeneratorExp_annotations); + PyObject *Await_annotations = PyDict_New(); + if (!Await_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Await_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Await_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Await_type, "_field_types", + Await_annotations) == 0; + if (!cond) { + Py_DECREF(Await_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Await_type, "__annotations__", + Await_annotations) == 0; + if (!cond) { + Py_DECREF(Await_annotations); + return 0; + } + Py_DECREF(Await_annotations); + PyObject *Yield_annotations = PyDict_New(); + if (!Yield_annotations) return 0; + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Yield_annotations); + return 0; + } + cond = PyDict_SetItemString(Yield_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Yield_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Yield_type, "_field_types", + Yield_annotations) == 0; + if (!cond) { + Py_DECREF(Yield_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Yield_type, "__annotations__", + Yield_annotations) == 0; + if (!cond) { + Py_DECREF(Yield_annotations); + return 0; + } + Py_DECREF(Yield_annotations); + PyObject *YieldFrom_annotations = PyDict_New(); + if (!YieldFrom_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(YieldFrom_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(YieldFrom_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->YieldFrom_type, "_field_types", + YieldFrom_annotations) == 0; + if (!cond) { + Py_DECREF(YieldFrom_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->YieldFrom_type, "__annotations__", + YieldFrom_annotations) == 0; + if (!cond) { + Py_DECREF(YieldFrom_annotations); + return 0; + } + Py_DECREF(YieldFrom_annotations); + PyObject *Compare_annotations = PyDict_New(); + if (!Compare_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Compare_annotations, "left", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + } + { + PyObject *type = state->cmpop_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + cond = PyDict_SetItemString(Compare_annotations, "ops", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + cond = PyDict_SetItemString(Compare_annotations, "comparators", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Compare_type, "_field_types", + Compare_annotations) == 0; + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Compare_type, "__annotations__", + Compare_annotations) == 0; + if (!cond) { + Py_DECREF(Compare_annotations); + return 0; + } + Py_DECREF(Compare_annotations); + PyObject *Call_annotations = PyDict_New(); + if (!Call_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Call_annotations, "func", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + cond = PyDict_SetItemString(Call_annotations, "args", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + } + { + PyObject *type = state->keyword_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + cond = PyDict_SetItemString(Call_annotations, "keywords", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Call_type, "_field_types", + Call_annotations) == 0; + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Call_type, "__annotations__", + Call_annotations) == 0; + if (!cond) { + Py_DECREF(Call_annotations); + return 0; + } + Py_DECREF(Call_annotations); + PyObject *FormattedValue_annotations = PyDict_New(); + if (!FormattedValue_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(FormattedValue_annotations, "value", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyLong_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(FormattedValue_annotations, "conversion", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + cond = PyDict_SetItemString(FormattedValue_annotations, "format_spec", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->FormattedValue_type, "_field_types", + FormattedValue_annotations) == 0; + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->FormattedValue_type, + "__annotations__", + FormattedValue_annotations) == 0; + if (!cond) { + Py_DECREF(FormattedValue_annotations); + return 0; + } + Py_DECREF(FormattedValue_annotations); + PyObject *JoinedStr_annotations = PyDict_New(); + if (!JoinedStr_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(JoinedStr_annotations); + return 0; + } + cond = PyDict_SetItemString(JoinedStr_annotations, "values", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(JoinedStr_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->JoinedStr_type, "_field_types", + JoinedStr_annotations) == 0; + if (!cond) { + Py_DECREF(JoinedStr_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->JoinedStr_type, "__annotations__", + JoinedStr_annotations) == 0; + if (!cond) { + Py_DECREF(JoinedStr_annotations); + return 0; + } + Py_DECREF(JoinedStr_annotations); + PyObject *Constant_annotations = PyDict_New(); + if (!Constant_annotations) return 0; + { + PyObject *type = (PyObject *)&PyBaseObject_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(Constant_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Constant_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Constant_annotations); + return 0; + } + cond = PyDict_SetItemString(Constant_annotations, "kind", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Constant_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Constant_type, "_field_types", + Constant_annotations) == 0; + if (!cond) { + Py_DECREF(Constant_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Constant_type, "__annotations__", + Constant_annotations) == 0; + if (!cond) { + Py_DECREF(Constant_annotations); + return 0; + } + Py_DECREF(Constant_annotations); + PyObject *Attribute_annotations = PyDict_New(); + if (!Attribute_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Attribute_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Attribute_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(Attribute_annotations, "attr", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Attribute_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Attribute_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Attribute_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Attribute_type, "_field_types", + Attribute_annotations) == 0; + if (!cond) { + Py_DECREF(Attribute_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Attribute_type, "__annotations__", + Attribute_annotations) == 0; + if (!cond) { + Py_DECREF(Attribute_annotations); + return 0; + } + Py_DECREF(Attribute_annotations); + PyObject *Subscript_annotations = PyDict_New(); + if (!Subscript_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Subscript_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Subscript_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Subscript_annotations, "slice", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Subscript_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Subscript_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Subscript_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Subscript_type, "_field_types", + Subscript_annotations) == 0; + if (!cond) { + Py_DECREF(Subscript_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Subscript_type, "__annotations__", + Subscript_annotations) == 0; + if (!cond) { + Py_DECREF(Subscript_annotations); + return 0; + } + Py_DECREF(Subscript_annotations); + PyObject *Starred_annotations = PyDict_New(); + if (!Starred_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Starred_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Starred_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Starred_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Starred_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Starred_type, "_field_types", + Starred_annotations) == 0; + if (!cond) { + Py_DECREF(Starred_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Starred_type, "__annotations__", + Starred_annotations) == 0; + if (!cond) { + Py_DECREF(Starred_annotations); + return 0; + } + Py_DECREF(Starred_annotations); + PyObject *Name_annotations = PyDict_New(); + if (!Name_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(Name_annotations, "id", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Name_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Name_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Name_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Name_type, "_field_types", + Name_annotations) == 0; + if (!cond) { + Py_DECREF(Name_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Name_type, "__annotations__", + Name_annotations) == 0; + if (!cond) { + Py_DECREF(Name_annotations); + return 0; + } + Py_DECREF(Name_annotations); + PyObject *List_annotations = PyDict_New(); + if (!List_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(List_annotations); + return 0; + } + cond = PyDict_SetItemString(List_annotations, "elts", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(List_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(List_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(List_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->List_type, "_field_types", + List_annotations) == 0; + if (!cond) { + Py_DECREF(List_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->List_type, "__annotations__", + List_annotations) == 0; + if (!cond) { + Py_DECREF(List_annotations); + return 0; + } + Py_DECREF(List_annotations); + PyObject *Tuple_annotations = PyDict_New(); + if (!Tuple_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(Tuple_annotations); + return 0; + } + cond = PyDict_SetItemString(Tuple_annotations, "elts", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Tuple_annotations); + return 0; + } + } + { + PyObject *type = state->expr_context_type; + Py_INCREF(type); + cond = PyDict_SetItemString(Tuple_annotations, "ctx", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Tuple_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Tuple_type, "_field_types", + Tuple_annotations) == 0; + if (!cond) { + Py_DECREF(Tuple_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Tuple_type, "__annotations__", + Tuple_annotations) == 0; + if (!cond) { + Py_DECREF(Tuple_annotations); + return 0; + } + Py_DECREF(Tuple_annotations); + PyObject *Slice_annotations = PyDict_New(); + if (!Slice_annotations) return 0; + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + cond = PyDict_SetItemString(Slice_annotations, "lower", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + cond = PyDict_SetItemString(Slice_annotations, "upper", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + cond = PyDict_SetItemString(Slice_annotations, "step", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->Slice_type, "_field_types", + Slice_annotations) == 0; + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Slice_type, "__annotations__", + Slice_annotations) == 0; + if (!cond) { + Py_DECREF(Slice_annotations); + return 0; + } + Py_DECREF(Slice_annotations); + PyObject *Load_annotations = PyDict_New(); + if (!Load_annotations) return 0; + cond = PyObject_SetAttrString(state->Load_type, "_field_types", + Load_annotations) == 0; + if (!cond) { + Py_DECREF(Load_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Load_type, "__annotations__", + Load_annotations) == 0; + if (!cond) { + Py_DECREF(Load_annotations); + return 0; + } + Py_DECREF(Load_annotations); + PyObject *Store_annotations = PyDict_New(); + if (!Store_annotations) return 0; + cond = PyObject_SetAttrString(state->Store_type, "_field_types", + Store_annotations) == 0; + if (!cond) { + Py_DECREF(Store_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Store_type, "__annotations__", + Store_annotations) == 0; + if (!cond) { + Py_DECREF(Store_annotations); + return 0; + } + Py_DECREF(Store_annotations); + PyObject *Del_annotations = PyDict_New(); + if (!Del_annotations) return 0; + cond = PyObject_SetAttrString(state->Del_type, "_field_types", + Del_annotations) == 0; + if (!cond) { + Py_DECREF(Del_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Del_type, "__annotations__", + Del_annotations) == 0; + if (!cond) { + Py_DECREF(Del_annotations); + return 0; + } + Py_DECREF(Del_annotations); + PyObject *And_annotations = PyDict_New(); + if (!And_annotations) return 0; + cond = PyObject_SetAttrString(state->And_type, "_field_types", + And_annotations) == 0; + if (!cond) { + Py_DECREF(And_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->And_type, "__annotations__", + And_annotations) == 0; + if (!cond) { + Py_DECREF(And_annotations); + return 0; + } + Py_DECREF(And_annotations); + PyObject *Or_annotations = PyDict_New(); + if (!Or_annotations) return 0; + cond = PyObject_SetAttrString(state->Or_type, "_field_types", + Or_annotations) == 0; + if (!cond) { + Py_DECREF(Or_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Or_type, "__annotations__", + Or_annotations) == 0; + if (!cond) { + Py_DECREF(Or_annotations); + return 0; + } + Py_DECREF(Or_annotations); + PyObject *Add_annotations = PyDict_New(); + if (!Add_annotations) return 0; + cond = PyObject_SetAttrString(state->Add_type, "_field_types", + Add_annotations) == 0; + if (!cond) { + Py_DECREF(Add_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Add_type, "__annotations__", + Add_annotations) == 0; + if (!cond) { + Py_DECREF(Add_annotations); + return 0; + } + Py_DECREF(Add_annotations); + PyObject *Sub_annotations = PyDict_New(); + if (!Sub_annotations) return 0; + cond = PyObject_SetAttrString(state->Sub_type, "_field_types", + Sub_annotations) == 0; + if (!cond) { + Py_DECREF(Sub_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Sub_type, "__annotations__", + Sub_annotations) == 0; + if (!cond) { + Py_DECREF(Sub_annotations); + return 0; + } + Py_DECREF(Sub_annotations); + PyObject *Mult_annotations = PyDict_New(); + if (!Mult_annotations) return 0; + cond = PyObject_SetAttrString(state->Mult_type, "_field_types", + Mult_annotations) == 0; + if (!cond) { + Py_DECREF(Mult_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Mult_type, "__annotations__", + Mult_annotations) == 0; + if (!cond) { + Py_DECREF(Mult_annotations); + return 0; + } + Py_DECREF(Mult_annotations); + PyObject *MatMult_annotations = PyDict_New(); + if (!MatMult_annotations) return 0; + cond = PyObject_SetAttrString(state->MatMult_type, "_field_types", + MatMult_annotations) == 0; + if (!cond) { + Py_DECREF(MatMult_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatMult_type, "__annotations__", + MatMult_annotations) == 0; + if (!cond) { + Py_DECREF(MatMult_annotations); + return 0; + } + Py_DECREF(MatMult_annotations); + PyObject *Div_annotations = PyDict_New(); + if (!Div_annotations) return 0; + cond = PyObject_SetAttrString(state->Div_type, "_field_types", + Div_annotations) == 0; + if (!cond) { + Py_DECREF(Div_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Div_type, "__annotations__", + Div_annotations) == 0; + if (!cond) { + Py_DECREF(Div_annotations); + return 0; + } + Py_DECREF(Div_annotations); + PyObject *Mod_annotations = PyDict_New(); + if (!Mod_annotations) return 0; + cond = PyObject_SetAttrString(state->Mod_type, "_field_types", + Mod_annotations) == 0; + if (!cond) { + Py_DECREF(Mod_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Mod_type, "__annotations__", + Mod_annotations) == 0; + if (!cond) { + Py_DECREF(Mod_annotations); + return 0; + } + Py_DECREF(Mod_annotations); + PyObject *Pow_annotations = PyDict_New(); + if (!Pow_annotations) return 0; + cond = PyObject_SetAttrString(state->Pow_type, "_field_types", + Pow_annotations) == 0; + if (!cond) { + Py_DECREF(Pow_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Pow_type, "__annotations__", + Pow_annotations) == 0; + if (!cond) { + Py_DECREF(Pow_annotations); + return 0; + } + Py_DECREF(Pow_annotations); + PyObject *LShift_annotations = PyDict_New(); + if (!LShift_annotations) return 0; + cond = PyObject_SetAttrString(state->LShift_type, "_field_types", + LShift_annotations) == 0; + if (!cond) { + Py_DECREF(LShift_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->LShift_type, "__annotations__", + LShift_annotations) == 0; + if (!cond) { + Py_DECREF(LShift_annotations); + return 0; + } + Py_DECREF(LShift_annotations); + PyObject *RShift_annotations = PyDict_New(); + if (!RShift_annotations) return 0; + cond = PyObject_SetAttrString(state->RShift_type, "_field_types", + RShift_annotations) == 0; + if (!cond) { + Py_DECREF(RShift_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->RShift_type, "__annotations__", + RShift_annotations) == 0; + if (!cond) { + Py_DECREF(RShift_annotations); + return 0; + } + Py_DECREF(RShift_annotations); + PyObject *BitOr_annotations = PyDict_New(); + if (!BitOr_annotations) return 0; + cond = PyObject_SetAttrString(state->BitOr_type, "_field_types", + BitOr_annotations) == 0; + if (!cond) { + Py_DECREF(BitOr_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->BitOr_type, "__annotations__", + BitOr_annotations) == 0; + if (!cond) { + Py_DECREF(BitOr_annotations); + return 0; + } + Py_DECREF(BitOr_annotations); + PyObject *BitXor_annotations = PyDict_New(); + if (!BitXor_annotations) return 0; + cond = PyObject_SetAttrString(state->BitXor_type, "_field_types", + BitXor_annotations) == 0; + if (!cond) { + Py_DECREF(BitXor_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->BitXor_type, "__annotations__", + BitXor_annotations) == 0; + if (!cond) { + Py_DECREF(BitXor_annotations); + return 0; + } + Py_DECREF(BitXor_annotations); + PyObject *BitAnd_annotations = PyDict_New(); + if (!BitAnd_annotations) return 0; + cond = PyObject_SetAttrString(state->BitAnd_type, "_field_types", + BitAnd_annotations) == 0; + if (!cond) { + Py_DECREF(BitAnd_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->BitAnd_type, "__annotations__", + BitAnd_annotations) == 0; + if (!cond) { + Py_DECREF(BitAnd_annotations); + return 0; + } + Py_DECREF(BitAnd_annotations); + PyObject *FloorDiv_annotations = PyDict_New(); + if (!FloorDiv_annotations) return 0; + cond = PyObject_SetAttrString(state->FloorDiv_type, "_field_types", + FloorDiv_annotations) == 0; + if (!cond) { + Py_DECREF(FloorDiv_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->FloorDiv_type, "__annotations__", + FloorDiv_annotations) == 0; + if (!cond) { + Py_DECREF(FloorDiv_annotations); + return 0; + } + Py_DECREF(FloorDiv_annotations); + PyObject *Invert_annotations = PyDict_New(); + if (!Invert_annotations) return 0; + cond = PyObject_SetAttrString(state->Invert_type, "_field_types", + Invert_annotations) == 0; + if (!cond) { + Py_DECREF(Invert_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Invert_type, "__annotations__", + Invert_annotations) == 0; + if (!cond) { + Py_DECREF(Invert_annotations); + return 0; + } + Py_DECREF(Invert_annotations); + PyObject *Not_annotations = PyDict_New(); + if (!Not_annotations) return 0; + cond = PyObject_SetAttrString(state->Not_type, "_field_types", + Not_annotations) == 0; + if (!cond) { + Py_DECREF(Not_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Not_type, "__annotations__", + Not_annotations) == 0; + if (!cond) { + Py_DECREF(Not_annotations); + return 0; + } + Py_DECREF(Not_annotations); + PyObject *UAdd_annotations = PyDict_New(); + if (!UAdd_annotations) return 0; + cond = PyObject_SetAttrString(state->UAdd_type, "_field_types", + UAdd_annotations) == 0; + if (!cond) { + Py_DECREF(UAdd_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->UAdd_type, "__annotations__", + UAdd_annotations) == 0; + if (!cond) { + Py_DECREF(UAdd_annotations); + return 0; + } + Py_DECREF(UAdd_annotations); + PyObject *USub_annotations = PyDict_New(); + if (!USub_annotations) return 0; + cond = PyObject_SetAttrString(state->USub_type, "_field_types", + USub_annotations) == 0; + if (!cond) { + Py_DECREF(USub_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->USub_type, "__annotations__", + USub_annotations) == 0; + if (!cond) { + Py_DECREF(USub_annotations); + return 0; + } + Py_DECREF(USub_annotations); + PyObject *Eq_annotations = PyDict_New(); + if (!Eq_annotations) return 0; + cond = PyObject_SetAttrString(state->Eq_type, "_field_types", + Eq_annotations) == 0; + if (!cond) { + Py_DECREF(Eq_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Eq_type, "__annotations__", + Eq_annotations) == 0; + if (!cond) { + Py_DECREF(Eq_annotations); + return 0; + } + Py_DECREF(Eq_annotations); + PyObject *NotEq_annotations = PyDict_New(); + if (!NotEq_annotations) return 0; + cond = PyObject_SetAttrString(state->NotEq_type, "_field_types", + NotEq_annotations) == 0; + if (!cond) { + Py_DECREF(NotEq_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->NotEq_type, "__annotations__", + NotEq_annotations) == 0; + if (!cond) { + Py_DECREF(NotEq_annotations); + return 0; + } + Py_DECREF(NotEq_annotations); + PyObject *Lt_annotations = PyDict_New(); + if (!Lt_annotations) return 0; + cond = PyObject_SetAttrString(state->Lt_type, "_field_types", + Lt_annotations) == 0; + if (!cond) { + Py_DECREF(Lt_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Lt_type, "__annotations__", + Lt_annotations) == 0; + if (!cond) { + Py_DECREF(Lt_annotations); + return 0; + } + Py_DECREF(Lt_annotations); + PyObject *LtE_annotations = PyDict_New(); + if (!LtE_annotations) return 0; + cond = PyObject_SetAttrString(state->LtE_type, "_field_types", + LtE_annotations) == 0; + if (!cond) { + Py_DECREF(LtE_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->LtE_type, "__annotations__", + LtE_annotations) == 0; + if (!cond) { + Py_DECREF(LtE_annotations); + return 0; + } + Py_DECREF(LtE_annotations); + PyObject *Gt_annotations = PyDict_New(); + if (!Gt_annotations) return 0; + cond = PyObject_SetAttrString(state->Gt_type, "_field_types", + Gt_annotations) == 0; + if (!cond) { + Py_DECREF(Gt_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Gt_type, "__annotations__", + Gt_annotations) == 0; + if (!cond) { + Py_DECREF(Gt_annotations); + return 0; + } + Py_DECREF(Gt_annotations); + PyObject *GtE_annotations = PyDict_New(); + if (!GtE_annotations) return 0; + cond = PyObject_SetAttrString(state->GtE_type, "_field_types", + GtE_annotations) == 0; + if (!cond) { + Py_DECREF(GtE_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->GtE_type, "__annotations__", + GtE_annotations) == 0; + if (!cond) { + Py_DECREF(GtE_annotations); + return 0; + } + Py_DECREF(GtE_annotations); + PyObject *Is_annotations = PyDict_New(); + if (!Is_annotations) return 0; + cond = PyObject_SetAttrString(state->Is_type, "_field_types", + Is_annotations) == 0; + if (!cond) { + Py_DECREF(Is_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->Is_type, "__annotations__", + Is_annotations) == 0; + if (!cond) { + Py_DECREF(Is_annotations); + return 0; + } + Py_DECREF(Is_annotations); + PyObject *IsNot_annotations = PyDict_New(); + if (!IsNot_annotations) return 0; + cond = PyObject_SetAttrString(state->IsNot_type, "_field_types", + IsNot_annotations) == 0; + if (!cond) { + Py_DECREF(IsNot_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->IsNot_type, "__annotations__", + IsNot_annotations) == 0; + if (!cond) { + Py_DECREF(IsNot_annotations); + return 0; + } + Py_DECREF(IsNot_annotations); + PyObject *In_annotations = PyDict_New(); + if (!In_annotations) return 0; + cond = PyObject_SetAttrString(state->In_type, "_field_types", + In_annotations) == 0; + if (!cond) { + Py_DECREF(In_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->In_type, "__annotations__", + In_annotations) == 0; + if (!cond) { + Py_DECREF(In_annotations); + return 0; + } + Py_DECREF(In_annotations); + PyObject *NotIn_annotations = PyDict_New(); + if (!NotIn_annotations) return 0; + cond = PyObject_SetAttrString(state->NotIn_type, "_field_types", + NotIn_annotations) == 0; + if (!cond) { + Py_DECREF(NotIn_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->NotIn_type, "__annotations__", + NotIn_annotations) == 0; + if (!cond) { + Py_DECREF(NotIn_annotations); + return 0; + } + Py_DECREF(NotIn_annotations); + PyObject *comprehension_annotations = PyDict_New(); + if (!comprehension_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(comprehension_annotations, "target", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(comprehension_annotations, "iter", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + cond = PyDict_SetItemString(comprehension_annotations, "ifs", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyLong_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(comprehension_annotations, "is_async", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->comprehension_type, "_field_types", + comprehension_annotations) == 0; + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->comprehension_type, "__annotations__", + comprehension_annotations) == 0; + if (!cond) { + Py_DECREF(comprehension_annotations); + return 0; + } + Py_DECREF(comprehension_annotations); + PyObject *ExceptHandler_annotations = PyDict_New(); + if (!ExceptHandler_annotations) return 0; + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + cond = PyDict_SetItemString(ExceptHandler_annotations, "type", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + cond = PyDict_SetItemString(ExceptHandler_annotations, "name", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + cond = PyDict_SetItemString(ExceptHandler_annotations, "body", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->ExceptHandler_type, "_field_types", + ExceptHandler_annotations) == 0; + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->ExceptHandler_type, "__annotations__", + ExceptHandler_annotations) == 0; + if (!cond) { + Py_DECREF(ExceptHandler_annotations); + return 0; + } + Py_DECREF(ExceptHandler_annotations); + PyObject *arguments_annotations = PyDict_New(); + if (!arguments_annotations) return 0; + { + PyObject *type = state->arg_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "posonlyargs", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->arg_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "args", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->arg_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "vararg", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->arg_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "kwonlyargs", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "kw_defaults", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->arg_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "kwarg", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyDict_SetItemString(arguments_annotations, "defaults", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->arguments_type, "_field_types", + arguments_annotations) == 0; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->arguments_type, "__annotations__", + arguments_annotations) == 0; + if (!cond) { + Py_DECREF(arguments_annotations); + return 0; + } + Py_DECREF(arguments_annotations); + PyObject *arg_annotations = PyDict_New(); + if (!arg_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(arg_annotations, "arg", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + cond = PyDict_SetItemString(arg_annotations, "annotation", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + cond = PyDict_SetItemString(arg_annotations, "type_comment", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->arg_type, "_field_types", + arg_annotations) == 0; + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->arg_type, "__annotations__", + arg_annotations) == 0; + if (!cond) { + Py_DECREF(arg_annotations); + return 0; + } + Py_DECREF(arg_annotations); + PyObject *keyword_annotations = PyDict_New(); + if (!keyword_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(keyword_annotations); + return 0; + } + cond = PyDict_SetItemString(keyword_annotations, "arg", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(keyword_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(keyword_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(keyword_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->keyword_type, "_field_types", + keyword_annotations) == 0; + if (!cond) { + Py_DECREF(keyword_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->keyword_type, "__annotations__", + keyword_annotations) == 0; + if (!cond) { + Py_DECREF(keyword_annotations); + return 0; + } + Py_DECREF(keyword_annotations); + PyObject *alias_annotations = PyDict_New(); + if (!alias_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(alias_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(alias_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(alias_annotations); + return 0; + } + cond = PyDict_SetItemString(alias_annotations, "asname", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(alias_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->alias_type, "_field_types", + alias_annotations) == 0; + if (!cond) { + Py_DECREF(alias_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->alias_type, "__annotations__", + alias_annotations) == 0; + if (!cond) { + Py_DECREF(alias_annotations); + return 0; + } + Py_DECREF(alias_annotations); + PyObject *withitem_annotations = PyDict_New(); + if (!withitem_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(withitem_annotations, "context_expr", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(withitem_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(withitem_annotations); + return 0; + } + cond = PyDict_SetItemString(withitem_annotations, "optional_vars", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(withitem_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->withitem_type, "_field_types", + withitem_annotations) == 0; + if (!cond) { + Py_DECREF(withitem_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->withitem_type, "__annotations__", + withitem_annotations) == 0; + if (!cond) { + Py_DECREF(withitem_annotations); + return 0; + } + Py_DECREF(withitem_annotations); + PyObject *match_case_annotations = PyDict_New(); + if (!match_case_annotations) return 0; + { + PyObject *type = state->pattern_type; + Py_INCREF(type); + cond = PyDict_SetItemString(match_case_annotations, "pattern", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + cond = PyDict_SetItemString(match_case_annotations, "guard", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + } + { + PyObject *type = state->stmt_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + cond = PyDict_SetItemString(match_case_annotations, "body", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->match_case_type, "_field_types", + match_case_annotations) == 0; + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->match_case_type, "__annotations__", + match_case_annotations) == 0; + if (!cond) { + Py_DECREF(match_case_annotations); + return 0; + } + Py_DECREF(match_case_annotations); + PyObject *MatchValue_annotations = PyDict_New(); + if (!MatchValue_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(MatchValue_annotations, "value", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchValue_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchValue_type, "_field_types", + MatchValue_annotations) == 0; + if (!cond) { + Py_DECREF(MatchValue_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchValue_type, "__annotations__", + MatchValue_annotations) == 0; + if (!cond) { + Py_DECREF(MatchValue_annotations); + return 0; + } + Py_DECREF(MatchValue_annotations); + PyObject *MatchSingleton_annotations = PyDict_New(); + if (!MatchSingleton_annotations) return 0; + { + PyObject *type = (PyObject *)&PyBaseObject_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(MatchSingleton_annotations, "value", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchSingleton_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchSingleton_type, "_field_types", + MatchSingleton_annotations) == 0; + if (!cond) { + Py_DECREF(MatchSingleton_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchSingleton_type, + "__annotations__", + MatchSingleton_annotations) == 0; + if (!cond) { + Py_DECREF(MatchSingleton_annotations); + return 0; + } + Py_DECREF(MatchSingleton_annotations); + PyObject *MatchSequence_annotations = PyDict_New(); + if (!MatchSequence_annotations) return 0; + { + PyObject *type = state->pattern_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchSequence_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchSequence_annotations, "patterns", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchSequence_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchSequence_type, "_field_types", + MatchSequence_annotations) == 0; + if (!cond) { + Py_DECREF(MatchSequence_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchSequence_type, "__annotations__", + MatchSequence_annotations) == 0; + if (!cond) { + Py_DECREF(MatchSequence_annotations); + return 0; + } + Py_DECREF(MatchSequence_annotations); + PyObject *MatchMapping_annotations = PyDict_New(); + if (!MatchMapping_annotations) return 0; + { + PyObject *type = state->expr_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchMapping_annotations, "keys", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + } + { + PyObject *type = state->pattern_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchMapping_annotations, "patterns", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchMapping_annotations, "rest", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchMapping_type, "_field_types", + MatchMapping_annotations) == 0; + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchMapping_type, "__annotations__", + MatchMapping_annotations) == 0; + if (!cond) { + Py_DECREF(MatchMapping_annotations); + return 0; + } + Py_DECREF(MatchMapping_annotations); + PyObject *MatchClass_annotations = PyDict_New(); + if (!MatchClass_annotations) return 0; + { + PyObject *type = state->expr_type; + Py_INCREF(type); + cond = PyDict_SetItemString(MatchClass_annotations, "cls", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + } + { + PyObject *type = state->pattern_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchClass_annotations, "patterns", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchClass_annotations, "kwd_attrs", type) + == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + } + { + PyObject *type = state->pattern_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchClass_annotations, "kwd_patterns", + type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchClass_type, "_field_types", + MatchClass_annotations) == 0; + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchClass_type, "__annotations__", + MatchClass_annotations) == 0; + if (!cond) { + Py_DECREF(MatchClass_annotations); + return 0; + } + Py_DECREF(MatchClass_annotations); + PyObject *MatchStar_annotations = PyDict_New(); + if (!MatchStar_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchStar_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchStar_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchStar_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchStar_type, "_field_types", + MatchStar_annotations) == 0; + if (!cond) { + Py_DECREF(MatchStar_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchStar_type, "__annotations__", + MatchStar_annotations) == 0; + if (!cond) { + Py_DECREF(MatchStar_annotations); + return 0; + } + Py_DECREF(MatchStar_annotations); + PyObject *MatchAs_annotations = PyDict_New(); + if (!MatchAs_annotations) return 0; + { + PyObject *type = state->pattern_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchAs_annotations, "pattern", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchAs_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchAs_type, "_field_types", + MatchAs_annotations) == 0; + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchAs_type, "__annotations__", + MatchAs_annotations) == 0; + if (!cond) { + Py_DECREF(MatchAs_annotations); + return 0; + } + Py_DECREF(MatchAs_annotations); + PyObject *MatchOr_annotations = PyDict_New(); + if (!MatchOr_annotations) return 0; + { + PyObject *type = state->pattern_type; + type = Py_GenericAlias((PyObject *)&PyList_Type, type); + cond = type != NULL; + if (!cond) { + Py_DECREF(MatchOr_annotations); + return 0; + } + cond = PyDict_SetItemString(MatchOr_annotations, "patterns", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(MatchOr_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->MatchOr_type, "_field_types", + MatchOr_annotations) == 0; + if (!cond) { + Py_DECREF(MatchOr_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->MatchOr_type, "__annotations__", + MatchOr_annotations) == 0; + if (!cond) { + Py_DECREF(MatchOr_annotations); + return 0; + } + Py_DECREF(MatchOr_annotations); + PyObject *TypeIgnore_annotations = PyDict_New(); + if (!TypeIgnore_annotations) return 0; + { + PyObject *type = (PyObject *)&PyLong_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeIgnore_annotations, "lineno", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeIgnore_annotations); + return 0; + } + } + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeIgnore_annotations, "tag", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeIgnore_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->TypeIgnore_type, "_field_types", + TypeIgnore_annotations) == 0; + if (!cond) { + Py_DECREF(TypeIgnore_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->TypeIgnore_type, "__annotations__", + TypeIgnore_annotations) == 0; + if (!cond) { + Py_DECREF(TypeIgnore_annotations); + return 0; + } + Py_DECREF(TypeIgnore_annotations); + PyObject *TypeVar_annotations = PyDict_New(); + if (!TypeVar_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeVar_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeVar_annotations); + return 0; + } + } + { + PyObject *type = state->expr_type; + type = _Py_union_type_or(type, Py_None); + cond = type != NULL; + if (!cond) { + Py_DECREF(TypeVar_annotations); + return 0; + } + cond = PyDict_SetItemString(TypeVar_annotations, "bound", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeVar_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->TypeVar_type, "_field_types", + TypeVar_annotations) == 0; + if (!cond) { + Py_DECREF(TypeVar_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->TypeVar_type, "__annotations__", + TypeVar_annotations) == 0; + if (!cond) { + Py_DECREF(TypeVar_annotations); + return 0; + } + Py_DECREF(TypeVar_annotations); + PyObject *ParamSpec_annotations = PyDict_New(); + if (!ParamSpec_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(ParamSpec_annotations, "name", type) == 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(ParamSpec_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->ParamSpec_type, "_field_types", + ParamSpec_annotations) == 0; + if (!cond) { + Py_DECREF(ParamSpec_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->ParamSpec_type, "__annotations__", + ParamSpec_annotations) == 0; + if (!cond) { + Py_DECREF(ParamSpec_annotations); + return 0; + } + Py_DECREF(ParamSpec_annotations); + PyObject *TypeVarTuple_annotations = PyDict_New(); + if (!TypeVarTuple_annotations) return 0; + { + PyObject *type = (PyObject *)&PyUnicode_Type; + Py_INCREF(type); + cond = PyDict_SetItemString(TypeVarTuple_annotations, "name", type) == + 0; + Py_DECREF(type); + if (!cond) { + Py_DECREF(TypeVarTuple_annotations); + return 0; + } + } + cond = PyObject_SetAttrString(state->TypeVarTuple_type, "_field_types", + TypeVarTuple_annotations) == 0; + if (!cond) { + Py_DECREF(TypeVarTuple_annotations); + return 0; + } + cond = PyObject_SetAttrString(state->TypeVarTuple_type, "__annotations__", + TypeVarTuple_annotations) == 0; + if (!cond) { + Py_DECREF(TypeVarTuple_annotations); + return 0; + } + Py_DECREF(TypeVarTuple_annotations); + + return 1; +} + + typedef struct { PyObject_HEAD @@ -860,7 +5026,7 @@ ast_type_init(PyObject *self, PyObject *args, PyObject *kw) Py_ssize_t i, numfields = 0; int res = -1; - PyObject *key, *value, *fields; + PyObject *key, *value, *fields, *remaining_fields = NULL; if (PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), state->_fields, &fields) < 0) { goto cleanup; } @@ -869,6 +5035,13 @@ ast_type_init(PyObject *self, PyObject *args, PyObject *kw) if (numfields == -1) { goto cleanup; } + remaining_fields = PySet_New(fields); + } + else { + remaining_fields = PySet_New(NULL); + } + if (remaining_fields == NULL) { + goto cleanup; } res = 0; /* if no error occurs, this stays 0 to the end */ @@ -888,6 +5061,11 @@ ast_type_init(PyObject *self, PyObject *args, PyObject *kw) goto cleanup; } res = PyObject_SetAttr(self, name, PyTuple_GET_ITEM(args, i)); + if (PySet_Discard(remaining_fields, name) < 0) { + res = -1; + Py_DECREF(name); + goto cleanup; + } Py_DECREF(name); if (res < 0) { goto cleanup; @@ -900,13 +5078,14 @@ ast_type_init(PyObject *self, PyObject *args, PyObject *kw) if (contains == -1) { res = -1; goto cleanup; - } else if (contains == 1) { - Py_ssize_t p = PySequence_Index(fields, key); + } + else if (contains == 1) { + int p = PySet_Discard(remaining_fields, key); if (p == -1) { res = -1; goto cleanup; } - if (p < PyTuple_GET_SIZE(args)) { + if (p == 0) { PyErr_Format(PyExc_TypeError, "%.400s got multiple values for argument '%U'", Py_TYPE(self)->tp_name, key); @@ -914,15 +5093,91 @@ ast_type_init(PyObject *self, PyObject *args, PyObject *kw) goto cleanup; } } + else if ( + PyUnicode_CompareWithASCIIString(key, "lineno") != 0 && + PyUnicode_CompareWithASCIIString(key, "col_offset") != 0 && + PyUnicode_CompareWithASCIIString(key, "end_lineno") != 0 && + PyUnicode_CompareWithASCIIString(key, "end_col_offset") != 0 + ) { + if (PyErr_WarnFormat( + PyExc_DeprecationWarning, 1, + "%.400s.__init__ got an unexpected keyword argument '%U'. " + "Support for arbitrary keyword arguments is deprecated " + "and will be removed in Python 3.15.", + Py_TYPE(self)->tp_name, key + ) < 0) { + res = -1; + goto cleanup; + } + } res = PyObject_SetAttr(self, key, value); if (res < 0) { goto cleanup; } } } + Py_ssize_t size = PySet_Size(remaining_fields); + PyObject *field_types = NULL, *remaining_list = NULL; + if (size > 0) { + if (!PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), &_Py_ID(_field_types), + &field_types)) { + res = -1; + goto cleanup; + } + remaining_list = PySequence_List(remaining_fields); + if (!remaining_list) { + goto set_remaining_cleanup; + } + for (Py_ssize_t i = 0; i < size; i++) { + PyObject *name = PyList_GET_ITEM(remaining_list, i); + PyObject *type = PyDict_GetItemWithError(field_types, name); + if (!type) { + if (!PyErr_Occurred()) { + PyErr_SetObject(PyExc_KeyError, name); + } + goto set_remaining_cleanup; + } + if (_PyUnion_Check(type)) { + // optional field + // do nothing, we'll have set a None default on the class + } + else if (Py_IS_TYPE(type, &Py_GenericAliasType)) { + // list field + PyObject *empty = PyList_New(0); + if (!empty) { + goto set_remaining_cleanup; + } + res = PyObject_SetAttr(self, name, empty); + Py_DECREF(empty); + if (res < 0) { + goto set_remaining_cleanup; + } + } + else { + // simple field (e.g., identifier) + if (PyErr_WarnFormat( + PyExc_DeprecationWarning, 1, + "%.400s.__init__ missing 1 required positional argument: '%U'. " + "This will become an error in Python 3.15.", + Py_TYPE(self)->tp_name, name + ) < 0) { + res = -1; + goto cleanup; + } + } + } + Py_DECREF(remaining_list); + Py_DECREF(field_types); + } cleanup: Py_XDECREF(fields); + Py_XDECREF(remaining_fields); return res; + set_remaining_cleanup: + Py_XDECREF(remaining_list); + Py_XDECREF(field_types); + res = -1; + goto cleanup; } /* Pickling support */ @@ -934,14 +5189,75 @@ ast_type_reduce(PyObject *self, PyObject *unused) return NULL; } - PyObject *dict; + PyObject *dict = NULL, *fields = NULL, *remaining_fields = NULL, + *remaining_dict = NULL, *positional_args = NULL; if (PyObject_GetOptionalAttr(self, state->__dict__, &dict) < 0) { return NULL; } + PyObject *result = NULL; if (dict) { - return Py_BuildValue("O()N", Py_TYPE(self), dict); + // Serialize the fields as positional args if possible, because if we + // serialize them as a dict, during unpickling they are set only *after* + // the object is constructed, which will now trigger a DeprecationWarning + // if the AST type has required fields. + if (PyObject_GetOptionalAttr((PyObject*)Py_TYPE(self), state->_fields, &fields) < 0) { + goto cleanup; + } + if (fields) { + Py_ssize_t numfields = PySequence_Size(fields); + if (numfields == -1) { + Py_DECREF(dict); + goto cleanup; + } + remaining_dict = PyDict_Copy(dict); + Py_DECREF(dict); + if (!remaining_dict) { + goto cleanup; + } + positional_args = PyList_New(0); + if (!positional_args) { + goto cleanup; + } + for (Py_ssize_t i = 0; i < numfields; i++) { + PyObject *name = PySequence_GetItem(fields, i); + if (!name) { + goto cleanup; + } + PyObject *value = PyDict_GetItemWithError(remaining_dict, name); + if (!value) { + if (PyErr_Occurred()) { + goto cleanup; + } + break; + } + if (PyList_Append(positional_args, value) < 0) { + goto cleanup; + } + if (PyDict_DelItem(remaining_dict, name) < 0) { + goto cleanup; + } + Py_DECREF(name); + } + PyObject *args_tuple = PyList_AsTuple(positional_args); + if (!args_tuple) { + goto cleanup; + } + result = Py_BuildValue("ONO", Py_TYPE(self), args_tuple, + remaining_dict); + } + else { + result = Py_BuildValue("O()N", Py_TYPE(self), dict); + } + } + else { + result = Py_BuildValue("O()", Py_TYPE(self)); } - return Py_BuildValue("O()", Py_TYPE(self)); +cleanup: + Py_XDECREF(fields); + Py_XDECREF(remaining_fields); + Py_XDECREF(remaining_dict); + Py_XDECREF(positional_args); + return result; } static PyMemberDef ast_type_members[] = { @@ -1939,6 +6255,9 @@ init_types(struct ast_state *state) "TypeVarTuple(identifier name)"); if (!state->TypeVarTuple_type) return -1; + if (!add_ast_annotations(state)) { + return -1; + } return 0; }

1 0

gh-112997: Don't log arguments in asyncio unless debugging (#115667)
by gvanrossum Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/5a1559d9493dd298a08c4be32b52295aa3… commit: 5a1559d9493dd298a08c4be32b52295aa3eb89e5 branch: main author: Pierre Ossman (ThinLinc team) <ossman(a)cendio.se> committer: gvanrossum <gvanrossum(a)gmail.com> date: 2024-02-27T17:39:08-08:00 summary: gh-112997: Don't log arguments in asyncio unless debugging (#115667) Nothing else in Python generally logs the contents of variables, so this can be very unexpected for developers and could leak sensitive information in to terminals and log files. files: A Misc/NEWS.d/next/Library/2024-02-19-16-53-48.gh-issue-112997.sYBXRZ.rst M Lib/asyncio/events.py M Lib/asyncio/format_helpers.py M Lib/test/test_asyncio/test_events.py diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py index 072a99fee123c3..680749325025db 100644 --- a/Lib/asyncio/events.py +++ b/Lib/asyncio/events.py @@ -54,7 +54,8 @@ def _repr_info(self): info.append('cancelled') if self._callback is not None: info.append(format_helpers._format_callback_source( - self._callback, self._args)) + self._callback, self._args, + debug=self._loop.get_debug())) if self._source_traceback: frame = self._source_traceback[-1] info.append(f'created at {frame[0]}:{frame[1]}') @@ -90,7 +91,8 @@ def _run(self): raise except BaseException as exc: cb = format_helpers._format_callback_source( - self._callback, self._args) + self._callback, self._args, + debug=self._loop.get_debug()) msg = f'Exception in callback {cb}' context = { 'message': msg, diff --git a/Lib/asyncio/format_helpers.py b/Lib/asyncio/format_helpers.py index 27d11fd4fa9553..93737b7708a67b 100644 --- a/Lib/asyncio/format_helpers.py +++ b/Lib/asyncio/format_helpers.py @@ -19,19 +19,26 @@ def _get_function_source(func): return None -def _format_callback_source(func, args): - func_repr = _format_callback(func, args, None) +def _format_callback_source(func, args, *, debug=False): + func_repr = _format_callback(func, args, None, debug=debug) source = _get_function_source(func) if source: func_repr += f' at {source[0]}:{source[1]}' return func_repr -def _format_args_and_kwargs(args, kwargs): +def _format_args_and_kwargs(args, kwargs, *, debug=False): """Format function arguments and keyword arguments. Special case for a single parameter: ('hello',) is formatted as ('hello'). + + Note that this function only returns argument details when + debug=True is specified, as arguments may contain sensitive + information. """ + if not debug: + return '()' + # use reprlib to limit the length of the output items = [] if args: @@ -41,10 +48,11 @@ def _format_args_and_kwargs(args, kwargs): return '({})'.format(', '.join(items)) -def _format_callback(func, args, kwargs, suffix=''): +def _format_callback(func, args, kwargs, *, debug=False, suffix=''): if isinstance(func, functools.partial): - suffix = _format_args_and_kwargs(args, kwargs) + suffix - return _format_callback(func.func, func.args, func.keywords, suffix) + suffix = _format_args_and_kwargs(args, kwargs, debug=debug) + suffix + return _format_callback(func.func, func.args, func.keywords, + debug=debug, suffix=suffix) if hasattr(func, '__qualname__') and func.__qualname__: func_repr = func.__qualname__ @@ -53,7 +61,7 @@ def _format_callback(func, args, kwargs, suffix=''): else: func_repr = repr(func) - func_repr += _format_args_and_kwargs(args, kwargs) + func_repr += _format_args_and_kwargs(args, kwargs, debug=debug) if suffix: func_repr += suffix return func_repr diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py index c92c88bd5b2429..5b9c871e1d1b5a 100644 --- a/Lib/test/test_asyncio/test_events.py +++ b/Lib/test/test_asyncio/test_events.py @@ -2250,7 +2250,7 @@ def test_handle_repr(self): h = asyncio.Handle(noop, (1, 2), self.loop) filename, lineno = test_utils.get_function_source(noop) self.assertEqual(repr(h), - '<Handle noop(1, 2) at %s:%s>' + '<Handle noop() at %s:%s>' % (filename, lineno)) # cancelled handle @@ -2268,14 +2268,14 @@ def test_handle_repr(self): # partial function cb = functools.partial(noop, 1, 2) h = asyncio.Handle(cb, (3,), self.loop) - regex = (r'^<Handle noop$1, 2$$3$ at %s:%s>$' + regex = (r'^<Handle noop at %s:%s>$' % (re.escape(filename), lineno)) self.assertRegex(repr(h), regex) # partial function with keyword args cb = functools.partial(noop, x=1) h = asyncio.Handle(cb, (2, 3), self.loop) - regex = (r'^<Handle noop$x=1$$2, 3$ at %s:%s>$' + regex = (r'^<Handle noop at %s:%s>$' % (re.escape(filename), lineno)) self.assertRegex(repr(h), regex) @@ -2316,6 +2316,24 @@ def test_handle_repr_debug(self): '<Handle cancelled noop(1, 2) at %s:%s created at %s:%s>' % (filename, lineno, create_filename, create_lineno)) + # partial function + cb = functools.partial(noop, 1, 2) + create_lineno = sys._getframe().f_lineno + 1 + h = asyncio.Handle(cb, (3,), self.loop) + regex = (r'^<Handle noop$1, 2$$3$ at %s:%s created at %s:%s>$' + % (re.escape(filename), lineno, + re.escape(create_filename), create_lineno)) + self.assertRegex(repr(h), regex) + + # partial function with keyword args + cb = functools.partial(noop, x=1) + create_lineno = sys._getframe().f_lineno + 1 + h = asyncio.Handle(cb, (2, 3), self.loop) + regex = (r'^<Handle noop$x=1$$2, 3$ at %s:%s created at %s:%s>$' + % (re.escape(filename), lineno, + re.escape(create_filename), create_lineno)) + self.assertRegex(repr(h), regex) + def test_handle_source_traceback(self): loop = asyncio.get_event_loop_policy().new_event_loop() loop.set_debug(True) diff --git a/Misc/NEWS.d/next/Library/2024-02-19-16-53-48.gh-issue-112997.sYBXRZ.rst b/Misc/NEWS.d/next/Library/2024-02-19-16-53-48.gh-issue-112997.sYBXRZ.rst new file mode 100644 index 00000000000000..4f97b2d6085ba0 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2024-02-19-16-53-48.gh-issue-112997.sYBXRZ.rst @@ -0,0 +1,2 @@ +Stop logging potentially sensitive callback arguments in :mod:`asyncio` +unless debug mode is active.

1 0

gh-114914: Avoid keeping dead StreamWriter alive (#115661)
by gvanrossum Feb. 28, 2024

Feb. 28, 2024

https://github.com/python/cpython/commit/a355f60b032306651ca27bc53bbb82eb51… commit: a355f60b032306651ca27bc53bbb82eb5106ff71 branch: main author: Pierre Ossman (ThinLinc team) <ossman(a)cendio.se> committer: gvanrossum <gvanrossum(a)gmail.com> date: 2024-02-27T17:27:44-08:00 summary: gh-114914: Avoid keeping dead StreamWriter alive (#115661) In some cases we might cause a StreamWriter to stay alive even when the application has dropped all references to it. This prevents us from doing automatical cleanup, and complaining that the StreamWriter wasn't properly closed. Fortunately, the extra reference was never actually used for anything so we can just drop it. files: A Misc/NEWS.d/next/Library/2024-02-19-15-52-30.gh-issue-114914.M5-1d8.rst M Lib/asyncio/streams.py M Lib/test/test_asyncio/test_streams.py diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index df58b7a799a5ad..3fe52dbac25c91 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -201,7 +201,6 @@ def __init__(self, stream_reader, client_connected_cb=None, loop=None): # is established. self._strong_reader = stream_reader self._reject_connection = False - self._stream_writer = None self._task = None self._transport = None self._client_connected_cb = client_connected_cb @@ -214,10 +213,8 @@ def _stream_reader(self): return None return self._stream_reader_wr() - def _replace_writer(self, writer): + def _replace_transport(self, transport): loop = self._loop - transport = writer.transport - self._stream_writer = writer self._transport = transport self._over_ssl = transport.get_extra_info('sslcontext') is not None @@ -239,11 +236,8 @@ def connection_made(self, transport): reader.set_transport(transport) self._over_ssl = transport.get_extra_info('sslcontext') is not None if self._client_connected_cb is not None: - self._stream_writer = StreamWriter(transport, self, - reader, - self._loop) - res = self._client_connected_cb(reader, - self._stream_writer) + writer = StreamWriter(transport, self, reader, self._loop) + res = self._client_connected_cb(reader, writer) if coroutines.iscoroutine(res): def callback(task): if task.cancelled(): @@ -405,7 +399,7 @@ async def start_tls(self, sslcontext, *, ssl_handshake_timeout=ssl_handshake_timeout, ssl_shutdown_timeout=ssl_shutdown_timeout) self._transport = new_transport - protocol._replace_writer(self) + protocol._replace_transport(new_transport) def __del__(self, warnings=warnings): if not self._transport.is_closing(): diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 210990593adfa9..bf123ebf9bd158 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -1130,6 +1130,31 @@ async def inner(httpd): self.assertEqual(messages, []) + def test_unclosed_server_resource_warnings(self): + async def inner(rd, wr): + fut.set_result(True) + with self.assertWarns(ResourceWarning) as cm: + del wr + gc.collect() + self.assertEqual(len(cm.warnings), 1) + self.assertTrue(str(cm.warnings[0].message).startswith("unclosed <StreamWriter")) + + async def outer(): + srv = await asyncio.start_server(inner, socket_helper.HOSTv4, 0) + async with srv: + addr = srv.sockets[0].getsockname() + with socket.create_connection(addr): + # Give the loop some time to notice the connection + await fut + + messages = [] + self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx)) + + fut = self.loop.create_future() + self.loop.run_until_complete(outer()) + + self.assertEqual(messages, []) + def _basetest_unhandled_exceptions(self, handle_echo): port = socket_helper.find_unused_port() diff --git a/Misc/NEWS.d/next/Library/2024-02-19-15-52-30.gh-issue-114914.M5-1d8.rst b/Misc/NEWS.d/next/Library/2024-02-19-15-52-30.gh-issue-114914.M5-1d8.rst new file mode 100644 index 00000000000000..f7d392c8bceea3 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2024-02-19-15-52-30.gh-issue-114914.M5-1d8.rst @@ -0,0 +1,2 @@ +Fix an issue where an abandoned :class:`StreamWriter` would not be garbage +collected.

1 0