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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
import sys
from contextlib import asynccontextmanager, contextmanager
from tempfile import TemporaryDirectory
from typing import AsyncContextManager, AsyncGenerator, ContextManager, Generator, Optional
import pytest
from apscheduler.abc import AsyncDataStore, DataStore, Serializer
from apscheduler.datastores.async_.sync_adapter import AsyncDataStoreAdapter
from apscheduler.datastores.sync.memory import MemoryDataStore
from apscheduler.serializers.cbor import CBORSerializer
from apscheduler.serializers.json import JSONSerializer
from apscheduler.serializers.pickle import PickleSerializer
if sys.version_info >= (3, 9):
from zoneinfo import ZoneInfo
else:
from backports.zoneinfo import ZoneInfo
@pytest.fixture(scope='session')
def timezone() -> ZoneInfo:
return ZoneInfo('Europe/Berlin')
@pytest.fixture(params=[
pytest.param(None, id='none'),
pytest.param(PickleSerializer, id='pickle'),
pytest.param(CBORSerializer, id='cbor'),
pytest.param(JSONSerializer, id='json')
])
def serializer(request) -> Optional[Serializer]:
return request.param() if request.param else None
@pytest.fixture
def anyio_backend() -> 'str':
return 'asyncio'
@contextmanager
def setup_memory_store() -> Generator[DataStore, None, None]:
yield MemoryDataStore()
@contextmanager
def setup_mongodb_store() -> Generator[DataStore, None, None]:
from pymongo import MongoClient
from apscheduler.datastores.sync.mongodb import MongoDBDataStore
with MongoClient(tz_aware=True, serverSelectionTimeoutMS=1000) as client:
yield MongoDBDataStore(client, start_from_scratch=True)
@contextmanager
def setup_sqlite_store() -> Generator[DataStore, None, None]:
from sqlalchemy.future import create_engine
from apscheduler.datastores.sync.sqlalchemy import SQLAlchemyDataStore
with TemporaryDirectory('sqlite_') as tempdir:
engine = create_engine(f'sqlite:///{tempdir}/test.db')
try:
yield SQLAlchemyDataStore(engine)
finally:
engine.dispose()
@contextmanager
def setup_psycopg2_store() -> Generator[DataStore, None, None]:
from sqlalchemy.future import create_engine
from apscheduler.datastores.sync.sqlalchemy import SQLAlchemyDataStore
engine = create_engine('postgresql+psycopg2://postgres:secret@localhost/testdb')
try:
yield SQLAlchemyDataStore(engine, start_from_scratch=True)
finally:
engine.dispose()
@contextmanager
def setup_mysql_store() -> Generator[DataStore, None, None]:
from sqlalchemy.future import create_engine
from apscheduler.datastores.sync.sqlalchemy import SQLAlchemyDataStore
engine = create_engine('mysql+pymysql://root:secret@localhost/testdb')
try:
yield SQLAlchemyDataStore(engine, start_from_scratch=True)
finally:
engine.dispose()
@asynccontextmanager
async def setup_asyncpg_store() -> AsyncGenerator[AsyncDataStore, None]:
from sqlalchemy.ext.asyncio import create_async_engine
from apscheduler.datastores.async_.sqlalchemy import SQLAlchemyDataStore
engine = create_async_engine('postgresql+asyncpg://postgres:secret@localhost/testdb',
future=True)
try:
yield SQLAlchemyDataStore(engine, start_from_scratch=True)
finally:
await engine.dispose()
@pytest.fixture(params=[
pytest.param(setup_memory_store, id='memory'),
pytest.param(setup_sqlite_store, id='sqlite'),
pytest.param(setup_mongodb_store, id='mongodb', marks=[pytest.mark.externaldb]),
pytest.param(setup_psycopg2_store, id='psycopg2', marks=[pytest.mark.externaldb]),
pytest.param(setup_mysql_store, id='mysql', marks=[pytest.mark.externaldb])
])
def setup_sync_store(request) -> ContextManager[DataStore]:
return request.param
@pytest.fixture(params=[
pytest.param(setup_asyncpg_store, id='asyncpg', marks=[pytest.mark.externaldb])
])
def setup_async_store(request) -> AsyncContextManager[AsyncDataStore]:
return request.param
@pytest.fixture(params=[
pytest.param(setup_memory_store, id='memory'),
pytest.param(setup_sqlite_store, id='sqlite'),
pytest.param(setup_mongodb_store, id='mongodb', marks=[pytest.mark.externaldb]),
pytest.param(setup_psycopg2_store, id='psycopg2', marks=[pytest.mark.externaldb]),
pytest.param(setup_mysql_store, id='mysql', marks=[pytest.mark.externaldb]),
pytest.param(setup_asyncpg_store, id='asyncpg', marks=[pytest.mark.externaldb])
])
async def datastore_cm(request):
cm = request.param()
if isinstance(cm, ContextManager):
with cm as store:
yield AsyncDataStoreAdapter(store)
else:
async with cm as store:
yield store
|
