56a6e2da93
- src/dist 산출물 분리 원칙 정리(.gitignore, .gitattributes) - 루트 및 주요 폴더(config/scripts/prompts/tests/src, 런타임 폴더 5종)에 안내용 README.md 추가 - CHANGELOG.md, LICENSE, docs/ops/05-release-and-versioning.md 추가 - docs/README.md 문서 지도 갱신
291 lines
11 KiB
Python
291 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
"""pytest 공통 픽스처.
|
|
|
|
이 파일이 제공하는 것 두 가지.
|
|
|
|
1. **네트워크 차단** — 실수로 실제 네트워크 요청이 나가는 사고를 막는다. 소스 API·agy·
|
|
웹훅 전부 이 프로젝트에서는 네트워크를 만지므로, 픽스처 전용 테스트가 아닌 한
|
|
실제 소켓 연결은 전부 예외로 막는다(``@pytest.mark.network`` 로 표시한 테스트만 예외).
|
|
2. **임시 프로젝트 트리** — DB·로그·리포트·백업 경로가 전부 ``tmp_path`` 밑을 가리키는
|
|
:class:`~dmf_crawler.config.Config` 를 만들어 준다. 실제 프로젝트 루트(``data/``,
|
|
``reports/`` 등)를 절대 건드리지 않기 위해서다.
|
|
|
|
다른 모듈이 아직 구현되지 않았을 수 있으므로, 이 파일 자체는 ``dmf_crawler.models`` ·
|
|
``dmf_crawler.config`` · ``dmf_crawler.paths`` 세 개(이미 존재가 확정된 모듈)에만 의존한다.
|
|
그 밖의 모듈에 의존하는 픽스처는 픽스처 본문 안에서 ``pytest.importorskip`` 을 쓴다.
|
|
"""
|
|
|
|
import dataclasses
|
|
import hashlib
|
|
import json
|
|
import socket
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterator
|
|
|
|
import pytest
|
|
|
|
from dmf_crawler.config import (
|
|
BackupCfg,
|
|
Config,
|
|
LoggingCfg,
|
|
ReportCfg,
|
|
StorageCfg,
|
|
)
|
|
from dmf_crawler.models import (
|
|
COMPARE_FIELDS,
|
|
COMPARE_KEY_FIELDS,
|
|
DmfRecord,
|
|
PermitParts,
|
|
)
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 경로 상수
|
|
# --------------------------------------------------------------------------
|
|
FIXTURES_DIR: Path = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 1. 네트워크 차단
|
|
# --------------------------------------------------------------------------
|
|
class BlockedNetworkError(RuntimeError):
|
|
"""테스트 중 실제 네트워크 연결 시도를 감지했을 때 던진다.
|
|
|
|
이 예외가 뜨면 테스트가 ``httpx.MockTransport`` 등으로 목킹하지 않고
|
|
진짜 소켓을 열려 했다는 뜻이다. 진짜 네트워크가 정말 필요하면
|
|
``@pytest.mark.network`` 로 표시하라(기본 실행에서 제외된다).
|
|
"""
|
|
|
|
|
|
def _blocked(*args: Any, **kwargs: Any) -> Any:
|
|
raise BlockedNetworkError(
|
|
"테스트에서 실제 네트워크 연결이 시도됐다. "
|
|
"httpx.MockTransport 나 monkeypatch 로 응답을 목킹하라. "
|
|
"진짜 네트워크가 필요한 테스트는 @pytest.mark.network 로 표시하라."
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def block_network(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
|
|
"""``network`` 마커가 없는 모든 테스트에서 실제 소켓 연결을 차단한다.
|
|
|
|
``socket.socket.connect`` / ``connect_ex`` 를 클래스 레벨에서 막으므로
|
|
``httpx``·``urllib``·표준 ``socket`` 모듈을 통한 어떤 실제 연결 시도도 걸린다.
|
|
로컬 파일 I/O·SQLite 는 소켓을 쓰지 않으므로 영향받지 않는다.
|
|
"""
|
|
if request.node.get_closest_marker("network") is not None:
|
|
yield
|
|
return
|
|
monkeypatch.setattr(socket.socket, "connect", _blocked, raising=True)
|
|
monkeypatch.setattr(socket.socket, "connect_ex", _blocked, raising=True)
|
|
monkeypatch.setattr(socket, "create_connection", _blocked, raising=True)
|
|
yield
|
|
|
|
|
|
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
|
|
"""``network`` 마커 테스트는 ``-m network`` 를 명시하지 않는 한 자동 제외한다.
|
|
|
|
``pyproject.toml`` 의 마커 설명("기본 실행에서는 제외한다")을 실제로 강제한다.
|
|
"""
|
|
markexpr = config.getoption("markexpr", default="")
|
|
if "network" in markexpr:
|
|
return
|
|
skip_network = pytest.mark.skip(reason="network 마커 테스트는 기본 실행에서 제외한다(-m network 로 실행)")
|
|
for item in items:
|
|
if item.get_closest_marker("network") is not None:
|
|
item.add_marker(skip_network)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 2. 임시 프로젝트 트리 · 설정
|
|
# --------------------------------------------------------------------------
|
|
@pytest.fixture
|
|
def project_dirs(tmp_path: Path) -> dict[str, Path]:
|
|
"""DB·로그·리포트·백업·원문 아카이브용 임시 디렉터리 묶음을 만든다.
|
|
|
|
Returns:
|
|
dict[str, Path]: ``data``·``raw``·``state``·``logs``·``reports``·``backup`` 경로.
|
|
"""
|
|
dirs = {
|
|
"data": tmp_path / "data",
|
|
"raw": tmp_path / "data" / "raw",
|
|
"state": tmp_path / "state",
|
|
"logs": tmp_path / "logs",
|
|
"reports": tmp_path / "reports",
|
|
"backup": tmp_path / "backup",
|
|
}
|
|
for d in dirs.values():
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return dirs
|
|
|
|
|
|
@pytest.fixture
|
|
def test_config(project_dirs: dict[str, Path]) -> Config:
|
|
"""모든 경로가 ``tmp_path`` 밑을 가리키는 :class:`Config` 를 만든다.
|
|
|
|
실제 프로젝트 루트의 ``data/``·``reports/``·``backup/``·``logs/`` 를
|
|
절대 건드리지 않는다. ``storage.sqlite_path`` 등은 절대경로이므로
|
|
``paths.resolve_under_root`` 를 거쳐도 그대로 쓰인다(절대경로는 통과).
|
|
"""
|
|
return Config().with_overrides(
|
|
storage=StorageCfg(sqlite_path=str(project_dirs["data"] / "dmf.sqlite3")),
|
|
report=ReportCfg(output_dir=str(project_dirs["reports"])),
|
|
backup=BackupCfg(dir=str(project_dirs["backup"])),
|
|
logging=LoggingCfg(dir=str(project_dirs["logs"])),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def db_conn(test_config: Config):
|
|
"""마이그레이션까지 적용된 SQLite 연결. ``storage.db`` 가 아직 없으면 스킵한다."""
|
|
db = pytest.importorskip("dmf_crawler.storage.db")
|
|
from dmf_crawler import paths
|
|
|
|
backup_dir = test_config.backup_dir()
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
conn = db.connect(test_config.db_path())
|
|
db.apply_migrations(conn, paths.MIGRATIONS_DIR, backup_dir)
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 3. 픽스처 파일 로딩 도우미
|
|
# --------------------------------------------------------------------------
|
|
def load_fixture_text(name: str) -> str:
|
|
"""``tests/fixtures/<name>`` 을 UTF-8 텍스트로 읽는다.
|
|
|
|
Args:
|
|
name: 픽스처 파일명(예: ``"api_page_ok.json"``).
|
|
|
|
Returns:
|
|
str: 파일 내용.
|
|
"""
|
|
return (FIXTURES_DIR / name).read_text(encoding="utf-8")
|
|
|
|
|
|
def load_fixture_json(name: str) -> Any:
|
|
"""``tests/fixtures/<name>`` 을 JSON 으로 파싱해 돌려준다.
|
|
|
|
Args:
|
|
name: 픽스처 파일명(예: ``"agy_envelope_clean.json"``).
|
|
|
|
Returns:
|
|
Any: 파싱된 JSON 값.
|
|
"""
|
|
return json.loads(load_fixture_text(name))
|
|
|
|
|
|
@pytest.fixture
|
|
def fixtures_dir() -> Path:
|
|
"""``tests/fixtures/`` 절대경로."""
|
|
return FIXTURES_DIR
|
|
|
|
|
|
@pytest.fixture
|
|
def load_fixture() -> Callable[[str], str]:
|
|
"""픽스처 파일을 텍스트로 읽는 함수를 돌려주는 픽스처."""
|
|
return load_fixture_text
|
|
|
|
|
|
@pytest.fixture
|
|
def load_fixture_dict() -> Callable[[str], Any]:
|
|
"""픽스처 파일을 JSON 으로 읽는 함수를 돌려주는 픽스처."""
|
|
return load_fixture_json
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 4. DmfRecord 최소 생성 팩토리 — diff · integrity · repo 테스트가 공유한다
|
|
# --------------------------------------------------------------------------
|
|
_UNIT_SEP = "\x1f"
|
|
|
|
|
|
def _hash16(parts: "Iterator[str] | list[str]") -> str:
|
|
"""모델 계약의 해시 규칙(``docs/design/02-data-model.md`` §4.3)과 같은 방식."""
|
|
joined = _UNIT_SEP.join(parts)
|
|
return hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def _simple_key(display: str) -> str:
|
|
"""테스트 전용 간이 매칭키. 영숫자·한글만 남기고 소문자화한다.
|
|
|
|
실제 ``normalize.matching_key`` 와 완전히 같지는 않지만(법인격 어휘 제거 등은
|
|
하지 않는다), 공백·구두점·대소문자 차이를 흡수한다는 핵심 성질은 같다 —
|
|
diff·integrity 테스트가 필요로 하는 것은 그 성질뿐이다.
|
|
"""
|
|
return "".join(ch for ch in display if ch.isalnum()).lower()
|
|
|
|
|
|
def _build_dmf_record(
|
|
permit_no: str = "20200101-1-A-1-1",
|
|
*,
|
|
dup_index: int = 0,
|
|
ingredient_name: str = "테스트성분",
|
|
applicant: str = "테스트제약",
|
|
manufacturer: str = "Test Manufacturer Co.",
|
|
manufacture_place: str = "1 Test Rd, Testville",
|
|
countries: tuple[str, ...] = ("미국",),
|
|
permit_date: str = "2020-01-01",
|
|
permit_no_raw: str | None = None,
|
|
) -> DmfRecord:
|
|
"""테스트용 :class:`DmfRecord` 를 만든다.
|
|
|
|
``content_hash``·``identity_hash`` 는 실제로 ``compare_view()``/``key_view()`` 에서
|
|
계산하므로, 두 레코드를 만들어 필드 하나를 바꾸면 해시도 정확히 그만큼만 바뀐다
|
|
(diff 테스트가 기대하는 성질).
|
|
|
|
Args:
|
|
permit_no: 정규화된 등록번호.
|
|
dup_index: 중복 그룹 내 순번(0 이면 접미사 없음).
|
|
ingredient_name: 성분명 표시값.
|
|
applicant: 신청인 표시값.
|
|
manufacturer: 제조소명 표시값.
|
|
manufacture_place: 제조소 소재지 표시값.
|
|
countries: 제조국가 튜플(이미 정렬된 값을 넣는다).
|
|
permit_date: 발급일자 ``YYYY-MM-DD``.
|
|
permit_no_raw: 원문 등록번호. 생략하면 ``permit_no`` 와 같다.
|
|
|
|
Returns:
|
|
DmfRecord: content_hash·identity_hash 까지 채워진 완결 레코드.
|
|
"""
|
|
dmf_key = permit_no if dup_index == 0 else f"{permit_no}#{dup_index + 1}"
|
|
sorted_countries = tuple(sorted(countries))
|
|
permit = PermitParts(raw=permit_no_raw or permit_no, normalized=permit_no, fmt="standard")
|
|
|
|
rec = DmfRecord(
|
|
dmf_key=dmf_key,
|
|
permit_no=permit_no,
|
|
permit_no_raw=permit_no_raw or permit_no,
|
|
ingredient_name=ingredient_name,
|
|
ingredient_key=_simple_key(ingredient_name),
|
|
ingredient_base=_simple_key(ingredient_name),
|
|
is_micronized=False,
|
|
applicant=applicant,
|
|
applicant_key=_simple_key(applicant),
|
|
manufacturer=manufacturer,
|
|
manufacturer_key=_simple_key(manufacturer),
|
|
manufacture_place=manufacture_place,
|
|
manufacture_place_key=_simple_key(manufacture_place),
|
|
countries=sorted_countries,
|
|
sites=(),
|
|
permit_date=permit_date,
|
|
permit=permit,
|
|
raw={},
|
|
dup_index=dup_index,
|
|
)
|
|
content_hash = _hash16(rec.compare_view()[f] for f in COMPARE_FIELDS)
|
|
identity_hash = _hash16(rec.key_view()[f] for f in COMPARE_KEY_FIELDS)
|
|
return dataclasses.replace(rec, content_hash=content_hash, identity_hash=identity_hash)
|
|
|
|
|
|
@pytest.fixture
|
|
def record_factory() -> Callable[..., DmfRecord]:
|
|
""":class:`DmfRecord` 를 만드는 팩토리 함수를 돌려주는 픽스처.
|
|
|
|
``test_diff.py``·``test_integrity.py``·``test_repo_idempotency.py`` 가 공유한다.
|
|
같은 인자로 두 번 호출하면 ``content_hash``·``identity_hash`` 까지 완전히 같은
|
|
값이 나온다(결정론적) — diff 의 "동일" 판정 테스트가 이 성질에 의존한다.
|
|
"""
|
|
return _build_dmf_record
|