Aislo/resources/tester/test_b06_bulk_designs.py

"""측점 설계 묶음 저장 — 한 행짜리 함수와 **같은 결과**를 내는지 본다.

왜 (2026-09-06) — [저장]이 측점마다 `SELECT`+`UPDATE` 두 왕복을 냈고, 원격 DB 라
22행이면 670ms 였다. 세 문장으로 묶었는데 값이 달라지면 안 된다.
DB 는 가짜 커서로 대신한다 — 검사 대상은 **어떤 SQL 을 몇 번 내고, 어떤 JSON 이 되는가**다.
"""

import asyncio
import json

from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs


class FakeCursor:
    def __init__(self, rows):
        self._rows = rows
        self.calls = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        return False

    async def execute(self, sql, params=None):
        self.calls.append((" ".join(sql.split()), list(params) if params else []))

    async def fetchall(self):
        return self._rows


class FakeConnection:
    def __init__(self, rows):
        self.cursor_obj = FakeCursor(rows)

    def cursor(self):
        return self.cursor_obj


def _rows():
    """측점 셋 — 하나는 design 이 이미 있고, 하나는 비었고, 하나는 문자열 JSON."""
    return [
        (11, 0.0, {"design": {"ground_type": "soil"}, "summary": "keep"}),
        (12, 20.0, {}),
        (13, 40.0, json.dumps({"design": {"paved": True}})),
    ]


def _written(connection):
    """UPDATE 문에 실린 (id, data) 짝을 돌려준다."""
    for sql, params in connection.cursor_obj.calls:
        if sql.startswith("UPDATE cross_sections SET data = CASE"):
            pairs = {}
            # params = [id, blob, id, blob, …, id, id, …]
            half = len(params) // 3 * 2
            for index in range(0, half, 2):
                pairs[params[index]] = json.loads(params[index + 1])
            return pairs
    return {}


def test_한_문장으로_여러_행을_쓴다():
    connection = FakeConnection(_rows())
    written = asyncio.run(
        merge_cross_section_designs(
            connection,
            route_id=7,
            entries=[(0.0, {"a": 1}), (20.0, {"b": 2}), (40.0, {"c": 3})],
            replace=True,
        )
    )
    assert written == 3
    sqls = [sql for sql, _ in connection.cursor_obj.calls]
    # SELECT 한 번 + UPDATE 한 번 = 왕복 두 번. 행 수와 무관해야 한다.
    assert len(sqls) == 2, sqls
    assert sqls[0].startswith("SELECT id, chainage_m, data")
    assert _written(connection) == {
        11: {"design": {"a": 1}, "summary": "keep"},
        12: {"design": {"b": 2}},
        13: {"design": {"c": 3}},
    }


def test_patch_는_기존_design_을_보존한다():
    connection = FakeConnection(_rows())
    asyncio.run(
        merge_cross_section_designs(
            connection,
            route_id=7,
            entries=[(0.0, {"cut_area_m2": 1.5}), (40.0, {"cut_area_m2": 2.5})],
            replace=False,
        )
    )
    written = _written(connection)
    # replace=False 는 키만 얹는다 — 옛 design 과 형제 키(summary)가 남아야 한다.
    assert written[11] == {
        "design": {"ground_type": "soil", "cut_area_m2": 1.5},
        "summary": "keep",
    }
    assert written[13] == {"design": {"paved": True, "cut_area_m2": 2.5}}


def test_측점_허용오차는_1cm():
    connection = FakeConnection(_rows())
    asyncio.run(
        merge_cross_section_designs(
            connection, route_id=7, entries=[(20.005, {"x": 1})], replace=True
        )
    )
    assert list(_written(connection)) == [12]

    far = FakeConnection(_rows())
    asyncio.run(
        merge_cross_section_designs(far, route_id=7, entries=[(20.5, {"x": 1})], replace=True)
    )
    # 1cm 밖이면 붙일 행이 없다 — project_id 가 없으므로 아무 것도 안 쓴다.
    assert not [sql for sql, _ in far.cursor_obj.calls if sql.startswith("UPDATE")]


def test_행이_없으면_project_id_로_새로_만든다():
    connection = FakeConnection(_rows())
    written = asyncio.run(
        merge_cross_section_designs(
            connection,
            route_id=7,
            entries=[(999.0, {"x": 1})],
            replace=True,
            project_id="11111111-2222-3333-4444-555555555555",
        )
    )
    assert written == 1
    inserts = [sql for sql, _ in connection.cursor_obj.calls if sql.startswith("INSERT")]
    assert len(inserts) == 1


def test_빈_목록은_왕복을_내지_않는다():
    connection = FakeConnection(_rows())
    assert (
        asyncio.run(merge_cross_section_designs(connection, route_id=7, entries=[], replace=True))
        == 0
    )
    assert connection.cursor_obj.calls == []