10793540350

Committed 10 Sep 2024 01:31PM UTC coverage: 93.551% (-0.3%) from 93.892%

Build # 10793540350

Build Type

push

github

Committed by

falvaradorodriguez

Commit Message

Fix cowswap test

Run Details

8777 of 9382 relevant lines covered (93.55%)

3.74 hits per line

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

84.72

/safe_eth/eth/django/models.py

import binascii
from typing import Optional, Union

from django.core import exceptions
from django.core.exceptions import ValidationError
from django.db import models
from django.utils.translation import gettext_lazy as _

from eth_typing import ChecksumAddress, HexAddress, HexStr
from eth_utils import to_normalized_address
from hexbytes import HexBytes

from ..utils import fast_bytes_to_checksum_address, fast_to_checksum_address
from .forms import EthereumAddressFieldForm, HexFieldForm, Keccak256FieldForm
from .validators import validate_address, validate_checksumed_address

try:
    from django.db import DefaultConnectionProxy

    connection = DefaultConnectionProxy()
except ImportError:
    from django.db import connections

    connection = connections["default"]


class EthereumAddressBinaryField(models.Field):
    """
    Stores Ethereum Addresses in binary. Requires keccak256 hashing to
    calculate the EIP55 checksummed address, and it can take a high impact
    on the CPU for a lot of addresses.
    """

    default_validators = [validate_checksumed_address]
    description = "Stores Ethereum Addresses (EIP55) in binary"
    default_error_messages = {
        "invalid": _('"%(value)s" value must be an EIP55 checksummed address.'),
    }

    def get_internal_type(self):
        return "BinaryField"

    def from_db_value(
        self, value: memoryview, expression, connection
    ) -> Optional[ChecksumAddress]:
        if value:
            return fast_bytes_to_checksum_address(value)
        return None

    def get_prep_value(self, value: ChecksumAddress) -> Optional[bytes]:
        if value:
            try:
                return HexBytes(to_normalized_address(value))
            except (TypeError, ValueError):
                raise exceptions.ValidationError(
                    self.error_messages["invalid"],
                    code="invalid",
                    params={"value": value},
                )
        return None

    def to_python(self, value) -> Optional[ChecksumAddress]:
        if value is not None:
            try:
                return fast_to_checksum_address(value)
            except ValueError:
                raise exceptions.ValidationError(
                    self.error_messages["invalid"],
                    code="invalid",
                    params={"value": value},
                )
        return None

    def formfield(self, **kwargs):
        defaults = {
            "form_class": EthereumAddressFieldForm,
            "max_length": 2 + 40,
        }
        defaults.update(kwargs)
        return super().formfield(**defaults)


class EthereumAddressFastBinaryField(EthereumAddressBinaryField):
    """
    Stores Ethereum Addresses in binary. It returns not EIP55 regular addresses,
    which is faster as not EIP55 checksum is involved.
    """

    default_validators = [validate_address]
    description = "Stores Ethereum Addresses in binary"
    default_error_messages = {
        "invalid": _('"%(value)s" value must be a valid address.'),
    }

    def from_db_value(
        self, value: memoryview, expression, connection
    ) -> Optional[ChecksumAddress]:
        if value:
            return ChecksumAddress(HexAddress(HexStr("0x" + bytes(value).hex())))

        return None

    def to_python(self, value) -> Optional[ChecksumAddress]:
        if value is not None:
            try:
                if isinstance(value, bytes):
                    if len(value) != 20:
                        raise ValueError(
                            "Cannot convert %s to a checksum address, 20 bytes were expected"
                        )
                return ChecksumAddress(
                    HexAddress(HexStr(to_normalized_address(value)[2:]))
                )
            except ValueError:
                raise exceptions.ValidationError(
                    self.error_messages["invalid"],
                    code="invalid",
                    params={"value": value},
                )
        return None


