Airflow with Test in CI

테스트란?

개발한 코드를 배포할 때 가장 중요한 것은?

→ 실제 서비스에서 제대로 동작하는 지 “테스트”하는 것

•

99.99999% 의 기업에서는 배포 전 테스트를 수행하며, 대부분 테스팅은 CI 파이프라인에서 자동화되어 있다.

Github Actions를 이용한 테스트 UI

•

통합 테스트의 경우 Github Actions를 통해 구현하기도 하지만, 테스트 단계에 따라 코드 레벨에서 많은 단위 테스트(unit test)를 구현하고, 직접 수행하며 실패시 디버깅을 한다.

Golang에서 테스트

•

golang에서는 테스트 프레임워크 내장

◦

go test 하면 현재 폴더의 *_test.go를 인식하고 일괄 실행

→ Python은 Pytest를 사용함

단위 테스트? 통합 테스트?

단위 테스트: 개별 Task 단위(단일 기능)로 테스트를 분할하여 수행하는 것

•

각 단위의 올바른 동작은 검증할 수 있지만, 여러 단위로 구성된 복잡한 시스템의 동작을 모두 검증하지는 않으므로 최종적으로는 통합 테스트를 작성해야함

왜 Airflow에서 테스트가 중요할까?

•

많은 외부 시스템과 통신이 발생하는 특성을 가짐

•

로직을 실제로 수행하는 Task를 시작하고 종료시키는 역할(오케스트레이션 시스템)

9.1.1 모든 DAG에 대한 무결성 테스트

무결성 테스트

모든 DAG의 무결성에 대해 DAG가 정상적으로 구현되었는지 확인

•

DAG에 사이클이 포함되어 있지는 않은지

•

DAG의 Task ID가 고유한지

•

등등등

테스트 구성

프로젝트의 최상단 디렉토리에 별도의 tests/ 디렉토리를 생성하여 검사 대상 코드를 그대로 복사하여 구성

필수는 아니지만…

•

pytest는 주어진 디렉토리에서 test_ || _test를 검색

•

테스트는 절대로 의존적이어선 안 된다.

테스트 예시

# DAG_PATH: 테스트를 할 DAG의 위치, DAG_FILES: DAG_PATH 안에 있는 파일들의 list
DAG_PATH = os.path.join(os.path.dirname(__file__), "..", "..", "dags/**.py")
DAG_FILES = glob.glob(DAG_PATH)

@pytest.mark.parametrize("dag_file", DAG_FILES)
def test_dag_integrity(dag_file):
    """Import DAG files and check for DAG."""
    module_name, _ = os.path.splitext(dag_file) # 모듈의 절대경로 포함 이름, 확장자
    module_path = os.path.join(DAG_PATH, dag_file) # 테스트할 DAG의 위치
    mod_spec = importlib.util.spec_from_file_location(module_name, module_path) # 모듈을 가져옴
    module = importlib.util.module_from_spec(mod_spec)
    mod_spec.loader.exec_module(module)
		
		# 파일 중 DAG 추출
    dag_objects = [var for var in vars(
        module).values() if isinstance(var, DAG)]
    # 검사 1: 모든 파이썬 파일에서 DAG 객체가 존재하는지 유효성 검사
		assert dag_objects
		# 검사 2: DAG 객체에 순환 주기 존재 여부 확인
    for dag in dag_objects:
        # Test cycles
        dag.test_cycle()
Python
복사

•

검사 1: 모든 파이썬 파일에서 DAG 객체가 존재하는지 유효성 검사

assert? raise?

◦

assert [조건], [에러 메시지]: 조건이 참이면 그대로 진행, 거짓이면 에러 메시지 출력 

◦

[조건]을 보장함!

•

검사 2: DAG 객체에 순환 주기 존재 여부 확인

•

그 외…

◦

DAG의 기능이나 구현과 관련된 검증을 할 수도 있고, 단순 파이썬을 이용한 테스트를 할 수도 있음!

◦

ex) 뭐하는 애일까요?

assert dag.dag_id.startswith((”import”, “export”))
Python
복사

실습

접속 및 실행

docker exec -it boaz-airflow-study-airflow-worker-1 /bin/bash
Shell
복사

cd /opt/airflow/dags/Chap09/tests/dags
pytest test_dag_integrity.py
Shell
복사

너무 길어요!!!

•

원래 결론은 마지막에 있슴..

