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
|
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
from pydantic import PostgresDsn
from generalresearch.managers.gr.business import (
BusinessAddressManager,
BusinessBankAccountManager,
BusinessManager,
)
from generalresearch.managers.gr.team import MembershipManager
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
from generalresearch.managers.gr.team import TeamManager
# === Msc ===
@pytest.fixture(scope="session")
def gr_redis_config_db() -> str:
# need to update 'databases' in /etc/redis/redis.conf
# or this won't work and you'll have no indication why ...
return str(randint(99, 1_023))
@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, gr_redis_config: RedisConfig) -> TeamManager:
assert gr_db.dsn.path
assert "/unittest-" in gr_db.dsn.path
from generalresearch.managers.gr.team import TeamManager
return TeamManager(pg_config=gr_db, redis_config=gr_redis_config)
@pytest.fixture(scope="session")
def gr_token_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)
@pytest.fixture(scope="session")
def gr_membership_manager(
gr_db: PostgresConfig,
) -> MembershipManager:
return MembershipManager(pg_config=gr_db)
|