class UnsignedDecimal(models.DecimalField):
    def deconstruct(self):
        name, path, args, kwargs = super().deconstruct()
        del kwargs["max_digits"]
        del kwargs["decimal_places"]
        return name, path, args, kwargs

    def from_db_value(self, value, expression, connection):
        if value is None:
            return value
        return int(value)

    def pre_save(self, model_instance, add):
        """
        Override pre_save to ensure that field is unsigned before save it
        :param model_instance:
        :param add:
        :return:
        """
        value = getattr(model_instance, self.attname)
        if value is not None and value < 0:
            raise ValidationError("Value must be an unsigned 256-bit integer")
        return super().pre_save(model_instance, add)


class Uint256Field(UnsignedDecimal):
    """
    Field to store ethereum uint256 values. Uses Decimal db type without decimals to store
    in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
    """

    description = _("Ethereum uint256 number")

    def __init__(self, *args, **kwargs):
        kwargs["max_digits"] = 78  # 2 ** 256 is 78 digits
        kwargs["decimal_places"] = 0
        super().__init__(*args, **kwargs)


class Uint96Field(UnsignedDecimal):
    """
    Field to store ethereum uint96 values. Uses Decimal db type without decimals to store
    in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
    """

    description = _("Ethereum uint96 number")

    def __init__(self, *args, **kwargs):
        kwargs["max_digits"] = 29  # 2 ** 96 is 29 digits
        kwargs["decimal_places"] = 0
        super().__init__(*args, **kwargs)


class Uint32Field(UnsignedDecimal):
    """
    Field to store ethereum uint32 values. Uses Decimal db type without decimals to store
    in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
    """

    description = _("Ethereum uint32 number")

    def __init__(self, *args, **kwargs):
        kwargs["max_digits"] = 10  # 2 ** 32 is 10 digits
        kwargs["decimal_places"] = 0
        super().__init__(*args, **kwargs)


class HexV2Field(models.BinaryField):
    def formfield(self, **kwargs):
        defaults = {
            "form_class": HexFieldForm,
        }
        defaults.update(kwargs)
        return super().formfield(**defaults)


class Keccak256Field(models.BinaryField):
    description = "Keccak256 hash stored as binary"
    default_error_messages = {
        "invalid": _('"%(value)s" hash must be a 32 bytes hexadecimal.'),
        "length": _('"%(value)s" hash must have exactly 32 bytes.'),
    }

    def _to_bytes(self, value) -> Optional[bytes]:
        if value is None:
            return None
        else:
            try:
                result = HexBytes(value)
                if len(result) != 32:
                    raise exceptions.ValidationError(
                        self.error_messages["length"],
                        code="length",
                        params={"value": value},
                    )
                return result
            except (ValueError, binascii.Error):
                raise exceptions.ValidationError(
                    self.error_messages["invalid"],
                    code="invalid",
                    params={"value": value},
                )

    def from_db_value(self, value: memoryview, expression, connection) -> Optional[str]:
        if value:
            return HexBytes(value.tobytes()).hex()
        return None

    def get_prep_value(self, value: Union[bytes, str]) -> Optional[bytes]:
        if value:
            return self._to_bytes(value)
        return None

    def value_to_string(self, obj):
        return str(self.value_from_object(obj))

    def to_python(self, value) -> Optional[bytes]:
        if value is not None:
            try:
                return self._to_bytes(value)
            except (ValueError, binascii.Error):
                raise exceptions.ValidationError(
                    self.error_messages["invalid"],
                    code="invalid",
                    params={"value": value},
                )
        return None

    def formfield(self, **kwargs):
        defaults = {
            "form_class": Keccak256FieldForm,
            "max_length": 2 + 64,
        }
        defaults.update(kwargs)
        return super().formfield(**defaults)


# --------- DEPRECATED, only for old migrations ----------------------
class EthereumAddressField(models.CharField):
    system_check_removed_details = {
        "msg": (
            "EthereumAddressField has been removed except for support in "
            "historical migrations."
        ),
        "hint": "Use EthereumAddressFastBinaryField instead.",
        "id": "fields.E4815",  # pick a unique ID for your field.
    }


