document-management/backend/tests/dms/test_system_management.py

"""系统管理功能（契约1.23）：运行日志、存储占用、预览缓存清理、管理位与接口。

引擎级测试不依赖数据库，始终执行；接口级测试沿用 B2 惯例，
提供 DMS_TEST_* 环境变量时执行，否则安全跳过。
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
from sqlalchemy import func, select

from dms.common.errors import MetricUnavailableError
from dms.extensions import db
from dms.models import AuditLog, MetricSample, Role, User
from dms.security.passwords import hash_password
from dms.services.system_management_service import (
    API_CONTRACT_VERSION,
    BACKEND_KIND,
    MetricSampler,
    PSUTIL_AVAILABLE,
    clear_preview_cache,
    log_file_path,
    read_log_tail,
    runtime_config,
    storage_usage,
    system_info,
)
from dms.storage.paths import STORAGE_SUBDIRECTORIES


SYSTEM_READONLY_URLS = [
    "/api/v1/system/metrics/current",
    "/api/v1/system/process",
    "/api/v1/system/storage-usage",
    "/api/v1/system/converter",
    "/api/v1/system/health",
    "/api/v1/system/info",
    "/api/v1/system/runtime-config",
    "/api/v1/system/logs",
]


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


# ===== 引擎级（不依赖数据库）==================================================


def test_log_file_path_disabled_values(flask_app, monkeypatch):
    with flask_app.app_context():
        for disabled in ("OFF", "NONE", ""):
            monkeypatch.setitem(flask_app.config, "DMS_LOG_FILE", disabled)
            assert log_file_path() is None


def test_read_log_tail_lines_and_level(flask_app, monkeypatch, tmp_path: Path):
    log = tmp_path / "dms.log"
    log.write_text(
        "2026-01-01 [INFO] one\n"
        "2026-01-01 [ERROR] two\n"
        "2026-01-01 [INFO] three\n"
        "2026-01-01 [ERROR] four\n",
        encoding="utf-8",
    )
    with flask_app.app_context():
        monkeypatch.setitem(flask_app.config, "DMS_LOG_FILE", str(log))
        tail = read_log_tail(lines=2)
        assert tail["lines"] == [
            "2026-01-01 [INFO] three",
            "2026-01-01 [ERROR] four",
        ]
        assert tail["truncated"] is False
        errors = read_log_tail(lines=10, level="error")
        assert errors["lines"] == [
            "2026-01-01 [ERROR] two",
            "2026-01-01 [ERROR] four",
        ]
        assert errors["truncated"] is True
        # 未知级别按无过滤处理
        bogus = read_log_tail(lines=10, level="NOPE")
        assert len(bogus["lines"]) == 4


def test_read_log_tail_missing_file(flask_app, monkeypatch, tmp_path: Path):
    with flask_app.app_context():
        monkeypatch.setitem(
            flask_app.config, "DMS_LOG_FILE", str(tmp_path / "absent.log")
        )
        payload = read_log_tail(lines=100)
        assert payload["lines"] == []
        assert payload["truncated"] is False


def test_clear_preview_cache_only_touches_preview(
    flask_app, monkeypatch, tmp_path: Path
):
    root = tmp_path / "storage"
    nested = root / "preview" / "aa"
    nested.mkdir(parents=True)
    (nested / "1.pdf").write_bytes(b"12345")
    (root / "preview" / "2.pdf").write_bytes(b"123456")
    original = root / "original"
    original.mkdir(parents=True)
    (original / "keep.pdf").write_bytes(b"x")
    with flask_app.app_context():
        monkeypatch.setitem(flask_app.config, "DMS_STORAGE_ROOT", str(root))
        assert clear_preview_cache() == {"removedFiles": 2, "removedBytes": 11}
        assert clear_preview_cache() == {"removedFiles": 0, "removedBytes": 0}
    assert (original / "keep.pdf").exists()
    assert not (root / "preview" / "aa").exists()


def test_storage_usage_covers_all_directories(
    flask_app, monkeypatch, tmp_path: Path
):
    root = tmp_path / "storage2"
    year_dir = root / "original" / "2026"
    year_dir.mkdir(parents=True)
    (year_dir / "a.pdf").write_bytes(b"12345")
    with flask_app.app_context():
        monkeypatch.setitem(flask_app.config, "DMS_STORAGE_ROOT", str(root))
        entries = {item["directory"]: item for item in storage_usage()}
    assert set(entries) == set(STORAGE_SUBDIRECTORIES)
    assert entries["original"] == {
        "directory": "original",
        "totalBytes": 5,
        "fileCount": 1,
    }
    assert entries["preview"]["totalBytes"] == 0
    assert entries["preview"]["fileCount"] == 0


def test_runtime_config_masks_secrets(flask_app):
    with flask_app.app_context():
        entries = {
            (item["category"], item["key"]): item["value"]
            for item in runtime_config()
        }
    assert entries[("认证", "JWT密钥")] in {"已配置", "未配置"}
    secret = str(flask_app.config.get("DMS_JWT_SECRET", ""))
    if secret:
        rendered = " ".join(str(value) for value in entries.values())
        assert secret not in rendered


def test_system_info_reports_python_backend(flask_app):
    with flask_app.app_context():
        info = system_info()
    assert BACKEND_KIND == "PYTHON"
    assert API_CONTRACT_VERSION == "1.30.0"
    assert info["backendKind"] == "PYTHON"
    assert info["contractVersion"] == "1.30.0"
    assert isinstance(info["storageRoot"], str)
    assert isinstance(info["workingDirectory"], str)
    assert info["monitoringEnabled"] is False


def test_metric_sampler_config_floors():
    sampler = MetricSampler(None, interval_seconds=5, retention_days=0)
    assert sampler._interval_seconds == 10
    assert sampler._retention_days == 1
    assert sampler.running is False


def test_monitoring_unavailable_without_psutil(flask_app, monkeypatch):
    import dms.services.system_management_service as service

    monkeypatch.setattr(service, "PSUTIL_AVAILABLE", False)
    with flask_app.app_context():
        with pytest.raises(MetricUnavailableError):
            service.current_metrics()
        with pytest.raises(MetricUnavailableError):
            service.process_metrics()


# ===== 接口级（B2 惯例：需 DMS_TEST_* 环境变量）===============================


def _expected_monitor_status() -> int:
    """psutil 缺失环境下监控接口按契约返回 503 METRIC_UNAVAILABLE。"""
    return 200 if PSUTIL_AVAILABLE else 503


@pytest.mark.parametrize("url", SYSTEM_READONLY_URLS)
def test_plain_user_forbidden_on_system_routes(b2_client, token_for, url):
    response = b2_client.get(url, headers=_auth(token_for("user")))
    assert response.status_code == 403


def test_plain_user_forbidden_on_maintenance_routes(
    b2_app, b2_client, token_for
):
    token = _auth(token_for("user"))
    assert (
        b2_client.post(
            "/api/v1/system/maintenance/preview-cache", headers=token
        ).status_code
        == 403
    )
    with b2_app.app_context():
        admin_id = db.session.scalar(
            select(User.id).where(User.username == "admin")
        )
    assert (
        b2_client.post(
            f"/api/v1/system/users/{admin_id}/force-offline", headers=token
        ).status_code
        == 403
    )
    assert (
        b2_client.post("/api/v1/backup/test-target", headers=token).status_code
        == 403
    )
    assert (
        b2_client.get(
            "/api/v1/system/logs/download", headers=token
        ).status_code
        == 403
    )
    assert (
        b2_client.get("/api/v1/audit/logs", headers=token).status_code == 403
    )


def test_admin_reads_system_routes(b2_client, login_token):
    headers = _auth(login_token)
    for url in SYSTEM_READONLY_URLS:
        response = b2_client.get(url, headers=headers)
        assert response.status_code == _expected_monitor_status(), url
    # 健康体检固定五项
    health = b2_client.get("/api/v1/system/health", headers=headers)
    items = health.get_json()["data"]["items"]
    assert {item["item"] for item in items} == {
        "DATABASE",
        "STORAGE_WRITABLE",
        "LIBREOFFICE",
        "BACKUP_TARGET",
        "STORAGE_ENCRYPTION",
    }


@pytest.mark.parametrize(
    "query",
    ["hours=0", "hours=721", "hours=abc", "hours=-1"],
)
def test_metrics_history_validates_hours(b2_client, login_token, query):
    response = b2_client.get(
        f"/api/v1/system/metrics/history?{query}",
        headers=_auth(login_token),
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.parametrize("query", ["lines=0", "lines=5001", "lines=abc"])
def test_system_logs_validates_lines(b2_client, login_token, query):
    response = b2_client.get(
        f"/api/v1/system/logs?{query}", headers=_auth(login_token)
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"


def test_metric_history_filters_by_hours(b2_app, b2_client, login_token):
    now = datetime.now(timezone.utc).replace(tzinfo=None)
    with b2_app.app_context():
        db.session.add_all(
            [
                MetricSample(
                    sampled_at=now - timedelta(minutes=5),
                    cpu_percent=11.0,
                    mem_used_bytes=1,
                    mem_total_bytes=2,
                ),
                MetricSample(
                    sampled_at=now - timedelta(hours=3),
                    cpu_percent=22.0,
                    mem_used_bytes=1,
                    mem_total_bytes=2,
                ),
                MetricSample(
                    sampled_at=now - timedelta(hours=48),
                    cpu_percent=99.0,
                    mem_used_bytes=1,
                    mem_total_bytes=2,
                ),
            ]
        )
        db.session.commit()
    response = b2_client.get(
        "/api/v1/system/metrics/history?hours=24", headers=_auth(login_token)
    )
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["hours"] == 24
    cpu_values = [entry["cpuPercent"] for entry in data["entries"]]
    assert 11.0 in cpu_values
    assert 22.0 in cpu_values
    assert 99.0 not in cpu_values


def test_metric_sampler_tick_writes_and_prunes(b2_app):
    if not PSUTIL_AVAILABLE:
        pytest.skip("psutil 未安装")
    now = datetime.now(timezone.utc).replace(tzinfo=None)
    with b2_app.app_context():
        db.session.add(
            MetricSample(
                sampled_at=now - timedelta(days=3),
                cpu_percent=88.0,
                mem_used_bytes=1,
                mem_total_bytes=2,
            )
        )
        db.session.commit()
    sampler = MetricSampler(b2_app, interval_seconds=10, retention_days=1)
    sampler._tick_once()
    with b2_app.app_context():
        cpu_values = db.session.scalars(select(MetricSample.cpu_percent)).all()
    assert 88.0 not in cpu_values
    assert len(cpu_values) == 1


def test_logs_download_404_when_disabled(b2_app, b2_client, login_token):
    b2_app.config["DMS_LOG_FILE"] = "OFF"
    response = b2_client.get(
        "/api/v1/system/logs/download", headers=_auth(login_token)
    )
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"


def test_logs_download_serves_file(
    b2_app, b2_client, login_token, tmp_path: Path
):
    log = tmp_path / "dms-server.log"
    log.write_text("2026-01-01 [INFO] hello\n", encoding="utf-8")
    b2_app.config["DMS_LOG_FILE"] = str(log)
    response = b2_client.get(
        "/api/v1/system/logs/download", headers=_auth(login_token)
    )
    assert response.status_code == 200
    assert "attachment" in response.headers["Content-Disposition"]
    assert log.name in response.headers["Content-Disposition"]
    b2_app.config["DMS_LOG_FILE"] = "OFF"


def test_clear_preview_cache_endpoint_audits(b2_app, b2_client, login_token):
    response = b2_client.post(
        "/api/v1/system/maintenance/preview-cache", headers=_auth(login_token)
    )
    assert response.status_code == 200
    assert response.get_json()["data"] == {
        "removedFiles": 0,
        "removedBytes": 0,
    }
    with b2_app.app_context():
        audit = db.session.scalar(
            select(AuditLog).where(
                AuditLog.action_type == "PREVIEW_CACHE_CLEAR"
            )
        )
    assert audit is not None
    assert audit.operation_result == "SUCCESS"


def test_force_offline_invalidates_target_token(
    b2_app, b2_client, login_token, b2_password
):
    victim_token = b2_client.post(
        "/api/v1/auth/login",
        json={
            "username": "user",
            "password": b2_password,
            "keepSignedIn": False,
        },
    ).get_json()["data"]["accessToken"]
    assert (
        b2_client.get(
            "/api/v1/users/me", headers=_auth(victim_token)
        ).status_code
        == 200
    )
    with b2_app.app_context():
        victim_id = db.session.scalar(
            select(User.id).where(User.username == "user")
        )
        before = db.session.scalar(
            select(User.auth_version).where(User.id == victim_id)
        )

    # 非法/不存在/自我保护
    assert (
        b2_client.post(
            "/api/v1/system/users/abc/force-offline", headers=_auth(login_token)
        ).status_code
        == 400
    )
    assert (
        b2_client.post(
            "/api/v1/system/users/999999/force-offline",
            headers=_auth(login_token),
        ).status_code
        == 404
    )
    with b2_app.app_context():
        admin_id = db.session.scalar(
            select(User.id).where(User.username == "admin")
        )
    self_offline = b2_client.post(
        f"/api/v1/system/users/{admin_id}/force-offline",
        headers=_auth(login_token),
    )
    assert self_offline.status_code == 403
    assert self_offline.get_json()["code"] == "USER_SELF_PROTECTION"

    response = b2_client.post(
        f"/api/v1/system/users/{victim_id}/force-offline",
        headers=_auth(login_token),
    )
    assert response.status_code == 200
    payload = response.get_json()["data"]
    assert payload["username"] == "user"
    assert payload["authVersion"] == before + 1

    # 旧令牌立即失效，且管理端令牌不受影响
    rejected = b2_client.get(
        "/api/v1/users/me", headers=_auth(victim_token)
    )
    assert rejected.status_code == 401
    assert rejected.get_json()["code"] == "AUTH_VERSION_MISMATCH"
    assert (
        b2_client.get("/api/v1/users/me", headers=_auth(login_token)).status_code
        == 200
    )
    with b2_app.app_context():
        audit = db.session.scalar(
            select(AuditLog).where(AuditLog.action_type == "FORCE_OFFLINE")
        )
    assert audit is not None
    assert audit.target_name == "user"


def test_backup_test_target_reports_and_audits(
    b2_app, b2_client, login_token, tmp_path: Path
):
    from dms.services.backup_service import BackupConfig, BackupEngine

    original = b2_app.extensions["dms_backup_engine"]
    original.stop_scheduler()
    engine = BackupEngine(
        storage_root=original.storage_root,
        business_timezone=b2_app.config["DMS_BUSINESS_TIMEZONE"],
        database_url=b2_app.config["SQLALCHEMY_DATABASE_URI"],
        backup_config_path=tmp_path / "dms-backup-config.json",
        env_path=tmp_path / ".env",
    )
    b2_app.extensions["dms_backup_engine"] = engine
    BackupConfig(local_path=str(tmp_path / "backup-root")).save(
        engine.backup_config_path
    )

    response = b2_client.post(
        "/api/v1/backup/test-target", headers=_auth(login_token)
    )
    assert response.status_code == 200
    assert response.get_json()["data"]["ok"] is True
    with b2_app.app_context():
        audit = db.session.scalar(
            select(AuditLog).where(
                AuditLog.action_type == "BACKUP_TEST_TARGET"
            )
        )
    assert audit is not None
    assert audit.operation_result == "SUCCESS"


def _make_role_with_bits(b2_app, name: str, **bits) -> Role:
    role = Role(role_name=name, sort_no=10, status="ENABLED", **bits)
    db.session.add(role)
    db.session.flush()
    return role


def _make_admin_user(b2_app, username: str, role: Role | None) -> User:
    org_name = "机关"
    user = User(
        username=username,
        password_hash=hash_password("Placeholder-Password-1"),
        real_name=username,
        organization_name=org_name,
        role_code="ADMIN",
        security_level="TOP_SECRET",
        status="ENABLED",
        role_id=role.id if role else None,
    )
    db.session.add(user)
    db.session.flush()
    return user


def test_audit_and_system_modules_follow_manage_bits(b2_app, b2_client):
    """管理位决定页签可见性与对应接口访问（契约1.23）。

    - 无角色受限管理员：仅后台管理+机构用户管理，审计/系统接口均 403；
    - manage_audit_log：补 AUDIT_LOG 页签与审计接口；
    - manage_system：补 SYSTEM_MANAGEMENT 页签与系统接口，审计仍 403。
    """
    from dms.services.user_service import allowed_modules_for_user

    with b2_app.app_context():
        none_role = _make_role_with_bits(b2_app, "无管理位角色")
        audit_role = _make_role_with_bits(
            b2_app, "审计管理位角色", manage_audit_log=True
        )
        system_role = _make_role_with_bits(
            b2_app, "系统管理位角色", manage_system=True
        )
        _make_admin_user(b2_app, "sys_none", none_role)
        _make_admin_user(b2_app, "sys_audit", audit_role)
        _make_admin_user(b2_app, "sys_system", system_role)
        db.session.commit()

        for username, expected in (
            ("sys_none", ["BACKEND_MANAGEMENT", "ORGANIZATION_USER_MANAGEMENT"]),
            (
                "sys_audit",
                [
                    "BACKEND_MANAGEMENT",
                    "AUDIT_LOG",
                    "ORGANIZATION_USER_MANAGEMENT",
                ],
            ),
            (
                "sys_system",
                [
                    "BACKEND_MANAGEMENT",
                    "ORGANIZATION_USER_MANAGEMENT",
                    "SYSTEM_MANAGEMENT",
                ],
            ),
        ):
            user = db.session.scalar(
                select(User).where(User.username == username)
            )
            assert allowed_modules_for_user(user) == expected

    from dms.security.jwt_tokens import issue_access_token

    def token_of(username: str) -> dict[str, str]:
        with b2_app.app_context():
            user = db.session.scalar(
                select(User).where(User.username == username)
            )
            token, _ = issue_access_token(user, keep_signed_in=False)
        return _auth(token)

    audit_url = "/api/v1/audit/logs?page=1&pageSize=10"
    system_url = "/api/v1/system/info"
    logs_download_url = "/api/v1/system/logs/download"
    assert (
        b2_client.get(audit_url, headers=token_of("sys_none")).status_code
        == 403
    )
    assert (
        b2_client.get(system_url, headers=token_of("sys_none")).status_code
        == 403
    )
    assert (
        b2_client.get(audit_url, headers=token_of("sys_audit")).status_code
        == 200
    )
    # 日志下载并入日志审计权限：审计位可访问（未启用日志文件时 404 而非 403）
    download = b2_client.get(
        logs_download_url, headers=token_of("sys_audit")
    )
    assert download.status_code in {200, 404}
    assert (
        b2_client.get(system_url, headers=token_of("sys_audit")).status_code
        == 403
    )
    assert (
        b2_client.get(system_url, headers=token_of("sys_system")).status_code
        == 200
    )
    assert (
        b2_client.get(audit_url, headers=token_of("sys_system")).status_code
        == 403
    )


def test_role_crud_roundtrip_manage_bits(b2_client, login_token):
    payload = {
        "roleName": "契约123管理位角色",
        "sortNo": 10,
        "status": "ENABLED",
        "remark": None,
        "isDefault": False,
        "manageCategoryEdit": False,
        "manageOrgEdit": False,
        "manageUserEdit": False,
        "manageSystemManage": True,
        "manageAuditLog": False,
        "manageRoleEdit": False,
        "managedOrgId": None,
    }
    created = b2_client.post(
        "/api/v1/roles", json=payload, headers=_auth(login_token)
    )
    assert created.status_code == 200
    role_id = created.get_json()["data"]["id"]

    detail = b2_client.get(
        f"/api/v1/roles/{role_id}", headers=_auth(login_token)
    )
    role = detail.get_json()["data"]
    assert role["manageSystemManage"] is True
    assert role["manageAuditLog"] is False
    assert role["manageRoleEdit"] is False

    updated = b2_client.put(
        f"/api/v1/roles/{role_id}",
        json={
            **payload,
            "roleName": "契约123管理位角色改",
            "manageAuditLog": True,
            "manageRoleEdit": True,
        },
        headers=_auth(login_token),
    )
    assert updated.status_code == 200
    assert updated.get_json()["data"]["manageAuditLog"] is True
    assert updated.get_json()["data"]["manageRoleEdit"] is True