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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
import datetime
try:
from psycopg2ct.compat import register
except ImportError:
pass
else:
register()
from pytest import mark, raises, fixture
from sqlalchemy.exc import StatementError
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy_utc import UtcDateTime, utc
Base = declarative_base()
Session = sessionmaker()
@fixture
def fx_connection(fx_engine):
connection = fx_engine.connect()
try:
transaction = connection.begin()
try:
metadata = Base.metadata
metadata.create_all(bind=connection)
yield connection
finally:
transaction.rollback()
finally:
connection.close()
@fixture
def fx_session(fx_connection):
session = Session(bind=fx_connection)
try:
yield session
finally:
session.close()
class UtcDateTimeTable(Base):
time = Column(UtcDateTime, primary_key=True)
__tablename__ = 'tb_utc_datetime'
class FixedOffset(datetime.tzinfo):
zero = datetime.timedelta(0)
def __init__(self, offset, name):
self.offset = offset
self.name = name
def utcoffset(self, _):
return self.offset
def tzname(self, _):
return self.name
def dst(self, _):
return self.zero
@mark.parametrize('tzinfo', [
utc,
FixedOffset(datetime.timedelta(hours=9), 'KST'),
])
def test_utc_datetime(fx_session, tzinfo):
aware_time = datetime.datetime.now(tzinfo).replace(microsecond=0)
e = UtcDateTimeTable(time=aware_time)
fx_session.add(e)
fx_session.flush()
saved_time, = fx_session.query(UtcDateTimeTable.time).one()
assert saved_time == aware_time
if fx_session.bind.dialect.name in ('sqlite', 'mysql'):
zero = datetime.timedelta(0)
assert saved_time.tzinfo.utcoffset(aware_time) == zero
assert saved_time.tzinfo.dst(aware_time) in (None, zero)
def test_utc_datetime_naive(fx_session):
with raises((ValueError, StatementError)):
a = UtcDateTimeTable(time=datetime.datetime.now())
fx_session.add(a)
fx_session.flush()
def test_utc_datetime_type(fx_session):
with raises((TypeError, StatementError)):
a = UtcDateTimeTable(time=str(datetime.datetime.now()))
fx_session.add(a)
fx_session.flush()
|
