Python 변환의 단위 테스트

Python 저장소는 pytest로 단위 테스트를 실행하고 CI checks에 결과를 포함할 수 있습니다. 이 저장소 테스트 경로는 배치 파이프라인을 대상으로 하며 스트리밍 파이프라인에는 지원하지 않습니다.

테스트는 실제 계산과 출력에 전달할 값을 검사하고, Foundry 데이터셋을 읽고 쓰는 핸들만 대체할 수 있습니다. CSV 입력과 기대 결과를 사용한 Polars, DuckDB, Pandas와 Spark 예제는 같은 필터를 실제 엔진에서 수행합니다.

CI 환경 설치

Checks의 condaPackRun은 해석한 환경의 패키지를 내려받아 압축 해제하고, checksum 또는 크기 검증 뒤 환경에 연결합니다. 패키지가 많을수록 설치 시간이 늘어납니다.

Run과 test의 환경 사양은 숨겨진 lockfile과 설치 캐시에 보관합니다. 설치 환경 캐시는 7일을 기준으로 사용하며 meta.yaml 변경 때 다시 구성합니다. 실제 잠금 파일의 이름과 재해석 조건은 사용 중인 템플릿에서 확인합니다.

테스트 플러그인

Transforms 저장소는 Python 하위 프로젝트의 build.gradle, Python library 저장소는 루트 build.gradle에 적용합니다.

Transforms 설정:

apply plugin: 'com.palantir.transforms.lang.python'
apply plugin: 'com.palantir.transforms.lang.python-defaults'
apply plugin: 'com.palantir.transforms.lang.pytest-defaults'

Library 설정:

apply plugin: 'com.palantir.transforms.lang.python-library'
apply plugin: 'com.palantir.transforms.lang.python-library-defaults'
apply plugin: 'com.palantir.transforms.lang.pytest-defaults'

condaLibraryPublish.onlyIf { versionDetails().commitDistance == 0 }

마지막 조건은 tag 이후 추가 commit이 없는 릴리스만 발행하는 예입니다. 테스트에서는 meta.yaml의 runtime 요구를 사용할 수 있고 추가 패키지는 Conda의 test.requires에 넣습니다.

로컬 Palantir 확장의 Palantir: Install Python environment로 .maestro를 만든 뒤 다음을 사용할 수 있습니다.

.maestro/bin/pytest

파일과 함수 발견

src/test 아래에 테스트를 두는 것이 권장됩니다. Pytest는 test_*.py 또는 *_test.py 파일의 test_ 함수들을 찾으며 Python unittest 방식의 테스트도 실행할 수 있습니다.

다음은 실패 메시지를 설명하기 위한 의도적인 실패 사례입니다. 정상 운영 테스트로 추가할 코드는 아닙니다.

def increment(number):
    return number + 1


def test_increment():
    assert increment(3) == 5

실제 계산은 4이고 기대값은 5이므로 실패하며 assertion 비교에서 assert 4 == 5를 확인할 수 있습니다.

DataFrame fixture의 기본 형태

다음은 두 행의 입력 프레임을 제공하는 pytest fixture입니다. 사용하는 엔진의 구현 하나를 선택하고 테스트 함수의 인자에 sample_frame을 선언하면 해당 프레임을 받을 수 있습니다. 업무 로직의 결과값은 아래의 CSV 테스트처럼 별도로 검사합니다.

Polars

import polars as pl
import pytest


@pytest.fixture
def sample_frame():
    return pl.DataFrame(
        [["a", 1], ["b", 2]], schema=["letter", "number"], orient="row"
    )

DuckDB

import duckdb
import pytest


@pytest.fixture
def sample_frame():
    with duckdb.connect() as connection:
        return connection.sql(
            "SELECT * FROM (VALUES ('a', 1), ('b', 2)) AS t(letter, number)"
        ).fetchdf()

Pandas

import pandas as pd
import pytest


@pytest.fixture
def sample_frame():
    return pd.DataFrame([["a", 1], ["b", 2]], columns=["letter", "number"])

Spark

import pytest


