테스트 코드가 없는 조직에서 자동화된 테스트 환경 구성하기

테스트 코드가 없는 조직에서 자동화된 테스트 환경 구성하기

안녕하세요. 실시간 결제 데이터를 관리하고 있는 주니어 개발자입니다.

사내에서 테스트 코드 없이 새로운 요구사항을 반영 한 결과물에 대한 QA 를 직접 두 눈으로 검증 해왔던 과거를 청산하고, 자동화된 테스트 환경을 구성하여 테스트를 보다 더 효율적으로 하기 위해 어떻게 적용 하였는지 공유합니다.

이 글을 읽는 예상 독자는 테스트 코드를 도입하고자 하는 개발자입니다.

그동안 왜 테스트 코드를 작성하지 않았는가?

레거시로 작성된 소스코드의 컨벤션을 지키면서 새로운 기능을 반영하도록 팀이 노력하고 있었습니다.

그렇게 컨벤션을 유지하면서 얻게 되는 장점은 서로 다른 프로젝트를 관리하더라도 코드를 읽는 비용은 상대적으로 저렴했기 때문이죠.

그런 프로젝트에서 테스트 코드를 도입하고자 시도했었을 때 이미 거대한 함수에서 소켓을 열고, 클라이언트 수신을 위한 while 문을 흘러 내부적으로 데이터를 처리하고 DB에 저장까지 하나의 함수에서 모두 이루어져있었습니다.

def main():
    context = daemon.DaemonContext(
        pidfile=pid_lock_file,
        files_preserve=stream_list
    )
      
    with context:
      mongo_client = MongoClient(...)
      mongo_db = mongo_client["example"]
      ...
    
      server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
      server_socket.bind((host, port))

      while true:
        client_socket, addr = server_socket.accept()
        ... 
        # 데이터 수신, 가공, 적재 프로세스

이런 코드 뭉치에서 테스트 코드를 작성할 엄두가 나지 않아 포기했죠.

그러던 어느 날, 신규 프로젝트에 투입 되면서 코드 개선 사항에 대해 충분히 의논하여 무너지지 않을 것 같던 레거시 컨벤션의 벽을 허물게 됩니다.

그렇게 새로운 방식으로 리팩터링을 결심하게 되었고 가장 첫번째로 자동화된 테스트 코드 도입을 적용하는 것이 목표였습니다.

테스트를 위한 코드 개선하기

함수 하나에서 여러가지 행위를 도맡아서 하던 전지전능한 권력을 모두 빼앗아 다른 객체에게 나눠주도록 하겠습니다.

객체로 추출하는 기준은 추상화 되지 않은 구현 부분이 핵심 로직에 직접 작성 되어있는 경우 모두 추상화와 의존 관계 주입을 통해 확장성을 확보합니다.

그 중 가장 쉽게 이해할 수 있는 예시 중 하나인 소켓을 열고 닫는 행위를 추상화합니다.

main 함수 내에서 DaemonContext 로 자식 프로세스 복제 과정과 무한 루프로 클라이언트가 접근할 수 있도록 계속 리스닝하는 구현 내용을 객체로 추출합니다.

class SocketService:
    def __init__(self, config: Config, callback_parse=None, callback_convert=None):
        self.realtime_receive_info = config

        self.log = logging.getLogger(__name__)
        self.callback_parse = callback_parse
        self.callback_convert = callback_convert

    def message_receive_process(self, client_socket) -> Tuple[Callable, str]:
        """
        소켓 서버에서 클라이언트 메세지 수신 및 응답
        """
        ...

    def convert_process(self, realtime_receive_data: bytes):
        """
        정상적인 메세지 수신으로 데이터 가공
        """
        ... # 실제 데이터 변환 로직 구현

    @staticmethod
    def invalid_data_receive_handler(*args):
        """
        비정상적인 메세지 수신으로 프로세스를 시작 지점으로 이동
        """
        raise ValueError("received invalid data!!!")

    def run_server(self):
        """
        실시간 수신 서버 실행
        """

class RealtimeMessageReceiverService(SocketService):
    def __init__(
            self,
            info,
            callback_parse: Optional[Parser] = None,
            callback_convert: Optional[Converter] = None
    ):
        super().__init__(info, callback_parse, callback_convert)
        self.log = logging.getLogger(__name__)

    @TimeOutHandler(10)
    def message_receive_process(self, client_socket):
      ... # 데이터 수신 구현

소켓과 실시간 데이터 연동 시스템은 강하게 결합 될 수 밖에 없는 구조로 판단하여 상속 관계를 맺도록 구성하고, 변할 수 있는 행위를 책임지는 객체는 모두 외부에서 주입하도록 변경 하였습니다.

