1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
import pytest
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy.orm import declarative_base
from geoalchemy2 import Geometry
from geoalchemy2 import WKBElement
from geoalchemy2 import WKTElement
metadata = MetaData()
Base = declarative_base(metadata=metadata)
class PickledLake(Base): # type: ignore
__tablename__ = "pickled_lake"
id = Column(Integer, primary_key=True)
geom = Column(Geometry(geometry_type="LINESTRING", srid=4326))
def __init__(self, geom):
self.geom = geom
class TestPickle:
@pytest.fixture
def setup_one_lake(self, session):
conn = session.bind
metadata.drop_all(conn, checkfirst=True)
metadata.create_all(conn)
lake = PickledLake(WKTElement("LINESTRING(0 0,1 1)", srid=4326))
session.add(lake)
session.flush()
session.expire(lake)
yield lake.id
session.rollback()
metadata.drop_all(session.bind, checkfirst=True)
def test_pickle_unpickle(self, session, setup_one_lake, dialect_name):
import pickle
lake_id = setup_one_lake
lake = session.get(PickledLake, lake_id)
assert isinstance(lake.geom, WKBElement)
data_desc = str(lake.geom)
pickled = pickle.dumps(lake)
unpickled = pickle.loads(pickled)
assert unpickled.geom.srid == 4326
assert str(unpickled.geom) == data_desc
if dialect_name == "mysql":
assert unpickled.geom.extended is False
else:
assert unpickled.geom.extended is True
|
