selfrelease
65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
"""导入管线集成/幂等性测试 — 对应 T-1.2 UT/E2E。"""
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
|
|
|
from app.etl import importer
|
|
|
|
|
|
class TestHeaderDetection(unittest.TestCase):
|
|
def test_find_header_row_skips_title(self):
|
|
rows = [["全国电力机车型号表"], [], ["型号", "首产时间", "停产时间", "生产商"]]
|
|
self.assertEqual(importer.find_header_row(rows), 2)
|
|
|
|
def test_build_column_map(self):
|
|
header = ["系列", "型号", "首产时间", "未知列X"]
|
|
cmap = importer.build_column_map(header)
|
|
self.assertEqual(cmap[0], "series")
|
|
self.assertEqual(cmap[1], "model_code")
|
|
self.assertEqual(cmap[2], "first_year")
|
|
self.assertTrue(cmap[3].startswith("raw::"))
|
|
|
|
|
|
class TestFullImport(unittest.TestCase):
|
|
"""跑真实 12 表导入，断言结果与幂等性。"""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp()
|
|
self.db = os.path.join(self.tmp, "t.db")
|
|
|
|
def test_import_and_idempotent(self):
|
|
if not os.path.isdir(importer.CSV_DIR):
|
|
self.skipTest("CSV 目录不存在")
|
|
r1 = importer.import_all(db_path=self.db)
|
|
self.assertGreater(r1["models"], 0)
|
|
self.assertGreater(r1["units"], 0)
|
|
conn = sqlite3.connect(self.db)
|
|
n1 = conn.execute("SELECT COUNT(*) FROM model").fetchone()[0]
|
|
conn.close()
|
|
# 再次导入：应重建而非翻倍（幂等）
|
|
importer.import_all(db_path=self.db)
|
|
conn = sqlite3.connect(self.db)
|
|
n2 = conn.execute("SELECT COUNT(*) FROM model").fetchone()[0]
|
|
conn.close()
|
|
self.assertEqual(n1, n2)
|
|
|
|
def test_numeric_split_persisted(self):
|
|
if not os.path.isdir(importer.CSV_DIR):
|
|
self.skipTest("CSV 目录不存在")
|
|
importer.import_all(db_path=self.db)
|
|
conn = sqlite3.connect(self.db)
|
|
row = conn.execute(
|
|
"SELECT max_speed_value, max_speed_unit FROM model "
|
|
"WHERE max_speed_value IS NOT NULL LIMIT 1").fetchone()
|
|
conn.close()
|
|
self.assertIsNotNone(row)
|
|
self.assertIsInstance(row[0], float)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|