gitea
5c9989f05e
/chart/report przekazywał pulę do warstwy danych, ale /chart/timeline (interpret=true) i /chart/prompt — a przez niego /chart/horoscope i /chart/horoscope/stream — budowały DataClient() bez puli, czyli szukały w całym udziale. TimelineRequest miał pole tenant i je gubił; PromptRequest nie miał go wcale. PromptRequest dostaje tenant (HoroscopeRequest dziedziczy), a wszystkie trzy miejsca budują DataClient(tenant=req.tenant). Testy wołają trasy przez TestClient z prawdziwym silnikiem i szpiegiem w miejscu DataClient: pula ma dotrzeć do każdego klienta, który naprawdę pytał bazę, a bez puli — zostać pusta (pełna aplikacja). Na kodzie sprzed poprawki pada dokładnie 7 tras, które gubiły pulę; report przechodzi. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
"""Pula konta (PRE-29) dociera do warstwy danych z KAŻDEJ trasy, która w niej szuka.
|
|
|
|
Trasy są wołane przez HTTP, a DataClient podmieniony na szpiega, który zapisuje,
|
|
z jaką pulą go zbudowano i czy naprawdę o coś zapytał. Celowo nie czytamy źródła:
|
|
napis `tenant=` w kodzie nie mówi, czy pula przeżyła drogę od żądania do bazy.
|
|
A pominięta pula nie daje błędu — daje cały udział, czyli cudze pliki.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from app import link_crypto, main
|
|
|
|
POOL = "konto-a"
|
|
PERIOD = {"from_date": "2024-01-01", "to_date": "2024-06-30"}
|
|
# Nieznany dostawca kończy horoskop na LLMError zaraz po zbudowaniu promptu —
|
|
# bez sieci i bez klucza API, a warstwa danych jest już wtedy przepytana.
|
|
NO_PROVIDER = {"provider": "brak-dostawcy"}
|
|
|
|
ROUTES = [
|
|
pytest.param("/chart/report", {}, id="report"),
|
|
pytest.param("/chart/timeline", {**PERIOD, "interpret": True}, id="timeline"),
|
|
pytest.param("/chart/prompt", {"profile": "natal"}, id="prompt-natal"),
|
|
pytest.param("/chart/prompt", {"profile": "period", **PERIOD}, id="prompt-period"),
|
|
pytest.param("/chart/horoscope", {"profile": "natal", **NO_PROVIDER}, id="horoscope-natal"),
|
|
pytest.param("/chart/horoscope", {"profile": "period", **PERIOD, **NO_PROVIDER},
|
|
id="horoscope-period"),
|
|
pytest.param("/chart/horoscope/stream", {"profile": "natal", **NO_PROVIDER},
|
|
id="stream-natal"),
|
|
pytest.param("/chart/horoscope/stream", {"profile": "period", **PERIOD, **NO_PROVIDER},
|
|
id="stream-period"),
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def clients(monkeypatch, own_engine) -> list:
|
|
"""Każdy DataClient zbudowany w trakcie żądania, w kolejności tworzenia."""
|
|
built: list = []
|
|
|
|
class SpyDataClient:
|
|
def __init__(self, base_url: str | None = None, tenant: str = "") -> None:
|
|
self.tenant = tenant
|
|
self.searches = 0
|
|
built.append(self)
|
|
|
|
def search(self, key, value, exact, limit, fields=None):
|
|
self.searches += 1
|
|
return {"provider": "szpieg", "total": 0, "rows": []}
|
|
|
|
monkeypatch.setattr(main, "DataClient", SpyDataClient)
|
|
monkeypatch.setattr(main, "get_engine", lambda: own_engine)
|
|
for env in ("INTERNAL_TOKEN", link_crypto.ENV_REQUIRED, link_crypto.ENV_PRESENTATION_LOGIC):
|
|
monkeypatch.delenv(env, raising=False)
|
|
return built
|
|
|
|
|
|
def _call(path: str, body: dict, reference_moment) -> None:
|
|
payload = {"when_utc": reference_moment.when_utc.isoformat(),
|
|
"lat": reference_moment.lat, "lon": reference_moment.lon, **body}
|
|
r = TestClient(main.app).post(path, json=payload)
|
|
assert r.status_code == 200, r.text
|
|
if path.endswith("/stream"):
|
|
# strumień zawsze odpowiada 200 — o porażce mówi dopiero ostatnia ramka
|
|
last = json.loads(r.text.strip().splitlines()[-1])
|
|
assert last["type"] == "result", last
|
|
|
|
|
|
@pytest.mark.parametrize("path, body", ROUTES)
|
|
def test_account_pool_reaches_the_data_layer(clients, reference_moment, path, body):
|
|
_call(path, {**body, "tenant": POOL}, reference_moment)
|
|
|
|
# Bez choćby jednego wyszukiwania test przechodziłby na pusto.
|
|
assert any(c.searches for c in clients), "trasa nie zapytała warstwy danych"
|
|
assert {c.tenant for c in clients} == {POOL}
|
|
|
|
|
|
@pytest.mark.parametrize("path, body", ROUTES)
|
|
def test_without_a_pool_the_route_searches_the_whole_share(clients, reference_moment, path, body):
|
|
"""Pełna aplikacja nie zna kont i nie wysyła puli — dla niej nic się nie zmienia."""
|
|
_call(path, body, reference_moment)
|
|
|
|
assert any(c.searches for c in clients)
|
|
assert {c.tenant for c in clients} == {""}
|