================================================ short test summary info ================================================
FAILED test_dag_integrity.py::test_dag_integrity[/opt/airflow/dags/Chap09/tests/dags/../../dags/bash_operator_no_command.py] - airflow.exceptions.AirflowException: missing keyword argument 'bash_command'
FAILED test_dag_integrity.py::test_dag_integrity[/opt/airflow/dags/Chap09/tests/dags/../../dags/dag_cycle.py] - AttributeError: 'DAG' object has no attribute 'test_cycle'
FAILED test_dag_integrity.py::test_dag_integrity[/opt/airflow/dags/Chap09/tests/dags/../../dags/duplicate_task_ids.py] - airflow.exceptions.DuplicateTaskIdFound: Task id 'task' has already been added to the DAG
FAILED test_dag_integrity.py::test_dag_integrity[/opt/airflow/dags/Chap09/tests/dags/../../dags/testme.py] - airflow.exceptions.DuplicateTaskIdFound: Task id 'test2' has already been added to the DAG
============================================ 4 failed, 11 warnings in 0.55s =============================================
Shell
복사

로그를 보고  같은 에러 찾기 

1) bash_operator_no_command.py

import airflow.utils.dates
from airflow import DAG
from airflow.operators.bash import BashOperator

dag = DAG(
    dag_id="chapter8_bash_operator_no_command",
    start_date=airflow.utils.dates.days_ago(3),
    schedule_interval=None,
)

BashOperator(task_id="this_should_fail", dag=dag)
Python
복사

Test failure log

2) dag_cycle.py

import airflow.utils.dates
from airflow import DAG
from airflow.operators.dummy import DummyOperator

dag = DAG(
    dag_id="chapter8_dag_cycle",
    start_date=airflow.utils.dates.days_ago(3),
    schedule_interval=None,
)

t1 = DummyOperator(task_id="t1", dag=dag)
t2 = DummyOperator(task_id="t2", dag=dag)
t3 = DummyOperator(task_id="t3", dag=dag)

t1 >> t2 >> t3 >> t1
Python
복사

Test failure log

3) duplicate_task_ids.py

import airflow.utils.dates
from airflow import DAG
from airflow.operators.dummy import DummyOperator

dag = DAG(
    dag_id="chapter8_duplicate_task_ids",
    start_date=airflow.utils.dates.days_ago(3),
    schedule_interval=None,
)

t1 = DummyOperator(task_id="task", dag=dag)
for i in range(5):
    DummyOperator(task_id="task", dag=dag) >> t1
Python
복사

Test failure log

4) testme.py

import airflow.utils.dates
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.dummy import DummyOperator

dag = DAG(
    dag_id="testme", start_date=airflow.utils.dates.days_ago(3), schedule_interval=None
)

t1 = DummyOperator(task_id="test", dag=dag)
for tasknr in range(5):
    BashOperator(task_id="test2", bash_command=f"echo '{tasknr}'", dag=dag) >> t1
Python
복사

Test failure log

9.1.2 CI/CD 파이프라인 설정하기

CI/CD 파이프라인?

코드 저장소를 통해 코드가 변경될 때 사전 정의된 스크립트를 실행하는 시스템

•

CI(Continuous integration): 변경된 코드가 코딩 표준과 테스트 조건을 준수하는 지 확인/검증

◦

Flake8, Pylint, Black 등의 코드 검사기를 통해 코드를 레포지토리에 푸시할 때 코드의 품질 확인 및 테스트 실행

•

CD(Continuous deployment): 사람의 간섭 없이 완전히 자동화된 코드를 프로덕션 시스템에 자동으로 배포하는 것

◦

수동으로 검증하고 배포하지 않고도 생산성 극대화

•

예시

github.com

https://github.com/MoaRoom/MoaRoom-Infra/blob/main/.github/workflows/control-image-build.yml

.github/workflows/control-image-build.yml

name: Docker Image CI(Control server)

on:
  push:
    paths:
      - "resources/images/control-ubuntu/**"
    branches:
      - "main"

jobs:
  build-and-push-image:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v1

      - name: Login to DockerHub
        uses: docker/login-action@v1
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}

      - name: Build and push
        uses: docker/build-push-action@v2
        with:
          context: ./resources/images/control-ubuntu/
          file: ./resources/images/control-ubuntu/Dockerfile
          platforms: linux/amd64
          push: true
          tags: ryann3/control-ubuntu:4.0
YAML
복사

main 브랜치에 머지된 커밋 단위로 workflow가 실행된다

◦

정의한 build-and-push-image job의 step 별로 실행된 것을 확인할 수 있다.

실패한 경우

Pytest with airflow

•

추후 테스트 예정!

name: Python tatic checks and tests for dags

