DataOps-platform/tests/test_hopms_import.py

import sys
from pathlib import Path

from docx import Document

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from scripts.import_hopms_dataset import (
    build_dry_run_report,
    import_tables,
    parse_hopms_docx,
)


def _write_sample_doc(path: Path) -> None:
    doc = Document()
    doc.add_paragraph("门急诊业务")
    doc.add_paragraph("患者基本信息")
    doc.add_paragraph("用于对患者在挂号环节产生的挂号数据进行上传。")
    table = doc.add_table(rows=3, cols=6)
    headers = ["字段中文名", "字段名", "类型", "字节", "填报 / 要求", "说明"]
    for idx, header in enumerate(headers):
        table.rows[0].cells[idx].text = header
    table.rows[1].cells[0].text = "医疗机构代码"
    table.rows[1].cells[1].text = "YLJGDM"
    table.rows[1].cells[2].text = "字符串"
    table.rows[1].cells[3].text = "22"
    table.rows[1].cells[4].text = "必填"
    table.rows[1].cells[5].text = "复合主键。医疗机构在HOPMs的唯一识别码"
    table.rows[2].cells[0].text = "挂号时间"
    table.rows[2].cells[1].text = "GHSJ"
    table.rows[2].cells[2].text = "日期时间"
    table.rows[2].cells[3].text = ""
    table.rows[2].cells[4].text = "应填"
    table.rows[2].cells[5].text = "患者挂号时间"

    doc.add_paragraph("CV5501.12 入院病情代码")
    dict_table = doc.add_table(rows=2, cols=3)
    for idx, header in enumerate(["值", "值含义", "说明"]):
        dict_table.rows[0].cells[idx].text = header
    dict_table.rows[1].cells[0].text = "1"
    dict_table.rows[1].cells[1].text = "有"
    dict_table.rows[1].cells[2].text = "示例代码"
    doc.save(path)


def test_parse_hopms_docx_extracts_field_definition_tables(tmp_path):
    docx_path = tmp_path / "sample.docx"
    _write_sample_doc(docx_path)

    tables = parse_hopms_docx(docx_path)

    assert len(tables) == 1
    table = tables[0]
    assert table["table_info"] == {
        "name_zh": "患者基本信息",
        "name_en": "patient_demographics",
        "description": "用于对患者在挂号环节产生的挂号数据进行上传。",
    }
    assert table["columns"][0] == {
        "name_zh": "医疗机构代码",
        "name_en": "YLJGDM",
        "data_type": "varchar(22)",
        "is_primary": True,
        "nullable": False,
        "comment": "复合主键。医疗机构在HOPMs的唯一识别码",
        "requirement": "必填",
        "length": "22",
    }
    assert table["columns"][1]["data_type"] == "timestamp"
    assert table["columns"][1]["nullable"] is True


def test_build_dry_run_report_shapes_backend_payload(tmp_path):
    docx_path = tmp_path / "sample.docx"
    _write_sample_doc(docx_path)
    tables = parse_hopms_docx(docx_path)

    report = build_dry_run_report(tables, source_path=docx_path)

    assert report["summary"]["table_count"] == 1
    assert report["summary"]["column_count"] == 2
    assert report["summary"]["tables_missing_name_en"] == 0
    payload = report["tables"][0]["backend_payload"]
    assert payload["name_zh"] == "患者基本信息"
    assert payload["name_en"] == "patient_demographics"
    assert payload["type"] == "table"
    assert payload["category"] == "HOPMs标准数据集"
    assert payload["parsed_data"] == [
        {
            "name_zh": "医疗机构代码",
            "name_en": "YLJGDM",
            "data_type": "varchar(22)",
            "describe": "复合主键。医疗机构在HOPMs的唯一识别码",
        },
        {
            "name_zh": "挂号时间",
            "name_en": "GHSJ",
            "data_type": "timestamp",
            "describe": "患者挂号时间",
        },
    ]


class _FakeResponse:
    def __init__(self, payload, status_code=200):
        self._payload = payload
        self.status_code = status_code
        self.text = str(payload)

    def raise_for_status(self):
        if self.status_code >= 400:
            raise RuntimeError(f"HTTP {self.status_code}")

    def json(self):
        return self._payload


