Architecture Patterns with Python(3장)

결합과 추상화

 

앞서 본 저장소 패턴은 저장소에 대한 추상화다.

B 컴포넌트가 깨지는 게 두려워서 A 컴포넌트를 변경할 수 없는 경우를 이 두 컴포넌트가 서로 결합되어 있다고 한다. 지역적인 결합은 좋다. 코드가 서로 함께 동작하고, 한 컴포넌트가 다른 컴포넌트를 지원하며 시계 나사처럼 서로 맞물려 돌아간다.

그러나 전역적인 결합은 성가신 존재일 수 있다. 코드를 변경하는 데 드는 비용을 증가시키며, 결합이 커지면 아예 코드를 변경할 수 없는 지경에 이르기도 한다.

추상화를 통해 세부 사항을 감추면 시스템 내의 결합 정도를 줄일 수 있다.

 

추상화는 보통 단순화 하는 것이므로, 어떠한 시스템 사이에 추상화가 들어감으로써 의존하는 의존성의 종류가 줄어든다. 

 

예를 들어, 원본(source)과 사본(destination) 으로 된 두 파일 디렉터리를 동기화 시키는 코드를 작성하고자 한다.

다음의 세 가지 조건을 만족시키는 코드를 작성해본다.

1. 원본에 파일이 있지만 사본에 없으면 파일을 원본에서 사본으로 복사한다.
2. 원본에 파일이 있지만 사본에 내용이 같은 파일과 이름이 다르면 사본의 파일 이름을 원본 파일 이름과 같게 변경한다.
3. 사본에 파일이 있지만 원본에 없는 경우 사본의 파일을 삭제한다.

 

위 조건 중에서 상대적으로 2번 조건이 까다롭다. 파일의 내용을 살펴보고 동일할 경우에 이름을 비교하고 수정해줘야하기 때문이다.

이 때, 해당 파일을 읽어온 뒤 hashlib 모듈의 md5 나 sha1 의 해시함수를 사용하여 내용을 비교할 수 있다. 아래 예시는 sha1 해시를 만드는 코드이다.

import hashlib

BLOCKSIZE = 65536


def hash_file(path):
    hasher = hashlib.sha1()
    with open(path, "rb") as file:
        buf = file.read(BLOCKSIZE)
        while buf:
            hasher.update(buf)
            buf = file.read(BLOCKSIZE)
    return hasher.hexdigest()

- BLOCKSIZE 는 해당 문서의 몇 번째 글자까지 한 번에 읽어올 지를 결정하기 위한 상수 값이다.

- hasher 라는 변수로 hash 객체를 생성하였다. 여기에 update 함수를 통해 읽어들인 파일의 내용을 hash 하여 계속 업데이트 한다.

- hash 객체에 대한 update 함수는 바이트에만 사용 가능하므로, file 을 읽어들일 때 rb 형식으로 읽어야 한다.

- while 문을 통해 파일 내의 내용을 전부 읽어들인 뒤, hexdigest 함수를 통해 해시 값을 반환한다.

 

이제 나머지 조건들도 만족하는 함수를 만든다.

import hashlib
import os
import shutil
from pathlib import Path

BLOCKSIZE = 65536


def sync(source, dest):
    # 원본 폴더의 자식들을 순회하면서 파일 이름과 해시의 사전을 만든다.
    source_hashes = {}
    for folder, _, files in os.walk(source):
        for fn in files:
            source_hashes[hash_file(Path(folder) / fn)] = fn

    seen = set()  # 사본 폴더에서 찾은 파일을 추적한다.

    # 사본 폴더 자식들을 순회하면서 파일 이름과 해시를 얻는다.
    for folder, _, files in os.walk(dest):
        for fn in files:
            dest_path = Path(folder) / fn
            dest_hash = hash_file(dest_path)
            seen.add(dest_hash)

            # 사본에는 있지만 원본에 없는 파일을 찾으면서 삭제한다.
            if dest_hash not in source_hashes:
                os.remove(dest_path)

            # 사본에 있는 파일이 원본과 다른 이름이면 원본 이름으로 바꾼다.
            elif dest_hash in source_hashes and fn != source_hashes[dest_hash]:
                shutil.move(dest_path, Path(folder) / source_hashes[dest_hash])

    # 원본에는 있지만 사본에는 없는 모든 파일들을 사본으로 복사한다.
    for src_hash, fn in source_hashes.items():
        if src_hash not in seen:
            shutil.copy(Path(source) / fn, Path(dest) / fn)


def hash_file(path):
    hasher = hashlib.sha1()
    with open(path, "rb") as file:
        buf = file.read(BLOCKSIZE)
        while buf:
            hasher.update(buf)
            buf = file.read(BLOCKSIZE)
    return hasher.hexdigest()

sync('source', 'destination')

 

위 코드를 보면 몇 가지 단계로 구분 지을 수 있어 보인다.

1. 파일 동기화를 위해 주요 로직이 수행되는 sync 함수
   1. 원본 폴더의 모든 파일들을 순회하면서 해시 값 생성 및 저장하는 부분
   2. 사본 폴더의 모든 자식들을 순회하면서 해시 값 생성 및 저장하는 부분
   3. 각각 조건을 비교하여 그에 대한 행위를 수행하는 부분(삭제, 이름수정, 복사)