@pytest.fixture
def sample_frame(spark_session):
    return spark_session.createDataFrame(
        [["a", 1], ["b", 2]], ["letter", "number"]
    )

spark_session은 pytest-transforms가 제공하는 session 범위 fixture입니다. 테스트마다 새 session을 만드는 것이 아니므로 time zone 같은 설정을 바꾸면 다음 테스트에도 영향을 줍니다. 변경 전 값을 저장하고 테스트가 끝나기 전에 복원해야 합니다.

CSV 입력과 기대 결과

다음 두 파일을 저장소의 transforms-python/src/test/resources/에 둡니다.

• aircraft_mock.csv는 가상 항공기 다섯 행입니다.
• expected_filtered_aircraft.csv는 독립적으로 정한 기대 결과입니다.

업무 조건은 number_of_seats > 300이고 operating_status == "Yes"입니다. 300석인 EEE1과 운항하지 않는 BBB1은 제외하고 CCC1만 남습니다. 기대 파일은 계산 결과를 복사해 만드는 것이 아니라 이 조건에 맞춘 명시적 테스트 사례입니다.

Polars 테스트, DuckDB 테스트, Pandas 테스트, Spark 테스트에서 실제 계산과 출력 호출의 전체 값을 비교합니다. 각 계산 함수는 runtime에서도 같은 구현으로 등록하며 테스트가 비공개 decorator 내부 속성에 의존하지 않도록 구성합니다.

결과와 coverage

Checks에는 각 테스트가 PASSED, FAILED 또는 SKIPPED로 표시되고 기본적으로 접혀 있습니다. 펼치면 결과와 StdOut 및 StdErr를 볼 수 있습니다.

Coverage가 필요하면 기존 meta.yaml에 다음 test 요구를 병합합니다.

test:
  requires:
    - pytest-cov

transforms-python/src/pytest.ini에 대상 패키지와 기준을 지정합니다.

[pytest]
addopts = --cov=myproject --cov-report term --cov-fail-under=100

예시는 100% 미만이면 checks를 실패시킵니다. 실제 기준은 --cov-fail-under로 정하며 패키지 이름을 바꿨다면 myproject도 맞춥니다.

병렬 테스트

기본 실행은 순차적입니다. pytest-xdist 기반의 여러 프로세스를 사용하려면 플러그인을 적용한 뒤 numProcesses를 설정합니다.

apply plugin: 'com.palantir.transforms.lang.python'
apply plugin: 'com.palantir.transforms.lang.python-defaults'
apply plugin: 'com.palantir.transforms.lang.pytest-defaults'

tasks.pytest {
    numProcesses "3"
}

대기 테스트는 사용 가능한 worker에 배정되며 실행 순서가 보장되지 않습니다. 다른 테스트가 먼저 전역 상태를 바꿔 줄 것이라는 가정을 제거해야 합니다.

Style 및 Spark 검사

PEP8과 Pylint는 다음 플러그인으로 켤 수 있습니다.

apply plugin: 'com.palantir.conda.pep8'
apply plugin: 'com.palantir.conda.pylint'

Pylint 설정은 src/.pylintrc에서 관리합니다. 예를 들어 문서 문자열 메시지를 요구하지 않는 정책은 다음과 같습니다.

[MESSAGES CONTROL]
disable =
    missing-module-docstring,
    missing-function-docstring

모든 Pylint 설정이 Foundry에서 지원되는 것은 아닙니다. 실제 checks에 반영되지 않는 기능에 의존하지 않아야 합니다.

Spark의 정확성, 성능과 보안상 흔한 실수를 찾는 플러그인도 있습니다.

apply plugin: 'com.palantir.transforms.lang.antipattern-linter'

테스트가 켜지면 CI 로그에서 :transforms-python:pytest를 확인할 수 있습니다. 작은 예제 행은 Workbook 콘솔의 collect로 얻을 수 있지만 전체 큰 데이터를 드라이버로 모으지 않도록 합니다. PySpark schema는 Dataset Preview의 Columns > Copy > Copy PySpark Schema에서 가져올 수 있습니다.