on:
  push:
    paths:
      - "dags/Chap09/**"
    branches:
      - "main"

jobs:
  test-and-check:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Setup Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.6.9"
      - name: Install Flake8
        run: pip install flake8
      - name: Run Flake8
        run: flake8
      - name: Install Pylint
        run: pip install pylint
      - name: Run Pylint
        run: find dags/Chap09/tests -name "*.py" | xargs pylint --output-format=colorized
      - name: Install Black
        run: pip install black
      - name: Run Black
        run: find dags/Chap09/tests -name "*.py" | xargs black --check
      - name: Install dependencies
        run: pip install apache-airflow pytest
      - name: Test DAG integrity
        run: pytest dags/Chap09/tests
YAML
복사

9.1.3 단위 테스트 작성하기

지난 시간 실습을 떠올려보자…

...
def get_ratings(self, start_date=None, end_date=None, batch_size=100):
    """
    Fetches ratings between the given start/end date.

    Parameters
    ----------
    start_date : str
        Start date to start fetching ratings from (inclusive). Expected
        format is YYYY-MM-DD (equal to Airflow's ds formats).
    end_date : str
        End date to fetching ratings up to (exclusive). Expected
        format is YYYY-MM-DD (equal to Airflow's ds formats).
    batch_size : int
        Size of the batches (pages) to fetch from the API. Larger values
        mean less requests, but more data transferred per request.
    """

    yield from self._get_with_pagination(
        endpoint="/ratings",
        params={"start_date": start_date, "end_date": end_date},
        batch_size=batch_size,
    )
...
Python
복사

•

get_ratings에서 batch size가 음수라면?!?! 

에헤이 조졌네~

사용자의 입력 값을 전달하기 전에 해당 값이 유효한지를 먼저 판단해볼까?

누가 회사에서 더 예쁨받을까요?

예시: 주어진 두 날짜 사이에 상위 N개의 인기 영화를 반환

import datetime as dt
import logging
import json
import os

from airflow import DAG
from airflow.operators.python import PythonOperator
from collections import defaultdict, Counter

from custom.hooks import MovielensHook


class MovielensPopulatriyOperator:
    def __init__(self, conn_id, start_date, end_date, min_ratings=4, top_n=5, **kwargs):
        super().__init__(**kwargs)
        self._conn_id = conn_id
        self._start_date = start_date
        self._end_date = end_date
        self._min_ratings = min_ratings
        self._top_n = top_n

    def execute(self, context):
        with MovielensHook(self._conn_id) as hook:
            ratings = hook.get_ratings(
                start_date=self._start_date, end_date=self._end_date,)
            rating_sums = defaultdict(Counter)
            for rating in ratings:
                rating_sums[rating["movieId"]].update(
                    count=1, rating=rating["rating"])
            averages = {
                movie_id: (rating_counter["rating"]/rating_counter["count"],
                           rating_counter["count"]
                           )
                for movie_id, rating_counter in rating_sums.items()
                if rating_counter["count"] >= self._min_ratings
            }
            return sorted(
                averages.items(),
                key=lambda x: x[1],
                reverse=True
            )[:self._top_n]
Python
복사

MovielensPopulatriyOperator의 정확성을 어떻게 테스트해야할까?

→ 주어진 값으로 오퍼레이터를 실행하고 결과가 예상대로인지 확인

•

이를 위하여 몇 가지 pytest 컴포넌트들이 필요함

9.1.4 Pytest 프로젝트 구성하기

•

BashOperator를 테스트하는 예제

def test_example():
	task = BashOperator(
		task_id="test", 
		bash_command="echo 'hello!'",
		xcom_push=True,
	)
result = task.execute(context={})
assert result == "hello!"
Python
복사

•

MovielensPopularityOperator에 적용

def test_movielenspopularityoperator(mocker):
    task = MovielensPopularityOperator(
        task_id="test_id",
        # conn_id="testconn", # 필수 인자이므로 없애면 test시 에러 발생
        start_date="2015-01-01",
        end_date="2015-01-03",
        top_n=5,
    )
    result = task.execute(context=None)
    assert len(result) == 5
Python
복사

테스트 결과값을 다른 쪽에 전달할 수 있을까?

•

테스트는 격리된 환경에서 진행되기 때문에 어떤 테스트의 결과는 다른 테스트에 영향을 줄 수 없다.

•

따라서 테스트 사이에 발생하는 정보를 목업(Mockup)을 이용하여 전달한다.

목업이란

특정 작업이나 객체를 모조로 만드는 것

