Thinh Lam
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""
|
|
``resolve_initiative_for_draft_case_key`` — evidence URLs may use ``sub-…`` or ``SUB-…`` instead of ``Initiative.case_code``.
|
|
|
|
Set INITIATIVE_DATABASE_URL to run (same as tests.test_applications_db_integration).
|
|
|
|
Run:
|
|
cd be0 && python -m unittest tests.test_evidence_initiative_resolution -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
_RUN_DB = os.getenv("INITIATIVE_DATABASE_URL", "").strip().lower().startswith("postgresql")
|
|
|
|
|
|
@unittest.skipUnless(
|
|
_RUN_DB,
|
|
"Set INITIATIVE_DATABASE_URL=postgresql+asyncpg://.../initiatives to run DB integration tests",
|
|
)
|
|
class EvidenceInitiativeResolutionTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self) -> None:
|
|
from src.initiative_db import engine as eng
|
|
|
|
await eng.dispose_engine()
|
|
await eng.init_engine()
|
|
|
|
async def asyncTearDown(self) -> None:
|
|
from src.initiative_db import engine as eng
|
|
|
|
await eng.dispose_engine()
|
|
|
|
async def test_resolves_by_public_submission_id_case_insensitive(self) -> None:
|
|
from sqlalchemy import delete
|
|
|
|
from src.initiative_db.engine import get_session
|
|
from src.initiative_db.models import Draft, Initiative, User
|
|
from src.initiative_db.submissions import resolve_initiative_for_draft_case_key
|
|
|
|
owner_id = uuid.uuid4()
|
|
owner_email = f"evtest-{owner_id.hex[:8]}@ump.edu.vn"
|
|
case_code = f"EVCASE-{uuid.uuid4().hex[:10]}"
|
|
submission_id = f"sub-{uuid.uuid4().hex[:16]}"
|
|
|
|
async with get_session() as session:
|
|
session.add(
|
|
User(
|
|
id=owner_id,
|
|
email=owner_email,
|
|
password_hash="-",
|
|
full_name="Evidence resolver test",
|
|
)
|
|
)
|
|
await session.flush()
|
|
ini = Initiative(
|
|
case_code=case_code,
|
|
owner_id=owner_id,
|
|
status="submitted",
|
|
submitted_at=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
|
|
)
|
|
session.add(ini)
|
|
await session.flush()
|
|
payload = {
|
|
"caseId": case_code,
|
|
"updatedAt": "2026-01-01T12:00:00Z",
|
|
"tabs": {},
|
|
"submissionRecord": {
|
|
"id": submission_id,
|
|
"submittedDate": "2026-01-01T12:00:00.000Z",
|
|
"name": "Test",
|
|
"author": {"id": case_code, "name": "T", "email": owner_email},
|
|
"status": "submitted",
|
|
"reviewStatus": "not_reviewed",
|
|
},
|
|
}
|
|
session.add(
|
|
Draft(
|
|
draft_code=f"DRAFT-{case_code}",
|
|
initiative_id=ini.id,
|
|
payload=payload,
|
|
version=1,
|
|
)
|
|
)
|
|
await session.commit()
|
|
ini_id = ini.id
|
|
|
|
async with get_session() as session:
|
|
upper_alias = "SUB-" + submission_id.split("-", 1)[1]
|
|
r1 = await resolve_initiative_for_draft_case_key(session, submission_id)
|
|
r2 = await resolve_initiative_for_draft_case_key(session, upper_alias)
|
|
self.assertIsNotNone(r1)
|
|
self.assertIsNotNone(r2)
|
|
assert r1 is not None and r2 is not None
|
|
self.assertEqual(r1.case_code, case_code)
|
|
self.assertEqual(r2.case_code, case_code)
|
|
|
|
async with get_session() as session:
|
|
await session.execute(delete(Draft).where(Draft.initiative_id == ini_id))
|
|
await session.execute(delete(Initiative).where(Initiative.id == ini_id))
|
|
await session.execute(delete(User).where(User.id == owner_id))
|
|
await session.commit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|