class _FakeSession:
    def __init__(self):
        self.calls = []

    def post(self, url, json=None, timeout=None):
        self.calls.append({"url": url, "json": json, "timeout": timeout})
        if url.endswith("/api/system/auth/login"):
            return _FakeResponse({"code": 200, "data": {"username": "testuser"}})
        if url.endswith("/api/bd/list"):
            return _FakeResponse({"code": 200, "data": {"records": []}})
        if url.endswith("/api/bd/save"):
            return _FakeResponse(
                {
                    "code": 200,
                    "data": {
                        "id": 123,
                        "name_zh": json["name_zh"],
                        "name_en": json["name_en"],
                    },
                }
            )
        if url.endswith("/api/bd/update"):
            return _FakeResponse(
                {
                    "code": 200,
                    "data": {
                        "id": json["id"],
                        "name_zh": json["name_zh"],
                        "name_en": json["name_en"],
                    },
                }
            )
        raise AssertionError(f"Unexpected URL: {url}")


def test_import_tables_logs_in_and_posts_backend_payload(tmp_path):
    docx_path = tmp_path / "sample.docx"
    _write_sample_doc(docx_path)
    report = build_dry_run_report(parse_hopms_docx(docx_path), source_path=docx_path)
    session = _FakeSession()

    result = import_tables(
        report["tables"],
        base_url="https://example.test/",
        username="testuser",
        password="testpassword123",
        session=session,
        skip_login=False,
    )

    assert result["summary"] == {
        "attempted": 1,
        "succeeded": 1,
        "failed": 0,
        "skipped": 0,
    }
    assert session.calls[0]["url"] == "https://example.test/api/system/auth/login"
    assert session.calls[0]["json"] == {
        "username": "testuser",
        "password": "testpassword123",
    }
    assert session.calls[1]["url"] == "https://example.test/api/bd/list"
    assert session.calls[1]["json"] == {
        "current": 1,
        "size": 10,
        "name_en": "patient_demographics",
    }
    assert session.calls[2]["url"] == "https://example.test/api/bd/save"
    assert session.calls[2]["json"]["name_en"] == "patient_demographics"
    assert session.calls[2]["json"]["parsed_data"][0]["name_en"] == "YLJGDM"


def test_import_tables_can_skip_login_when_endpoint_is_broken(tmp_path):
    docx_path = tmp_path / "sample.docx"
    _write_sample_doc(docx_path)
    report = build_dry_run_report(parse_hopms_docx(docx_path), source_path=docx_path)
    session = _FakeSession()

    result = import_tables(
        report["tables"],
        base_url="https://example.test/",
        username="",
        password="",
        session=session,
        skip_login=True,
    )

    assert result["summary"] == {
        "attempted": 1,
        "succeeded": 1,
        "failed": 0,
        "skipped": 0,
    }
    assert len(session.calls) == 2
    assert session.calls[0]["url"] == "https://example.test/api/bd/list"
    assert session.calls[1]["url"] == "https://example.test/api/bd/save"


class _ExistingDomainSession(_FakeSession):
    def post(self, url, json=None, timeout=None):
        if url.endswith("/api/bd/list"):
            self.calls.append({"url": url, "json": json, "timeout": timeout})
            return _FakeResponse(
                {
                    "code": 200,
                    "data": {
                        "records": [
                            {
                                "id": 456,
                                "name_zh": "患者基本信息",
                                "name_en": "patient_demographics",
                            }
                        ]
                    },
                }
            )
        return super().post(url, json=json, timeout=timeout)


def test_import_tables_skips_existing_domain_by_name_en(tmp_path):
    docx_path = tmp_path / "sample.docx"
    _write_sample_doc(docx_path)
    report = build_dry_run_report(parse_hopms_docx(docx_path), source_path=docx_path)
    session = _ExistingDomainSession()

    result = import_tables(
        report["tables"],
        base_url="https://example.test/",
        session=session,
    )

    assert result["summary"] == {
        "attempted": 1,
        "succeeded": 0,
        "failed": 0,
        "skipped": 1,
    }
    assert len(session.calls) == 1
    assert result["results"][0]["status"] == "skipped"
    assert result["results"][0]["existing_id"] == 456