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

from __future__ import annotations

from sqlalchemy import Computed, UniqueConstraint

import dms.models  # noqa: F401
from dms.common.enums import (
    AllowedModule,
    AuditAction,
    AuditResult,
    AuditTarget,
    ContentExtractStatus,
    RoleCode,
    SecurityLevel,
    SECURITY_LEVEL_VALUES,
)
from dms.extensions import db


EXPECTED_TABLES = {
    "sys_organization",
    "sys_user",
    "doc_category",
    "doc_document",
    "doc_attachment_binding",
    "doc_document_relation",
    "doc_sub_plan_binding",
    "doc_permission",
    "sys_audit_log",
    "sys_role",
    "sys_role_permission",
    "sys_ip_entry",
    "sys_org_ip_assignment",
}
BUSINESS_TABLES = EXPECTED_TABLES - {"sys_audit_log"}
COMMON_COLUMNS = {
    "id",
    "created_by",
    "updated_by",
    "created_at",
    "updated_at",
    "deleted_at",
    "is_deleted",
    "row_version",
}


def test_expected_models_are_registered():
    assert set(db.metadata.tables) == EXPECTED_TABLES
    assert len(db.metadata.tables) == 13


def test_business_tables_have_common_columns():
    for table_name in BUSINESS_TABLES:
        columns = {column.name for column in db.metadata.tables[table_name].columns}
        assert COMMON_COLUMNS <= columns


def test_audit_log_has_append_only_exception_fields():
    columns = {
        column.name for column in db.metadata.tables["sys_audit_log"].columns
    }
    assert "created_at" in columns
    assert {"updated_at", "deleted_at", "is_deleted", "row_version"}.isdisjoint(
        columns
    )


def test_auth_version_field_exists_with_default_zero():
    column = db.metadata.tables["sys_user"].c.auth_version
    assert column.nullable is False
    assert str(column.server_default.arg) == "0"


def test_attachment_type_field_exists():
    column = db.metadata.tables["doc_document"].c.attachment_type
    assert column.nullable is True


def test_active_marker_columns_are_virtual_generated_columns():
    for table_name in (
        "doc_attachment_binding",
        "doc_permission",
        "sys_role",
        "sys_role_permission",
        "sys_ip_entry",
        "sys_org_ip_assignment",
        "sys_user",
        "doc_sub_plan_binding",
    ):
        column = db.metadata.tables[table_name].c.active_marker
        assert isinstance(column.computed, Computed)
        assert "IF(is_deleted = 0, 1, NULL)" in str(column.computed.sqltext)
        assert column.computed.persisted is False


def test_role_default_marker_is_virtual_generated_column():
    column = db.metadata.tables["sys_role"].c.default_marker
    assert isinstance(column.computed, Computed)
    assert "IF(is_default = 1 AND is_deleted = 0, 1, NULL)" in str(
        column.computed.sqltext
    )
    assert column.computed.persisted is False


def _unique_column_sets(table_name: str) -> set[tuple[str, ...]]:
    table = db.metadata.tables[table_name]
    return {
        tuple(column.name for column in constraint.columns)
        for constraint in table.constraints
        if isinstance(constraint, UniqueConstraint)
    }


def test_active_binding_unique_constraint():
    assert (
        "main_document_id",
        "attachment_document_id",
        "active_marker",
    ) in _unique_column_sets("doc_attachment_binding")


def test_active_permission_unique_constraint():
    assert (
        "document_id",
        "subject_type",
        "subject_id",
        "active_marker",
    ) in _unique_column_sets("doc_permission")


def test_role_tables_unique_constraints():
    assert ("role_name", "active_marker") in _unique_column_sets("sys_role")
    assert ("default_marker",) in _unique_column_sets("sys_role")
    assert (
        "role_id",
        "security_level",
        "dept_scope",
        "active_marker",
    ) in _unique_column_sets("sys_role_permission")


def test_ip_tables_unique_constraints():
    assert ("ip_value", "active_marker") in _unique_column_sets("sys_ip_entry")
    assert (
        "organization_id",
        "ip_entry_id",
        "active_marker",
    ) in _unique_column_sets("sys_org_ip_assignment")


def test_sub_plan_binding_unique_constraint_and_indexes():
    assert (
        "main_document_id",
        "sub_document_id",
        "active_marker",
    ) in _unique_column_sets("doc_sub_plan_binding")
    names = {
        index.name
        for index in db.metadata.tables["doc_sub_plan_binding"].indexes
    }
    assert {
        "ix_doc_sub_plan_binding_main_active_sort",
        "ix_doc_sub_plan_binding_sub_active",
    } <= names


