feat: add standard system initialization tooling
This commit is contained in:
汤学会
@@ -0,0 +1,152 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
class SystemInitializerCliTest(unittest.TestCase):
|
||||
def test_parse_fresh_arguments(self) -> None:
|
||||
from scripts.system_initialize import parse_args
|
||||
|
||||
args = parse_args(
|
||||
[
|
||||
"fresh",
|
||||
"--company-name",
|
||||
"百华",
|
||||
"--admin-name",
|
||||
"超级管理员",
|
||||
"--admin-phone",
|
||||
"13800000000",
|
||||
"--admin-password",
|
||||
"secret123",
|
||||
"--smart-operation-report",
|
||||
"off",
|
||||
"--delete-files",
|
||||
"--allow-any-database",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.mode, "fresh")
|
||||
self.assertEqual(args.company_name, "百华")
|
||||
self.assertEqual(args.admin_name, "超级管理员")
|
||||
self.assertEqual(args.admin_phone, "13800000000")
|
||||
self.assertEqual(args.admin_password, "secret123")
|
||||
self.assertEqual(args.smart_operation_report, "off")
|
||||
self.assertTrue(args.delete_files)
|
||||
self.assertTrue(args.allow_any_database)
|
||||
self.assertFalse(hasattr(args, "confirm_reset"))
|
||||
|
||||
def test_parse_reset_arguments(self) -> None:
|
||||
from scripts.system_initialize import parse_args
|
||||
|
||||
args = parse_args(
|
||||
[
|
||||
"reset",
|
||||
"--company-name",
|
||||
"百华",
|
||||
"--admin-name",
|
||||
"超级管理员",
|
||||
"--admin-phone",
|
||||
"13800000000",
|
||||
"--admin-password",
|
||||
"secret123",
|
||||
"--confirm-reset",
|
||||
"--smart-operation-report",
|
||||
"off",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.mode, "reset")
|
||||
self.assertEqual(args.company_name, "百华")
|
||||
self.assertEqual(args.admin_phone, "13800000000")
|
||||
self.assertTrue(args.confirm_reset)
|
||||
self.assertEqual(args.smart_operation_report, "off")
|
||||
|
||||
def test_main_fresh_creates_database_then_initializes_without_real_db(self) -> None:
|
||||
from scripts import system_initialize
|
||||
|
||||
settings = SimpleNamespace(
|
||||
mysql_host="127.0.0.1",
|
||||
mysql_port=3306,
|
||||
mysql_database="erp_test",
|
||||
mysql_dsn="mysql+pymysql://root:secret@127.0.0.1:3306/erp_test",
|
||||
)
|
||||
result = SimpleNamespace(
|
||||
backup_path=None,
|
||||
summary_path="/tmp/summary.json",
|
||||
seeded={"admin_username": "13800000000"},
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(system_initialize, "get_settings", return_value=settings),
|
||||
patch.object(system_initialize, "create_engine", return_value="engine") as create_engine_mock,
|
||||
patch.object(system_initialize, "create_database_if_missing", return_value="erp_test") as create_db_mock,
|
||||
patch.object(system_initialize, "fresh_initialize_database", return_value=result) as fresh_mock,
|
||||
patch.object(system_initialize, "reset_existing_database_with_backup") as reset_mock,
|
||||
patch("sys.stdout", new_callable=io.StringIO) as stdout,
|
||||
):
|
||||
exit_code = system_initialize.main(
|
||||
[
|
||||
"fresh",
|
||||
"--company-name",
|
||||
"百华",
|
||||
"--admin-name",
|
||||
"超级管理员",
|
||||
"--admin-phone",
|
||||
"13800000000",
|
||||
"--admin-password",
|
||||
"secret123",
|
||||
"--smart-operation-report",
|
||||
"on",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(exit_code, 0)
|
||||
create_db_mock.assert_called_once_with(settings)
|
||||
create_engine_mock.assert_called_once_with(settings.mysql_dsn, pool_pre_ping=True, future=True)
|
||||
fresh_mock.assert_called_once()
|
||||
reset_mock.assert_not_called()
|
||||
self.assertIn("mode=fresh", stdout.getvalue())
|
||||
self.assertIn("database=127.0.0.1:3306/erp_test", stdout.getvalue())
|
||||
self.assertIn("summary_path=/tmp/summary.json", stdout.getvalue())
|
||||
self.assertIn("admin_username=13800000000", stdout.getvalue())
|
||||
|
||||
def test_main_reset_calls_backup_reset_without_swallowing_service_errors(self) -> None:
|
||||
from scripts import system_initialize
|
||||
|
||||
settings = SimpleNamespace(
|
||||
mysql_host="127.0.0.1",
|
||||
mysql_port=3306,
|
||||
mysql_database="erp_test",
|
||||
mysql_dsn="mysql+pymysql://root:secret@127.0.0.1:3306/erp_test",
|
||||
)
|
||||
|
||||
with (
|
||||
patch.object(system_initialize, "get_settings", return_value=settings),
|
||||
patch.object(system_initialize, "create_engine", return_value="engine"),
|
||||
patch.object(
|
||||
system_initialize,
|
||||
"reset_existing_database_with_backup",
|
||||
side_effect=ValueError("Refusing to reset database without --confirm-reset"),
|
||||
),
|
||||
):
|
||||
with self.assertRaisesRegex(ValueError, "confirm-reset"):
|
||||
system_initialize.main(
|
||||
[
|
||||
"reset",
|
||||
"--company-name",
|
||||
"百华",
|
||||
"--admin-name",
|
||||
"超级管理员",
|
||||
"--admin-phone",
|
||||
"13800000000",
|
||||
"--admin-password",
|
||||
"secret123",
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,97 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import create_engine, inspect
|
||||
|
||||
import app.models # noqa: F401
|
||||
from app.models.base import Base
|
||||
|
||||
|
||||
MINIAPP_TABLES = {
|
||||
"attendance_points",
|
||||
"personnel",
|
||||
"person_roles",
|
||||
"person_attendance_points",
|
||||
"products",
|
||||
"equipment",
|
||||
"work_schedules",
|
||||
"work_sessions",
|
||||
"work_session_devices",
|
||||
"production_reports",
|
||||
"production_report_items",
|
||||
"report_audit_logs",
|
||||
"mold_lock_feedbacks",
|
||||
"device_qrcodes",
|
||||
"device_qrcode_batch_tasks",
|
||||
"notices",
|
||||
"notice_points",
|
||||
"reconciliation_ledger_entries",
|
||||
}
|
||||
|
||||
REQUIRED_ERP_TABLES = {
|
||||
"sys_department",
|
||||
"hr_employee",
|
||||
"sys_role",
|
||||
"sys_permission",
|
||||
"sys_role_permission",
|
||||
"sys_user",
|
||||
"sys_user_role",
|
||||
"sys_system_config",
|
||||
"sys_ai_assistant_config",
|
||||
"md_unit",
|
||||
"md_item_category",
|
||||
"md_item",
|
||||
"md_material",
|
||||
"md_product",
|
||||
"md_bom",
|
||||
"md_bom_item",
|
||||
"wh_warehouse",
|
||||
"wh_location",
|
||||
"wh_stock_lot",
|
||||
"wh_stock_balance",
|
||||
"wh_inventory_txn",
|
||||
"pp_material_issue",
|
||||
"pp_material_issue_item",
|
||||
"pp_production_batch_ledger",
|
||||
"pp_production_batch_ledger_txn",
|
||||
"document_archives",
|
||||
}
|
||||
|
||||
|
||||
def _erp_model_table_names() -> set[str]:
|
||||
models_dir = Path(__file__).resolve().parents[1] / "app" / "models"
|
||||
table_names: set[str] = set()
|
||||
for model_file in models_dir.glob("*.py"):
|
||||
if model_file.name in {"__init__.py", "base.py", "miniapp.py"}:
|
||||
continue
|
||||
for line in model_file.read_text(encoding="utf-8").splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("__tablename__"):
|
||||
table_names.add(stripped.split("=", 1)[1].strip().strip("\"'"))
|
||||
return table_names
|
||||
|
||||
|
||||
def test_app_models_registers_erp_and_miniapp_tables() -> None:
|
||||
expected_tables = _erp_model_table_names() | REQUIRED_ERP_TABLES | MINIAPP_TABLES
|
||||
|
||||
missing_tables = expected_tables - set(Base.metadata.tables)
|
||||
|
||||
assert missing_tables == set()
|
||||
|
||||
|
||||
def test_reset_table_groups_are_registered_in_metadata() -> None:
|
||||
from app.services.system_initializer import ACCOUNT_RESET_TABLES, BUSINESS_RESET_TABLES, MINIAPP_RESET_TABLES, SKELETON_TABLES
|
||||
|
||||
expected_tables = set(BUSINESS_RESET_TABLES) | set(MINIAPP_RESET_TABLES) | set(ACCOUNT_RESET_TABLES) | set(SKELETON_TABLES)
|
||||
|
||||
assert expected_tables.issubset(set(Base.metadata.tables))
|
||||
|
||||
|
||||
def test_registered_models_create_all_in_sqlite_memory() -> None:
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
|
||||
Base.metadata.create_all(engine)
|
||||
|
||||
created_tables = set(inspect(engine).get_table_names())
|
||||
assert (MINIAPP_TABLES | REQUIRED_ERP_TABLES).issubset(created_tables)
|
||||
@@ -0,0 +1,328 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from sqlalchemy import create_engine, inspect, select, text
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
import app.models # noqa: F401
|
||||
from app.models.base import Base
|
||||
from app.models.org import User
|
||||
from app.services.auth import verify_password
|
||||
|
||||
|
||||
def _options(*, confirm_reset: bool = False, mode: str = "reset"):
|
||||
from app.services.system_initializer import SystemInitializeOptions
|
||||
|
||||
return SystemInitializeOptions(
|
||||
mode=mode,
|
||||
company_name="百华",
|
||||
admin_name="超级管理员",
|
||||
admin_phone="13800000000",
|
||||
admin_password="secret123",
|
||||
confirm_reset=confirm_reset,
|
||||
)
|
||||
|
||||
|
||||
class SystemInitializerResetTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
||||
Base.metadata.create_all(self.engine)
|
||||
self.SessionLocal = sessionmaker(bind=self.engine, autoflush=False, autocommit=False, future=True)
|
||||
self.db: Session = self.SessionLocal()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.db.close()
|
||||
|
||||
def _seed_dirty_data(self) -> None:
|
||||
self.db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO sys_department
|
||||
(id, dept_code, dept_name, org_node_type, dept_type, status, sort_no, created_at, updated_at)
|
||||
VALUES
|
||||
(100, 'DIRTY_ROOT', '旧公司', 'COMPANY', 'ADMIN', 'ACTIVE', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
)
|
||||
self.db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO hr_employee
|
||||
(id, employee_code, employee_name, dept_id, mobile, is_operator, is_workshop_staff, status, created_at, updated_at)
|
||||
VALUES
|
||||
(100, 'EMP_OLD', '旧管理员', 100, '13900000000', 0, 0, 'ACTIVE', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
)
|
||||
self.db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO sys_user
|
||||
(id, username, password_hash, employee_id, dept_id, nickname, is_super_admin, status, created_at, updated_at)
|
||||
VALUES
|
||||
(100, '13900000000', 'old', 100, 100, '旧管理员', 0, 'ACTIVE', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
)
|
||||
self.db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO personnel (phone, name, is_temporary, created_at, updated_at)
|
||||
VALUES ('13900000000', '旧小程序用户', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
)
|
||||
self.db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO md_customer
|
||||
(id, customer_code, customer_name, status, created_at, updated_at)
|
||||
VALUES
|
||||
(100, 'C_OLD', '旧客户', 'ACTIVE', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
)
|
||||
self.db.commit()
|
||||
|
||||
def test_reset_requires_confirm_reset_and_does_not_clear_data(self) -> None:
|
||||
from app.services.system_initializer import reset_existing_database
|
||||
|
||||
self._seed_dirty_data()
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "confirm-reset"):
|
||||
reset_existing_database(self.db, _options(confirm_reset=False))
|
||||
|
||||
self.db.rollback()
|
||||
self.assertEqual(self.db.execute(text("SELECT COUNT(*) FROM sys_user")).scalar_one(), 1)
|
||||
self.assertEqual(self.db.execute(text("SELECT COUNT(*) FROM md_customer")).scalar_one(), 1)
|
||||
|
||||
def test_reset_clears_business_account_and_miniapp_data_then_recreates_admin(self) -> None:
|
||||
from app.services.system_initializer import reset_existing_database
|
||||
|
||||
self._seed_dirty_data()
|
||||
|
||||
result = reset_existing_database(self.db, _options(confirm_reset=True))
|
||||
self.db.commit()
|
||||
|
||||
self.assertEqual(result.mode, "reset")
|
||||
self.assertEqual(result.counts_before["sys_user"], 1)
|
||||
self.assertEqual(result.counts_before["md_customer"], 1)
|
||||
self.assertEqual(result.counts_before["personnel"], 1)
|
||||
self.assertEqual(result.counts_after["md_customer"], 0)
|
||||
self.assertEqual(result.counts_after["sys_user"], 1)
|
||||
self.assertEqual(result.counts_after["personnel"], 1)
|
||||
self.assertEqual(result.seeded["admin_username"], "13800000000")
|
||||
|
||||
self.assertEqual(self.db.execute(text("SELECT COUNT(*) FROM md_customer")).scalar_one(), 0)
|
||||
self.assertIsNone(self.db.scalar(select(User).where(User.username == "13900000000")))
|
||||
admin = self.db.scalar(select(User).where(User.username == "13800000000"))
|
||||
self.assertIsNotNone(admin)
|
||||
assert admin is not None
|
||||
self.assertEqual(admin.is_super_admin, 1)
|
||||
self.assertTrue(verify_password("secret123", admin.password_hash))
|
||||
|
||||
def test_reset_keeps_sqlite_foreign_keys_enabled(self) -> None:
|
||||
from app.services.system_initializer import reset_existing_database
|
||||
|
||||
self.db.execute(text("PRAGMA foreign_keys = ON"))
|
||||
self._seed_dirty_data()
|
||||
|
||||
reset_existing_database(self.db, _options(confirm_reset=True))
|
||||
self.db.commit()
|
||||
|
||||
self.assertEqual(self.db.execute(text("PRAGMA foreign_keys")).scalar_one(), 1)
|
||||
|
||||
def test_write_summary_report_writes_expected_json(self) -> None:
|
||||
from app.services.system_initializer import SystemInitializeResult, write_summary_report
|
||||
|
||||
result = SystemInitializeResult(
|
||||
mode="reset",
|
||||
database="erp_test",
|
||||
backup_path="/tmp/backup.sql",
|
||||
counts_before={"sys_user": 2},
|
||||
counts_after={"sys_user": 1},
|
||||
seeded={"admin_username": "13800000000"},
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
summary_path = write_summary_report(Path(tmpdir), result)
|
||||
payload = json.loads(Path(summary_path).read_text(encoding="utf-8"))
|
||||
|
||||
self.assertEqual(payload["mode"], "reset")
|
||||
self.assertEqual(payload["database"], "erp_test")
|
||||
self.assertEqual(payload["backup_path"], "/tmp/backup.sql")
|
||||
self.assertEqual(payload["summary_path"], summary_path)
|
||||
self.assertEqual(payload["counts_before"], {"sys_user": 2})
|
||||
self.assertEqual(payload["counts_after"], {"sys_user": 1})
|
||||
self.assertEqual(payload["seeded"], {"admin_username": "13800000000"})
|
||||
self.assertRegex(payload["created_at"], r"^\d{4}-\d{2}-\d{2}T")
|
||||
|
||||
def test_default_output_dir_uses_cleanup_backups_folder(self) -> None:
|
||||
from app.services.system_initializer import default_output_dir
|
||||
|
||||
self.assertEqual(default_output_dir().name, "cleanup_backups")
|
||||
self.assertEqual(default_output_dir().parent.name, "outputs")
|
||||
|
||||
def test_backup_mysql_database_uses_defaults_file_without_password_argument(self) -> None:
|
||||
from app.services.system_initializer import backup_mysql_database
|
||||
|
||||
class Settings:
|
||||
mysql_host = "127.0.0.1"
|
||||
mysql_port = 3306
|
||||
mysql_user = "root"
|
||||
mysql_password = "secret"
|
||||
mysql_database = "jiaheng_erp"
|
||||
|
||||
commands: list[list[str]] = []
|
||||
|
||||
def fake_run(command: list[str], check: bool) -> None:
|
||||
commands.append(command)
|
||||
defaults_arg = next(part for part in command if part.startswith("--defaults-extra-file="))
|
||||
defaults_path = Path(defaults_arg.split("=", 1)[1])
|
||||
self.assertTrue(defaults_path.exists())
|
||||
self.assertIn("password=secret", defaults_path.read_text(encoding="utf-8"))
|
||||
self.assertEqual(defaults_path.stat().st_mode & 0o777, 0o600)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with patch("app.services.system_initializer.subprocess.run", side_effect=fake_run):
|
||||
backup_path = backup_mysql_database(Settings(), tmpdir)
|
||||
|
||||
self.assertTrue(backup_path.endswith(".sql"))
|
||||
self.assertEqual(len(commands), 1)
|
||||
self.assertFalse(any("secret" in part for part in commands[0]))
|
||||
defaults_arg = next(part for part in commands[0] if part.startswith("--defaults-extra-file="))
|
||||
self.assertFalse(Path(defaults_arg.split("=", 1)[1]).exists())
|
||||
|
||||
|
||||
class SystemInitializerFreshTest(unittest.TestCase):
|
||||
def test_fresh_create_all_builds_key_tables_and_seeds_skeleton(self) -> None:
|
||||
from app.services.system_initializer import fresh_initialize_database
|
||||
|
||||
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
||||
|
||||
result = fresh_initialize_database(engine, _options(mode="fresh"))
|
||||
|
||||
created_tables = set(inspect(engine).get_table_names())
|
||||
self.assertIn("sys_user", created_tables)
|
||||
self.assertIn("md_customer", created_tables)
|
||||
self.assertIn("attendance_points", created_tables)
|
||||
self.assertEqual(result.mode, "fresh")
|
||||
self.assertEqual(result.counts_after["sys_user"], 1)
|
||||
self.assertEqual(result.counts_after["wh_warehouse"], 6)
|
||||
self.assertEqual(result.seeded["admin_username"], "13800000000")
|
||||
|
||||
with Session(engine, future=True) as db:
|
||||
self.assertIsNotNone(db.scalar(select(User).where(User.username == "13800000000")))
|
||||
|
||||
|
||||
class SystemInitializerFileCleanupTest(unittest.TestCase):
|
||||
def test_delete_managed_files_removes_only_children_and_keeps_directories(self) -> None:
|
||||
from app.services.system_initializer import delete_managed_files
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
root = Path(tmpdir)
|
||||
archive_dir = root / "document_archives"
|
||||
photo_dir = root / "uploads" / "logistics"
|
||||
nested_dir = archive_dir / "batch"
|
||||
archive_dir.mkdir(parents=True)
|
||||
photo_dir.mkdir(parents=True)
|
||||
nested_dir.mkdir()
|
||||
(archive_dir / "a.pdf").write_text("pdf", encoding="utf-8")
|
||||
(nested_dir / "nested.pdf").write_text("nested", encoding="utf-8")
|
||||
(photo_dir / "b.png").write_text("png", encoding="utf-8")
|
||||
|
||||
summary = delete_managed_files([archive_dir, photo_dir])
|
||||
|
||||
self.assertEqual(summary["deleted_file_count"], 3)
|
||||
self.assertEqual(summary["deleted_dir_count"], 1)
|
||||
self.assertTrue(archive_dir.exists())
|
||||
self.assertTrue(photo_dir.exists())
|
||||
self.assertEqual(list(archive_dir.iterdir()), [])
|
||||
self.assertEqual(list(photo_dir.iterdir()), [])
|
||||
|
||||
def test_reset_with_backup_skips_file_cleanup_by_default(self) -> None:
|
||||
from app.services import system_initializer
|
||||
from app.services.system_initializer import SystemInitializeResult, reset_existing_database_with_backup
|
||||
|
||||
class Settings:
|
||||
mysql_database = "erp_test"
|
||||
|
||||
reset_result = SystemInitializeResult(
|
||||
mode="reset",
|
||||
counts_before={"sys_user": 2},
|
||||
counts_after={"sys_user": 1},
|
||||
seeded={"admin_username": "13800000000"},
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with (
|
||||
patch.object(system_initializer, "backup_mysql_database", return_value="/tmp/backup.sql"),
|
||||
patch.object(system_initializer, "reset_existing_database", return_value=reset_result),
|
||||
patch.object(system_initializer, "delete_managed_files") as delete_mock,
|
||||
patch.object(system_initializer, "Session") as session_mock,
|
||||
):
|
||||
session_mock.return_value.__enter__.return_value = MagicMock()
|
||||
result = reset_existing_database_with_backup(
|
||||
engine=object(),
|
||||
settings=Settings(),
|
||||
options=_options(confirm_reset=True),
|
||||
output_dir=tmpdir,
|
||||
)
|
||||
|
||||
delete_mock.assert_not_called()
|
||||
self.assertNotIn("deleted_file_count", result.seeded)
|
||||
|
||||
def test_reset_with_backup_includes_file_cleanup_when_requested(self) -> None:
|
||||
from app.services import system_initializer
|
||||
from app.services.system_initializer import SystemInitializeResult, SystemInitializeOptions, reset_existing_database_with_backup
|
||||
|
||||
class Settings:
|
||||
mysql_database = "erp_test"
|
||||
|
||||
reset_result = SystemInitializeResult(
|
||||
mode="reset",
|
||||
counts_before={"sys_user": 2},
|
||||
counts_after={"sys_user": 1},
|
||||
seeded={"admin_username": "13800000000"},
|
||||
)
|
||||
options = SystemInitializeOptions(
|
||||
mode="reset",
|
||||
company_name="百华",
|
||||
admin_name="超级管理员",
|
||||
admin_phone="13800000000",
|
||||
admin_password="secret123",
|
||||
confirm_reset=True,
|
||||
delete_files=True,
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
with (
|
||||
patch.object(system_initializer, "backup_mysql_database", return_value="/tmp/backup.sql"),
|
||||
patch.object(system_initializer, "reset_existing_database", return_value=reset_result),
|
||||
patch.object(
|
||||
system_initializer,
|
||||
"delete_managed_files",
|
||||
return_value={"deleted_file_count": 3, "deleted_dir_count": 2},
|
||||
) as delete_mock,
|
||||
patch.object(system_initializer, "Session") as session_mock,
|
||||
):
|
||||
session_mock.return_value.__enter__.return_value = MagicMock()
|
||||
result = reset_existing_database_with_backup(
|
||||
engine=object(),
|
||||
settings=Settings(),
|
||||
options=options,
|
||||
output_dir=tmpdir,
|
||||
)
|
||||
|
||||
delete_mock.assert_called_once()
|
||||
self.assertEqual(result.seeded["deleted_file_count"], 3)
|
||||
self.assertEqual(result.seeded["deleted_dir_count"], 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,177 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from sqlalchemy import create_engine, select, text
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
|
||||
|
||||
import app.models # noqa: E402,F401
|
||||
from app.models.base import Base # noqa: E402
|
||||
from app.models.master_data import Warehouse # noqa: E402
|
||||
from app.models.miniapp import MiniAppPersonAttendancePoint, MiniAppPersonnel, MiniAppPersonRole # noqa: E402
|
||||
from app.models.operations import WarehouseLocation # noqa: E402
|
||||
from app.models.org import AiAssistantConfig, Department, Employee, Permission, Role, RolePermission, SystemConfig, User, UserRole # noqa: E402
|
||||
from app.services.auth import verify_password # noqa: E402
|
||||
from app.services.system_permissions import MENU_PERMISSION_TREE # noqa: E402
|
||||
|
||||
|
||||
class SystemInitializerSeedTest(unittest.TestCase):
|
||||
def test_table_groups_have_no_duplicates_and_include_expected_tables(self) -> None:
|
||||
from app.services.system_initializer import (
|
||||
ACCOUNT_RESET_TABLES,
|
||||
BUSINESS_RESET_TABLES,
|
||||
MINIAPP_RESET_TABLES,
|
||||
RESET_TABLES_IN_DELETE_ORDER,
|
||||
SKELETON_TABLES,
|
||||
)
|
||||
|
||||
all_tables = BUSINESS_RESET_TABLES + MINIAPP_RESET_TABLES + ACCOUNT_RESET_TABLES + SKELETON_TABLES
|
||||
reset_tables = BUSINESS_RESET_TABLES + MINIAPP_RESET_TABLES + ACCOUNT_RESET_TABLES
|
||||
|
||||
self.assertEqual(len(all_tables), len(set(all_tables)))
|
||||
self.assertEqual(set(reset_tables).issubset(set(RESET_TABLES_IN_DELETE_ORDER)), True)
|
||||
self.assertEqual(len(RESET_TABLES_IN_DELETE_ORDER), len(set(RESET_TABLES_IN_DELETE_ORDER)))
|
||||
self.assertIn("so_sales_order", BUSINESS_RESET_TABLES)
|
||||
self.assertIn("wh_stock_lot", BUSINESS_RESET_TABLES)
|
||||
self.assertIn("pp_production_batch_ledger", BUSINESS_RESET_TABLES)
|
||||
self.assertIn("production_reports", MINIAPP_RESET_TABLES)
|
||||
self.assertIn("work_sessions", MINIAPP_RESET_TABLES)
|
||||
self.assertIn("sys_user", ACCOUNT_RESET_TABLES)
|
||||
self.assertIn("sys_permission", SKELETON_TABLES)
|
||||
self.assertIn("wh_warehouse", SKELETON_TABLES)
|
||||
|
||||
|
||||
class SystemInitializerSkeletonSeedTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
||||
Base.metadata.create_all(engine)
|
||||
self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
||||
self.db: Session = self.SessionLocal()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.db.close()
|
||||
|
||||
def test_seed_system_skeleton_creates_required_defaults(self) -> None:
|
||||
from app.services.system_initializer import SystemInitializeOptions, seed_system_skeleton
|
||||
|
||||
options = SystemInitializeOptions(
|
||||
mode="reset",
|
||||
company_name="百华",
|
||||
admin_name="超级管理员",
|
||||
admin_phone="13800000000",
|
||||
admin_password="secret123",
|
||||
smart_operation_report_enabled=False,
|
||||
confirm_reset=True,
|
||||
)
|
||||
|
||||
result = seed_system_skeleton(self.db, options)
|
||||
self.db.commit()
|
||||
|
||||
admin_user = self.db.scalar(select(User).where(User.username == "13800000000"))
|
||||
self.assertIsNotNone(admin_user)
|
||||
assert admin_user is not None
|
||||
self.assertEqual(admin_user.is_super_admin, 1)
|
||||
self.assertTrue(verify_password("secret123", admin_user.password_hash))
|
||||
|
||||
admin_employee = self.db.scalar(select(Employee).where(Employee.mobile == "13800000000"))
|
||||
self.assertIsNotNone(admin_employee)
|
||||
assert admin_employee is not None
|
||||
self.assertEqual(admin_employee.employee_name, "超级管理员")
|
||||
|
||||
root = self.db.scalar(select(Department).where(Department.dept_code == "ORG_ROOT"))
|
||||
self.assertIsNotNone(root)
|
||||
assert root is not None
|
||||
self.assertEqual(root.dept_name, "百华")
|
||||
|
||||
expected_permission_codes = set()
|
||||
|
||||
def walk(nodes: list[tuple[str, str, list]]) -> None:
|
||||
for code, _name, children in nodes:
|
||||
expected_permission_codes.add(code)
|
||||
walk(children)
|
||||
|
||||
walk(MENU_PERMISSION_TREE)
|
||||
actual_permission_codes = set(self.db.scalars(select(Permission.permission_code)).all())
|
||||
self.assertEqual(expected_permission_codes, actual_permission_codes)
|
||||
|
||||
admin_role = self.db.scalar(select(Role).where(Role.role_code == "ADMIN"))
|
||||
self.assertIsNotNone(admin_role)
|
||||
assert admin_role is not None
|
||||
self.assertEqual(
|
||||
self.db.query(RolePermission).filter(RolePermission.role_id == admin_role.id).count(),
|
||||
len(expected_permission_codes),
|
||||
)
|
||||
role_names = {role.role_code: role.role_name for role in self.db.scalars(select(Role)).all()}
|
||||
self.assertEqual(
|
||||
role_names,
|
||||
{
|
||||
"ADMIN": "超级管理员",
|
||||
"PURCHASER": "采购专员",
|
||||
"WAREHOUSE": "仓库管理人员",
|
||||
"SALES": "销售人员",
|
||||
},
|
||||
)
|
||||
self.assertEqual(self.db.query(UserRole).count(), 1)
|
||||
|
||||
self.assertEqual(self.db.query(Warehouse).count(), 6)
|
||||
self.assertEqual(self.db.query(WarehouseLocation).count(), 6)
|
||||
self.assertEqual({row.warehouse_type for row in self.db.scalars(select(Warehouse)).all()}, {"RAW", "SEMI", "FINISHED", "AUX", "SCRAP", "RETURN"})
|
||||
self.assertEqual(self.db.query(WarehouseLocation).filter(WarehouseLocation.is_default == 1).count(), 6)
|
||||
|
||||
prefix = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == "RAW_MATERIAL_LOT_PREFIX"))
|
||||
self.assertIsNotNone(prefix)
|
||||
assert prefix is not None
|
||||
self.assertEqual(prefix.config_value, "YL")
|
||||
smart = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == "SMART_OPERATION_REPORT_ENABLED"))
|
||||
self.assertIsNotNone(smart)
|
||||
assert smart is not None
|
||||
self.assertEqual(smart.config_value, "关闭")
|
||||
ai_config = self.db.scalar(select(AiAssistantConfig))
|
||||
self.assertIsNotNone(ai_config)
|
||||
assert ai_config is not None
|
||||
self.assertEqual(ai_config.assistant_name, "百华工艺助手")
|
||||
|
||||
units = dict(self.db.execute(text("SELECT unit_code, unit_name FROM md_unit")).all())
|
||||
self.assertEqual(units, {"KG": "kg", "PCS": "件", "SET": "套"})
|
||||
categories = dict(self.db.execute(text("SELECT category_code, category_name FROM md_item_category")).all())
|
||||
self.assertEqual(categories, {"RAW": "原材料", "SEMI": "半成品", "FINISHED": "成品", "AUX": "辅料", "SCRAP": "废料"})
|
||||
|
||||
miniapp_admin = self.db.get(MiniAppPersonnel, "13800000000")
|
||||
self.assertIsNotNone(miniapp_admin)
|
||||
assert miniapp_admin is not None
|
||||
self.assertEqual(miniapp_admin.name, "超级管理员")
|
||||
self.assertIsNotNone(self.db.get(MiniAppPersonRole, {"phone": "13800000000", "role": "admin"}))
|
||||
self.assertIsNotNone(
|
||||
self.db.get(
|
||||
MiniAppPersonAttendancePoint,
|
||||
{"phone": "13800000000", "attendance_point_name": "百华"},
|
||||
)
|
||||
)
|
||||
|
||||
self.assertEqual(result.seeded["warehouse_count"], 6)
|
||||
self.assertEqual(result.seeded["admin_username"], "13800000000")
|
||||
|
||||
def test_seed_system_skeleton_defaults_smart_operation_report_to_enabled(self) -> None:
|
||||
from app.services.system_initializer import SystemInitializeOptions, seed_system_skeleton
|
||||
|
||||
options = SystemInitializeOptions(
|
||||
mode="fresh",
|
||||
company_name="百华",
|
||||
admin_name="超级管理员",
|
||||
admin_phone="13800000000",
|
||||
admin_password="secret123",
|
||||
)
|
||||
|
||||
result = seed_system_skeleton(self.db, options)
|
||||
self.db.commit()
|
||||
|
||||
smart = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == "SMART_OPERATION_REPORT_ENABLED"))
|
||||
self.assertIsNotNone(smart)
|
||||
assert smart is not None
|
||||
self.assertEqual(smart.config_value, "开启")
|
||||
self.assertEqual(result.seeded["smart_operation_report_enabled"], True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user