2. 해당 파일의 해시 값을 생성해주는 hash_file 함수

이에 따른 테스트 코드를 아래와 같이 작성해 볼 수 있다.

import tempfile
from pathlib import Path
import shutil

from ..sync import sync


def test_when_a_file_exists_in_the_source_but_not_the_destination():
    try:
        source = tempfile.mkdtemp()
        dest = tempfile.mkdtemp()

        content = "I am a very useful file"
        (Path(source) / 'my-file').write_text(content)

        sync(source, dest)

        expected = Path(dest) / 'my-file'
        assert expected.exists()
        assert expected.read_text() == content

    finally:
        shutil.rmtree(source)
        shutil.rmtree(dest)

def test_when_a_file_has_been_renamed_in_the_source():
    try:
        source = tempfile.mkdtemp()
        dest = tempfile.mkdtemp()

        content = "I am a file that was renamed"
        source_path = Path(source) / 'source-filename'
        old_dest_path = Path(dest) / 'dest-filename'
        expected_dest_path = Path(dest) / 'source-filename'
        source_path.write_text(content)
        old_dest_path.write_text(content)

        sync(source, dest)

        assert old_dest_path.exists() is False
        assert expected_dest_path.read_text() == content

    finally:
        shutil.rmtree(source)
        shutil.rmtree(dest)

 

위의 두 테스트 코드는 각각 

 - 원본에는 있지만 사본 폴더에는 없는 경우

 - 원본과 사본에 동일한 파일이 있지만 파일명이 서로 다른 경우

 

이 두가지를 테스트 하는 코드이다. 상당히 많은 과정이 필요하다. 테스트 커버리지를 적절히 달성하고 버그를 찾기 위해서는 더 많은 테스트를 해야 하지만, 이렇게 번거로운 테스트 코드를 작성해야 한다면 추가 테스트 작업이 더 큰 일로 느껴질 것이다.

또한 위 코드는 확장성이 좋지 않다. 실제 파일 시스템의 변경 없이 어떤 작업을 수행해야 할지만 표시해주는 --dry-run 플래그를 구현한다고 가정해보면, 코드의 변화가 필요하다.

 

테스트하기 쉽도록 코드를 작성하려면 기능별로 구분할 필요가 있다.

다음의 세 가지 서로 다른 책임을 갖는 코드로 구분할 수 있다.

1. os.walk 함수를 통해 파일 시스템 정보를 얻고, 파일의 해시 값을 생성한다. 이는 원본/사본 디렉터리에서 모두 비슷하게 수행된다.
2. 파일이 새 파일인지, 이름이 변경된 파일인지, 중복된 파일인지 결정한다.
3. 원본과 사본을 일치시키기 위해 파일을 복사하거나 옮기거나 삭제한다.

이 세 가지 부분에 대해 단순화된 추상화를 가져야 한다.

 

첫 번째 원본/사본과 각각의 해시 값은 딕셔너리를 사용하여 추상화할 수 있다.

source_files = {'hash1': 'path1', 'hash2': 'path2'}
dest_files = {'hash1': 'path1', 'hash2': 'path2'}

 

그렇다면 실제 파일의 이동이나 복사, 삭제 같은 파일 시스템 연산을 어떤 식으로 추상화할 수 있는가?

이 때,  무엇을 원하는가와 어떻게 이를 달성할 지를 분리한다.

