"""Drzewo astroklienta budowane tak samo, jak buduje je Dockerfile. Ta usługa nie ma własnego kompletu plików: bierze rdzeń z warstwy prezentacji, usuwa z niego to, czego ten produkt nie ma (usun.txt), i nakłada własne. Testy muszą patrzeć na WYNIK tego złożenia, a nie na katalog `app/` w repozytorium — inaczej sprawdzałyby garść plików, a nie produkt. """ from __future__ import annotations import json import pathlib import re import shutil import subprocess import sys import pytest USLUGA = pathlib.Path(__file__).resolve().parents[1] RDZEN = USLUGA.parents[0] / "presentation" / "app" def zloz(cel: pathlib.Path) -> pathlib.Path: """Rdzeń minus usun.txt plus nakładka — dokładnie jak w Dockerfile.""" app = cel / "app" shutil.copytree(RDZEN, app, ignore=shutil.ignore_patterns("__pycache__", "*.pyc")) for linia in (USLUGA / "usun.txt").read_text(encoding="utf-8").splitlines(): sciezka = linia.strip() if not sciezka: continue do_usuniecia = app / sciezka assert do_usuniecia.exists(), f"usun.txt wskazuje nieistniejące: {sciezka}" shutil.rmtree(do_usuniecia) if do_usuniecia.is_dir() else do_usuniecia.unlink() shutil.copytree(USLUGA / "app", app, dirs_exist_ok=True, ignore=shutil.ignore_patterns("__pycache__", "*.pyc")) return cel @pytest.fixture(scope="session") def drzewo(tmp_path_factory) -> pathlib.Path: return zloz(tmp_path_factory.mktemp("astroklient")) SONDA = ''' import json, os, hashlib, secrets, re sol = secrets.token_bytes(16) dk = hashlib.scrypt(b"tajne", salt=sol, n=2**14, r=8, p=1, dklen=32) os.environ.update(ASTROKLIENT_USERS="klient:scrypt$%s$%s" % (sol.hex(), dk.hex()), SESSION_SECRET="t"*64, COOKIES_INSECURE="1", RATE_LIMIT_PER_MIN="0") from starlette.testclient import TestClient from app.main import app from app import security, session, features from app.clients import logic_client c = TestClient(app, follow_redirects=False) H = {"Cookie": f"{session.COOKIE}={security.issue_session('klient')}"} strona = c.get("/", headers=H).text zebrane = [] logic_client.link_crypto.call_json = lambda *a, **kw: (zebrane.append(kw.get("payload")), {})[1] c.post("/interpret", headers=H, data={"date":"1984-04-30","time":"11:20", "tz_offset":"2","lat":"50","lon":"19","tenant":"ktos-inny"}) print("WYNIK:" + json.dumps({ "ekrany": [f.label for f in features.SCREENS], "uprawnienia": sorted(security._uprawnienia()), "dziala": {a: c.get(a, headers=H).status_code for a in ("/", "/interpret", "/timeline", "/synastry", "/significators", "/pliki")}, "nie_ma": {a: c.get(a, headers=H).status_code for a in ("/compile", "/settings", "/accounts", "/files", "/horoscope/stream", "/compile/pdf")}, "nawigacja": re.findall(r"", strona, re.S)[0].count('(.*?)", strona) or [None, "?"])[1], "nazwa_w_tytule": (re.search(r"(.*?) ·", strona) or [None, "?"])[1], "rdzen_w_tresci": "astrololo" in strona.lower(), "dokumentacja": {a: c.get(a, headers=H).status_code for a in ("/docs", "/openapi.json", "/redoc")}, "logowanie_dziala": bool(security.verify("klient", "tajne")), "zle_haslo_odmowa": security.verify("klient", "inne") is None, })) ''' # Generowania tekstu w tym obrazie nie ma, ale most modułów zostaje — więc jego # transport musi nieść pulę z sesji tak samo jak wywołania ekranów. Osobna sonda, # bo dokłada trasę, której w produkcie nie ma: w sondzie produktu zafałszowałaby # każde pytanie o to, co produkt zawiera. SONDA_MODULU = ''' import json, os, hashlib, secrets sol = secrets.token_bytes(16) dk = hashlib.scrypt(b"tajne", salt=sol, n=2**14, r=8, p=1, dklen=32) os.environ.update(ASTROKLIENT_USERS="klient:scrypt$%s$%s" % (sol.hex(), dk.hex()), SESSION_SECRET="t"*64, COOKIES_INSECURE="1", RATE_LIMIT_PER_MIN="0") from fastapi.responses import StreamingResponse from starlette.testclient import TestClient from app.main import app, logic from app import security, session, features from app.clients import logic_client wyslane = {} def call_json(client, method, url, payload=None, **kw): wyslane[url.split("/chart/", 1)[1]] = (payload or {}).get("tenant") return {} def stream_lines(client, url, payload=None, **kw): wyslane[url.split("/chart/", 1)[1]] = (payload or {}).get("tenant") yield '{"type": "result", "result": {}}' logic_client.link_crypto.call_json = call_json logic_client.link_crypto.stream_lines = stream_lines # Trasa w miejscu modułu: woła ten sam transport co moduł generowania, a strumień # konsumuje równie leniwie — dopiero przy wysyłaniu odpowiedzi, nie w obsłudze trasy. features.ROUTES[("POST", "/sonda/modul")] = None @app.post("/sonda/modul") def sonda_modul(): podrobione = {"profile": "natal", "tenant": "ktos-inny"} logic.wywolaj("/chart/prompt", podrobione, 5) logic.wywolaj("/chart/horoscope", podrobione, 5) def relay(): yield from logic.strumien("/chart/horoscope/stream", podrobione) return StreamingResponse(relay(), media_type="application/x-ndjson") c = TestClient(app, follow_redirects=False) H = {"Cookie": f"{session.COOKIE}={security.issue_session('klient')}"} r = c.post("/sonda/modul", headers=H) print("WYNIK:" + json.dumps({"status": r.status_code, "pule": wyslane})) ''' def _uruchom(drzewo: pathlib.Path, sonda: str) -> dict: """Aplikacja uruchomiona w OSOBNYM PROCESIE. Importy Pythona są zapamiętywane w procesie, więc sprawdzanie „produktu bez modułu" tam, gdzie moduł już się zaimportował, dałoby wynik fałszywie pozytywny.""" r = subprocess.run([sys.executable, "-c", sonda], cwd=drzewo, capture_output=True, text=True, env={"PYTHONPATH": str(drzewo), "PATH": "/usr/bin:/bin"}) linia = next((l for l in r.stdout.splitlines() if l.startswith("WYNIK:")), None) assert linia, f"sonda nie wystartowała:\n{r.stdout[-2000:]}\n{r.stderr[-2000:]}" return json.loads(linia[len("WYNIK:"):]) @pytest.fixture(scope="module") def wynik(drzewo): return _uruchom(drzewo, SONDA) @pytest.fixture(scope="module") def wynik_modulu(drzewo): return _uruchom(drzewo, SONDA_MODULU)