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
|
import logging
import time
from datetime import datetime, timezone
from typing import TYPE_CHECKING
import pytest
from generalresearch.managers.spectrum.survey import (
SpectrumCriteriaManager,
SpectrumSurveyManager,
)
from generalresearch.models.spectrum.survey import SpectrumSurvey
from generalresearch.sql_helper import SqlHelper
from .surveys_json import CONDITIONS, SURVEYS_JSON
if TYPE_CHECKING:
from generalresearch.config import GRLBaseSettings
@pytest.fixture(scope="session")
def spectrum_rw(settings: "GRLBaseSettings") -> SqlHelper:
logging.info(f"{settings.spectrum_rw_db=}")
assert settings.spectrum_rw_db is not None
assert "/unittest-" in settings.spectrum_rw_db.path
return SqlHelper(
dsn=settings.spectrum_rw_db,
read_timeout=2,
write_timeout=1,
connect_timeout=2,
)
@pytest.fixture(scope="session")
def spectrum_criteria_manager(spectrum_rw: SqlHelper) -> SpectrumCriteriaManager:
assert "/unittest-" in spectrum_rw.dsn.path
return SpectrumCriteriaManager(spectrum_rw)
@pytest.fixture(scope="session")
def spectrum_survey_manager(spectrum_rw: SqlHelper) -> SpectrumSurveyManager:
assert "/unittest-" in spectrum_rw.dsn.path
return SpectrumSurveyManager(spectrum_rw)
@pytest.fixture(scope="session")
def setup_spectrum_surveys(
spectrum_rw: SqlHelper, spectrum_survey_manager, spectrum_criteria_manager
) -> None:
now = datetime.now(timezone.utc)
# make sure these example surveys exist in db
surveys = [SpectrumSurvey.model_validate_json(x) for x in SURVEYS_JSON]
for s in surveys:
s.modified_api = datetime.now(tz=timezone.utc)
spectrum_survey_manager.create_or_update(surveys)
spectrum_criteria_manager.update(CONDITIONS)
# and make sure they have allocation for 687
spectrum_rw.execute_sql_query(
f"""
INSERT IGNORE INTO `{spectrum_rw.db}`.spectrum_supplier
(supplier_id, name, api_key, secret_key, username, password)
VALUES (%s, %s, %s, %s, %s, %s)""",
["687", "GRL", "x", "x", "x", "x"],
commit=True,
)
supplier687_pk = spectrum_rw.execute_sql_query(
f"""
select id from `{spectrum_rw.db}`.spectrum_supplier where supplier_id = '687'"""
)[0]["id"]
conn = spectrum_rw.make_connection()
c = conn.cursor()
c.executemany(
f"""
INSERT IGNORE INTO `{spectrum_rw.db}`.spectrum_surveysupplier
(created, surveySig, supplier_id, survey_id)
VALUES (%s, %s, %s, %s)
""",
[[now, "xxx", supplier687_pk, s.survey_id] for s in surveys],
)
conn.commit()
# Wait a second to make sure the spectrum-grpc pulls these from the db into global-vars
time.sleep(1)
|
