92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
from src.poc4.qa_history import evaluate_sql, load_benchmark_questions
|
|
from src.poc4.qa_history_store import (
|
|
REFERENCE_STATUSES,
|
|
QaHistoryStoreError,
|
|
_choice,
|
|
_normalize_oracle_dsn,
|
|
schema_statements,
|
|
)
|
|
|
|
|
|
class QaHistoryTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
benchmark = Path(__file__).parents[1] / "config" / "smilegate_qa_benchmark.json"
|
|
cls.questions = {item.question_code: item for item in load_benchmark_questions(benchmark)}
|
|
|
|
def test_customer_excel_benchmark_contains_all_47_cases(self) -> None:
|
|
self.assertEqual(47, len(self.questions))
|
|
self.assertIn("STD-01", self.questions)
|
|
self.assertIn("CZN-19", self.questions)
|
|
|
|
def test_supported_query_passes_when_required_terms_are_present(self) -> None:
|
|
judgment = evaluate_sql(
|
|
self.questions["STD-13"],
|
|
"SELECT SUM(PAYMT_AMT) FROM COMN_SALES_TXN",
|
|
execution_succeeded=True,
|
|
)
|
|
self.assertEqual("PASS", judgment.status)
|
|
|
|
def test_monthly_au_with_au_flag_fails(self) -> None:
|
|
judgment = evaluate_sql(
|
|
self.questions["STD-27"],
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM CZN_COMN_USER_MST
|
|
WHERE AU_FLAG = 1
|
|
AND BASE_DT = (SELECT MAX(BASE_DT) FROM CZN_COMN_USER_MST)
|
|
AND LAST_CONN_DT >= ADD_MONTHS(BASE_DT, -1)
|
|
AND STD_USER_YN = 'Y'
|
|
AND EXPT_USER_YN = 'N'
|
|
""",
|
|
execution_succeeded=True,
|
|
)
|
|
self.assertEqual("FAIL", judgment.status)
|
|
self.assertIn("AU_FLAG", judgment.reason)
|
|
|
|
def test_unsupported_game_requires_safe_alias_lookup(self) -> None:
|
|
safe = evaluate_sql(
|
|
self.questions["STD-02"],
|
|
"SELECT GAME_ID FROM COMN_GAME_ALIAS_BAS WHERE GAME_NM LIKE '%버블리즈%'",
|
|
execution_succeeded=True,
|
|
)
|
|
unsafe = evaluate_sql(
|
|
self.questions["STD-02"],
|
|
"SELECT COUNT(*) FROM CZN_COMN_USER_MST WHERE GAME_ID = 'STOVE_CHAOSZERO'",
|
|
execution_succeeded=True,
|
|
)
|
|
self.assertEqual("PASS", safe.status)
|
|
self.assertEqual("FAIL", unsafe.status)
|
|
|
|
def test_free_text_is_review_not_automatic_pass(self) -> None:
|
|
judgment = evaluate_sql(None, "SELECT 1 FROM DUAL", execution_succeeded=True)
|
|
self.assertEqual("REVIEW", judgment.status)
|
|
|
|
def test_jdbc_url_wallet_is_normalized_for_python_driver(self) -> None:
|
|
self.assertEqual(
|
|
("sgmpaipoc_medium", "/home/opc/wallet/sgmpaipoc"),
|
|
_normalize_oracle_dsn(
|
|
"jdbc:oracle:thin:@sgmpaipoc_medium?TNS_ADMIN=/home/opc/wallet/sgmpaipoc"
|
|
),
|
|
)
|
|
|
|
def test_schema_defines_two_history_tables_and_indexes(self) -> None:
|
|
statements = "\n".join(schema_statements())
|
|
self.assertIn("CREATE TABLE SG_AI_QA_QUESTION", statements)
|
|
self.assertIn("CREATE TABLE SG_AI_QA_ANSWER", statements)
|
|
self.assertIn("answer_seq NUMBER GENERATED ALWAYS AS IDENTITY", statements)
|
|
|
|
def test_few_shot_status_is_normalized_and_unknown_status_is_rejected(self) -> None:
|
|
self.assertEqual("APPROVED", _choice(" approved ", REFERENCE_STATUSES, "검색 상태"))
|
|
with self.assertRaises(QaHistoryStoreError):
|
|
_choice("LIVE", REFERENCE_STATUSES, "검색 상태")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|