실제 운영 환경에서는 예상되지만 테스트 중에는 발생하지 않는 데이터베이스에 대한 호출을 실제로 발생시키지 않고 특정 값을 반환하도록 파이썬 코드로 지시하여 임의의 값을 전달함

•

외부 시스템이 실제로 연결하지 않고도 테스트를 개발하고 실행할 수 있음

def test_movielenspopularityoperator(mocker):
    mock_get = mocker.patch.object(
        MovielensHook, # BaseHook으로 하면 안 됨(정의 위치 x 호출 위치 o)
        "get_connection",
        return_value=Connection(
            conn_id="test", login="airflow", password="airflow"),
    )
    task = MovielensPopularityOperator(
        task_id="test_id",
        conn_id="test",
        start_date="2015-01-01",
        end_date="2015-01-03",
        top_n=5,
    )
    result = task.execute(context=None)
    assert len(result) == 5
    assert mock_get.call_count == 1 # 한 번만 호출된 것인지 확인
    mock_get.assert_called_with("test") # 예상되는 conn_id로 검증
Python
복사

cd /opt/airflow/dags/Chap09/tests/dags/chapter9/custom
pytest test_operators.py
Shell
복사

Test log

9.1.5 디스크 파일로 테스트하기

JSON 리스트를 가진 파일을 읽고 이를 CSV 형식으로 쓰는 오퍼레이터!

Operator

class JsonToCsvOperator(BaseOperator):
    def __init__(self,input_path, output_path, **kwargs):
        super().__init__(**kwargs)
        self._input_path = input_path
        self._output_path = output_path
    def execute(self, context):
        with open(self._input_path, "r") as json_file:
            data=json.load(json_file)
        columns={key for row in data for key in row.keys()}
        
        with open(self._output_path, "w") as csv_file:
            writer=csv.DictWriter(csv_file, fieldnames=columns)
            writer.writeheader()
            writer.writerows(data)
Python
복사

Test

•

픽스처(fixture): 선언한 함수를 테스트 함수에서 인자로 받아 사용할 수 있게 해주는 기능(ex. tmp_path 인수)

1) 테스트 함수 A를 작성하고

2) A에 fixture 를 선언하면,

3) 다른 테스트 함수에서 A의 결과를 받아와 사용할 수 있다.

def test_json_to_csv_operator(tmp_path: Path):
    print(tmp_path.as_posix())

    input_path = tmp_path / "input.json"
    output_path = tmp_path / "output.csv"

    # Write input data to tmp path
    input_data = [
        {"name": "bob", "age": "41", "sex": "M"},
        {"name": "alice", "age": "24", "sex": "F"},
        {"name": "carol", "age": "60", "sex": "F"},
    ]
    with open(input_path, "w") as f:
        f.write(json.dumps(input_data))

    # Run task
    operator = JsonToCsvOperator(
        task_id="test", input_path=input_path, output_path=output_path
    )
    operator.execute(context={})

    # Read result
    with open(output_path, "r") as f:
        reader = csv.DictReader(f)
        result = [dict(row) for row in reader]

    # Assert
    assert result == input_data
Python
복사

cd /opt/airflow/dags/Chap09/tests/airflowbook/operators
pytest test_json_to_csv_operator.py
Shell
복사

Test log

파일 이름이 static이라니.. 너무 dumb 해요!

→ 다음 섹션을 대기하세요

9.2 테스트에서 DAG 및 테스크 콘텍스트로 작업하기

•

오퍼레이터는 보통 실행을 위해 더 많은 콘텍스트(변수 템플릿) 또는 작업 인스턴스 콘텍스트가 필요하다

Task가 실행될 때 수행되는 단계

•

operator.execute(context={}) 형태보다 더 실제 시스템에 가깝게 Task를 실행해야한다

•

위 그림에서 output_path 인수를 /output/{{ ds }}.json 형태로 제공할 수 있지만 context={}으로 테스트하면 ds 변수를 사용할 수 없다.

⇒ 따라서 airflow 자체에서 Task를 시작할 때 사용하는 실제 메서드인 operator.run()을 호출함

•

기본 DAG

dag = DAG(
        "test_dag",
        default_args={"owner": "airflow", "start_date": datetime.datetime(2019, 1, 1)},
        schedule_interval="@daily",
    )
Python
복사

•

템플릿 변수를 할당하기 위해 DAG로 테스트

