generated from kgod/ai-review-template
88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.main import create_app
|
|
from app.offerpai_auth import OfferPaiIdentity
|
|
from app.postgres_database import PostgresDatabase
|
|
from app.services import RuleBasedEntryExpander, RuleBasedExperienceExtractor, RuleBasedResumeRewriter
|
|
from app.settings import Settings
|
|
|
|
|
|
class StaticIdentityProvider:
|
|
def authenticate(self, _token: str) -> OfferPaiIdentity:
|
|
return OfferPaiIdentity(
|
|
user_id="2081575100391407617",
|
|
mobile_number="13421012384",
|
|
nick="用户2384",
|
|
)
|
|
|
|
|
|
def test_create_app_uses_postgres_when_database_path_is_not_supplied(monkeypatch) -> None:
|
|
schema = f"test_runtime_{uuid4().hex}"
|
|
database_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]
|
|
monkeypatch.setenv("RESUME_AGENT_DATABASE_SCHEMA", schema)
|
|
monkeypatch.setenv("RESUME_AGENT_DEFAULT_TIER", "free")
|
|
application = create_app(
|
|
extractor=RuleBasedExperienceExtractor(),
|
|
rewriter=RuleBasedResumeRewriter(),
|
|
expander=RuleBasedEntryExpander(),
|
|
settings=Settings(
|
|
llm_provider="rule",
|
|
database_url=database_url,
|
|
offerpai_auth_required=False,
|
|
),
|
|
)
|
|
try:
|
|
assert isinstance(application.state.database, PostgresDatabase)
|
|
with TestClient(application) as client:
|
|
response = client.post("/ai-api/resume-agent/sessions", json={})
|
|
assert response.status_code == 201
|
|
session_id = response.json()["session_id"]
|
|
assert application.state.database.get_session(session_id) is not None
|
|
finally:
|
|
application.state.database.engine.dispose()
|
|
engine = create_engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_postgres_restores_session_by_external_user_id(monkeypatch) -> None:
|
|
schema = f"test_external_identity_{uuid4().hex}"
|
|
database_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]
|
|
monkeypatch.setenv("RESUME_AGENT_DATABASE_SCHEMA", schema)
|
|
application = create_app(
|
|
extractor=RuleBasedExperienceExtractor(),
|
|
rewriter=RuleBasedResumeRewriter(),
|
|
expander=RuleBasedEntryExpander(),
|
|
settings=Settings(
|
|
llm_provider="rule",
|
|
database_url=database_url,
|
|
offerpai_auth_required=True,
|
|
),
|
|
offerpai_identity_provider=StaticIdentityProvider(),
|
|
)
|
|
headers = {"Authorization": "Bearer header.payload.signature-value"}
|
|
try:
|
|
with TestClient(application) as client:
|
|
first = client.post("/ai-api/resume-agent/sessions", json={}, headers=headers)
|
|
second = client.post("/ai-api/resume-agent/sessions", json={}, headers=headers)
|
|
assert first.status_code == 201
|
|
assert second.status_code == 201
|
|
assert second.json()["session_id"] == first.json()["session_id"]
|
|
finally:
|
|
application.state.database.engine.dispose()
|
|
engine = create_engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
|
|
finally:
|
|
engine.dispose()
|