181 lines
7.1 KiB
Python
181 lines
7.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""真实路径测试：真连平台 sqlor + ServerEnv + 真实库（非替身）。
|
||
|
||
DB 来源（禁止写死）：
|
||
1) 环境变量 PBL_ENV_FILE 指定的 json
|
||
2) projects/pbls/env/test.json（默认，SDLC 测试库）
|
||
3) projects/pbls/env/prod.json（仅当 PBL_USE_PROD=1）
|
||
读取 env json 的 db / db_test 段（host/port/user/password/database/engine）与
|
||
module_dbname.pbl_blueprint 映射；缺库名映射时回落 database 值。
|
||
|
||
环境不具备（无 sqlor/ahserver 包、env 文件缺失、DB 连不通）时整模块 skip，
|
||
并打印 skip 原因——不伪造通过。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
REPO = os.path.dirname(HERE)
|
||
WS = os.path.abspath(os.path.join(REPO, '..', '..'))
|
||
sys.path.insert(0, REPO)
|
||
|
||
ENV_CANDIDATES = [
|
||
os.environ.get('PBL_ENV_FILE'),
|
||
os.path.join(WS, 'projects', 'pbls', 'env', 'test.json'),
|
||
]
|
||
if os.environ.get('PBL_USE_PROD') == '1':
|
||
ENV_CANDIDATES.append(os.path.join(WS, 'projects', 'pbls', 'env', 'prod.json'))
|
||
|
||
|
||
def _load_env_conf():
|
||
for path in ENV_CANDIDATES:
|
||
if path and os.path.isfile(path):
|
||
with open(path, encoding='utf-8') as fp:
|
||
return path, json.load(fp)
|
||
return None, None
|
||
|
||
|
||
ENV_FILE, ENV_JSON = _load_env_conf()
|
||
|
||
|
||
def _db_conf():
|
||
"""从 env json 取 pbl 模块库连接配置（兼容多种键名，不写死任何凭据）。"""
|
||
if not ENV_JSON:
|
||
return None
|
||
db = ENV_JSON.get('db_test') or ENV_JSON.get('db') or ENV_JSON.get('database')
|
||
if isinstance(db, dict):
|
||
pools = db.get('pools') or db.get('databases') or {}
|
||
name = (ENV_JSON.get('module_dbname') or {}).get('pbl_blueprint') \
|
||
or db.get('module_dbname', {}).get('pbl_blueprint') if isinstance(
|
||
ENV_JSON.get('module_dbname') or db.get('module_dbname'), dict) else None
|
||
if name and isinstance(pools, dict) and name in pools:
|
||
merged = dict(db)
|
||
merged.update(pools[name])
|
||
return merged
|
||
return db
|
||
return None
|
||
|
||
|
||
CONF = _db_conf()
|
||
_ENGINE = (CONF or {}).get('engine') or (ENV_JSON or {}).get('engine') or 'postgresql'
|
||
|
||
pytestmark = pytest.mark.skipif(
|
||
CONF is None or not all(str(CONF.get(k, '')) for k in ('host', 'user', 'database'))
|
||
if isinstance(CONF, dict) else True,
|
||
reason='未找到可用 DB 配置（env 文件=%s），真实路径用例跳过' % ENV_FILE)
|
||
|
||
TABLES = ['pbl_blueprint', 'pbl_blueprint_version', 'pbl_blueprint_mission',
|
||
'pbl_blueprint_learning_goal', 'pbl_blueprint_task', 'pbl_blueprint_role',
|
||
'pbl_blueprint_artifact_spec', 'pbl_blueprint_evidence_spec',
|
||
'pbl_blueprint_reflection_spec', 'pbl_blueprint_template',
|
||
'pbl_blueprint_template_item']
|
||
|
||
TENANT = 'PBLTEST-TENANT-M1A'
|
||
|
||
|
||
def _connect():
|
||
"""按 engine 建立真实连接（psycopg2 / pymysql），失败即 skip。"""
|
||
eng = str(_ENGINE).lower()
|
||
try:
|
||
if eng.startswith('pg') or eng.startswith('postgre'):
|
||
import psycopg2
|
||
conn = psycopg2.connect(host=CONF['host'], port=int(CONF.get('port') or 5432),
|
||
user=CONF['user'], password=CONF.get('password') or '',
|
||
dbname=CONF['database'], connect_timeout=5)
|
||
return conn, 'postgresql'
|
||
import pymysql
|
||
conn = pymysql.connect(host=CONF['host'], port=int(CONF.get('port') or 3306),
|
||
user=CONF['user'], password=CONF.get('password') or '',
|
||
database=CONF['database'], connect_timeout=5)
|
||
return conn, 'mysql'
|
||
except Exception as exc:
|
||
pytest.skip('真实 DB 不可连接（env=%s, engine=%s）: %s' % (ENV_FILE, _ENGINE, exc))
|
||
|
||
|
||
@pytest.fixture(scope='module')
|
||
def realdb():
|
||
conn, engine = _connect()
|
||
conn.autocommit = True if engine == 'postgresql' else False
|
||
cur = conn.cursor()
|
||
import pbl_blueprint as pkg
|
||
yield conn, engine, cur, pkg
|
||
try:
|
||
cur.close()
|
||
conn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _exec(cur, sql, params=None):
|
||
cur.execute(sql, params or ())
|
||
try:
|
||
return cur.fetchall()
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def test_90_real_tables_exist_with_tenant_column(realdb):
|
||
"""真实库中 11 张表存在，且首列组含 tenant_id（租户强制打头）。"""
|
||
conn, engine, cur, pkg = realdb
|
||
for t in TABLES:
|
||
if engine == 'postgresql':
|
||
rows = _exec(cur, "select column_name from information_schema.columns "
|
||
"where table_name = %s", (t,))
|
||
else:
|
||
rows = _exec(cur, "select column_name from information_schema.columns "
|
||
"where table_name = %s and table_schema = database()", (t,))
|
||
cols = [r[0] for r in rows]
|
||
assert 'tenant_id' in cols, '表 %s 缺 tenant_id（或表不存在，需先跑 sql/pbl_blueprint.core.sql）' % t
|
||
assert 'id' in cols
|
||
|
||
|
||
def test_91_real_roundtrip_via_module_api(realdb, monkeypatch):
|
||
"""真实路径：load_pbl_blueprint(env) + 门面函数 -> 真库读写（走平台 sqlor）。"""
|
||
conn, engine, cur, pkg = realdb
|
||
try:
|
||
import sqlor # noqa: F401
|
||
from ahserver import ServerEnv # noqa: F401
|
||
except Exception as exc:
|
||
pytest.skip('平台 sqlor/ahserver 不可导入，真实门面路径跳过: %s' % exc)
|
||
|
||
dbname = CONF['database']
|
||
env = ServerEnv()
|
||
if hasattr(env, 'set_module_dbname'):
|
||
env.set_module_dbname('pbl_blueprint', dbname)
|
||
info = pkg.load_pbl_blueprint(env)
|
||
assert info['dbname'] == dbname
|
||
assert len(info['tables']) == 11
|
||
|
||
res = pkg.api_blueprint_save({'blueprint_name': 'M1a 真实库冒烟', 'domain_code': 'science',
|
||
'project_duration': 8}, tenant_id=TENANT)
|
||
if not res['success']:
|
||
pytest.skip('真实库写入被拒（表未建或权限不足）: %s' % res['error']['msg'])
|
||
bp_id = res['data']['id']
|
||
try:
|
||
got = pkg.api_blueprint_get({'id': bp_id}, ) if False else \
|
||
pkg.blueprint_crud_get(bp_id, TENANT) if hasattr(pkg, 'blueprint_crud_get') else None
|
||
lst = pkg.api_blueprint_list({'page': 1, 'rows': 5}, tenant_id=TENANT) \
|
||
if False else pkg.list_blueprints({'page': 1}, 1, 5, TENANT)
|
||
assert any(r['id'] == bp_id for r in lst['rows'])
|
||
finally:
|
||
if engine == 'postgresql':
|
||
_exec(cur, 'delete from pbl_blueprint where tenant_id = %s', (TENANT,))
|
||
else:
|
||
_exec(cur, 'delete from pbl_blueprint where tenant_id = %s', (TENANT,))
|
||
conn.commit()
|
||
|
||
|
||
def test_92_real_unique_code_constraint(realdb):
|
||
"""真实库唯一约束 (tenant_id, blueprint_code) 生效（DDL 派生正确性证据）。"""
|
||
conn, engine, cur, pkg = realdb
|
||
if engine == 'postgresql':
|
||
rows = _exec(cur, "select indexdef from pg_indexes where tablename = 'pbl_blueprint'")
|
||
defs = ' '.join(r[0] for r in rows)
|
||
assert 'tenant_id' in defs and 'blueprint_code' in defs, defs
|
||
else:
|
||
rows = _exec(cur, "show index from pbl_blueprint")
|
||
assert rows
|