이렇게 개선된 코드는 소켓과 관련된 정보를 테스트할 때 보다 수월했죠.

@pytest.fixture
def socket_server(schema, callback_parse, callback_convert):
    return RealtimeMessageReceiverService(schema.receiver_schema, callback_parse, callback_convert)

def test_message_receive_success(socket_server, credit_data):
    client_socket = MagicMock()
    message_length = 1234
    client_socket.recv.return_value = credit_data

    _, next_step_argument = socket_server.message_receive_process(client_socket)

    assert message_length == socket_server.realtime_receive_info.receive_info.data_length
    assert socket_server.check_receive_data_length(credit_data)

    response_message = socket_server.response_message(client_message, True)

    client_socket.sendall.assert_called_once_with(response_message)
    client_socket.close.assert_called_once()

def test_message_receive_failed(socket_server):
    client_socket = MagicMock()
    client_message = "test message".encode("euc-kr")
    client_socket.recv.return_value = client_message
    replace_bytes_start_index = 0
    replace_bytes_end_index = 2

    def client_message_response(message, start_idx, end_idx):
        return message[:start_idx] + b"ff" + message[end_idx:]

    actual_response_message = socket_server.response_message(client_message, False)
    expect_client_message_response = client_message_response(
        client_message, replace_bytes_start_index, replace_bytes_end_index
    )

    assert expect_client_message_response == actual_response_message

    next_step_method, receive_message = socket_server.message_receive_process(client_socket)

    assert "" == receive_message

    with pytest.raises(ValueError):
        next_step_method(receive_message)

def test_occured_timeout_error():
    expect_time = 1
    real_wait_time = 2

    @TimeOutHandler(expect_time)
    def wait_for_timeout_test(second):
        time.sleep(second)

    # Note: 타임아웃 설정 시간 보다 사용 함수가 더 긴 시간을 사용할 때
    with pytest.raises(TimeOutException):
        wait_for_timeout_test(real_wait_time)

1. 실시간 데이터를 성공적으로 처리하는가
2. 실시간 데이터가 정의 되지 않은 요청인 경우 실패로 처리하는가
3. 소켓이 데이터 처리 과정이 일정 시간을 넘기면 타임아웃 처리 되는가

이렇게 소켓 서버를 제어하면서 클라이언트가 요청할 때 전송하는 바이트 데이터를 기준으로 검증해야 할 부분들을 테스트 할 수 있게 됩니다.

여러 고객사의 다른 전문 양식을 어떻게 테스트 해야할까?

위에서 객체를 추출하면서 단위 테스트가 가능하게 되어 첫번째 목표였던 “테스트 코드 도입” 을 이루었습니다.

이제 신규 프로젝트의 목적이었던 신규 고객사 데이터 연동이 목적이었는데요. 
그럼, 여러 개의 고객사 코드를 어떻게 테스트해야할지 또 프로덕션 코드를 어떻게 구성해야 할지 고민이 됩니다.

스프링이 자바에서 DI 컨테이너를 통해 의존성 관리를 편리하게 만들었던 것처럼, 파이썬에서도 충분히 관리할 수 있을 것 같았죠.

그렇게 FastAPI 에서 자주 사용하는 dependency_injector 라이브러리를 적용합니다.

컨테이너 도입

class ContainerFactory:
    _containers: Dict[str, Dict[str, Type[containers.DeclarativeContainer]]] = defaultdict(dict)

    @classmethod
    def register_container(cls, brand: str, van: str, container_class: Type[containers.DeclarativeContainer]):
        cls._containers[brand.lower()][van.lower()] = container_class

    @classmethod
    def create_container(cls, args) -> containers.DeclarativeContainer:
        brand_key: str = args.brand.lower()
        van_key: str = args.van.lower()

        if brand_key not in cls._containers:
            raise ValueError(f"Unsupported brand: {args.brand}")

        if van_key not in cls._containers[brand_key]:
            raise ValueError(f"Unsupported van: {args.van}")

        container_class = cls._containers[brand_key][van_key]
        return container_class()

def register_container(brand, van):
    def wrapper(cls):
        ContainerFactory.register_container(brand, van, cls)
        return cls
    return wrapper

컨테이너 팩토리를 만들고, 데코레이터를 구성하여 앞으로 작성되는 컨테이너를 자동으로 등록하도록 구성합니다.

컨테이너를 계층별로 쌓아 코어 계층, 고객사 고유 계층, 통신 채널 사 계층으로 모두 분리한 뒤 계층을 참조하도록 구성합니다.

