IntenseWebs/veilid
3
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-11-22 00:47:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Run Black on veilid-python

Browse Source
This commit is contained in:
Teknique 2023-07-22 10:06:46 -07:00
parent 86a5accb1a
commit d9d03885d9
11 changed files with 168 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
veilid-python/pyproject.toml
View File

@ -17,3 +17,9 @@ pytest-asyncio = "^0.21.0"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.black]
line-length = 99
[tool.mypy]
check_untyped_defs = true

4
veilid-python/tests/test_basic.py
View File

@ -27,9 +27,7 @@ async def test_get_node_id(api_connection: veilid.VeilidAPI):
@pytest.mark.asyncio
async def test_fail_connect():
with pytest.raises(socket.gaierror) as exc:
await veilid.json_api_connect(
"fuahwelifuh32luhwafluehawea", 1, simple_update_callback
)
await veilid.json_api_connect("fuahwelifuh32luhwafluehawea", 1, simple_update_callback)
assert exc.value.errno == socket.EAI_NONAME

6
veilid-python/tests/test_crypto.py
View File

@ -12,11 +12,10 @@ async def test_best_crypto_system(api_connection: veilid.VeilidAPI):
async with cs:
assert await cs.default_salt_length() == 16
@pytest.mark.asyncio
async def test_get_crypto_system(api_connection: veilid.VeilidAPI):
cs: CryptoSystem = await api_connection.get_crypto_system(
veilid.CryptoKind.CRYPTO_KIND_VLD0
)
cs: CryptoSystem = await api_connection.get_crypto_system(veilid.CryptoKind.CRYPTO_KIND_VLD0)
async with cs:
assert await cs.default_salt_length() == 16
@ -45,4 +44,3 @@ async def test_hash_and_verify_password(api_connection: veilid.VeilidAPI):
# Password mismatch
phash2 = await cs.hash_password(b"abc1234", salt)
assert not await cs.verify_password(b"abc12345", phash)

4
veilid-python/tests/test_dht.py
View File

@ -47,7 +47,9 @@ async def test_delete_dht_record_nonexistent(api_connection: veilid.VeilidAPI):
async def test_create_delete_dht_record_simple(api_connection: veilid.VeilidAPI):
rc = await api_connection.new_routing_context()
async with rc:
rec = await rc.create_dht_record(veilid.DHTSchema.dflt(1), veilid.CryptoKind.CRYPTO_KIND_VLD0)
rec = await rc.create_dht_record(
veilid.DHTSchema.dflt(1), veilid.CryptoKind.CRYPTO_KIND_VLD0
)
await rc.close_dht_record(rec.key)
await rc.delete_dht_record(rec.key)

32
veilid-python/tests/test_routing_context.py
View File

@ -34,9 +34,7 @@ async def test_routing_contexts(api_connection: veilid.VeilidAPI):
rc = await (await api_connection.new_routing_context()).with_custom_privacy(
veilid.SafetySelection.safe(
veilid.SafetySpec(
None, 2, veilid.Stability.RELIABLE, veilid.Sequencing.ENSURE_ORDERED
)
veilid.SafetySpec(None, 2, veilid.Stability.RELIABLE, veilid.Sequencing.ENSURE_ORDERED)
)
)
await rc.release()
@ -117,14 +115,10 @@ async def test_routing_context_app_call_loopback():
# send an app message to our own private route
request = b"abcd1234"
app_call_task = asyncio.create_task(
rc.app_call(prr, request), name="app call task"
)
app_call_task = asyncio.create_task(rc.app_call(prr, request), name="app call task")
# we should get the same request back
update: veilid.VeilidUpdate = await asyncio.wait_for(
app_call_queue.get(), timeout=10
)
update: veilid.VeilidUpdate = await asyncio.wait_for(app_call_queue.get(), timeout=10)
appcall = update.detail
assert isinstance(appcall, veilid.VeilidAppCall)
@ -164,9 +158,9 @@ async def test_routing_context_app_message_loopback_big_packets():
await api.debug("purge routes")
# make a routing context that uses a safety route
rc = await (
await (await api.new_routing_context()).with_privacy()
).with_sequencing(veilid.Sequencing.ENSURE_ORDERED)
rc = await (await (await api.new_routing_context()).with_privacy()).with_sequencing(
veilid.Sequencing.ENSURE_ORDERED
)
async with rc:
# make a new local private route
prl, blob = await api.new_private_route()
@ -223,14 +217,12 @@ async def test_routing_context_app_call_loopback_big_packets():
# purge routes to ensure we start fresh
await api.debug("purge routes")
app_call_task = asyncio.create_task(
app_call_queue_task_handler(api), name="app call task"
)
app_call_task = asyncio.create_task(app_call_queue_task_handler(api), name="app call task")
# make a routing context that uses a safety route
rc = await (
await (await api.new_routing_context()).with_privacy()
).with_sequencing(veilid.Sequencing.ENSURE_ORDERED)
rc = await (await (await api.new_routing_context()).with_privacy()).with_sequencing(
veilid.Sequencing.ENSURE_ORDERED
)
async with rc:
# make a new local private route
prl, blob = await api.new_private_route()
@ -249,9 +241,7 @@ async def test_routing_context_app_call_loopback_big_packets():
app_call_task.cancel()
@pytest.mark.skipif(
os.getenv("NOSKIP") != "1", reason="unneeded test, only for performance check"
)
@pytest.mark.skipif(os.getenv("NOSKIP") != "1", reason="unneeded test, only for performance check")
@pytest.mark.asyncio
async def test_routing_context_app_message_loopback_bandwidth():
app_message_queue: asyncio.Queue = asyncio.Queue()

