"""Niezmiennik: KAŻDE wyjście HTTP w dół niesie token międzywarstwowy (LOG-32) oraz klucz szyfrujący łącze (PRE-16). Powód istnienia tego testu: token dodano do klienta na gałęzi, która odbiła się od mastera zanim powstały metody `prompt()` i `horoscope()`. Git zmergował obie zmiany czysto (różne linie), ale nowe metody wyszły BEZ tokenu — i dostawały 401 dopiero na produkcji. Zwykły test jednej metody by tego nie złapał, więc sprawdzamy regułę strukturalnie. Po dołożeniu szyfrowania ta sama klasa błędu ma gorszy objaw: wywołanie bez `link=` nie wywala się widocznie, tylko po cichu wysyła treść JAWNYM tekstem. Dlatego dla wywołań przez `link_crypto` wymagamy obu argumentów naraz. """ import ast import pathlib CLIENT = pathlib.Path(__file__).resolve().parents[1] / "app" / "clients" / "logic_client.py" HTTP_VERBS = ("post", "get", "put", "patch", "delete") # Wszystkie drogi w dół przez link_crypto — także strumień postępu (stream_lines), # który dołożono po scaleniu okna postępu z szyfrowaniem. Każda musi nieść i token, # i klucz łącza. LINK_CALLS = ("call", "call_json", "stream_lines") def _http_calls(path: pathlib.Path) -> list[tuple[str, int, bool, bool]]: """(opis, linia, czy_ma_headers, czy_wymaga_i_ma_link) dla każdego wyjścia w dół.""" tree = ast.parse(path.read_text(encoding="utf-8")) out = [] for node in ast.walk(tree): if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): continue target = node.func.value has_headers = any(kw.arg == "headers" for kw in node.keywords) has_link = any(kw.arg == "link" for kw in node.keywords) if (node.func.attr in HTTP_VERBS and isinstance(target, ast.Name) and target.id == "client"): # Surowe wywołanie httpx — omija szyfrowanie, więc dopuszczalne tylko # dla ścieżek wyjętych spod niego (patrz link_crypto.PUBLIC_PATHS). out.append((f"client.{node.func.attr}()", node.lineno, has_headers, True)) elif (node.func.attr in LINK_CALLS and isinstance(target, ast.Name) and target.id == "link_crypto"): out.append((f"link_crypto.{node.func.attr}()", node.lineno, has_headers, has_link)) return out def test_client_module_exists(): assert CLIENT.is_file() def test_every_outbound_call_sends_auth_header(): calls = _http_calls(CLIENT) assert calls, "nie znaleziono żadnego wywołania HTTP — test przestał cokolwiek pilnować" missing = [f"{CLIENT.name}:{line} {what}" for what, line, ok, _ in calls if not ok] assert not missing, ( "Wywołania w dół bez tokenu międzywarstwowego (dostaną 401 przy włączonej " "ochronie): " + ", ".join(missing) ) def test_every_outbound_call_passes_link_key(): """Brak `link=` nie boli od razu — po prostu treść leci jawnym tekstem.""" calls = _http_calls(CLIENT) missing = [f"{CLIENT.name}:{line} {what}" for what, line, _, ok in calls if not ok] assert not missing, ( "Wywołania w dół bez klucza łącza — poszłyby NIEZASZYFROWANE: " + ", ".join(missing) ) def test_raw_http_calls_only_on_paths_exempt_from_encryption(): """Surowe `client.get/post` wolno wołać wyłącznie tam, gdzie szyfrowania nie ma z założenia (`/health` dla sond k8s). Każde inne to obejście PRE-16.""" import re from app import link_crypto source = CLIENT.read_text(encoding="utf-8").splitlines() offenders = [] for what, line, _, _ in _http_calls(CLIENT): if not what.startswith("client."): continue url = re.search(r'f"\{self\.base_url\}([^"]*)"', source[line - 1]) if url is None or url.group(1) not in link_crypto.PUBLIC_PATHS: offenders.append(f"{CLIENT.name}:{line} {what}") assert not offenders, ( "Surowe wywołania HTTP poza ścieżkami wyjętymi spod szyfrowania: " + ", ".join(offenders) ) def test_auth_headers_helper_is_lazy(): """Token czytany przy wywołaniu, nie przy imporcie — inaczej pod wystartowałby z pustym tokenem, gdyby zmienna pojawiła się później.""" import os from app.clients.logic_client import _auth_headers old = os.environ.get("INTERNAL_TOKEN") try: os.environ["INTERNAL_TOKEN"] = "abc" assert _auth_headers() == {"X-Astrololo-Token": "abc"} os.environ.pop("INTERNAL_TOKEN") assert _auth_headers() == {} # ochrona wyłączona = brak nagłówka finally: if old is not None: os.environ["INTERNAL_TOKEN"] = old else: os.environ.pop("INTERNAL_TOKEN", None) # --------------------------------------------- wybor modelu musi dojsc do logiki # Ta sama klasa bledu co przy tokenie: dokladajac nowa sciezke latwo zapomniec # przekazac parametr, a objaw (cichy powrot do modelu domyslnego) jest niewidoczny. def _calls_to(path_fragment: str) -> list[int]: """Linie wywolan logic.(...) w handlerach prezentacji.""" main = CLIENT.parent.parent / "main.py" tree = ast.parse(main.read_text(encoding="utf-8")) out = [] for node in ast.walk(tree): if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == path_fragment and isinstance(node.func.value, ast.Name) and node.func.value.id == "logic"): out.append(node.lineno) return out def _has_kwarg(main_src: str, lineno: int, name: str) -> bool: tree = ast.parse(main_src) for node in ast.walk(tree): if isinstance(node, ast.Call) and node.lineno == lineno: return any(kw.arg == name for kw in node.keywords) return False def test_every_llm_call_passes_selected_model(): main = CLIENT.parent.parent / "main.py" src = main.read_text(encoding="utf-8") missing = [] for method in ("prompt", "horoscope"): for line in _calls_to(method): if not _has_kwarg(src, line, "model"): missing.append(f"main.py:{line} logic.{method}()") assert not missing, ( "Wywołania bez wybranego modelu — po cichu użyją domyślnego: " + ", ".join(missing) ) def test_every_llm_call_passes_provider(): main = CLIENT.parent.parent / "main.py" src = main.read_text(encoding="utf-8") missing = [] for method in ("prompt", "horoscope"): for line in _calls_to(method): if not _has_kwarg(src, line, "provider"): missing.append(f"main.py:{line} logic.{method}()") assert not missing, "Wywołania bez dostawcy: " + ", ".join(missing)