26725838468

Committed 31 May 2026 10:06PM UTC coverage: 78.709% (+0.008%) from 78.701%

Build # 26725838468

Build Type

Pull #63

github

Committed by

web-flow

Commit Message

Merge ad1ebc8ca into 519fc2167

Pull Request Pull Request #63: Addition of a v3 package

Coverage Stats

808 of 1023 new or added lines in 7 files covered. (78.98%)

2403 of 3053 relevant lines covered (78.71%)

4.44 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.87

/javaobj/v3/transformers.py

#!/usr/bin/env python3
"""
Defines the object transformers for javaobj v3

:authors: Thomas Calmant
:license: Apache License 2.0
:version: 0.5.0
:status: Alpha

..

    Copyright 2026 Thomas Calmant

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

# Standard library
import functools
import struct
from typing import TYPE_CHECKING, Any

# Numpy (optional)
try:
    import numpy  # type: ignore[import-untyped]
except ImportError:
    numpy = None  # type: ignore[assignment]

# Javaobj
from ..constants import TerminalCode, TypeCode
from .beans import BlockData, JavaClassDesc, JavaInstance
from .reader import DataReader

if TYPE_CHECKING:
    from .parser import JavaStreamParser

# ------------------------------------------------------------------------------

# Module version
__version_info__ = (0, 5, 0)
__version__ = ".".join(str(x) for x in __version_info__)

# Documentation strings format
__docformat__ = "restructuredtext en"

# ------------------------------------------------------------------------------

__all__ = [
    "ObjectTransformer",
    "DefaultObjectTransformer",
    "NumpyArrayTransformer",
]


# ------------------------------------------------------------------------------
# Base transformer interface
# ------------------------------------------------------------------------------


class ObjectTransformer:
    """
    Base class for v3 object transformers.

    Override any combination of the three hook methods to customise how
    specific Java classes are represented in Python.  Returning ``None``
    from any method signals that this transformer does not handle the case
    and the next transformer (or the default behaviour) should be tried.
    """

    def create_instance(self, classdesc: JavaClassDesc) -> JavaInstance | None:
        """
        Returns a custom :class:`~javaobj.v3.beans.JavaInstance` subclass
        for the given class descriptor, or ``None`` to use the default
        :class:`~javaobj.v3.beans.JavaInstance`.

        The parser will set ``.handle``, ``.classdesc``, ``.field_data``
        and ``.annotations`` on the returned object after this call.
        """
        return None

    def load_array(
        self,
        reader: DataReader,
        type_code: TypeCode,
        size: int,
    ) -> bytes | list[Any] | None:
        """
        Reads and returns the content of a Java array of *size* elements.

        Returns ``None`` to fall back to the default element-by-element
        reading logic.
        """
        return None

    def load_custom_writeObject(
        self,
        parser: "JavaStreamParser",
        reader: DataReader,
        class_name: str,
    ) -> Any | None:
        """
        Handles the content of a class that uses a custom ``writeObject``
        / ``readExternal`` method unknown to the default transformers.

        Returns ``None`` to indicate that this transformer cannot handle
        the class.
        """
        return None


# ------------------------------------------------------------------------------
# Collection / primitive transformer classes
# ------------------------------------------------------------------------------


class JavaList(list, JavaInstance):
    """Python list backed by a Java ArrayList or LinkedList."""

    HANDLED_CLASSES: tuple[str, ...] = (
        "java.util.ArrayList",
        "java.util.LinkedList",
    )

    def __init__(self) -> None:
        list.__init__(self)
        JavaInstance.__init__(self)

    def load_from_instance(self) -> bool:
        for cd, ann_list in self.annotations.items():
            if cd.name in self.HANDLED_CLASSES:
                # The first annotation entry is the capacity int; skip it.
                self.extend(a for a in ann_list[1:])
                return True
        return False


@functools.total_ordering
class JavaPrimitiveClass(JavaInstance):
    """
    Base for Java wrapper classes that box a single primitive value
    (Boolean, Integer, Long …).
    """

    HANDLED_CLASSES: str | tuple[str, ...] = ()

    def __init__(self) -> None:
        JavaInstance.__init__(self)
        self.value: Any = None

    def __str__(self) -> str:
        return str(self.value)

    def __repr__(self) -> str:
        return repr(self.value)

    def __hash__(self) -> int:
        return hash(self.value)

    def __eq__(self, other: object) -> bool:
        return self.value == other  # type: ignore[no-any-return]

    def __lt__(self, other: object) -> bool:
        return self.value < other  # type: ignore[operator]

    def load_from_instance(self) -> bool:
        for fields in self.field_data.values():
            for f, v in fields.items():
                if f.name == "value":
                    self.value = v
                    return True
        return False


class JavaBool(JavaPrimitiveClass):
    """Represents a Java ``Boolean`` wrapper object."""

    HANDLED_CLASSES = "java.lang.Boolean"

    def __bool__(self) -> bool:
        return bool(self.value)


class JavaInt(JavaPrimitiveClass):
    """Represents a Java ``Integer`` or ``Long`` wrapper object."""

    HANDLED_CLASSES = ("java.lang.Integer", "java.lang.Long")

    def __int__(self) -> int:
        return int(self.value)


class JavaMap(dict, JavaInstance):
    """Python dict backed by a Java HashMap or TreeMap."""

    HANDLED_CLASSES: tuple[str, ...] = (
        "java.util.HashMap",
        "java.util.TreeMap",
    )

    def __init__(self) -> None:
        dict.__init__(self)
        JavaInstance.__init__(self)

    def load_from_instance(self) -> bool:
        for cd, ann_list in self.annotations.items():
            if cd.name in self.HANDLED_CLASSES:
                # Annotation[0] is load-factor/capacity; skip it.
                it = iter(ann_list[1:])
                for key, value in zip(it, it):
                    self[key] = value
                return True
        return False


class JavaLinkedHashMap(JavaMap):
    """Java LinkedHashMap with custom block-data serialization."""

    HANDLED_CLASSES = ("java.util.LinkedHashMap",)

    def load_from_blockdata(self, parser: "JavaStreamParser", reader: DataReader) -> bool:
        # Read HashMap capacity / load-factor fields
        self.buckets: int = reader.read_int()
        self.size: int = reader.read_int()

        for _ in range(self.size):
            key_opcode = reader.read_byte()
            key = parser._read_content(key_opcode, block_data_allowed=True)

            val_opcode = reader.read_byte()
            value = parser._read_content(val_opcode, block_data_allowed=True)
            self[key] = value

        end_code = reader.read_byte()
        if end_code != TerminalCode.TC_ENDBLOCKDATA:
            raise ValueError(f"Expected TC_ENDBLOCKDATA, got 0x{end_code:02x}")
        final_byte = reader.read_byte()
        if final_byte != 0:
            raise ValueError(f"Expected trailing 0x00, got 0x{final_byte:02x}")
        return True


class JavaSet(set, JavaInstance):
    """Python set backed by a Java HashSet or LinkedHashSet."""

    HANDLED_CLASSES: tuple[str, ...] = (
        "java.util.HashSet",
        "java.util.LinkedHashSet",
    )

    def __init__(self) -> None:
        set.__init__(self)
        JavaInstance.__init__(self)

    def load_from_instance(self) -> bool:
        for cd, ann_list in self.annotations.items():
            if cd.name in self.HANDLED_CLASSES:
                # ann_list[0] is load-factor/capacity; skip it.
                self.update(a for a in ann_list[1:])
                return True
        return False


class JavaTreeSet(JavaSet):
    """Python set backed by a Java TreeSet."""

    HANDLED_CLASSES = ("java.util.TreeSet",)

    def load_from_instance(self) -> bool:
        for cd, ann_list in self.annotations.items():
            if cd.name in self.HANDLED_CLASSES:
                # ann_list[0] is comparator, ann_list[1] is size; skip both.
                self.update(a for a in ann_list[2:])
                return True
        return False


def _read_struct_from_bytes(data: bytes, fmt: str) -> tuple[tuple[Any, ...], bytes]:
    """Helper: unpacks *fmt* from the start of *data* and returns remaining."""
    size = struct.calcsize(fmt)
    values = struct.unpack(fmt, data[:size])
    return values, data[size:]


class JavaTime(JavaInstance):
    """
    Represents instances of the ``java.time`` package serialised via the
    ``java.time.Ser`` proxy class.
    """

    HANDLED_CLASSES = ("java.time.Ser",)

    DURATION_TYPE = 1
    INSTANT_TYPE = 2
    LOCAL_DATE_TYPE = 3
    LOCAL_TIME_TYPE = 4
    LOCAL_DATE_TIME_TYPE = 5
    ZONE_DATE_TIME_TYPE = 6
    ZONE_REGION_TYPE = 7
    ZONE_OFFSET_TYPE = 8
    OFFSET_TIME_TYPE = 9
    OFFSET_DATE_TIME_TYPE = 10
    YEAR_TYPE = 11
    YEAR_MONTH_TYPE = 12
    MONTH_DAY_TYPE = 13
    PERIOD_TYPE = 14

    def __init__(self) -> None:
        JavaInstance.__init__(self)
        self.type: int = -1
        self.year: int | None = None
        self.month: int | None = None
        self.day: int | None = None
        self.hour: int | None = None
        self.minute: int | None = None
        self.second: int | None = None
        self.nano: int | None = None
        self.offset: int | None = None
        self.zone: str | None = None

    def __str__(self) -> str:
        return (
            f"JavaTime(type=0x{self.type:x}, "
            f"year={self.year}, month={self.month}, day={self.day}, "
            f"hour={self.hour}, minute={self.minute}, second={self.second}, "
            f"nano={self.nano}, offset={self.offset}, zone={self.zone})"
        )

    def load_from_blockdata(self, parser: "JavaStreamParser", reader: DataReader) -> bool:
        # Block data is handled entirely inside load_from_instance via
        # the annotations.  Accept the call and let load_from_instance do
        # the real work.
        return True

    def load_from_instance(self) -> bool:
        for cd, ann_list in self.annotations.items():
            if cd.name not in self.HANDLED_CLASSES:
                continue
            if not ann_list or not isinstance(ann_list[0], BlockData):
                return False

            # The raw bytes are stored in the BlockData annotation.
            content: bytes = ann_list[0].data
            (self.type,), content = _read_struct_from_bytes(content, ">b")

            handlers = {
                self.DURATION_TYPE: self._do_duration,
                self.INSTANT_TYPE: self._do_instant,
                self.LOCAL_DATE_TYPE: self._do_local_date,
                self.LOCAL_DATE_TIME_TYPE: self._do_local_date_time,
                self.LOCAL_TIME_TYPE: self._do_local_time,
                self.ZONE_DATE_TIME_TYPE: self._do_zoned_date_time,
                self.ZONE_OFFSET_TYPE: self._do_zone_offset,
                self.ZONE_REGION_TYPE: self._do_zone_region,
                self.OFFSET_TIME_TYPE: self._do_offset_time,
                self.OFFSET_DATE_TIME_TYPE: self._do_offset_date_time,
                self.YEAR_TYPE: self._do_year,
                self.YEAR_MONTH_TYPE: self._do_year_month,
                self.MONTH_DAY_TYPE: self._do_month_day,
                self.PERIOD_TYPE: self._do_period,
            }
            handler = handlers.get(self.type)
            if handler is not None:
                handler(content)
            return True
        return False

    # ------------------------------------------------------------------
    # Internal time-type handlers
    # ------------------------------------------------------------------

    def _do_duration(self, data: bytes) -> bytes:
        (self.second, self.nano), data = _read_struct_from_bytes(data, ">qi")
        return data

    def _do_instant(self, data: bytes) -> bytes:
        (self.second, self.nano), data = _read_struct_from_bytes(data, ">qi")
        return data

    def _do_local_date(self, data: bytes) -> bytes:
        (self.year, self.month, self.day), data = _read_struct_from_bytes(data, ">ibb")
        return data

    def _do_local_time(self, data: bytes) -> bytes:
        (hour,), data = _read_struct_from_bytes(data, ">b")
        minute = second = nano = 0

        if hour < 0:
            hour = ~hour
        else:
            (minute,), data = _read_struct_from_bytes(data, ">b")
            if minute < 0:
                minute = ~minute
            else:
                (second,), data = _read_struct_from_bytes(data, ">b")
                if second < 0:
                    second = ~second
                else:
                    (nano,), data = _read_struct_from_bytes(data, ">i")

        self.hour, self.minute, self.second, self.nano = (
            hour,
            minute,
            second,
            nano,
        )
        return data

    def _do_local_date_time(self, data: bytes) -> bytes:
        data = self._do_local_date(data)
        data = self._do_local_time(data)
        return data

    def _do_zoned_date_time(self, data: bytes) -> bytes:
        data = self._do_local_date_time(data)
        data = self._do_zone_offset(data)
        data = self._do_zone_region(data)
        return data

    def _do_zone_offset(self, data: bytes) -> bytes:
        (offset_byte,), data = _read_struct_from_bytes(data, ">b")
        if offset_byte == 127:
            (self.offset,), data = _read_struct_from_bytes(data, ">i")
        else:
            self.offset = offset_byte * 900
        return data

    def _do_zone_region(self, data: bytes) -> bytes:
        # 2-byte length + UTF-8 string (standard UTF-8, not modified)
        (length,), data = _read_struct_from_bytes(data, ">H")
        self.zone = data[:length].decode("utf-8")
        return data[length:]

    def _do_offset_time(self, data: bytes) -> bytes:
        data = self._do_local_time(data)
        data = self._do_zone_offset(data)
        return data

    def _do_offset_date_time(self, data: bytes) -> bytes:
        data = self._do_local_date_time(data)
        data = self._do_zone_offset(data)
        return data

    def _do_year(self, data: bytes) -> bytes:
        (self.year,), data = _read_struct_from_bytes(data, ">i")
        return data

    def _do_year_month(self, data: bytes) -> bytes:
        (self.year, self.month), data = _read_struct_from_bytes(data, ">ib")
        return data

    def _do_month_day(self, data: bytes) -> bytes:
        (self.month, self.day), data = _read_struct_from_bytes(data, ">bb")
        return data

    def _do_period(self, data: bytes) -> bytes:
        (self.year, self.month, self.day), data = _read_struct_from_bytes(data, ">iii")
        return data


# ------------------------------------------------------------------------------
# DefaultObjectTransformer
# ------------------------------------------------------------------------------


class DefaultObjectTransformer(ObjectTransformer):
    """
    Built-in transformer that covers the most common Java standard-library
    classes.

    Handled classes
    ~~~~~~~~~~~~~~~
    * ``java.lang.Boolean``, ``java.lang.Integer``, ``java.lang.Long``
    * ``java.util.ArrayList``, ``java.util.LinkedList``
    * ``java.util.HashMap``, ``java.util.TreeMap``, ``java.util.LinkedHashMap``
    * ``java.util.HashSet``, ``java.util.LinkedHashSet``, ``java.util.TreeSet``
    * ``java.time.Ser``
    """

    _KNOWN_TRANSFORMERS: tuple[type[JavaInstance], ...] = (
        JavaBool,
        JavaInt,
        JavaList,
        JavaMap,
        JavaLinkedHashMap,
        JavaSet,
        JavaTreeSet,
        JavaTime,
    )

    def __init__(self) -> None:
        self._type_mapper: dict[str, type[JavaInstance]] = {}
        for klass in self._KNOWN_TRANSFORMERS:
            handled = klass.HANDLED_CLASSES  # type: ignore[attr-defined]
            if isinstance(handled, str):
                self._type_mapper[handled] = klass
            else:
                for name in handled:
                    self._type_mapper[name] = klass

    def create_instance(self, classdesc: JavaClassDesc) -> JavaInstance | None:
        """
        Returns a specialised :class:`JavaInstance` subclass for known Java
        types, or ``None`` for unknown types.
        """
        mapped = self._type_mapper.get(classdesc.name)
        if mapped is None:
            return None
        instance = mapped()
        instance.classdesc = classdesc
        return instance

    def handles(self, class_name: str) -> bool:
        """Returns ``True`` if this transformer knows how to handle *class_name*."""
        return class_name in self._type_mapper


# ------------------------------------------------------------------------------
# NumpyArrayTransformer
# ------------------------------------------------------------------------------


class NumpyArrayTransformer(ObjectTransformer):
    """
    Loads primitive Java arrays as NumPy arrays when *numpy* is available.

    NumPy dtype mapping (corrected from v1/v2):
        * ``TYPE_CHAR`` → ``>u2`` (2-byte unsigned, UTF-16; **not** ``b``)
        * ``TYPE_BYTE`` → ``B``  (unsigned byte)
        * All other types use their natural NumPy big-endian counterparts.
    """

    NUMPY_TYPE_MAP: dict[TypeCode, str] = {
        TypeCode.TYPE_BYTE: "B",
        TypeCode.TYPE_CHAR: ">u2",  # Fixed: Java char = 2-byte unsigned
        TypeCode.TYPE_DOUBLE: ">d",
        TypeCode.TYPE_FLOAT: ">f",
        TypeCode.TYPE_INTEGER: ">i",
        TypeCode.TYPE_LONG: ">q",
        TypeCode.TYPE_SHORT: ">h",
        TypeCode.TYPE_BOOLEAN: ">B",
    }

    def load_array(
        self,
        reader: DataReader,
        type_code: TypeCode,
        size: int,
    ) -> Any | None:
        """
        Reads *size* elements from the stream as a NumPy array.

        Returns ``None`` if NumPy is not installed or the element type has
        no NumPy mapping.
        """
        if numpy is None:
            return None
        dtype = self.NUMPY_TYPE_MAP.get(type_code)
        if dtype is None:
            return None
        return numpy.fromfile(reader._fd, dtype=dtype, count=size)

1	#!/usr/bin/env python3
2	"""
3	Defines the object transformers for javaobj v3
4
5	:authors: Thomas Calmant
6	:license: Apache License 2.0
7	:version: 0.5.0
8	:status: Alpha
9
10	..
11
12	Copyright 2026 Thomas Calmant
13
14	Licensed under the Apache License, Version 2.0 (the "License");
15	you may not use this file except in compliance with the License.
16	You may obtain a copy of the License at
17
18	http://www.apache.org/licenses/LICENSE-2.0
19
20	Unless required by applicable law or agreed to in writing, software
21	distributed under the License is distributed on an "AS IS" BASIS,
22	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
23	See the License for the specific language governing permissions and
24	limitations under the License.
25	"""
26
27	# Standard library
28	import functools	3✔
29	import struct	3✔
30	from typing import TYPE_CHECKING, Any	3✔
31
32	# Numpy (optional)
33	try:	3✔
34	import numpy # type: ignore[import-untyped]	3✔
35	except ImportError:	3✔
36	numpy = None # type: ignore[assignment]	3✔
37
38	# Javaobj
39	from ..constants import TerminalCode, TypeCode	3✔
40	from .beans import BlockData, JavaClassDesc, JavaInstance	3✔
41	from .reader import DataReader	3✔
42
43	if TYPE_CHECKING:
44	from .parser import JavaStreamParser
45
46	# ------------------------------------------------------------------------------
47
48	# Module version
49	__version_info__ = (0, 5, 0)	3✔
50	__version__ = ".".join(str(x) for x in __version_info__)	3✔
51
52	# Documentation strings format
53	__docformat__ = "restructuredtext en"	3✔
54
55	# ------------------------------------------------------------------------------
56
57	__all__ = [	3✔
58	"ObjectTransformer",
59	"DefaultObjectTransformer",
60	"NumpyArrayTransformer",
61	]
62
63
64	# ------------------------------------------------------------------------------
65	# Base transformer interface
66	# ------------------------------------------------------------------------------
67
68
69	class ObjectTransformer:	3✔
70	"""
71	Base class for v3 object transformers.
72
73	Override any combination of the three hook methods to customise how
74	specific Java classes are represented in Python. Returning ``None``
75	from any method signals that this transformer does not handle the case
76	and the next transformer (or the default behaviour) should be tried.
77	"""
78
79	def create_instance(self, classdesc: JavaClassDesc) -> JavaInstance \| None:	3✔
80	"""
81	Returns a custom :class:`~javaobj.v3.beans.JavaInstance` subclass
82	for the given class descriptor, or ``None`` to use the default
83	:class:`~javaobj.v3.beans.JavaInstance`.
84
85	The parser will set ``.handle``, ``.classdesc``, ``.field_data``
86	and ``.annotations`` on the returned object after this call.
87	"""
NEW 88	return None	×
89
90	def load_array(	3✔
91	self,
92	reader: DataReader,
93	type_code: TypeCode,
94	size: int,
95	) -> bytes \| list[Any] \| None:
96	"""
97	Reads and returns the content of a Java array of size elements.
98
99	Returns ``None`` to fall back to the default element-by-element
100	reading logic.
101	"""
102	return None	3✔
103
104	def load_custom_writeObject(	3✔
105	self,
106	parser: "JavaStreamParser",
107	reader: DataReader,
108	class_name: str,
109	) -> Any \| None:
110	"""
111	Handles the content of a class that uses a custom ``writeObject``
112	/ ``readExternal`` method unknown to the default transformers.
113
114	Returns ``None`` to indicate that this transformer cannot handle
115	the class.
116	"""
NEW 117	return None	×
118
119
120	# ------------------------------------------------------------------------------
121	# Collection / primitive transformer classes
122	# ------------------------------------------------------------------------------
123
124
125	class JavaList(list, JavaInstance):	3✔
126	"""Python list backed by a Java ArrayList or LinkedList."""
127
128	HANDLED_CLASSES: tuple[str, ...] = (	3✔
129	"java.util.ArrayList",
130	"java.util.LinkedList",
131	)
132
133	def __init__(self) -> None:	3✔
134	list.__init__(self)	3✔
135	JavaInstance.__init__(self)	3✔
136
137	def load_from_instance(self) -> bool:	3✔
138	for cd, ann_list in self.annotations.items():	3✔
139	if cd.name in self.HANDLED_CLASSES:	3✔
140	# The first annotation entry is the capacity int; skip it.
141	self.extend(a for a in ann_list[1:])	3✔
142	return True	3✔
NEW 143	return False	×
144
145
146	@functools.total_ordering	3✔
147	class JavaPrimitiveClass(JavaInstance):	3✔
148	"""
149	Base for Java wrapper classes that box a single primitive value
150	(Boolean, Integer, Long …).
151	"""
152
153	HANDLED_CLASSES: str \| tuple[str, ...] = ()	3✔
154
155	def __init__(self) -> None:	3✔
156	JavaInstance.__init__(self)	3✔
157	self.value: Any = None	3✔
158
159	def __str__(self) -> str:	3✔
NEW 160	return str(self.value)	×
161
162	def __repr__(self) -> str:	3✔
NEW 163	return repr(self.value)	×
164
165	def __hash__(self) -> int:	3✔
166	return hash(self.value)	3✔
167
168	def __eq__(self, other: object) -> bool:	3✔
169	return self.value == other # type: ignore[no-any-return]	3✔
170
171	def __lt__(self, other: object) -> bool:	3✔
NEW 172	return self.value < other # type: ignore[operator]	×
173
174	def load_from_instance(self) -> bool:	3✔
175	for fields in self.field_data.values():	3✔
176	for f, v in fields.items():	3✔
177	if f.name == "value":	3✔
178	self.value = v	3✔
179	return True	3✔
NEW 180	return False	×
181
182
183	class JavaBool(JavaPrimitiveClass):	3✔
184	"""Represents a Java ``Boolean`` wrapper object."""
185
186	HANDLED_CLASSES = "java.lang.Boolean"	3✔
187
188	def __bool__(self) -> bool:	3✔
NEW 189	return bool(self.value)	×
190
191
192	class JavaInt(JavaPrimitiveClass):	3✔
193	"""Represents a Java ``Integer`` or ``Long`` wrapper object."""
194
195	HANDLED_CLASSES = ("java.lang.Integer", "java.lang.Long")	3✔
196
197	def __int__(self) -> int:	3✔
NEW 198	return int(self.value)	×
199
200
201	class JavaMap(dict, JavaInstance):	3✔
202	"""Python dict backed by a Java HashMap or TreeMap."""
203
204	HANDLED_CLASSES: tuple[str, ...] = (	3✔
205	"java.util.HashMap",
206	"java.util.TreeMap",
207	)
208
209	def __init__(self) -> None:	3✔
210	dict.__init__(self)	3✔
211	JavaInstance.__init__(self)	3✔
212
213	def load_from_instance(self) -> bool:	3✔
214	for cd, ann_list in self.annotations.items():	3✔
215	if cd.name in self.HANDLED_CLASSES:	3✔
216	# Annotation[0] is load-factor/capacity; skip it.
217	it = iter(ann_list[1:])	3✔
218	for key, value in zip(it, it):	3✔
219	self[key] = value	3✔
220	return True	3✔
NEW 221	return False	×
222
223
224	class JavaLinkedHashMap(JavaMap):	3✔
225	"""Java LinkedHashMap with custom block-data serialization."""
226
227	HANDLED_CLASSES = ("java.util.LinkedHashMap",)	3✔
228
229	def load_from_blockdata(self, parser: "JavaStreamParser", reader: DataReader) -> bool:	3✔
230	# Read HashMap capacity / load-factor fields
NEW 231	self.buckets: int = reader.read_int()	×
NEW 232	self.size: int = reader.read_int()	×
233
NEW 234	for _ in range(self.size):	×
NEW 235	key_opcode = reader.read_byte()	×
NEW 236	key = parser._read_content(key_opcode, block_data_allowed=True)	×
237
NEW 238	val_opcode = reader.read_byte()	×
NEW 239	value = parser._read_content(val_opcode, block_data_allowed=True)	×
NEW 240	self[key] = value	×
241
NEW 242	end_code = reader.read_byte()	×
NEW 243	if end_code != TerminalCode.TC_ENDBLOCKDATA:	×
NEW 244	raise ValueError(f"Expected TC_ENDBLOCKDATA, got 0x{end_code:02x}")	×
NEW 245	final_byte = reader.read_byte()	×
NEW 246	if final_byte != 0:	×
NEW 247	raise ValueError(f"Expected trailing 0x00, got 0x{final_byte:02x}")	×
NEW 248	return True	×
249
250
251	class JavaSet(set, JavaInstance):	3✔
252	"""Python set backed by a Java HashSet or LinkedHashSet."""
253
254	HANDLED_CLASSES: tuple[str, ...] = (	3✔
255	"java.util.HashSet",
256	"java.util.LinkedHashSet",
257	)
258
259	def __init__(self) -> None:	3✔
260	set.__init__(self)	3✔
261	JavaInstance.__init__(self)	3✔
262
263	def load_from_instance(self) -> bool:	3✔
264	for cd, ann_list in self.annotations.items():	3✔
265	if cd.name in self.HANDLED_CLASSES:	3✔
266	# ann_list[0] is load-factor/capacity; skip it.
267	self.update(a for a in ann_list[1:])	3✔
268	return True	3✔
NEW 269	return False	×
270
271
272	class JavaTreeSet(JavaSet):	3✔
273	"""Python set backed by a Java TreeSet."""
274
275	HANDLED_CLASSES = ("java.util.TreeSet",)	3✔
276
277	def load_from_instance(self) -> bool:	3✔
278	for cd, ann_list in self.annotations.items():	3✔
279	if cd.name in self.HANDLED_CLASSES:	3✔
280	# ann_list[0] is comparator, ann_list[1] is size; skip both.
281	self.update(a for a in ann_list[2:])	3✔
282	return True	3✔
NEW 283	return False	×
284
285
286	def _read_struct_from_bytes(data: bytes, fmt: str) -> tuple[tuple[Any, ...], bytes]:	3✔
287	"""Helper: unpacks fmt from the start of data and returns remaining."""
288	size = struct.calcsize(fmt)	3✔
289	values = struct.unpack(fmt, data[:size])	3✔
290	return values, data[size:]	3✔
291
292
293	class JavaTime(JavaInstance):	3✔
294	"""
295	Represents instances of the ``java.time`` package serialised via the
296	``java.time.Ser`` proxy class.
297	"""
298
299	HANDLED_CLASSES = ("java.time.Ser",)	3✔
300
301	DURATION_TYPE = 1	3✔
302	INSTANT_TYPE = 2	3✔
303	LOCAL_DATE_TYPE = 3	3✔
304	LOCAL_TIME_TYPE = 4	3✔
305	LOCAL_DATE_TIME_TYPE = 5	3✔
306	ZONE_DATE_TIME_TYPE = 6	3✔
307	ZONE_REGION_TYPE = 7	3✔
308	ZONE_OFFSET_TYPE = 8	3✔
309	OFFSET_TIME_TYPE = 9	3✔
310	OFFSET_DATE_TIME_TYPE = 10	3✔
311	YEAR_TYPE = 11	3✔
312	YEAR_MONTH_TYPE = 12	3✔
313	MONTH_DAY_TYPE = 13	3✔
314	PERIOD_TYPE = 14	3✔
315
316	def __init__(self) -> None:	3✔
317	JavaInstance.__init__(self)	3✔
318	self.type: int = -1	3✔
319	self.year: int \| None = None	3✔
320	self.month: int \| None = None	3✔
321	self.day: int \| None = None	3✔
322	self.hour: int \| None = None	3✔
323	self.minute: int \| None = None	3✔
324	self.second: int \| None = None	3✔
325	self.nano: int \| None = None	3✔
326	self.offset: int \| None = None	3✔
327	self.zone: str \| None = None	3✔
328
329	def __str__(self) -> str:	3✔
NEW 330	return (	×
331	f"JavaTime(type=0x{self.type:x}, "
332	f"year={self.year}, month={self.month}, day={self.day}, "
333	f"hour={self.hour}, minute={self.minute}, second={self.second}, "
334	f"nano={self.nano}, offset={self.offset}, zone={self.zone})"
335	)
336
337	def load_from_blockdata(self, parser: "JavaStreamParser", reader: DataReader) -> bool:	3✔
338	# Block data is handled entirely inside load_from_instance via
339	# the annotations. Accept the call and let load_from_instance do
340	# the real work.
341	return True	3✔
342
343	def load_from_instance(self) -> bool:	3✔
344	for cd, ann_list in self.annotations.items():	3✔
345	if cd.name not in self.HANDLED_CLASSES:	3✔
NEW 346	continue	×
347	if not ann_list or not isinstance(ann_list[0], BlockData):	3✔
NEW 348	return False	×
349
350	# The raw bytes are stored in the BlockData annotation.
351	content: bytes = ann_list[0].data	3✔
352	(self.type,), content = _read_struct_from_bytes(content, ">b")	3✔
353
354	handlers = {	3✔
355	self.DURATION_TYPE: self._do_duration,
356	self.INSTANT_TYPE: self._do_instant,
357	self.LOCAL_DATE_TYPE: self._do_local_date,
358	self.LOCAL_DATE_TIME_TYPE: self._do_local_date_time,
359	self.LOCAL_TIME_TYPE: self._do_local_time,
360	self.ZONE_DATE_TIME_TYPE: self._do_zoned_date_time,
361	self.ZONE_OFFSET_TYPE: self._do_zone_offset,
362	self.ZONE_REGION_TYPE: self._do_zone_region,
363	self.OFFSET_TIME_TYPE: self._do_offset_time,
364	self.OFFSET_DATE_TIME_TYPE: self._do_offset_date_time,
365	self.YEAR_TYPE: self._do_year,
366	self.YEAR_MONTH_TYPE: self._do_year_month,
367	self.MONTH_DAY_TYPE: self._do_month_day,
368	self.PERIOD_TYPE: self._do_period,
369	}
370	handler = handlers.get(self.type)	3✔
371	if handler is not None:	3✔
372	handler(content)	3✔
373	return True	3✔
NEW 374	return False	×
375
376	# ------------------------------------------------------------------
377	# Internal time-type handlers
378	# ------------------------------------------------------------------
379
380	def _do_duration(self, data: bytes) -> bytes:	3✔
381	(self.second, self.nano), data = _read_struct_from_bytes(data, ">qi")	3✔
382	return data	3✔
383
384	def _do_instant(self, data: bytes) -> bytes:	3✔
385	(self.second, self.nano), data = _read_struct_from_bytes(data, ">qi")	3✔
386	return data	3✔
387
388	def _do_local_date(self, data: bytes) -> bytes:	3✔
389	(self.year, self.month, self.day), data = _read_struct_from_bytes(data, ">ibb")	3✔
390	return data	3✔
391
392	def _do_local_time(self, data: bytes) -> bytes:	3✔
393	(hour,), data = _read_struct_from_bytes(data, ">b")	3✔
394	minute = second = nano = 0	3✔
395
396	if hour < 0:	3✔
NEW 397	hour = ~hour	×
398	else:
399	(minute,), data = _read_struct_from_bytes(data, ">b")	3✔
400	if minute < 0:	3✔
NEW 401	minute = ~minute	×
402	else:
403	(second,), data = _read_struct_from_bytes(data, ">b")	3✔
404	if second < 0:	3✔
NEW 405	second = ~second	×
406	else:
407	(nano,), data = _read_struct_from_bytes(data, ">i")	3✔
408
409	self.hour, self.minute, self.second, self.nano = (	3✔
410	hour,
411	minute,
412	second,
413	nano,
414	)
415	return data	3✔
416
417	def _do_local_date_time(self, data: bytes) -> bytes:	3✔
418	data = self._do_local_date(data)	3✔
419	data = self._do_local_time(data)	3✔
420	return data	3✔
421
422	def _do_zoned_date_time(self, data: bytes) -> bytes:	3✔
423	data = self._do_local_date_time(data)	3✔
424	data = self._do_zone_offset(data)	3✔
425	data = self._do_zone_region(data)	3✔
426	return data	3✔
427
428	def _do_zone_offset(self, data: bytes) -> bytes:	3✔
429	(offset_byte,), data = _read_struct_from_bytes(data, ">b")	3✔
430	if offset_byte == 127:	3✔
NEW 431	(self.offset,), data = _read_struct_from_bytes(data, ">i")	×
432	else:
433	self.offset = offset_byte * 900	3✔
434	return data	3✔
435
436	def _do_zone_region(self, data: bytes) -> bytes:	3✔
437	# 2-byte length + UTF-8 string (standard UTF-8, not modified)
438	(length,), data = _read_struct_from_bytes(data, ">H")	3✔
439	self.zone = data[:length].decode("utf-8")	3✔
440	return data[length:]	3✔
441
442	def _do_offset_time(self, data: bytes) -> bytes:	3✔
NEW 443	data = self._do_local_time(data)	×
NEW 444	data = self._do_zone_offset(data)	×
NEW 445	return data	×
446
447	def _do_offset_date_time(self, data: bytes) -> bytes:	3✔
NEW 448	data = self._do_local_date_time(data)	×
NEW 449	data = self._do_zone_offset(data)	×
NEW 450	return data	×
451
452	def _do_year(self, data: bytes) -> bytes:	3✔
NEW 453	(self.year,), data = _read_struct_from_bytes(data, ">i")	×
NEW 454	return data	×
455
456	def _do_year_month(self, data: bytes) -> bytes:	3✔
NEW 457	(self.year, self.month), data = _read_struct_from_bytes(data, ">ib")	×
NEW 458	return data	×
459
460	def _do_month_day(self, data: bytes) -> bytes:	3✔
NEW 461	(self.month, self.day), data = _read_struct_from_bytes(data, ">bb")	×
NEW 462	return data	×
463
464	def _do_period(self, data: bytes) -> bytes:	3✔
NEW 465	(self.year, self.month, self.day), data = _read_struct_from_bytes(data, ">iii")	×
NEW 466	return data	×
467
468
469	# ------------------------------------------------------------------------------
470	# DefaultObjectTransformer
471	# ------------------------------------------------------------------------------
472
473
474	class DefaultObjectTransformer(ObjectTransformer):	3✔
475	"""
476	Built-in transformer that covers the most common Java standard-library
477	classes.
478
479	Handled classes
480	~~~~~~~~~~~~~~~
481	* ``java.lang.Boolean``, ``java.lang.Integer``, ``java.lang.Long``
482	* ``java.util.ArrayList``, ``java.util.LinkedList``
483	* ``java.util.HashMap``, ``java.util.TreeMap``, ``java.util.LinkedHashMap``
484	* ``java.util.HashSet``, ``java.util.LinkedHashSet``, ``java.util.TreeSet``
485	* ``java.time.Ser``
486	"""
487
488	_KNOWN_TRANSFORMERS: tuple[type[JavaInstance], ...] = (	3✔
489	JavaBool,
490	JavaInt,
491	JavaList,
492	JavaMap,
493	JavaLinkedHashMap,
494	JavaSet,
495	JavaTreeSet,
496	JavaTime,
497	)
498
499	def __init__(self) -> None:	3✔
500	self._type_mapper: dict[str, type[JavaInstance]] = {}	3✔
501	for klass in self._KNOWN_TRANSFORMERS:	3✔
502	handled = klass.HANDLED_CLASSES # type: ignore[attr-defined]	3✔
503	if isinstance(handled, str):	3✔
504	self._type_mapper[handled] = klass	3✔
505	else:
506	for name in handled:	3✔
507	self._type_mapper[name] = klass	3✔
508
509	def create_instance(self, classdesc: JavaClassDesc) -> JavaInstance \| None:	3✔
510	"""
511	Returns a specialised :class:`JavaInstance` subclass for known Java
512	types, or ``None`` for unknown types.
513	"""
514	mapped = self._type_mapper.get(classdesc.name)	3✔
515	if mapped is None:	3✔
516	return None	3✔
517	instance = mapped()	3✔
518	instance.classdesc = classdesc	3✔
519	return instance	3✔
520
521	def handles(self, class_name: str) -> bool:	3✔
522	"""Returns ``True`` if this transformer knows how to handle class_name."""
NEW 523	return class_name in self._type_mapper	×
524
525
526	# ------------------------------------------------------------------------------
527	# NumpyArrayTransformer
528	# ------------------------------------------------------------------------------
529
530
531	class NumpyArrayTransformer(ObjectTransformer):	3✔
532	"""
533	Loads primitive Java arrays as NumPy arrays when numpy is available.
534
535	NumPy dtype mapping (corrected from v1/v2):
536	* ``TYPE_CHAR`` → ``>u2`` (2-byte unsigned, UTF-16; not ``b``)
537	* ``TYPE_BYTE`` → ``B`` (unsigned byte)
538	* All other types use their natural NumPy big-endian counterparts.
539	"""
540
541	NUMPY_TYPE_MAP: dict[TypeCode, str] = {	3✔
542	TypeCode.TYPE_BYTE: "B",
543	TypeCode.TYPE_CHAR: ">u2", # Fixed: Java char = 2-byte unsigned
544	TypeCode.TYPE_DOUBLE: ">d",
545	TypeCode.TYPE_FLOAT: ">f",
546	TypeCode.TYPE_INTEGER: ">i",
547	TypeCode.TYPE_LONG: ">q",
548	TypeCode.TYPE_SHORT: ">h",
549	TypeCode.TYPE_BOOLEAN: ">B",
550	}
551
552	def load_array(	3✔
553	self,
554	reader: DataReader,
555	type_code: TypeCode,
556	size: int,
557	) -> Any \| None:
558	"""
559	Reads size elements from the stream as a NumPy array.
560
561	Returns ``None`` if NumPy is not installed or the element type has
562	no NumPy mapping.
563	"""
NEW 564	if numpy is None:	×
NEW 565	return None	×
NEW 566	dtype = self.NUMPY_TYPE_MAP.get(type_code)	×
NEW 567	if dtype is None:	×
NEW 568	return None	×
NEW 569	return numpy.fromfile(reader._fd, dtype=dtype, count=size)	×

tcalmant / python-javaobj / 26725838468

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous