Files
specklepy/tests/integration/test_serialization.py
KatKatKateryna 57e176af91 typo
2024-02-12 15:23:37 +00:00

99 lines
3.9 KiB
Python

import json
import pytest
from specklepy.api import operations
from specklepy.objects import Base
from specklepy.objects.fakemesh import FakeMesh
from specklepy.objects.geometry import Point
from specklepy.transports.memory import MemoryTransport
from specklepy.transports.server import ServerTransport
@pytest.mark.run(order=5)
class TestSerialization:
def test_serialize(self, base):
serialized = operations.serialize(base)
deserialized = operations.deserialize(serialized)
assert base.get_id() == deserialized.get_id()
assert base.units == "millimetres"
assert isinstance(base.test_bases[0], Base)
assert base["@revit_thing"].speckle_type == "SpecialRevitFamily"
assert base["@detach"].name == deserialized["@detach"].name
def test_detaching(self, mesh):
transport = MemoryTransport()
serialized = operations.serialize(mesh, [transport])
deserialized = operations.deserialize(serialized, transport)
serialized_dict = json.loads(serialized)
assert serialized_dict["detach_this"]["speckle_type"] == "reference"
assert serialized_dict["@detach"]["speckle_type"] == "reference"
assert serialized_dict["origin"]["speckle_type"] == "reference"
assert serialized_dict["@detached_list"][-1]["speckle_type"] == "reference"
assert mesh.get_id() == deserialized.get_id()
def test_chunking(self, mesh):
transport = MemoryTransport()
serialized = operations.serialize(mesh, [transport])
deserialized = operations.deserialize(serialized, transport)
serialized_dict = json.loads(serialized)
assert len(serialized_dict["vertices"]) == 3
assert len(serialized_dict["@(100)colours"]) == 3
assert len(serialized_dict["@()default_chunk"]) == 1
assert serialized_dict["vertices"][0]["speckle_type"] == "reference"
assert serialized_dict["@(100)colours"][0]["speckle_type"] == "reference"
assert serialized_dict["@()default_chunk"][0]["speckle_type"] == "reference"
assert mesh.get_id() == deserialized.get_id()
def test_send_and_receive(self, client, sample_stream, mesh):
transport = ServerTransport(client=client, stream_id=sample_stream.id)
hash = operations.send(mesh, transports=[transport])
# also try constructing server transport with token and url
transport = ServerTransport(
stream_id=sample_stream.id, token=client.account.token, url=client.url
)
# use a fresh memory transport to force receiving from remote
received = operations.receive(
hash, remote_transport=transport, local_transport=MemoryTransport()
)
assert isinstance(received, FakeMesh)
assert received.vertices == mesh.vertices
assert isinstance(received.origin, Point)
assert received.origin.x == mesh.origin.x
# not comparing hashes as order is not guaranteed back from server
mesh.id = hash # populate with decomposed id for use in proceeding tests
def test_receive_local(self, client, mesh):
hash = operations.send(mesh) # defaults to SQLiteTransport
received = operations.receive(hash)
assert isinstance(received, Base)
assert mesh.get_id() == received.get_id()
def test_unknown_type(self):
unknown = '{"speckle_type": "mysterious.type"}'
deserialised = operations.deserialize(unknown)
assert isinstance(deserialised, Base)
assert deserialised.speckle_type == "mysterious.type"
def test_no_speckle_type(self):
untyped = '{"foo": "bar"}'
deserialised = operations.deserialize(untyped)
assert deserialised == {"foo": "bar"}
def test_big_int(self):
big_int = '{"big": ' + str(2**64) + "}"
deserialised = operations.deserialize(big_int)
assert deserialised == {"big": 2**64}