class 코어_컨테이너(containers.DeclarativeContainer):
    args = providers.Object(_args)
    
    db = ...
    schema = ...
    socket_daemon = ...

class 고객사_컨테이너(containers.DeclarativeContainer):
    core: Container = providers.Container(
        Container,
    )

    repository = providers.Factory(
        Repository,
        database=core.db,
    )

    collection_meta_data = providers.Resource(
        Metadata,
        client=core.db,
    )

@register_container(고객사명, 채널사명)
class 통신_채널사_컨테이너(고객사_컨테이너):
    realtime_parser_callback = providers.Factory(
        RealtimeParserCallback,
        db=TcbBaseContainer.core.db,
        parse_schema=고객사_컨테이너.core.schema.provided.realtime_parser_schema,
    )
    realtime_converter_callback = providers.Factory(
        RealtimeConverterCallback,
        service=convert_service
    )

    realtime_message_receiver_service = providers.Factory(
        RealtimeMessageReceiverService,
        info=TcbBaseContainer.core.schema.provided.receiver_schema,
        callback_parse=realtime_parser_callback,
        callback_convert=realtime_converter_callback
    )

이렇게 구성한 컨테이너에서 동일한 고객사지만 통신하는 채널사만 변경 되는 경우, 통신 채널사 컨테이너의 내용만 수정하면 되겠죠.

Pytest 는 패키지에 근접해 있는 conftest.py 를 참조하려고 하는 특징이 있습니다.

그렇기에 고유하게 패키지를 구성하고, 내부적으로 다른 conftest 를 관리하면 고객사마다 필요한 Fixture 를 구성할 수 있으며 시스템 전체적으로 필요한 데이터베이스 같은 객체는 테스트 루트에서 관리하며 불필요한 코드를 줄일 수 있게 됩니다.

def 고객사_컨테이너(args)
  container = ContainerFactory.create_container(args)
  container.init_resources()
  return container

def 테스트_코드1(고객사_컨테이너, credit_data, mock_database, setup_and_teardown):

  # Given
  service = container.service()
  request = DataServiceRequest(credit_data)
  mock_collection = mock_database["test"]

  # When
  service.convert_from(request)

  # Then
  result = list(mock_collection.find())
  assert len(result) == 1
  assert result[0]["검증할 데이터"] == "검증 내용"

이렇게 두 눈으로 직접 확인하던 검증 작업이 테스트 케이스를 추가함으로 필수적으로 검증해야 할 요소는 기능적으로 보장받을 수 있게 됩니다.

만약 이 테스트 코드 품질이 너무 떨어져 거짓 양성을 만들어내지만 않는다면 말이죠.

부록

컨테이너가 각자 다른 브랜드의 객체를 초기화 하기 때문에, 결국 원하는 브랜드의 컨테이너가 초기화 되었는지 확인하려면 데이터를 전송 해봐야 했었는데요.

그래서 매번 소켓 서버를 기동할 때 어떤 객체가 컨테이너화 되어 초기화 되었는지 로그로 남기도록 설정하였습니다.

class ContainerLog:
    def __init__(self, container):
        self.container = container
        self.log = logging.getLogger(__name__)

    def log_container_dependencies(self):
        """컨테이너의 모든 의존성 객체를 로그로 출력"""
        if not self.container:
            return

        self.log.info("=" * 50)
        self.log.info("Container Dependencies Overview")
        self.log.info("=" * 50)
        self._log_simple_providers()

    def _log_simple_providers(self):
        """컨테이너의 프로바이더를 간단한 형태로 로깅"""
        for name, provider in vars(self.container).items():
            if isinstance(provider, providers.Provider):
                if name.startswith('_'):
                    continue

                class_name = self._get_provider_class_name(provider)
                if class_name:
                    self.log.info(f"\t-> {name}: {class_name}")

            if isinstance(provider, providers.Container):
                self.log.info(f"\t-> {name}: Container({provider.providers})")
                self.core_container_logging(provider)

    def core_container_logging(self, container_provider):
        for sub_name, sub_item in container_provider.providers.items():
            if sub_name.startswith('_'):
                continue

            if hasattr(sub_item, "overridden") and sub_item.overridden:
                overridden = sub_item.overridden[0]
                self.log.info(f"\t\t{sub_name}: {overridden.kwargs}")
                continue

            if hasattr(sub_item, 'cls') and sub_item.cls:
                if hasattr(sub_item.cls, '__name__'):
                    self.log.info(f"\t\t{sub_name}: {sub_item.cls.__name__}")
                    continue

            if hasattr(sub_item, 'provides') and sub_item.provides:
                # 다른 형태의 프로바이더
                self.log.info(f"\t\t{sub_name}: {sub_item.provides}")

    @staticmethod
    def _get_provider_class_name(provider) -> Optional[str]:
        try:
            if hasattr(provider, 'cls') and provider.cls:
                if hasattr(provider.cls, '__name__'):
                    return provider.cls.__name__
            elif hasattr(provider, 'provides') and provider.provides:
                # 다른 형태의 프로바이더
                return str(provider.provides)
            elif hasattr(provider, "overridden") and provider.overridden:
                overridden = provider.overridden[0]
                return f"{overridden.cls.__name__}({overridden.kwargs})"
            return None
        except Exception:
            return None