("COPY", "sourcepath", "destpath")
("MOVE", "old", "new)

 

파일 시스템을 표현하는 source, dest 에 해당하는 각각의 딕셔너리를 입력받고, 수행할 동작을 표현하는 문자열 튜플로 이루어진 리스트를 예상하는 출력을 지정할 수 있다.

 

어떠한 주어진 실제 파일 시스템에 대해 함수를 실행하면 어떤 일이 일어나는지 검사
--> 어떤 주어진 파일 시스템의 추상화에 대해 함수를 실행하면 어떤 추상화된 동작이 일어나는지 검사

 

def test_when_a_file_exists_in_the_source_but_not_the_destination():
    src_hashes = {'hash1': 'fn1'}
    dst_hashes = {}
    expected_actions = [('COPY', '/src/fn1', '/dst/fn1')]
    ...


def test_when_a_file_has_been_renamed_in_the_source():
    src_hashes = {'hash1': 'fn1'}
    dst_hashes = {'hash1': 'fn2'}
    expected_actions = [('MOVE', '/dst/fn2', '/dst/fn1')]
    ...

 

추상화된 함수와 그에 따르는 추상화된 예상 동작을 비교하기만 하면 된다. 테스트 코드가 매우 짧고 간단해졌다.

 

그렇다면 이렇게 작동하기 위해서는 어떻게 코드를 바꿔야 하는가?

외부 상태에 대해 어떠한 의존성도 가지지 않는 코드의 핵을 만들고, 외부 입력에 대해 이 핵이 어떻게 반응하는지를 살펴보는 방식을 취한다.

 

def sync(source, dest):
    # 명령형 쉘 1단계 : 입력 수집
    source_hashes = read_paths_and_hashes(source)
    dest_hashes = read_paths_and_hashes(dest)

    # 명령형 쉘 2단계 : 함수형 핵 호출
    actions = determine_actions(source_hashes, dest_hashes, source, dest)

    # 명령형 쉘 3단계 : 출력 적용
    for action, *paths in actions:
        if action == "copy":
            shutil.copyfile(*paths)
        if action == "move":
            shutil.move(*paths)
        if action == "delete":
            os.remove(paths[0])


def read_paths_and_hashes(root):
    hashes = {}
    for folder, _, files in os.walk(root):
        for fn in files:
            hashes[hash_file(Path(folder) / fn)] = fn
    return hashes


def hash_file(path):
    hasher = hashlib.sha1()
    with open(path, "rb") as file:
        buf = file.read(BLOCKSIZE)
        while buf:
            hasher.update(buf)
            buf = file.read(BLOCKSIZE)
    return hasher.hexdigest()


def determine_actions(src_hashes, dst_hashes, src_folder, dst_folder):
    for sha, filename in src_hashes.items():
        if sha not in dst_hashes:
            sourcepath = Path(src_folder) / filename
            destpath = Path(dst_folder) / filename
            yield 'copy', sourcepath, destpath

        elif dst_hashes[sha] != filename:
            olddestpath = Path(dst_folder) / dst_hashes[sha]
            newdestpath = Path(dst_folder) / filename
            yield 'move', olddestpath, newdestpath

    for sha, filename in dst_hashes.items():
        if sha not in src_hashes:
            yield 'delete', Path(dst_folder) / filename

 

실질적인 비즈니스 로직은 determine_actions 에 속한다. 테스트도 추상화된 동작을 반환하는 이 함수만을 사용하여 수행할 수 있다.

def test_when_a_file_exists_in_the_source_but_not_the_destination():
    src_hashes = {'hash1': 'fn1'}
    dst_hashes = {}
    actions = determine_actions(src_hashes, dst_hashes, Path('/src'), Path('/dst'))
    assert list(actions) == [('copy', Path('/src/fn1'), Path('/dst/fn1'))]


def test_when_a_file_has_been_renamed_in_the_source():
    src_hashes = {'hash1': 'fn1'}
    dst_hashes = {'hash1': 'fn2'}
    actions = determine_actions(src_hashes, dst_hashes, Path('/src'), Path('/dst'))
    assert list(actions) == [('move', Path('/dst/fn2'), Path('/dst/fn1'))]

 

프로그램의 로직과 저수준 I/O 세부 사항 사이의 얽힘을 풀었기 때문에 쉽게 코드의 핵 부분을 테스트할 수 있다.

이런 접근 방법을 사용하면 테스트 코드가 주 진입 함수인 sync() 를 테스트하지 않고 determine_actions() 라는 더 저수준 함수를 테스트하게 된다. sync() 가 아주 단순하므로 이런 식의 테스트로 충분하다고 결정할 수 있다.

 

테스트에서 mock.patch를 사용하지 않는 이유는 다음과 같다.

1. 단위 테스트는 수행할 수 있지만 설계를 개선하는 데는 도움이 되지 못한다. mock.patch를 사용하면 코드가 --dry-run 플래그에 대해 동작하지 않고, FTP 서버에 접속해 동작하지 못한다.
2. mock 을 사용한 테스트는 코드베이스의 구현 세부 사항에 더 밀접히 결합된다. mock 이 shutil.copy 에 올바른 인수를 넘겼는가 등 여러 요소 사이의 상호검증을 하기 때문이다. 
3. mock 을 과용하면 테스트 코드가 복잡해져서 테스트 코드를 보고 동작을 알아내기가 어려워진다.

 

비즈니스 로직과 지저분한 I/O 사이의 인터페이스를 단순화하여 시스템을 더 쉽게 테스트 및 유지보수 할 수 있다. 올바른 추상화를 위해 점검해볼 수 있는 사항들은 아래와 같다.

 

• 지저분한 시스템의 상태를 표현할 수 있는 익숙한 파이썬 객체가 있는가? 이를 활용해 시스템 상태를 반환하는 단일 함수를 만들어라
  • 위의 예에서 각 상황에 따라 copy, move, delete 등의 상태를 반환하는 determine_actions 함수
• 시스템의 구성 요소 중 어떤 부분에 선을 그을 수 있는가? 선을 통해 각각의 추상화 사이의 연결고리를 어떻게 만들 수 있는가?
• 시스템의 여러 부분을 서로 다른 책임을 갖는 구성 요소로 나누는 합리적인 방법은 무엇인가?
  • 위의 예에서 각각의 코드가 갖는 특징에 따라 파일의 정보와 해시값 / 추상화된 상태값 / 실행 으로 구분지을 수 있었다.
• 어떤 의존 관계가 존재하는가? 핵심 비즈니스 로직은 무엇인가?
  • 예시에서는 최초 주어진 3가지 조건에 따라 상태값을 반환하는 로직이 핵심 비즈니스 로직이다.