107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
|
"""只读核验真实 SQL Server 2008 R2；2022 PoC 不能获得兼容通过结果。"""
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import json
|
|||
|
|
import sys
|
|||
|
|
from datetime import datetime
|
|||
|
|
from decimal import Decimal
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pyodbc
|
|||
|
|
|
|||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "backend"))
|
|||
|
|
|
|||
|
|
from sqlalchemy import URL, create_engine
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS
|
|||
|
|
from app.services.tplus.reader import TPlusReader
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
"""检查版本、中文、金额、日期和写权限，结果不包含凭据或客户记录。"""
|
|||
|
|
parser = argparse.ArgumentParser(description="SQL Server 2008 R2 兼容性验收")
|
|||
|
|
parser.add_argument("--connection-file", required=True, type=Path)
|
|||
|
|
parser.add_argument("--report", required=True, type=Path)
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
config = json.loads(args.connection_file.read_text(encoding="utf-8-sig"))
|
|||
|
|
connection_string = (
|
|||
|
|
"DRIVER={"
|
|||
|
|
+ config["driver"]
|
|||
|
|
+ "};SERVER="
|
|||
|
|
+ config["host"]
|
|||
|
|
+ ","
|
|||
|
|
+ str(config["port"])
|
|||
|
|
+ ";DATABASE="
|
|||
|
|
+ config["database"]
|
|||
|
|
+ ";UID="
|
|||
|
|
+ config["user"]
|
|||
|
|
+ ";PWD={"
|
|||
|
|
+ config["password"].replace("}", "}}")
|
|||
|
|
+ "};"
|
|||
|
|
)
|
|||
|
|
with pyodbc.connect(connection_string, timeout=10) as connection:
|
|||
|
|
connection.timeout = 15
|
|||
|
|
cursor = connection.cursor()
|
|||
|
|
version = str(
|
|||
|
|
cursor.execute(
|
|||
|
|
"SELECT CAST(SERVERPROPERTY('ProductVersion') AS varchar(80))"
|
|||
|
|
).fetchone()[0]
|
|||
|
|
)
|
|||
|
|
values = cursor.execute(
|
|||
|
|
"SELECT N'中文核验', CAST('12345.6789' AS decimal(19,4)), "
|
|||
|
|
"CAST('2008-01-02T03:04:05' AS datetime)"
|
|||
|
|
).fetchone()
|
|||
|
|
permissions = cursor.execute(
|
|||
|
|
"SELECT HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','INSERT'), "
|
|||
|
|
"HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','UPDATE'), "
|
|||
|
|
"HAS_PERMS_BY_NAME(DB_NAME(),'DATABASE','DELETE')"
|
|||
|
|
).fetchone()
|
|||
|
|
checks = {
|
|||
|
|
"sqlserver_2008r2": version.startswith("10.50."),
|
|||
|
|
"unicode_roundtrip": values[0] == "中文核验",
|
|||
|
|
"decimal_exact": values[1] == Decimal("12345.6789"),
|
|||
|
|
"datetime_exact": values[2] == datetime(2008, 1, 2, 3, 4, 5),
|
|||
|
|
"no_dml_permissions": all(x == 0 for x in permissions),
|
|||
|
|
}
|
|||
|
|
report = {
|
|||
|
|
"server_version": version,
|
|||
|
|
"driver_name": connection.getinfo(pyodbc.SQL_DRIVER_NAME),
|
|||
|
|
"driver_version": connection.getinfo(pyodbc.SQL_DRIVER_VER),
|
|||
|
|
"checks": checks,
|
|||
|
|
"compatible": all(checks.values()),
|
|||
|
|
}
|
|||
|
|
url = URL.create(
|
|||
|
|
"mssql+pyodbc",
|
|||
|
|
username=config["user"],
|
|||
|
|
password=config["password"],
|
|||
|
|
host=config["host"],
|
|||
|
|
port=config["port"],
|
|||
|
|
database=config["database"],
|
|||
|
|
query={"driver": config["driver"]},
|
|||
|
|
)
|
|||
|
|
engine = create_engine(url, hide_parameters=True)
|
|||
|
|
adapter_checks = {}
|
|||
|
|
try:
|
|||
|
|
with Session(engine) as session:
|
|||
|
|
reader = TPlusReader(session)
|
|||
|
|
for mapping in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS):
|
|||
|
|
try:
|
|||
|
|
reader.fetch_entity(mapping.entity_type, max_rows=1)
|
|||
|
|
adapter_checks[mapping.entity_type] = True
|
|||
|
|
except Exception as exc:
|
|||
|
|
adapter_checks[mapping.entity_type] = False
|
|||
|
|
report.setdefault("errors", {})[mapping.entity_type] = type(exc).__name__
|
|||
|
|
finally:
|
|||
|
|
engine.dispose()
|
|||
|
|
report["adapter_queries"] = adapter_checks
|
|||
|
|
report["compatible"] = report["compatible"] and all(adapter_checks.values())
|
|||
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|||
|
|
print(json.dumps(report, ensure_ascii=False))
|
|||
|
|
return 0 if report["compatible"] else 2
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|