Files
Aislo/resources/tester/test_763_2_content_key.py
T
eomsangdonandClaude Opus 5.5 6af3f990ff fix(763-2): 같은 서피스 = 파일 내용 지문 — 비용면 · 계류 · 초기 종단 · 초기 종단 목표 = 고른 지형
- `file_digest()`(derived_store) — blake2b 내용 지문 · 한 프로세스 안 경로 · 시각 · 크기로 기억
- 비용면 서명 · 계류 지문 · 초기 종단 `stat_marks` = 내용 지문(옮겨 심기 · 동기화 · 다시 짓기로 시각만 바뀌면 같은 열쇠)
- 초기 종단 표식 = 확정 지형이 읽는 파일만(`_source_npz_paths` · `*_{필터}*.npz` glob 걷음)
- `balanced_profile_of` 에 selection — 추천 초기 종단 목표 계획고 = 지반선과 같은 고른 지형(3h)
- 열쇠 꼴이 바뀌어 기존 추천 · 초기 종단 저장본은 다시 셈(763-5 몫)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 13:23:31 +09:00

117 lines
4.5 KiB
Python

"""763-2 같은 서피스 = 내용 지문 — 시각만 바뀌면 같은 열쇠 · 한 바이트 바꿈 = 다른 열쇠 ·
초기 종단 표식 = 확정 지형이 읽는 파일만 · 초기 종단 목표 = 고른 지형.
./venv/Scripts/python.exe -m pytest resources/tester/test_763_2_content_key.py -q
"""
import asyncio
import os
import shutil
from pathlib import Path
from uuid import uuid4
from test_763_1_confirmed_terrain import _models
from B05_Profile import B05_Profile_Engine_InitialProfile as ip
from B05_Profile import B05_Profile_Engine_Solver as solver
from B05_Profile import B05_Profile_Router_Replan as replan
from B05_Profile import B05_Profile_Router_RouteInitial as router
from common_util.common_util_derived_store import file_digest
ROOT = Path(__file__).resolve().parents[2]
def _retime(folder: Path, seconds: int) -> None:
for path in folder.iterdir():
os.utime(path, (seconds, seconds))
def _flip_byte(path: Path) -> None:
data = bytearray(path.read_bytes())
data[-1] ^= 0xFF
path.write_bytes(bytes(data))
def test_파일_지문_내용만(tmp_path) -> None:
a, b = tmp_path / "a.bin", tmp_path / "b.bin"
a.write_bytes(b"terrain" * 1000)
b.write_bytes(b"terrain" * 1000)
os.utime(b, (1_000_000, 1_000_000))
assert file_digest(a) == file_digest(b)
_flip_byte(b)
assert file_digest(a) != file_digest(b)
assert file_digest(tmp_path / "none.bin") is None
def test_비용면_서명_시각만_바뀌면_같음(tmp_path) -> None:
(tmp_path / "one").mkdir()
first = _models(tmp_path / "one")
for method in ("implicit", "dtm"):
solver._source_npz_paths(first, "classification", method, False) # 굳힌 격자 지음
second = tmp_path / "two"
shutil.copytree(first, second)
_retime(second, 2_000_000_000) # 옮겨 심기 · 동기화 = 시각만 바뀜
for method in ("implicit", "dtm"):
one = solver._cost_surface_signature(first, "classification", method, False)
two = solver._cost_surface_signature(second, "classification", method, False)
assert one == two
_flip_byte(second / "dtm_classification.npz")
assert solver._cost_surface_signature(
first, "classification", "dtm", False
) != solver._cost_surface_signature(second, "classification", "dtm", False)
# 초기 종단 표식도 내용
names = ["implicit_classification.npz", "implicit_classification_grid.npz"]
assert ip.stat_marks([first / n for n in names]) == ip.stat_marks([second / n for n in names])
def test_초기_종단_표식_확정_지형_파일만() -> None:
source = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial_Initial.py").read_text(
encoding="utf-8"
)
assert "models.glob(" not in source
assert '"models": ip.stat_marks(\n _source_npz_paths(' in source
# 목표 = 고른 지형(지반선 표본기와 같은 selection)
assert "request.criteria, curves, selection\n" in source
def test_균형_종단은_받은_지형으로(monkeypatch) -> None:
import B05_Profile.B05_Profile_Engine_Grade as grade
import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
import common_util.common_util_surface_sampler as sampler
opened: list[tuple] = []
async def road(_pid):
return "간선", 20, "일반"
async def standard(_pid):
return None
def build(_models, filter_key, method, smooth):
opened.append((filter_key, method, smooth))
return object()
def route_profile(*_a, **_k):
return {
"stations_m": [0.0],
"plan_m": [0.0],
"cut_m3": 0,
"fill_m3": 0,
"mass_end_m3": 0,
"seconds": 0,
}
def no_db():
raise AssertionError("고른 지형이 있으면 확정 params 를 다시 읽지 않음")
monkeypatch.setattr(replan, "_road_settings", road)
monkeypatch.setattr(router, "project_standard", standard)
monkeypatch.setattr(router, "get_db_pool", no_db)
monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
monkeypatch.setattr(sampler, "build_surface_sampler", build)
monkeypatch.setattr(profile, "route_profile", route_profile)
sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
points = [{"x": 0.0, "y": 0.0}, {"x": 10.0, "y": 0.0}]
got = asyncio.run(router.balanced_profile_of(uuid4(), Path("."), points, None, None, sheet))
assert got is not None and opened == [("sheet_laplace", "dtm", True)]