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
144
145
|
from __future__ import annotations
import subprocess
from collections.abc import Callable, Generator
from random import randint
from typing import TYPE_CHECKING
import pytest
import redis
import redis.asyncio as redis_async
from pydantic import PostgresDsn
from redis import Redis
from generalresearch.managers.gr.business import (
BusinessAddressManager,
BusinessBankAccountManager,
BusinessManager,
)
from generalresearch.pg_helper import PostgresConfig
from generalresearch.redis_helper import RedisConfig
if TYPE_CHECKING:
from generalresearch.config import GRLBaseSettings
from generalresearch.managers.gr.authentication import GRTokenManager, GRUserManager
# === Msc ===
@pytest.fixture(scope="session")
def gr_redis_config_db() -> str:
return str(randint(99, 1_023))
@pytest.fixture(scope="session")
def gr_redis(settings: GRLBaseSettings) -> Redis:
assert "unittest" in str(settings.testing_redis) or "127.0.0.1" in str(
settings.testing_redis
)
return Redis.from_url(
url=str(settings.gr_redis),
decode_responses=True,
socket_timeout=settings.redis_timeout,
socket_connect_timeout=settings.redis_timeout,
)
@pytest.fixture
def gr_redis_async(settings: GRLBaseSettings) -> redis_async.Redis:
assert "unittest" in str(settings.testing_redis) or "127.0.0.1" in str(
settings.testing_redis
)
return redis_async.Redis.from_url(
str(settings.testing_redis),
decode_responses=True,
socket_timeout=0.20,
socket_connect_timeout=0.20,
)
@pytest.fixture(scope="session")
def gr_redis_config(
settings: GRLBaseSettings, gr_redis_config_db: str
) -> Generator[RedisConfig]:
assert "unittest" in str(settings.testing_redis) or "127.0.0.1" in str(
settings.testing_redis
)
uri = f"redis://{settings.testing_redis}/{gr_redis_config_db}"
res = subprocess.run(
["redis-cli", "-u", uri, "SET", "jenkins_lock", "1", "NX", "EX", "3600"],
check=True,
text=True,
capture_output=True,
)
if res.stdout.strip() != "OK":
raise ValueError("Redis already locked... aborting.")
yield RedisConfig(
dsn=uri,
decode_responses=True,
socket_timeout=settings.redis_timeout,
socket_connect_timeout=settings.redis_timeout,
)
r = redis.from_url(uri)
r.flushdb()
@pytest.fixture(scope="session")
def gr_db(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig:
_dsn = django_db_factory("gr.common")
return PostgresConfig(
dsn=_dsn,
connect_timeout=1,
statement_timeout=5,
)
# === Managers ===
@pytest.fixture(scope="session")
def gr_user_manager(
gr_db: PostgresConfig, gr_redis_config: RedisConfig
) -> GRUserManager:
assert gr_db.dsn.path
assert "/unittest-" in gr_db.dsn.path
from generalresearch.managers.gr.authentication import GRUserManager
return GRUserManager(pg_config=gr_db, redis_config=gr_redis_config)
@pytest.fixture(scope="session")
def gr_team_manager(gr_db: PostgresConfig) -> GRTokenManager:
assert gr_db.dsn.path
assert "/unittest-" in gr_db.dsn.path
from generalresearch.managers.gr.authentication import GRTokenManager
return GRTokenManager(pg_config=gr_db)
@pytest.fixture(scope="session")
def gr_business_manager(
gr_db: PostgresConfig, gr_redis_config: RedisConfig
) -> BusinessManager:
return BusinessManager(pg_config=gr_db, redis_config=gr_redis_config)
@pytest.fixture(scope="session")
def gr_business_bank_account_manager(
gr_db: PostgresConfig,
) -> BusinessBankAccountManager:
return BusinessBankAccountManager(pg_config=gr_db)
@pytest.fixture(scope="session")
def gr_business_address_manager(
gr_db: PostgresConfig,
) -> BusinessAddressManager:
return BusinessAddressManager(pg_config=gr_db)
|
