202 lines
8.3 KiB
Python
202 lines
8.3 KiB
Python
"""
|
||
数据库 URL 构建单元测试（D1 多库支持）
|
||
=============================================================
|
||
|
||
验证 _build_database_url() 对 mysql / postgresql / mssql / sqlite / oracle
|
||
五类驱动的 URL 构建正确，且 sqlite 无 host/port/账号分支。
|
||
"""
|
||
|
||
import config.database as db_mod
|
||
from src.common.constants.common import DEFAULT_PASSWORD
|
||
|
||
|
||
def _url(monkeypatch, driver, **overrides):
|
||
defaults = {
|
||
'DB_DRIVER': driver,
|
||
'DB_HOST': '127.0.0.1',
|
||
'DB_PORT': 1,
|
||
'DB_DATABASE': 'appdb',
|
||
'DB_USERNAME': 'user',
|
||
'DB_PASSWORD': 'pass',
|
||
'DB_SSL': False,
|
||
}
|
||
defaults.update(overrides)
|
||
for key, value in defaults.items():
|
||
monkeypatch.setattr(db_mod, key, value)
|
||
return db_mod._build_database_url()
|
||
|
||
|
||
class TestBuildDatabaseUrl:
|
||
def test_mysql_default(self, monkeypatch):
|
||
assert _url(monkeypatch, 'mysql') == 'mysql+pymysql://user:pass@127.0.0.1:1/appdb?charset=utf8mb4'
|
||
|
||
def test_postgresql(self, monkeypatch):
|
||
assert _url(monkeypatch, 'postgresql') == 'postgresql+psycopg2://user:pass@127.0.0.1:1/appdb'
|
||
|
||
def test_pgsql_alias(self, monkeypatch):
|
||
assert _url(monkeypatch, 'pgsql') == 'postgresql+psycopg2://user:pass@127.0.0.1:1/appdb'
|
||
|
||
def test_mssql(self, monkeypatch):
|
||
assert _url(monkeypatch, 'mssql') == 'mssql+pymssql://user:pass@127.0.0.1:1/appdb'
|
||
|
||
def test_sqlite_file_path(self, monkeypatch):
|
||
# Windows 反斜杠路径应规范化为正斜杠，且无 host/账号
|
||
assert _url(monkeypatch, 'sqlite', DB_DATABASE=r'D:\data\app.db') == 'sqlite:///D:/data/app.db'
|
||
|
||
def test_sqlite_memory(self, monkeypatch):
|
||
assert _url(monkeypatch, 'sqlite', DB_DATABASE=':memory:') == 'sqlite://'
|
||
|
||
def test_oracle_service_name(self, monkeypatch):
|
||
# Oracle 无"数据库名"，DB_DATABASE 即 SERVICE_NAME，走 ?service_name= 查询参数
|
||
assert _url(monkeypatch, 'oracle') == 'oracle+oracledb://user:pass@127.0.0.1:1/?service_name=appdb'
|
||
|
||
def test_unknown_driver_falls_back_mysql(self, monkeypatch):
|
||
assert _url(monkeypatch, 'db2') == 'mysql+pymysql://user:pass@127.0.0.1:1/appdb?charset=utf8mb4'
|
||
|
||
def test_driver_default_ports(self):
|
||
# DB_PORT 未显式配置时按驱动回退（_DEFAULT_PORT 表）
|
||
assert db_mod._DEFAULT_PORT['oracle'] == 1521
|
||
assert db_mod._DEFAULT_PORT['mysql'] == 3306
|
||
assert db_mod._DEFAULT_PORT['postgresql'] == 5432
|
||
assert db_mod._DEFAULT_PORT['pgsql'] == 5432
|
||
assert db_mod._DEFAULT_PORT['mssql'] == 1433
|
||
assert db_mod._DEFAULT_PORT['sqlite'] == 0
|
||
|
||
|
||
class TestSeedData:
|
||
"""种子数据（D3）：幂等写入 admin/角色/菜单"""
|
||
|
||
def test_seed_creates_admin_role_menu(self, app):
|
||
from scripts.seed_data import seed_all
|
||
from src.modules.system.user.repository import UserRepository
|
||
from src.modules.system.role.repository import RoleRepository
|
||
from src.modules.system.menu.repository import MenuRepository
|
||
|
||
with app.app_context():
|
||
seed_all()
|
||
admin = UserRepository.first_by({'username': 'admin'})
|
||
assert admin is not None
|
||
assert admin.password != DEFAULT_PASSWORD # 已加密
|
||
assert admin.salt
|
||
role = RoleRepository.first_by({'code': 'admin'})
|
||
assert role is not None
|
||
assert MenuRepository.first_by({'name': '系统管理', 'parent_id': 0}) is not None
|
||
|
||
def test_seed_is_idempotent(self, app):
|
||
from scripts.seed_data import seed_all
|
||
from src.modules.system.user.repository import UserRepository
|
||
|
||
with app.app_context():
|
||
seed_all()
|
||
seed_all()
|
||
admins = UserRepository.find_list(filters={'username': 'admin'})
|
||
assert len(admins) == 1
|
||
|
||
|
||
class TestNormalizeDriver:
|
||
"""驱动别名归一化（_normalize_driver，与 migrate_db.py 共用）"""
|
||
|
||
def test_canonical(self):
|
||
from config.database import _normalize_driver
|
||
assert _normalize_driver('mysql') == 'mysql'
|
||
assert _normalize_driver('postgresql') == 'postgresql'
|
||
assert _normalize_driver('mssql') == 'mssql'
|
||
assert _normalize_driver('sqlite') == 'sqlite'
|
||
assert _normalize_driver('oracle') == 'oracle'
|
||
|
||
def test_aliases(self):
|
||
from config.database import _normalize_driver
|
||
assert _normalize_driver('postgres') == 'postgresql'
|
||
assert _normalize_driver('pgsql') == 'postgresql'
|
||
assert _normalize_driver('sqlserver') == 'mssql'
|
||
assert _normalize_driver('sqlite3') == 'sqlite'
|
||
|
||
def test_upper_case(self):
|
||
from config.database import _normalize_driver
|
||
assert _normalize_driver('PostgreSQL') == 'postgresql'
|
||
|
||
|
||
class TestSqliteUrl:
|
||
"""SQLite 连接 URL 构建（_sqlite_url，与 migrate_db.py 共用）"""
|
||
|
||
def test_memory(self):
|
||
from config.database import _sqlite_url
|
||
assert _sqlite_url(':memory:') == 'sqlite://'
|
||
|
||
def test_windows_path(self):
|
||
from config.database import _sqlite_url
|
||
assert _sqlite_url(r'D:\data\app.db') == 'sqlite:///D:/data/app.db'
|
||
|
||
|
||
class TestMigrateDbUrl:
|
||
"""跨库迁移脚本 migrate_db.py 的 URL 构建（显式参数版，M1）"""
|
||
|
||
def test_make_url_mysql(self):
|
||
from scripts.migrate_db import make_url
|
||
assert make_url('mysql', '10.0.0.1', 3306, 'src_db', 'root', 'pwd') \
|
||
== 'mysql+pymysql://root:pwd@10.0.0.1:3306/src_db?charset=utf8mb4'
|
||
|
||
def test_make_url_postgresql(self):
|
||
from scripts.migrate_db import make_url
|
||
assert make_url('postgresql', '10.0.0.1', 5432, 'dst_db', 'pg', 'pwd') \
|
||
== 'postgresql+psycopg2://pg:pwd@10.0.0.1:5432/dst_db'
|
||
|
||
def test_make_url_oracle_service_name(self):
|
||
from scripts.migrate_db import make_url
|
||
assert make_url('oracle', '10.0.0.1', 1521, 'XEPDB1', 'system', 'pwd') \
|
||
== 'oracle+oracledb://system:pwd@10.0.0.1:1521/?service_name=XEPDB1'
|
||
|
||
def test_make_url_mssql(self):
|
||
from scripts.migrate_db import make_url
|
||
assert make_url('mssql', '10.0.0.1', 1433, 'dst_db', 'sa', 'pwd') \
|
||
== 'mssql+pymssql://sa:pwd@10.0.0.1:1433/dst_db?charset=utf8'
|
||
|
||
def test_make_url_sqlite_path(self):
|
||
from scripts.migrate_db import make_url
|
||
assert make_url('sqlite', '', None, r'D:\data\app.db', '', '') == 'sqlite:///D:/data/app.db'
|
||
|
||
def test_make_url_driver_alias(self):
|
||
from scripts.migrate_db import make_url
|
||
# 别名：postgres→postgresql、sqlserver→mssql
|
||
assert make_url('postgres', '10.0.0.1', 5432, 'dst_db', 'pg', 'pwd') \
|
||
== 'postgresql+psycopg2://pg:pwd@10.0.0.1:5432/dst_db'
|
||
assert make_url('sqlserver', '10.0.0.1', 1433, 'dst_db', 'sa', 'pwd') \
|
||
== 'mssql+pymssql://sa:pwd@10.0.0.1:1433/dst_db?charset=utf8'
|
||
|
||
|
||
class TestFriendlyDbError:
|
||
"""数据库连接异常友好提示（翻译认证失败/库不存在/拒连）"""
|
||
|
||
def test_auth_failed_gbk(self):
|
||
from config import friendly_db_error
|
||
gbk = '用户 "postgres" Password 认证失败'.encode('gbk')
|
||
e = UnicodeDecodeError('utf-8', gbk, 0, len(gbk), 'invalid continuation byte')
|
||
assert '认证失败' in friendly_db_error(e)
|
||
assert 'DB_PASSWORD' in friendly_db_error(e)
|
||
|
||
def test_db_not_exist(self):
|
||
from config import friendly_db_error
|
||
assert '数据库不存在' in friendly_db_error(Exception('database "x" does not exist'))
|
||
|
||
def test_connection_refused(self):
|
||
from config import friendly_db_error
|
||
assert '连接被拒绝' in friendly_db_error(Exception('Connection refused'))
|
||
|
||
def test_oracle_auth_failed(self):
|
||
from config import friendly_db_error
|
||
assert '认证失败' in friendly_db_error(Exception('ORA-01017: invalid username/password; logon denied'))
|
||
|
||
def test_oracle_service_name_error(self):
|
||
from config import friendly_db_error
|
||
msg = friendly_db_error(Exception('ORA-12514: TNS:listener does not currently know of service requested'))
|
||
assert '服务名' in msg
|
||
assert 'SERVICE_NAME' in msg
|
||
|
||
def test_oracle_connection_refused(self):
|
||
from config import friendly_db_error
|
||
assert '连接被拒绝' in friendly_db_error(Exception('ORA-12170: TNS:Connect timeout occurred'))
|
||
|
||
def test_unknown_error(self):
|
||
from config import friendly_db_error
|
||
assert 'boom' in friendly_db_error(Exception('boom'))
|