def test_movielens_operator(tmp_path: Path, mocker: MockFixture):
    mocker.patch.object(
        MovielensHook,
        "get_connection",
        return_value=Connection(conn_id="test", login="airflow", password="airflow"),
    )
    dag = DAG(
        "test_dag",
        default_args={"owner": "airflow", "start_date": datetime.datetime(2019, 1, 1)},
        schedule_interval="@daily",
    )

		task = MovielensDownloadOperator(
        task_id="test",
        conn_id="testconn",
        start_date="{{ prev_ds }}",
        end_date="{{ ds }}",
        output_path=str(tmp_path / "{{ ds }}.json"),
        dag=dag,
    )

    dag.clear()
    task.run(
        start_date=dag.default_args["start_date"],
        end_date=dag.default_args["start_date"],
        ignore_ti_state=True,
    )
Python
복사

재사용

DAG를 사용하는 테스트가 여러번 있는 경우 Pytest와 함께 재사용할 수 있다

•

conftest.py 사용하여 (하위)디렉토리의 여러 파일 재사용

@pytest.fixture
def test_dag():
    return DAG(
        "test_dag",
        default_args={"owner": "airflow",
                      "start_date": datetime.datetime(2015, 1, 1)},
        schedule_interval="@daily",
    )
Python
복사

•

테스트에 fixture 포함하여 필요 객체 생성

◦

task.run(): DAG의 schedule_interval에 따라 지정된 두 날짜 사이에 실행할 Task의 인스턴스 스케줄 주기 계산

def test_movielens_operator(tmp_path: Path, mocker: MockFixture):
    mocker.patch.object(
        MovielensHook,
        "get_connection",
        return_value=Connection(
            conn_id="test", login="airflow", password="airflow"),
    )
    dag = DAG(
        "test_dag",
        default_args={"owner": "airflow",
                      "start_date": datetime.datetime(2019, 1, 1)},
        schedule_interval="@daily",
    )

    task = MovielensDownloadOperator(
        task_id="test",
        conn_id="testconn",
        start_date="{{ prev_ds }}",
        end_date="{{ ds }}",
        output_path=str(tmp_path / "{{ ds }}.json"),
        dag=dag,
    )

    dag.clear()
    task.run(
        start_date=dag.default_args["start_date"],
        end_date=dag.default_args["start_date"],
        ignore_ti_state=True,
    )
Python
복사

9.2.1 외부 시스템 작업

MovieLens 평점을 읽고 결과를 Postgres 데이터베이스에 저장하는 오퍼레이터를 만들어보자.

class MovielensToPostgresOperator(BaseOperator):
    template_fields=("_start_date", "_end_date", "_insert_query")
    
    def __init__(self,movielens_conn_id, start_date, end_date, insert_query, postgres_conn_id, **kwargs):
        super().__init__(**kwargs)
        self._movielens_conn_id = movielens_conn_id
        self._start_date = start_date
        self._end_date = end_date
        self._insert_query = insert_query
        self._postgres_conn_id = postgres_conn_id
    
		def execute(self, context):
        with MovielensHook(self._movielens_conn_id) as movielens_hook:
            ratings=list(movielens_hook.get_ratings(start_date=self._start_date, end_date=self._end_date))
        postgres_hook=PostgresHook(postgres_conn_id=self._postgres_conn_id)
        insert_queries=[
            self._insert_query.format(",".join([str(_[1]) for _ in sorted(rating.items())])
        ) for rating in ratings
        ]
        postgres_hook.run(insert_queries)
Python
복사