def test_portal_user_external_id_unique_constraint():
    assert ("external_id", "active_marker") in _unique_column_sets("sys_user")


def test_no_foreign_key_uses_cascade_and_audit_has_no_foreign_keys():
    for table in db.metadata.tables.values():
        for foreign_key in table.foreign_keys:
            assert foreign_key.ondelete in {"RESTRICT", "NO ACTION"}
            assert foreign_key.ondelete != "CASCADE"
    assert not db.metadata.tables["sys_audit_log"].foreign_keys


def test_created_and_updated_by_have_no_foreign_keys():
    for table_name in BUSINESS_TABLES:
        table = db.metadata.tables[table_name]
        assert not table.c.created_by.foreign_keys
        assert not table.c.updated_by.foreign_keys


def test_required_permanent_unique_codes_exist():
    assert ("org_code", "active_marker") in _unique_column_sets("sys_organization")
    assert ("username",) in _unique_column_sets("sys_user")
    assert ("category_code", "active_marker") in _unique_column_sets("doc_category")


def test_critical_indexes_exist():
    expected = {
        "doc_document": {
            "ix_doc_document_type_active_updated",
            "ix_doc_document_file_hash",
            "ix_doc_document_attachment_type_active_updated",
        },
        "sys_audit_log": {
            "ix_sys_audit_log_created_at",
            "ix_sys_audit_log_user_created",
            "ix_sys_audit_log_organization_created",
            "ix_sys_audit_log_action_created",
            "ix_sys_audit_log_target_created",
            "ix_sys_audit_log_result_created",
            "ix_sys_audit_log_request_id",
        },
    }
    for table_name, names in expected.items():
        actual = {index.name for index in db.metadata.tables[table_name].indexes}
        assert names <= actual


def test_permission_model_has_no_inherited_source_column():
    assert "permission_source" not in db.metadata.tables["doc_permission"].c


def test_contract_enums_are_exact():
    assert {item.value for item in RoleCode} == {"USER", "ADMIN"}
    assert {item.value for item in ContentExtractStatus} == {
        "SUCCESS",
        "EMPTY",
        "UNSUPPORTED",
        "FAILED",
    }
    assert {item.value for item in AllowedModule} == {
        "DOCUMENT_BROWSER",
        "BACKEND_MANAGEMENT",
        "AUDIT_LOG",
        "ORGANIZATION_USER_MANAGEMENT",
    }
    assert {item.value for item in AuditResult} == {"SUCCESS", "FAILURE"}
    assert {item.value for item in AuditAction} == {
        "LOGIN",
        "LOGOUT",
        "VIEW_DOCUMENT",
        "DOWNLOAD_DOCUMENT",
        "UPLOAD_DOCUMENT",
        "BATCH_IMPORT",
        "EDIT_DOCUMENT",
        "DELETE_DOCUMENT",
        "CREATE_CATEGORY",
        "EDIT_CATEGORY",
        "DELETE_CATEGORY",
        "CHANGE_PERMISSION",
        "BIND_ATTACHMENT",
        "UNBIND_ATTACHMENT",
        "RESTORE_DOCUMENT",
        "CREATE_ORGANIZATION",
        "EDIT_ORGANIZATION",
        "DELETE_ORGANIZATION",
        "CREATE_USER",
        "EDIT_USER",
        "DELETE_USER",
        "RESET_USER_PASSWORD",
        "REORDER_DOCUMENT",
        "REORDER_ATTACHMENT",
        "REORDER_CATEGORY",
        "CREATE_ROLE",
        "EDIT_ROLE",
        "DELETE_ROLE",
        "CHANGE_ROLE_PERMISSION",
        "CHANGE_ROLE_USER",
        "DOCUMENT_PASSWORD_SET",
        "DOCUMENT_PASSWORD_REMOVE",
        "CREATE_IP_ENTRY",
        "EDIT_IP_ENTRY",
        "DELETE_IP_ENTRY",
        "ASSIGN_ORG_IP",
    }
    assert {item.value for item in AuditTarget} == {
        "AUTH",
        "CATEGORY",
        "DOCUMENT",
        "ATTACHMENT",
        "PERMISSION",
        "ATTACHMENT_BINDING",
        "ORGANIZATION",
        "USER",
        "ROLE",
        "IP_ENTRY",
    }
    assert SECURITY_LEVEL_VALUES == {
        SecurityLevel.PUBLIC: 10,
        SecurityLevel.INTERNAL: 20,
        SecurityLevel.SECRET: 30,
        SecurityLevel.CONFIDENTIAL: 40,
        SecurityLevel.TOP_SECRET: 50,
    }