mirror of
https://github.com/superdesigndev/treg.git
synced 2026-10-02 03:24:35 +08:00
test: stop tests leaking cwd, logging config and env to later tests
- test_cmd_run_linux_hands_off_to_treg_run_user let the isolated hand-off chdir to /tmp with exec stubbed out; record the cwd with monkeypatch.chdir so it is restored. - test_call_response_limits started uvicorn with its default log config, leaving uvicorn.error at CRITICAL and hiding the fault the analytics uvicorn test expects; pass log_config=None. - scripts/dump_surface.py rewrote TREG_DATABASE_URL when imported by the surface snapshot test; skip its env setup once treg is loaded. - Import verified_signup from conftest, not tests.conftest, which ran conftest a second time. - A teardown hook fails any test that leaves the working directory changed, so the next leak points at its cause.
This commit is contained in:
@@ -6,13 +6,19 @@ from __future__ import annotations
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _configure_test_environment() -> None:
|
||||
"""Match tests/conftest.py before importing treg, whose settings load at import time."""
|
||||
"""Match tests/conftest.py before importing treg, whose settings load at import time.
|
||||
|
||||
Imported by the test suite, treg is already loaded with conftest's environment; rewriting it
|
||||
here would only leave the process's variables disagreeing with the engine it already built."""
|
||||
if "treg" in sys.modules:
|
||||
return
|
||||
worker = os.environ.get("PYTEST_XDIST_WORKER", "")
|
||||
db_dir = os.path.join(tempfile.gettempdir(), "treg-tests")
|
||||
os.makedirs(db_dir, exist_ok=True)
|
||||
|
||||
@@ -541,3 +541,21 @@ def contactout_platform(monkeypatch):
|
||||
get_settings.cache_clear()
|
||||
yield
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
_SESSION_CWD = os.getcwd()
|
||||
|
||||
|
||||
@pytest.hookimpl(wrapper=True)
|
||||
def pytest_runtest_teardown(item, nextitem):
|
||||
"""Every test ends in the directory the run started in. A test that changes it (a bare
|
||||
`os.chdir`, or product code that chdirs while a stub stands in for `exec`) makes later tests
|
||||
fail on relative paths, far from the cause and only in some orders. Runs after every fixture
|
||||
has torn down, so `monkeypatch.chdir` restores first; the leaking test errors here instead."""
|
||||
result = yield
|
||||
if os.getcwd() != _SESSION_CWD:
|
||||
leaked = os.getcwd()
|
||||
os.chdir(_SESSION_CWD)
|
||||
raise AssertionError(f"{item.nodeid} left the working directory at {leaked}; "
|
||||
"use monkeypatch.chdir so it is restored")
|
||||
return result
|
||||
|
||||
@@ -408,7 +408,7 @@ async def test_each_team_pays_full_price_once_per_question(clients: AsyncClient,
|
||||
""""Second call per TEAM": another team's first call on a question already in the archive is
|
||||
a hit at FULL price (the archive saved treg a vendor call, not the team its first price);
|
||||
only that team's own second call is a repeat."""
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
r1 = await clients.get(f"/call/{EP}?aweme_id=7&count=5")
|
||||
await archive.drain()
|
||||
live = int(r1.headers["X-Treg-Cost-Micro"])
|
||||
@@ -508,7 +508,7 @@ async def test_an_own_key_answer_is_recorded_and_served_back_free(
|
||||
|
||||
async def test_an_own_key_answer_is_the_orgs_question_and_never_another_teams(
|
||||
clients: AsyncClient, own_key_serve, monkeypatch):
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
_vendor_says(monkeypatch, OWN)
|
||||
await _own_key(clients)
|
||||
await clients.get(f"/call/{EP}?aweme_id=7&count=5")
|
||||
@@ -547,7 +547,7 @@ async def test_an_own_key_answer_is_the_orgs_question_and_never_another_teams(
|
||||
|
||||
async def test_a_judged_licence_does_not_share_an_own_key_answer_but_an_endpoint_declaration_does(
|
||||
clients: AsyncClient, own_key_serve, monkeypatch):
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
entry = catalog_store.load().by_id[EP]
|
||||
monkeypatch.setitem(entry, "cache", {"mode": "transient", "license_quote": "q",
|
||||
"source_url": "u", "checked": "2026-09-14"})
|
||||
@@ -579,7 +579,7 @@ async def test_an_own_account_answer_is_the_connections_and_a_reconnect_starts_o
|
||||
"""`scope: own_account`: the answer is about the credential's account. It is keyed to the
|
||||
connection (org + bound secrets); a public answer for the same URL is never consulted, and a
|
||||
new connection (a new secret) never sees the old one's history."""
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
entry = catalog_store.load().by_id[EP]
|
||||
# A legacy/public answer for the same URL, recorded before the scope flips.
|
||||
other = await verified_signup(clients, json={"email": "stranger@example.com"})
|
||||
@@ -629,7 +629,7 @@ async def test_the_refresh_worker_never_re_asks_a_private_question(
|
||||
|
||||
async def test_a_platform_answer_serves_an_own_key_caller_free(
|
||||
clients: AsyncClient, own_key_serve, monkeypatch):
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
other = await verified_signup(clients, json={"email": "stranger@example.com"})
|
||||
headers = {"X-Treg-Token": other.json()["token"]}
|
||||
_vendor_says(monkeypatch, PLAT)
|
||||
|
||||
@@ -125,7 +125,10 @@ async def wire(clients, monkeypatch):
|
||||
sock.bind(('127.0.0.1', 0))
|
||||
sock.listen()
|
||||
port = sock.getsockname()[1]
|
||||
server = uvicorn.Server(uvicorn.Config(app, lifespan='off', log_level='critical', ws='none'))
|
||||
# log_config=None: uvicorn's default dictConfig and log_level would reconfigure the process's
|
||||
# `uvicorn` loggers for every later test (uvicorn.error left at CRITICAL hides server faults).
|
||||
server = uvicorn.Server(uvicorn.Config(app, lifespan='off', log_config=None, access_log=False,
|
||||
ws='none'))
|
||||
task = asyncio.create_task(server.serve(sockets=[sock]))
|
||||
try:
|
||||
async with asyncio.timeout(10):
|
||||
|
||||
@@ -633,6 +633,9 @@ def test_cmd_run_linux_hands_off_to_treg_run_user(monkeypatch):
|
||||
# On Linux with local-run set up, the member never fetches the credential: it execs
|
||||
# `sudo -u treg-run <runner>` and passes its own token through the environment.
|
||||
from treg import cli as cli_mod
|
||||
# From a cwd treg-run can't traverse, the hand-off chdirs to /tmp before exec; recording the
|
||||
# cwd here makes monkeypatch restore it, since the stubbed exec returns into this process.
|
||||
monkeypatch.chdir(os.getcwd())
|
||||
monkeypatch.setattr(cli_mod.sys, "platform", "linux")
|
||||
monkeypatch.setattr(cli_mod.os.path, "exists", lambda p: p == cli_mod._RUNNER_PATH)
|
||||
seen = {}
|
||||
|
||||
@@ -249,7 +249,7 @@ async def test_billed_oauth_answers_are_the_orgs_own_in_the_archive(
|
||||
"""Metered (treg's app pays X) but fetched with the org's OWN token: keyed to the org,
|
||||
served back to the org at the repeat price, never to a stranger."""
|
||||
from sqlalchemy import select
|
||||
from tests.conftest import verified_signup
|
||||
from conftest import verified_signup
|
||||
from treg import archive
|
||||
from treg.models import ArchiveSnapshot
|
||||
await _connect_x(clients)
|
||||
|
||||
Reference in New Issue
Block a user