86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
import pytest
|
|
from psycopg.conninfo import conninfo_to_dict
|
|
|
|
from app.infra.db.project_routing import (
|
|
ActiveProjectRouting,
|
|
activate_project_routing,
|
|
get_active_project_routing,
|
|
get_project_pgconn_string,
|
|
get_project_timescale_pgconn_string,
|
|
)
|
|
|
|
|
|
def _routing(project_code: str = "project_a") -> ActiveProjectRouting:
|
|
return ActiveProjectRouting(
|
|
project_code=project_code,
|
|
business_dsn=(
|
|
"postgresql://biz_user:biz_password@biz.example:5432/biz_database"
|
|
"?sslmode=require"
|
|
),
|
|
timescale_dsn=(
|
|
"postgresql://ts_user:ts_password@timescale.example:5433/ts_database"
|
|
"?sslmode=require"
|
|
),
|
|
)
|
|
|
|
|
|
def test_project_database_uses_exact_routing_dsn_for_project_code() -> None:
|
|
routing = _routing()
|
|
|
|
with activate_project_routing(routing):
|
|
assert get_project_pgconn_string("project_a") == routing.business_dsn
|
|
assert (
|
|
get_project_timescale_pgconn_string("project_a")
|
|
== routing.timescale_dsn
|
|
)
|
|
|
|
|
|
def test_business_template_keeps_server_and_timescale_ignores_legacy_db_name() -> None:
|
|
with activate_project_routing(_routing()):
|
|
business = conninfo_to_dict(get_project_pgconn_string("project_a_template"))
|
|
timescale = conninfo_to_dict(
|
|
get_project_timescale_pgconn_string("temporary_scheme")
|
|
)
|
|
|
|
assert business == {
|
|
"user": "biz_user",
|
|
"password": "biz_password",
|
|
"dbname": "project_a_template",
|
|
"host": "biz.example",
|
|
"port": "5432",
|
|
"sslmode": "require",
|
|
}
|
|
assert timescale == {
|
|
"user": "ts_user",
|
|
"password": "ts_password",
|
|
"dbname": "ts_database",
|
|
"host": "timescale.example",
|
|
"port": "5433",
|
|
"sslmode": "require",
|
|
}
|
|
|
|
|
|
def test_project_routing_is_nested_and_request_local() -> None:
|
|
first = _routing("project_a")
|
|
second = _routing("project_b")
|
|
|
|
assert get_active_project_routing() is None
|
|
with activate_project_routing(first):
|
|
assert get_active_project_routing() is first
|
|
with activate_project_routing(second):
|
|
assert get_active_project_routing() is second
|
|
assert get_active_project_routing() is first
|
|
assert get_active_project_routing() is None
|
|
|
|
|
|
def test_timescale_access_requires_iot_routing_in_project_request() -> None:
|
|
business_only = _routing()
|
|
business_only = ActiveProjectRouting(
|
|
project_code=business_only.project_code,
|
|
business_dsn=business_only.business_dsn,
|
|
)
|
|
|
|
with activate_project_routing(business_only):
|
|
with pytest.raises(RuntimeError, match="TimescaleDB routing is not configured"):
|
|
get_project_timescale_pgconn_string()
|