[INFO|container_log.py:54] 2025-09-15 09:36:39,821 >       args: Namespace(brand='고객사', van='통신_채널사', target='서버_환경', oper='start')
[INFO|container_log.py:54] 2025-09-15 09:36:39,821 >       mongo_client_manager: <class 'utils.database.MongoClientManager'>
[INFO|container_log.py:54] 2025-09-15 09:36:39,821 >       mongo_database: <function Container.<lambda> at 0x1026a2ee0>
[INFO|container_log.py:49] 2025-09-15 09:36:39,821 >       schema: Schema
[INFO|container_log.py:44] 2025-09-15 09:36:39,821 >       realtime_parser_callback: {'db': <dependency_injector.providers.Resource(<function Container.<lambda> at 0x1026a2ee0>) at 0x10292dc10>, 'parse_schema': AttributeGetter("realtime_parser_schema")}
[INFO|container_log.py:44] 2025-09-15 09:36:39,821 >       realtime_converter_callback: {'service': <dependency_injector.providers.Factory(<class 'src.converter.application.sales_list_convert_service.SalesListConvertService'>) at 0x1029345e0>}
[INFO|container_log.py:49] 2025-09-15 09:36:39,821 >       realtime_message_receiver_service: RealtimeMessageReceiverService
[INFO|container_log.py:49] 2025-09-15 09:36:39,821 >       socket_daemon: SocketDaemon
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> repository: Repository
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> collection_meta_data: <class 'conf.metadata.metadata.CollectionMasterData'>
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> van_info: VanInfo(name='채널통신사1', code='1')
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> context: <class 'conf.context.ApplicationContext'>
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> credit_service: 고객사SalesListCreditService
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> payment_service_factory: PaymentServiceFactory
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> payment_resolver: 고객사PaymentResolver
[INFO|container_log.py:31] 2025-09-15 09:36:39,821 >    -> convert_service: SalesListConvertService
[INFO|daemonize.py:57] 2025-09-15 09:36:39,825 > Daemon process successfully created with PID: 85492
[INFO|daemonize.py:60] 2025-09-15 09:36:39,826 > Attempting to initialize container resources (e.g., DB connection)...
[INFO|database.py:86] 2025-09-15 09:36:39,892 > Connect to MongoClientManager(host=('1.2.3.4', 27017), db_name=DB) database
[INFO|daemonize.py:62] 2025-09-15 09:36:40,627 > Container resources initialized successfully.

마무리하며

테스트 코드를 쌓아가는 과정에서 초기에는 모든 객체에 대한 단위 테스트를 구현했었으나, 추후 신규 브랜드가 늘어나면서 중복 코드가 많이 발생하는 느낌이 조금은 불쾌 했었습니다.

이로인해 바이트 데이터를 요청했을 때 어떻게 변환 되는지에 대한 통합 테스트를 바탕으로 구성하여 대부분의 기능을 검증하면서 중복 코드도 줄일 수 있었죠.

또한 TimeOutHandler 데코레이터 내부적으로 시그널을 통해 관리하기 때문에 모든 테스트를 동작 시킬 시 브랜드가 늘어날 수록 1초에 지연이 발생합니다.

그렇기에 전체 테스트를 자주 실행 시킬수록 수 초를 기다려야 하는 상황이 되는데요.

그래서 초기에는 테스트 수행 범위를 통신하는 채널사 기준으로 범위를 좁혀 빠르게 테스트를 수행 시키고, 모든 개발이 완료 되었을 때 전체 테스트를 진행하여 보다 더 활발하게 테스트를 실행하였습니다.

결론적으로 페인 포인트를 따라서 해결하고자 하는 목표를 이루다보니 프로덕션 코드에서 개선점을 찾고, 더 나은 방식은 없을지 고민 해볼 수 있는 시간을 보냈습니다.

감사합니다.