zhanghongyu_0626
70aa861983
- Added new modules for advisor compliance, KYC sessions, and script templates, enhancing the advisor agent's capabilities. - Implemented a comprehensive API structure under the `/api/advisor-agent` prefix, ensuring clear organization and access to new features. - Established database models and repositories for compliance rules and KYC sessions, facilitating robust data management. - Integrated exception handling and response models to improve error management and user feedback. - Updated settings to include new configurations for compliance and KYC features, ensuring flexibility and adaptability. This update significantly expands the advisor agent's functionality, providing essential tools for compliance and customer interaction while maintaining a structured API design.
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)
|