6
veilid-python/tests/test_table_db.py
View File

@ -8,6 +8,7 @@ from veilid.api import CryptoSystem
TEST_DB = "__pytest_db"
TEST_NONEXISTENT_DB = "__pytest_nonexistent_db"
@pytest.mark.asyncio
async def test_delete_table_db_nonexistent(api_connection: veilid.VeilidAPI):
deleted = await api_connection.delete_table_db(TEST_NONEXISTENT_DB)
@ -30,6 +31,7 @@ async def test_open_delete_table_db(api_connection: veilid.VeilidAPI):
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
@pytest.mark.asyncio
async def test_open_twice_table_db(api_connection: veilid.VeilidAPI):
# delete test db if it exists
@ -71,6 +73,7 @@ async def test_open_twice_table_db_store_load(api_connection: veilid.VeilidAPI):
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted
@pytest.mark.asyncio
async def test_open_twice_table_db_store_delete_load(api_connection: veilid.VeilidAPI):
# delete test db if it exists
@ -80,7 +83,6 @@ async def test_open_twice_table_db_store_delete_load(api_connection: veilid.Veil
async with tdb:
tdb2 = await api_connection.open_table_db(TEST_DB, 1)
async with tdb2:
# store into first db copy
await tdb.store(b"asdf", b"1234")
# delete from second db copy and clean up
@ -113,7 +115,6 @@ async def test_resize_table_db(api_connection: veilid.VeilidAPI):
# reopen the db with fewer columns
tdb = await api_connection.open_table_db(TEST_DB, 1)
async with tdb:
# Should fail access to second column
with pytest.raises(veilid.VeilidAPIErrorGeneric) as exc:
await tdb.load(b"qwer", col=1)
@ -124,4 +125,3 @@ async def test_resize_table_db(api_connection: veilid.VeilidAPI):
# now delete should succeed
deleted = await api_connection.delete_table_db(TEST_DB)
assert deleted

26
veilid-python/veilid/api.py
View File

@ -6,7 +6,6 @@ from .state import VeilidState
class RoutingContext(ABC):
async def __aenter__(self) -> Self:
return self
@ -27,7 +26,9 @@ class RoutingContext(ABC):
pass
@abstractmethod
async def with_custom_privacy(self, safety_selection: types.SafetySelection, release = True) -> Self:
async def with_custom_privacy(
self, safety_selection: types.SafetySelection, release=True
) -> Self:
pass
@abstractmethod
@ -35,9 +36,7 @@ class RoutingContext(ABC):
pass
@abstractmethod
async def app_call(
self, target: types.TypedKey | types.RouteId, request: bytes
) -> bytes:
async def app_call(self, target: types.TypedKey | types.RouteId, request: bytes) -> bytes:
pass
@abstractmethod
@ -166,7 +165,6 @@ class TableDb(ABC):
class CryptoSystem(ABC):
async def __aenter__(self) -> Self:
return self
@ -183,9 +181,7 @@ class CryptoSystem(ABC):
pass
@abstractmethod
async def cached_dh(
self, key: types.PublicKey, secret: types.SecretKey
) -> types.SharedSecret:
async def cached_dh(self, key: types.PublicKey, secret: types.SecretKey) -> types.SharedSecret:
pass
@abstractmethod
@ -211,9 +207,7 @@ class CryptoSystem(ABC):
pass
@abstractmethod
async def derive_shared_secret(
self, password: bytes, salt: bytes
) -> types.SharedSecret:
async def derive_shared_secret(self, password: bytes, salt: bytes) -> types.SharedSecret:
pass
@abstractmethod
@ -233,9 +227,7 @@ class CryptoSystem(ABC):
pass
@abstractmethod
async def validate_key_pair(
self, key: types.PublicKey, secret: types.SecretKey
) -> bool:
async def validate_key_pair(self, key: types.PublicKey, secret: types.SecretKey) -> bool:
pass
@abstractmethod
@ -255,9 +247,7 @@ class CryptoSystem(ABC):
pass
@abstractmethod
async def verify(
self, key: types.PublicKey, data: bytes, signature: types.Signature
):
async def verify(self, key: types.PublicKey, data: bytes, signature: types.Signature):
pass
@abstractmethod

123
veilid-python/veilid/json_api.py
View File

@ -6,20 +6,45 @@ from typing import Awaitable, Callable, Optional, Self
from jsonschema import exceptions, validators
from . import schema
from .api import (CryptoSystem, RoutingContext, TableDb, TableDbTransaction,
VeilidAPI)
from .api import CryptoSystem, RoutingContext, TableDb, TableDbTransaction, VeilidAPI
from .error import raise_api_result
from .operations import (CryptoSystemOperation, Operation,
RoutingContextOperation, TableDbOperation,
TableDbTransactionOperation)
from .operations import (
CryptoSystemOperation,
Operation,
RoutingContextOperation,
TableDbOperation,
TableDbTransactionOperation,
)
from .state import VeilidState, VeilidUpdate
from .types import (CryptoKey, CryptoKeyDistance, CryptoKind,
DHTRecordDescriptor, DHTSchema, HashDigest, KeyPair,
NewPrivateRouteResult, Nonce, OperationId, PublicKey,
RouteId, SafetySelection, SecretKey, Sequencing, SharedSecret, Signature,
Stability, Timestamp, TypedKey, TypedKeyPair,
TypedSignature, ValueData, ValueSubkey, VeilidJSONEncoder,
VeilidVersion, urlsafe_b64decode_no_pad)
from .types import (
CryptoKey,
CryptoKeyDistance,
CryptoKind,
DHTRecordDescriptor,
DHTSchema,
HashDigest,
KeyPair,
NewPrivateRouteResult,
Nonce,
OperationId,
PublicKey,
RouteId,
SafetySelection,
SecretKey,
Sequencing,
SharedSecret,
Signature,
Stability,
Timestamp,
TypedKey,
TypedKeyPair,
TypedSignature,
ValueData,
ValueSubkey,
VeilidJSONEncoder,
VeilidVersion,
urlsafe_b64decode_no_pad,
)
##############################################################
@ -200,10 +225,7 @@ class _JsonVeilidAPI(VeilidAPI):
self.writer.write(reqbytes)
async def send_ndjson_request(
self,
op: Operation,
validate: Optional[Callable[[dict, dict], None]] = None,
**kwargs
self, op: Operation, validate: Optional[Callable[[dict, dict], None]] = None, **kwargs
) -> dict:
# Get next id
await self.lock.acquire()
@ -249,9 +271,7 @@ class _JsonVeilidAPI(VeilidAPI):
return response
async def control(self, args: list[str]) -> str:
return raise_api_result(
await self.send_ndjson_request(Operation.CONTROL, args=args)
)
return raise_api_result(await self.send_ndjson_request(Operation.CONTROL, args=args))
async def get_state(self) -> VeilidState:
return VeilidState.from_json(
@ -266,9 +286,7 @@ class _JsonVeilidAPI(VeilidAPI):
async def new_private_route(self) -> tuple[RouteId, bytes]:
return NewPrivateRouteResult.from_json(
raise_api_result(
await self.send_ndjson_request(Operation.NEW_PRIVATE_ROUTE)
)
raise_api_result(await self.send_ndjson_request(Operation.NEW_PRIVATE_ROUTE))
).to_tuple()
async def new_custom_private_route(
@ -288,17 +306,13 @@ class _JsonVeilidAPI(VeilidAPI):
async def import_remote_private_route(self, blob: bytes) -> RouteId:
return RouteId(
raise_api_result(
await self.send_ndjson_request(
Operation.IMPORT_REMOTE_PRIVATE_ROUTE, blob=blob
)
await self.send_ndjson_request(Operation.IMPORT_REMOTE_PRIVATE_ROUTE, blob=blob)
)
)
async def release_private_route(self, route_id: RouteId):
raise_api_result(
await self.send_ndjson_request(
Operation.RELEASE_PRIVATE_ROUTE, route_id=route_id
)
await self.send_ndjson_request(Operation.RELEASE_PRIVATE_ROUTE, route_id=route_id)
)
async def app_call_reply(self, call_id: OperationId, message: bytes):
@ -309,9 +323,7 @@ class _JsonVeilidAPI(VeilidAPI):
)
async def new_routing_context(self) -> RoutingContext:
rc_id = raise_api_result(
await self.send_ndjson_request(Operation.NEW_ROUTING_CONTEXT)
)
rc_id = raise_api_result(await self.send_ndjson_request(Operation.NEW_ROUTING_CONTEXT))
return _JsonRoutingContext(self, rc_id)
async def open_table_db(self, name: str, column_count: int) -> TableDb:
@ -334,9 +346,7 @@ class _JsonVeilidAPI(VeilidAPI):
return _JsonCryptoSystem(self, cs_id)
async def best_crypto_system(self) -> CryptoSystem:
cs_id = raise_api_result(
await self.send_ndjson_request(Operation.BEST_CRYPTO_SYSTEM)
)
cs_id = raise_api_result(await self.send_ndjson_request(Operation.BEST_CRYPTO_SYSTEM))
return _JsonCryptoSystem(self, cs_id)
async def verify_signatures(
@ -375,27 +385,19 @@ class _JsonVeilidAPI(VeilidAPI):
map(
lambda x: TypedKeyPair(x),
raise_api_result(
await self.send_ndjson_request(
Operation.GENERATE_KEY_PAIR, kind=kind
)
await self.send_ndjson_request(Operation.GENERATE_KEY_PAIR, kind=kind)
),
)
)
async def now(self) -> Timestamp:
return Timestamp(
raise_api_result(await self.send_ndjson_request(Operation.NOW))
)
return Timestamp(raise_api_result(await self.send_ndjson_request(Operation.NOW)))
async def debug(self, command: str) -> str:
return raise_api_result(
await self.send_ndjson_request(Operation.DEBUG, command=command)
)
return raise_api_result(await self.send_ndjson_request(Operation.DEBUG, command=command))
async def veilid_version_string(self) -> str:
return raise_api_result(
await self.send_ndjson_request(Operation.VEILID_VERSION_STRING)
)
return raise_api_result(await self.send_ndjson_request(Operation.VEILID_VERSION_STRING))
async def veilid_version(self) -> VeilidVersion:
v = await self.send_ndjson_request(Operation.VEILID_VERSION)
@ -424,9 +426,7 @@ class _JsonRoutingContext(RoutingContext):
if not self.done:
# attempt to clean up server-side anyway
self.api.send_one_way_ndjson_request(
Operation.ROUTING_CONTEXT,
rc_id=self.rc_id,
rc_op=RoutingContextOperation.RELEASE
Operation.ROUTING_CONTEXT, rc_id=self.rc_id, rc_op=RoutingContextOperation.RELEASE
)
# complain
@ -442,7 +442,7 @@ class _JsonRoutingContext(RoutingContext):
Operation.ROUTING_CONTEXT,
validate=validate_rc_op,
rc_id=self.rc_id,
rc_op=RoutingContextOperation.RELEASE
rc_op=RoutingContextOperation.RELEASE,
)
self.done = True
@ -664,7 +664,9 @@ class _JsonTableDbTransaction(TableDbTransaction):
)
# complain
raise AssertionError("Should have committed or rolled back transaction before dropping object")
raise AssertionError(
"Should have committed or rolled back transaction before dropping object"
)
def is_done(self) -> bool:
return self.done
@ -736,12 +738,9 @@ class _JsonTableDb(TableDb):
def __del__(self):
if not self.done:
# attempt to clean up server-side anyway
self.api.send_one_way_ndjson_request(
Operation.TABLE_DB,
db_id=self.db_id,
db_op=TableDbOperation.RELEASE
Operation.TABLE_DB, db_id=self.db_id, db_op=TableDbOperation.RELEASE
)
# complain
@ -757,11 +756,10 @@ class _JsonTableDb(TableDb):
Operation.TABLE_DB,
validate=validate_db_op,
db_id=self.db_id,
db_op=TableDbOperation.RELEASE
db_op=TableDbOperation.RELEASE,
)
self.done = True
async def get_column_count(self) -> int:
return raise_api_result(
await self.api.send_ndjson_request(
@ -859,12 +857,9 @@ class _JsonCryptoSystem(CryptoSystem):
def __del__(self):
if not self.done:
# attempt to clean up server-side anyway
self.api.send_one_way_ndjson_request(
Operation.CRYPTO_SYSTEM,
cs_id=self.cs_id,
cs_op=CryptoSystemOperation.RELEASE
Operation.CRYPTO_SYSTEM, cs_id=self.cs_id, cs_op=CryptoSystemOperation.RELEASE
)
# complain
@ -880,7 +875,7 @@ class _JsonCryptoSystem(CryptoSystem):
Operation.CRYPTO_SYSTEM,
validate=validate_cs_op,
cs_id=self.cs_id,
cs_op=CryptoSystemOperation.RELEASE
cs_op=CryptoSystemOperation.RELEASE,
)
self.done = True
@ -1142,9 +1137,7 @@ class _JsonCryptoSystem(CryptoSystem):
)
)
async def crypt_no_auth(
self, body: bytes, nonce: Nonce, shared_secret: SharedSecret
) -> bytes:
async def crypt_no_auth(self, body: bytes, nonce: Nonce, shared_secret: SharedSecret) -> bytes:
return urlsafe_b64decode_no_pad(
raise_api_result(
await self.api.send_ndjson_request(

6
veilid-python/veilid/operations.py
View File

@ -1,6 +1,7 @@
from enum import StrEnum
from typing import Self
class Operation(StrEnum):
CONTROL = "Control"
GET_STATE = "GetState"
@ -28,6 +29,7 @@ class Operation(StrEnum):
VEILID_VERSION_STRING = "VeilidVersionString"
VEILID_VERSION = "VeilidVersion"
class RoutingContextOperation(StrEnum):
INVALID_ID = "InvalidId"
RELEASE = "Release"
@ -45,6 +47,7 @@ class RoutingContextOperation(StrEnum):
WATCH_DHT_VALUES = "WatchDhtValues"
CANCEL_DHT_WATCH = "CancelDhtWatch"
class TableDbOperation(StrEnum):
INVALID_ID = "InvalidId"
RELEASE = "Release"
@ -55,6 +58,7 @@ class TableDbOperation(StrEnum):
LOAD = "Load"
DELETE = "Delete"
class TableDbTransactionOperation(StrEnum):
INVALID_ID = "InvalidId"
COMMIT = "Commit"
@ -62,6 +66,7 @@ class TableDbTransactionOperation(StrEnum):
STORE = "Store"
DELETE = "Delete"
class CryptoSystemOperation(StrEnum):
INVALID_ID = "InvalidId"
RELEASE = "Release"
@ -86,6 +91,7 @@ class CryptoSystemOperation(StrEnum):
ENCRYPT_AEAD = "EncryptAead"
CRYPT_NO_AUTH = "CryptNoAuth"
class RecvMessageType(StrEnum):
RESPONSE = "Response"
UPDATE = "Update"

27
veilid-python/veilid/state.py
View File

@ -2,9 +2,18 @@ from enum import StrEnum
from typing import Optional, Self
from .config import VeilidConfig
from .types import (ByteCount, RouteId, Timestamp, TimestampDuration, TypedKey,
ValueData, ValueSubkey, VeilidLogLevel, OperationId,
urlsafe_b64decode_no_pad)
from .types import (
ByteCount,
RouteId,
Timestamp,
TimestampDuration,
TypedKey,
ValueData,
ValueSubkey,
VeilidLogLevel,
OperationId,
urlsafe_b64decode_no_pad,
)
class AttachmentState(StrEnum):
@ -200,9 +209,7 @@ class PeerTableData:
@classmethod
def from_json(cls, j: dict) -> Self:
"""JSON object hook"""
return cls(
j["node_ids"], j["peer_address"], PeerStats.from_json(j["peer_stats"])
)
return cls(j["node_ids"], j["peer_address"], PeerStats.from_json(j["peer_stats"]))
class VeilidStateNetwork:
@ -276,9 +283,7 @@ class VeilidLog:
message: str
backtrace: Optional[str]
def __init__(
self, log_level: VeilidLogLevel, message: str, backtrace: Optional[str]
):
def __init__(self, log_level: VeilidLogLevel, message: str, backtrace: Optional[str]):
self.log_level = log_level
self.message = message
self.backtrace = backtrace
@ -349,9 +354,7 @@ class VeilidValueChange:
count: int
value: ValueData
def __init__(
self, key: TypedKey, subkeys: list[ValueSubkey], count: int, value: ValueData
):
def __init__(self, key: TypedKey, subkeys: list[ValueSubkey], count: int, value: ValueData):
self.key = key
self.subkeys = subkeys
self.count = count

38
veilid-python/veilid/types.py
View File

@ -1,8 +1,8 @@
import base64
import json
from enum import StrEnum
from typing import Any, Optional, Self, Tuple
from functools import total_ordering
from typing import Any, Optional, Self, Tuple
####################################################################
@ -83,6 +83,7 @@ class SafetySelectionKind(StrEnum):
UNSAFE = "Unsafe"
SAFE = "Safe"
####################################################################
@ -249,7 +250,12 @@ class VeilidVersion:
return False
def __eq__(self, other):
return isinstance(other, VeilidVersion) and self.data == other.data and self.seq == other.seq and self.writer == other.writer
return (
isinstance(other, VeilidVersion)
and self.data == other.data
and self.seq == other.seq
and self.writer == other.writer
)
@property
def major(self):
@ -319,8 +325,7 @@ class DHTSchema:
if DHTSchemaKind(j["kind"]) == DHTSchemaKind.SMPL:
return cls.smpl(
j["o_cnt"],
[DHTSchemaSMPLMember.from_json(member)
for member in j["members"]],
[DHTSchemaSMPLMember.from_json(member) for member in j["members"]],
)
raise Exception("Unknown DHTSchema kind", j["kind"])
@ -351,8 +356,7 @@ class DHTRecordDescriptor:
return cls(
TypedKey(j["key"]),
PublicKey(j["owner"]),
None if j["owner_secret"] is None else SecretKey(
j["owner_secret"]),
None if j["owner_secret"] is None else SecretKey(j["owner_secret"]),
DHTSchema.from_json(j["schema"]),
)
@ -387,7 +391,12 @@ class ValueData:
return False
def __eq__(self, other):
return isinstance(other, ValueData) and self.data == other.data and self.seq == other.seq and self.writer == other.writer
return (
isinstance(other, ValueData)
and self.data == other.data
and self.seq == other.seq
and self.writer == other.writer
)
@classmethod
def from_json(cls, j: dict) -> Self:
@ -403,13 +412,20 @@ class ValueData:
####################################################################
class SafetySpec:
preferred_route: Optional[RouteId]
hop_count: int
stability: Stability
sequencing: Sequencing
def __init__(self, preferred_route: Optional[RouteId], hop_count: int, stability: Stability, sequencing: Sequencing):
def __init__(
self,
preferred_route: Optional[RouteId],
hop_count: int,
stability: Stability,
sequencing: Sequencing,
):
self.preferred_route = preferred_route
self.hop_count = hop_count
self.stability = stability
@ -417,10 +433,12 @@ class SafetySpec:
@classmethod
def from_json(cls, j: dict) -> Self:
return cls(RouteId(j["preferred_route"]) if "preferred_route" in j else None,
return cls(
RouteId(j["preferred_route"]) if "preferred_route" in j else None,
j["hop_count"],
Stability(j["stability"]),
Sequencing(j["sequencing"]))
Sequencing(j["sequencing"]),
)
def to_json(self) -> dict:
return self.__dict__