class EthereumAddressV2Field(EthereumAddressBinaryField):
    system_check_removed_details = {
        "msg": (
            "EthereumAddressV2Field has been removed except for support in "
            "historical migrations."
        ),
        "hint": "Use EthereumAddressBinaryField instead.",
        "id": "fields.E4816",  # pick a unique ID for your field.
    }


class Sha3HashField(models.CharField):
    system_check_removed_details = {
        "msg": (
            "Sha3HashField has been removed except for support in "
            "historical migrations."
        ),
        "hint": "Use Keccak256Field instead.",
        "id": "fields.E4817",  # pick a unique ID for your field.
    }


class HexField(models.CharField):
    system_check_removed_details = {
        "msg": (
            "HexField has been removed except for support in " "historical migrations."
        ),
        "hint": "Use HexV2Field instead.",
        "id": "fields.E4818",  # pick a unique ID for your field.
    }

1	import binascii	4✔
2	from typing import Optional, Union	4✔
3
4	from django.core import exceptions	4✔
5	from django.core.exceptions import ValidationError	4✔
6	from django.db import models	4✔
7	from django.utils.translation import gettext_lazy as _	4✔
8
9	from eth_typing import ChecksumAddress, HexAddress, HexStr	4✔
10	from eth_utils import to_normalized_address	4✔
11	from hexbytes import HexBytes	4✔
12
13	from ..utils import fast_bytes_to_checksum_address, fast_to_checksum_address	4✔
14	from .forms import EthereumAddressFieldForm, HexFieldForm, Keccak256FieldForm	4✔
15	from .validators import validate_address, validate_checksumed_address	4✔
16
17	try:	4✔
18	from django.db import DefaultConnectionProxy	4✔
19
20	connection = DefaultConnectionProxy()	×
21	except ImportError:	4✔
22	from django.db import connections	4✔
23
24	connection = connections["default"]	4✔
25
26
27	class EthereumAddressBinaryField(models.Field):	4✔
28	"""
29	Stores Ethereum Addresses in binary. Requires keccak256 hashing to
30	calculate the EIP55 checksummed address, and it can take a high impact
31	on the CPU for a lot of addresses.
32	"""
33
34	default_validators = [validate_checksumed_address]	4✔
35	description = "Stores Ethereum Addresses (EIP55) in binary"	4✔
36	default_error_messages = {	4✔
37	"invalid": _('"%(value)s" value must be an EIP55 checksummed address.'),
38	}
39
40	def get_internal_type(self):	4✔
41	return "BinaryField"	4✔
42
43	def from_db_value(	4✔
44	self, value: memoryview, expression, connection
45	) -> Optional[ChecksumAddress]:
46	if value:	4✔
47	return fast_bytes_to_checksum_address(value)	4✔
48	return None	4✔
49
50	def get_prep_value(self, value: ChecksumAddress) -> Optional[bytes]:	4✔
51	if value:	4✔
52	try:	4✔
53	return HexBytes(to_normalized_address(value))	4✔
54	except (TypeError, ValueError):	4✔
55	raise exceptions.ValidationError(	4✔
56	self.error_messages["invalid"],
57	code="invalid",
58	params={"value": value},
59	)
60	return None	4✔
61
62	def to_python(self, value) -> Optional[ChecksumAddress]:	4✔
63	if value is not None:	4✔
64	try:	4✔
65	return fast_to_checksum_address(value)	4✔
66	except ValueError:	×
67	raise exceptions.ValidationError(	×
68	self.error_messages["invalid"],
69	code="invalid",
70	params={"value": value},
71	)
72	return None	×
73
74	def formfield(self, **kwargs):	4✔
75	defaults = {	×
76	"form_class": EthereumAddressFieldForm,
77	"max_length": 2 + 40,
78	}
79	defaults.update(kwargs)	×
80	return super().formfield(**defaults)	×
81
82
83	class EthereumAddressFastBinaryField(EthereumAddressBinaryField):	4✔
84	"""
85	Stores Ethereum Addresses in binary. It returns not EIP55 regular addresses,
86	which is faster as not EIP55 checksum is involved.
87	"""
88
89	default_validators = [validate_address]	4✔
90	description = "Stores Ethereum Addresses in binary"	4✔
91	default_error_messages = {	4✔
92	"invalid": _('"%(value)s" value must be a valid address.'),
93	}
94
95	def from_db_value(	4✔
96	self, value: memoryview, expression, connection
97	) -> Optional[ChecksumAddress]:
98	if value:	4✔
99	return ChecksumAddress(HexAddress(HexStr("0x" + bytes(value).hex())))	4✔
100
101	return None	4✔
102
103	def to_python(self, value) -> Optional[ChecksumAddress]:	4✔
104	if value is not None:	4✔
105	try:	4✔
106	if isinstance(value, bytes):	4✔
107	if len(value) != 20:	×
108	raise ValueError(	×
109	"Cannot convert %s to a checksum address, 20 bytes were expected"
110	)
111	return ChecksumAddress(	4✔
112	HexAddress(HexStr(to_normalized_address(value)[2:]))
113	)
114	except ValueError:	×
115	raise exceptions.ValidationError(	×
116	self.error_messages["invalid"],
117	code="invalid",
118	params={"value": value},
119	)
120	return None	×
121
122
123	class UnsignedDecimal(models.DecimalField):	4✔
124	def deconstruct(self):	4✔
125	name, path, args, kwargs = super().deconstruct()	4✔
126	del kwargs["max_digits"]	4✔
127	del kwargs["decimal_places"]	4✔
128	return name, path, args, kwargs	4✔
129
130	def from_db_value(self, value, expression, connection):	4✔
131	if value is None:	4✔
132	return value	4✔
133	return int(value)	4✔
134
135	def pre_save(self, model_instance, add):	4✔
136	"""
137	Override pre_save to ensure that field is unsigned before save it
138	:param model_instance:
139	:param add:
140	:return:
141	"""
142	value = getattr(model_instance, self.attname)	4✔
143	if value is not None and value < 0:	4✔
144	raise ValidationError("Value must be an unsigned 256-bit integer")	4✔
145	return super().pre_save(model_instance, add)	4✔
146
147
148	class Uint256Field(UnsignedDecimal):	4✔
149	"""
150	Field to store ethereum uint256 values. Uses Decimal db type without decimals to store
151	in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
152	"""
153
154	description = _("Ethereum uint256 number")	4✔
155
156	def __init__(self, args, *kwargs):	4✔
157	kwargs["max_digits"] = 78 # 2 ** 256 is 78 digits	4✔
158	kwargs["decimal_places"] = 0	4✔
159	super().__init__(args, *kwargs)	4✔
160
161
162	class Uint96Field(UnsignedDecimal):	4✔
163	"""
164	Field to store ethereum uint96 values. Uses Decimal db type without decimals to store
165	in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
166	"""
167
168	description = _("Ethereum uint96 number")	4✔
169
170	def __init__(self, args, *kwargs):	4✔
171	kwargs["max_digits"] = 29 # 2 ** 96 is 29 digits	4✔
172	kwargs["decimal_places"] = 0	4✔
173	super().__init__(args, *kwargs)	4✔
174
175
176	class Uint32Field(UnsignedDecimal):	4✔
177	"""
178	Field to store ethereum uint32 values. Uses Decimal db type without decimals to store
179	in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
180	"""
181
182	description = _("Ethereum uint32 number")	4✔
183
184	def __init__(self, args, *kwargs):	4✔
185	kwargs["max_digits"] = 10 # 2 ** 32 is 10 digits	4✔
186	kwargs["decimal_places"] = 0	4✔
187	super().__init__(args, *kwargs)	4✔
188
189
190	class HexV2Field(models.BinaryField):	4✔
191	def formfield(self, **kwargs):	4✔
192	defaults = {	×
193	"form_class": HexFieldForm,
194	}
195	defaults.update(kwargs)	×
196	return super().formfield(**defaults)	×
197
198
199	class Keccak256Field(models.BinaryField):	4✔
200	description = "Keccak256 hash stored as binary"	4✔
201	default_error_messages = {	4✔
202	"invalid": _('"%(value)s" hash must be a 32 bytes hexadecimal.'),
203	"length": _('"%(value)s" hash must have exactly 32 bytes.'),
204	}
205
206	def _to_bytes(self, value) -> Optional[bytes]:	4✔
207	if value is None:	4✔
208	return None	×
209	else:
210	try:	4✔
211	result = HexBytes(value)	4✔
212	if len(result) != 32:	4✔
213	raise exceptions.ValidationError(	4✔
214	self.error_messages["length"],
215	code="length",
216	params={"value": value},
217	)
218	return result	4✔
219	except (ValueError, binascii.Error):	4✔
220	raise exceptions.ValidationError(	4✔
221	self.error_messages["invalid"],
222	code="invalid",
223	params={"value": value},
224	)
225
226	def from_db_value(self, value: memoryview, expression, connection) -> Optional[str]:	4✔
227	if value:	4✔
228	return HexBytes(value.tobytes()).hex()	4✔
229	return None	4✔
230
231	def get_prep_value(self, value: Union[bytes, str]) -> Optional[bytes]:	4✔
232	if value:	4✔
233	return self._to_bytes(value)	4✔
234	return None	4✔
235
236	def value_to_string(self, obj):	4✔
237	return str(self.value_from_object(obj))	4✔
238
239	def to_python(self, value) -> Optional[bytes]:	4✔
240	if value is not None:	4✔
241	try:	4✔
242	return self._to_bytes(value)	4✔
243	except (ValueError, binascii.Error):	×
244	raise exceptions.ValidationError(	×
245	self.error_messages["invalid"],
246	code="invalid",
247	params={"value": value},
248	)
249	return None	×
250
251	def formfield(self, **kwargs):	4✔
252	defaults = {	×
253	"form_class": Keccak256FieldForm,
254	"max_length": 2 + 64,
255	}
256	defaults.update(kwargs)	×
257	return super().formfield(**defaults)	×
258
259
260	# --------- DEPRECATED, only for old migrations ----------------------
261	class EthereumAddressField(models.CharField):	4✔
262	system_check_removed_details = {	4✔
263	"msg": (
264	"EthereumAddressField has been removed except for support in "
265	"historical migrations."
266	),
267	"hint": "Use EthereumAddressFastBinaryField instead.",
268	"id": "fields.E4815", # pick a unique ID for your field.
269	}
270
271
272	class EthereumAddressV2Field(EthereumAddressBinaryField):	4✔
273	system_check_removed_details = {	4✔
274	"msg": (
275	"EthereumAddressV2Field has been removed except for support in "
276	"historical migrations."
277	),
278	"hint": "Use EthereumAddressBinaryField instead.",
279	"id": "fields.E4816", # pick a unique ID for your field.
280	}
281
282
283	class Sha3HashField(models.CharField):	4✔
284	system_check_removed_details = {	4✔
285	"msg": (
286	"Sha3HashField has been removed except for support in "
287	"historical migrations."
288	),
289	"hint": "Use Keccak256Field instead.",
290	"id": "fields.E4817", # pick a unique ID for your field.
291	}
292
293
294	class HexField(models.CharField):	4✔
295	system_check_removed_details = {	4✔
296	"msg": (
297	"HexField has been removed except for support in " "historical migrations."
298	),
299	"hint": "Use HexV2Field instead.",
300	"id": "fields.E4818", # pick a unique ID for your field.
301	}

safe-global / safe-eth-py / 10793540350

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