56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
|
"""Second pass: auth + stale advisor imports in sprint tests."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[2] / "tests"
|
||
|
|
|
||
|
|
LOGIN_BLOCK = re.compile(
|
||
|
|
r"def login_token\(username: str, password: str\) -> str:\n"
|
||
|
|
r" response = client\.post\(\n"
|
||
|
|
r' "/api/v1/auth/login",\n'
|
||
|
|
r' json=\{"username": username, "password": password\},\n'
|
||
|
|
r" \)\n"
|
||
|
|
r" assert response\.status_code == 200\n"
|
||
|
|
r' return response\.json\(\)\["data"\]\["access_token"\]\n',
|
||
|
|
re.MULTILINE,
|
||
|
|
)
|
||
|
|
|
||
|
|
LOGIN_INLINE = re.compile(
|
||
|
|
r'client\.post\(\s*\n?\s*"/api/v1/auth/login",\s*\n?\s*json=\{"username": "[^"]+", "password": "[^"]+"\},?\s*\)',
|
||
|
|
re.MULTILINE,
|
||
|
|
)
|
||
|
|
|
||
|
|
REPLACEMENTS = [
|
||
|
|
("from app.config.database import AgentSessionLocal", "from app.advisor_db import AgentSessionLocal"),
|
||
|
|
("from app.config.database import agent_engine", ""),
|
||
|
|
("from app.model.entities import ComplianceRule", "from app.model.entities_advisor import ComplianceRule"),
|
||
|
|
("app.config.database", "app.advisor_db"),
|
||
|
|
]
|
||
|
|
|
||
|
|
for path in sorted(ROOT.glob("test_sprint*.py")) + [ROOT / "test_demo_kyc_advisor_mapping.py"]:
|
||
|
|
if not path.exists():
|
||
|
|
continue
|
||
|
|
text = path.read_text(encoding="utf-8")
|
||
|
|
orig = text
|
||
|
|
for old, new in REPLACEMENTS:
|
||
|
|
text = text.replace(old, new)
|
||
|
|
text = LOGIN_BLOCK.sub(
|
||
|
|
"from tests.advisor_test_utils import login_staff_token\n\n"
|
||
|
|
"def login_token(username: str, password: str) -> str:\n"
|
||
|
|
" from tests.advisor_test_utils import STAFF_ADVISOR, STAFF_COMPLIANCE\n"
|
||
|
|
" actor = STAFF_COMPLIANCE if username == \"compliance_test\" else STAFF_ADVISOR\n"
|
||
|
|
" return login_staff_token(client, actor_id=actor)\n",
|
||
|
|
text,
|
||
|
|
)
|
||
|
|
|
||
|
|
def _inline_login(match: re.Match[str]) -> str:
|
||
|
|
return 'client.post("/api/auth/login", json={"actor_id": "STAFF-10086", "token_type": "staff"})'
|
||
|
|
|
||
|
|
text = LOGIN_INLINE.sub(_inline_login, text)
|
||
|
|
text = re.sub(r"\n\n\n+", "\n\n", text)
|
||
|
|
if text != orig:
|
||
|
|
path.write_text(text, encoding="utf-8")
|
||
|
|
print("updated", path.name)
|