_insert_query.format(",".join([str(_[1]) for _ in sorted(rating.items())])

테스트 단계에서 Postgres 데이터베이스에 접근할 수 없다면?

•

pytest에서 테스트를 할 수 있도록 도커 컨테이너의 제어가 가능한 이런저런 파이썬 패키지들이 있다.

◦

pytest-docker-tools

◦

테스트 도커 컨테이너를 생성할 수 있는 여러 함수 제공

→ 도커 컨테이너로 테스트하면 목업 환경을 구성할 필요 없이 실제 훅 메서드를 사용할 수 있다!

테스트

postgres_image = fetch(repository="postgres:11.1-alpine")
postgres = container(
    image="{postgres_image.id}",
    environment={"POSTGRES_USER": "testuser", "POSTGRES_PASSWORD": "testpass"},
    ports={"5432/tcp": None}, # 컨테이너의 5432 포트를 호스트의 임의의 열린 포트(None)으로 매핑
    volumes={
        os.path.join(os.path.dirname(__file__), "postgres-init.sql"): {
            "bind": "/docker-entrypoint-initdb.d/postgres-init.sql"
        }
    },
)


def test_movielens_to_postgres_operator(mocker, test_dag, postgres):
    mocker.patch.object(
        MovielensHook,
        "get_connection",
        return_value=Connection(
            conn_id="test", login="airflow", password="airflow"),
    )
    mocker.patch.object(
        PostgresHook,
        "get_connection",
        return_value=Connection(
            conn_id="postgres",
            conn_type="postgres",
            host="localhost",
            login="testuser",
            password="testpass",
            port=postgres.ports["5432/tcp"][0], # 호스트에 할당된 포트
        ),
    )

    task = MovielensToPostgresOperator(
        task_id="test",
        movielens_conn_id="movielens_id",
        start_date="{{ prev_ds }}",
        end_date="{{ ds }}",
        postgres_conn_id="postgres_id",
        insert_query=(
            "INSERT INTO movielens (movieId,rating,ratingTimestamp,userId,scrapeTime) "
            "VALUES ({0}, '{{ macros.datetime.now() }}')"
        ),
        dag=test_dag,
    )

    pg_hook = PostgresHook()

    row_count = pg_hook.get_first("SELECT COUNT(*) FROM movielens")[0]
    assert row_count == 0

    pytest.helpers.run_airflow_task(task, test_dag)

    row_count = pg_hook.get_first("SELECT COUNT(*) FROM movielens")[0]
    assert row_count > 0
Python
복사

•

pytest_docker_tools의 container 함수도 fixture이기 때문에 테스트를 위해 인수로 제공해야 호출할 수 있다.

•

/docker-entrypoint-initdb.d 디렉터리에 배치하여 시작 스크립트로 컨테이너를 초기화할 수 있음(컨테이너 부팅 시 실행)

cd /opt/airflow/dags/Chap09/tests/airflowbook/operators
pytest test_movielens_operator_postgres.py
Shell
복사

•

우리 환경은 docker 내부 airflow이므로 제대로 실행하기 위해서 DinD 패턴 구현이 필요함

Using Docker-in-Docker for your CI or testing environment? Think twice.

The primary purpose of Docker-in-Docker was to help with the development of Docker itself. Many people use it to run CI (e.g. with Jenkins), which seems fine at first, but they run into many “interesting” problems that can be avoided by bind-mounting the Docker socket into your Jenkins container instead.

https://jpetazzo.github.io/2015/09/03/do-not-use-docker-in-docker-for-ci/

◦

docker-compose.yaml의 volumes에 로컬의 docker.sock mount해서 소켓을 공유하는 방식으로 진행

Test log

9.3 개발을 위해 테스트 사용하기

•

다들 쩜 찍어서 디버깅 해보셨죠?

•

PyCharm, VSCode 등 웬만한 IDE는 다 가능합니다

•

일종의 breakpoint를 만드는 건데, 해당 코드 라인까지 도달하면

변수의 현재 상태(값)

해당 시점에서 코드 실행 가능(다음 줄, …)

•

IDE에서 사용할 수 없는 경우 pdb(Python Debugger)라는 내장 디버거가 있어서 디버그하려는 위치에 다음 코드 라인을 추가하면 됨

import pdb; pdb.set_trace()
Python
복사

•

airflow에 적용하기 위해서는 airflow tasks test를 이용하여 메타스토어에 실행 기록을 등록하지 않고 Task를 실행하여 테스트!

airflow tasks test [dagid] [taskid] [execution date]
Python
복사

9.3.1 DAG 완료 테스트하기

•

Task 인스턴스 콘텍스트를 포함한 테스트, 로컬 파일 시스템을 이요한 오퍼레이터, 도커를 이용한 외부 시스템을 사용하는 오퍼레이터 테스트 등등…

•

이는 모두 단일 오퍼레이터 테스트에 중점을 두지만…

워크플로 개발에서 가장 중요한 것은 모든 구성 요건이 잘 맞는지 확인하는 것 아닌가요?

•

그러나 이것은 대답하기 어렵습니다… 

•

개인 정보 보호 규정이나 데이터 크기로 인해 개발 환경에서 완벽한 프로덕션 환경을 시뮬레이션하기는 어려움(데이터가 페타바이트 단위라면..?)

•

따라서 sw를 개발하고 검증하는 데 사용할 수 있는 가능한 한 구현할 수 있는 프로덕션 환경을 만들어서 테스트함

9.4 Whirl을 이용한 프로덕션 환경 에뮬레이션

whirl

•

도커에서 운영 환경의 모든 구성 요소를 시뮬레이션

•

도커 컴포즈로 구성 요소 관리

•

그러나 모든 것이 도커 이미지로 제공되지는 않는다는 단점이 있음

9.5 DTAP 환경 생성

DTAP란?

Development, Test, Acceptance, Production을 분리하는 시스템

격리된 DTAP 환경을 새로 설정, 구현하여 테스트 (실제로 가능할런지는…)

Github Action 맛보기

•

책에는 이론만 있지만…

책에 있던 거

•

이건 현실적으로 지금 우리가 하기 어렵습니다…

name: Python static checks and tests for dags

on:
  push:
    paths:
      - "dags/Chap09/**"
    branches:
      - "main"

jobs:
  test-and-check:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Setup Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.6.9"
      - name: Install Flake8
        run: pip install flake8
      - name: Run Flake8
        run: flake8
      - name: Install Pylint
        run: pip install pylint
      - name: Run Pylint
        run: find dags/Chap09/tests -name "*.py" | xargs pylint --output-format=colorized
      - name: Install Black
        run: pip install black
      - name: Run Black
        run: find dags/Chap09/tests -name "*.py" | xargs black --check
      - name: Install dependencies
        run: pip install apache-airflow pytest
      - name: Test DAG integrity
        run: pytest dags/Chap09/tests
YAML
복사

Black

•

코드 스타일 통일하기

•

다들 사용하는 코드 포맷터도 다르고(아예 안 쓰실 수도 있고) 해서 패스

Black으로 파이썬 코드 스타일 통일하기

Engineering Blog by Dale Seo

https://www.daleseo.com/python-black/

Pylint

•

린트(lint) 또는 린터(linter)라고 부름

•

에러, 스타일 등 검사

[Python]Pylint로 파이썬 코드 품질 관리하기

Pylint파이썬 개발자들이 가장 수월하게 코딩을 할 수 있도록 도와주는 도구 중 하나가 바로 정적 분석 도구이다. Pylint는 파이썬 소스 코드에서 에러, 버그, 스타일 문제 등을 감지하는 인기 있는 정적 분석 도구 중 하나이다. 이 글에서는 Pylint를 설치하고 사용하는 방법, 다양한 옵션 그리고 실제 사용 예시를 통해 Pylint로 파이썬 코드의 품질을 효과적으로 관리하는 방법을 알아본다.Pylint 설치Pylint는 pip를 사용해 간편하게 설치할 수 있다. 터미널이나 명령 프롬프트에 다음 명령어를 입력하자.pip install pylint Pylint 사용법Pylint를 사용하여 소스 코드를 검사하려면, 아래와 같은 명령어를 사용한다.pylint myfile.py 명령을 실행하면, Pylin..

https://gr-st-dev.tistory.com/129

Flake8

•

pep8 기반 코딩 컨벤션을 준수

python flake8 기본 사용법

flake8은 pep8 코딩컨벤션을 준수하는 Lint를 위한 패키지이다. 정적으로 코드를 검사를 해주면 수정은 해주지 않는다. 수정을 해주는 패키지는 black으로 다음번에 다룰 예정이다. flake8 설치 pip install flake8 flake 사용법 아래와 같이 파일명을 입력하거나 파일명을 입력하지 않으면 폴더 전체를 검사한다. flake8 # 폴더 전체 검사 flake8 file_name.py # 한개의 파일만 검사 flake8 설정파일 .flake8 file을 생성해주어 아래와 같은 내용을 입력하면 제외할 폴더와 파일은 제외를 하고 또한 무시할 에러메시지도 설정 할 수 있따.. [flake8] exclude = .git, .gitignore, *.pot, *.py[co], __pycache_..

https://bitrader.tistory.com/340

비교

파이썬 코드 스타일 리팩토링 모듈 정리(flake8, black, isort, pylint)

안녕하세요. 파이썬에서 코드 리팩토링을 도와주는 대표적인 툴 몇 가지에 대하여 사용 예시를 중심으로 정리를 해보도록 하겠습니다. 다음과 같은 main.py 파일을 기준으로 사용을 해보도록 하겠습니다. import numpy as np import pandas as pd def ABc(a): b = np.array(a) c = b + 1 return b - 1 if __name__=='__main__': print(ABc([1, 2, 3])) pandas 모듈과 변수 c가 미사용 되었으며, 적절하지 못한 줄바꿈, 메소드명, operator 사이에 공백이 없는 현상 등이 현재 코드 내에 존재하고 있습니다. 1. flake8 모듈 첫 번째 코드 스타일 리팩토링 모듈인 flake8을 설치하고 사용해보도록 하겠습..

https://jimmy-ai2.tistory.com/22

사용?

저희 회사는 Python 기반 모노레포이다보니 세가지를 전부 사용하고 있슴니다

•

모노레포?

◦

여러 개의 프로젝트를 한 저장소에 저장하는 sw 개발 전략

▪

중복되는 코드

▪

서로 다른 패키지 의존성

▪

서로 의존하는 프로젝트들끼리의 리팩터링 비용

▪

코드가 저장소마다 상황이 다르기에 개발자들 사이의 협업도 어려워지는 문제

모노레포(monorepo) 란?

모노레포가 뭘까? 두 개 이상의 프로젝트 코드를 하나의 버전 관리 저장소(repository)에서 관리하는 방법을 말합니다. 모노레포가 왜 나왔을까? 모노레포(mono repo) 개념이 처음 등장한 이유는 여러 가지가 있지만, 대표적인 이유 중 하나는 큰 규모의 소프트웨어 개발 프로젝트에서 발생하는 문제를 해결하기 위해서입니다. 큰 규모의 프로젝트에서는 여러 개의 코드 저장소가 생성됩니다. 표면적으로 봤을 때는 큰 문제가 없으나 다음과 같은 문제점들 때문에 프로젝트 관리에 어려움이 발생할 수 있습니다. 중복되는 코드 서로 다른 패키지 의존성 서로 의존하는 프로젝트들끼리의 리팩터링 비용 코드가 저장소마다 상황이 다르기에 개발자들 사이의 협업도 어려워지는 문제 이러한 문제를 해결하기 위해 모든 프로젝트 코드..

https://com789.tistory.com/44

Pantsbuild를 활용하여 대규모 Python 프로젝트를 모노리포로 이전하기, 김준기 - PyCon Korea 2022

청력에 상관없이 모두가 발표를 접할 수 있도록 파이콘 한국 2022 모든 영상에는 자막이 제공됩니다. 자막 보기 설정을 켜주세요(선택 : 한국어). Every video in PyCon Korea 2022 has subtitles. Please turn on the subtitle view setting.(Language: Korean) PyCon Korea 2022 프로그램 자세히 보기 : https://2022.pycon.kr/program/talks

https://www.youtube.com/watch?v=r2FpfmcoL5M

테스트

name: Python mini test for dags

on:
  push:
    paths:
      - "dags/Chap09/**"
    branches:
      - "main"

jobs:
  test-and-check:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v2
      - name: Setup Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.6.9"
      - name: Install dependencies
        run: pip install apache-airflow pytest
      - name: Test DAG integrity
        run: cd dags/Chap09/tests/dags && pytest test_dag_integrity.py
YAML
복사

•

이미 workflow에 추가해놨슴다

•

dags/Chap09 밑에 변경사항이 생길 때마다 워크플로우가 트리거 됩니다

CI 단계

github action 전용 pytest를 이용한 airflow integrity test

ubuntu - python 기본 환경을 기반으로 git action yaml을 작성.

이 때, airflow를 설치한 후, airflow metadatabase를 init 해준다

그리고 필요한 커넥션, variable들을 import할 수 있게 스크립트를 작성해준 후,

해당 스크립트를 깃 액션에서 사용할 수 있도록 한다

해당 부분은 자세한 설명은 생략.

Test 단계

아래는 pytest를 이용한 integrity test 단계

import pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dag_bag():
    dagbag = DagBag(dag_folder='dags', include_examples=False)
    return dagbag

def test_dags_load_with_no_errors(dag_bag):
    print(dag_bag.dags)
    assert dag_bag.dags is not None
    assert len(dag_bag.import_errors) == 0


def test_dag_structure(dag_bag):
    for dag_id, dag in dag_bag.dags.items():
        assert dag.tasks, f"DAG {dag_id} has no tasks"

def test_schedule_interval(dag_bag):
    for dag_id, dag in dag_bag.dags.items():
        assert dag.schedule_interval, f"DAG {dag_id} has no schedule_interval"

def test_default_args(dag_bag):
    for dag_id, dag in dag_bag.dags.items():
        assert dag.default_args, f"DAG {dag_id} has no default_args"
        # Further checks can be added here, e.g., checking for specific default arguments

def test_task_dependencies(dag_bag):
    for dag_id, dag in dag_bag.dags.items():
        for task in dag.tasks:
            assert task.downstream_list or task.upstream_list, f"Task {task.task_id} in DAG {dag_id} has no dependencies"
JavaScript
복사

dag의 정합성을 체크해주는 단계이다.

참고 링크