MQTT 테스트 자동화 도구 개발기 (1) — 프로젝트 설계와 테스트 인프라 구축

시리즈: Part 1: 설계와 인프라 · Part 2: 35개 테스트 케이스 · Part 3: GUI와 배포

왜 MQTT 테스트 자동화가 필요했나?

IoT 프로젝트에서 AWS IoT Core를 MQTT 브로커로 도입하면서, 디바이스가 보내는 다양한 입력값에 대한 체계적인 검증이 필요해졌습니다.

처음엔 MQTTX 같은 GUI 도구로 수동 테스트를 했는데, 문제가 금방 드러났습니다:

결국 코드로 테스트를 작성하고, 한 번의 실행으로 전체 검증 + 리포트 생성까지 자동화하기로 했습니다.

기술 스택 선정

각 기술을 선택한 이유를 정리하면:

☕ Spring Boot가 테스트 프레임워크에 필요한가? — 싶을 수 있지만, application.yml로 엔드포인트·인증서 경로·테스트 파라미터 23개를 한곳에서 관리하고, 프로파일(test)로 환경을 분리할 수 있어서 상당히 편했습니다.

AWS IoT Core의 제약사항

테스트 설계에 앞서 AWS IoT Core의 제약사항을 정리했습니다. 일반적인 MQTT 브로커(Mosquitto 등)와 다른 점이 꽤 있습니다:

⚠️ 가장 임팩트가 큰 제약은 QoS 2 미지원이에요. QoS 2로 publish하면 브로커가 연결 자체를 끊어버리는데, 이런 동작을 테스트로 검증하는 것이 중요했습니다.

TLS/X.509 인증서 연동

AWS IoT Core는 X.509 인증서 기반 mTLS(Mutual TLS) 인증을 요구합니다. 아래 시퀀스 다이어그램으로 TLS 핸드셰이크 흐름을 살펴보겠습니다:

이 인증 과정을 코드로 구현한 것이 MqttClientManager입니다:

// MqttClientManager.java — TLS SSLSocketFactory 생성
public SSLSocketFactory createSSLSocketFactory(String caPath, String certPath, String keyPath) throws Exception {
    // 1. CA 인증서 로드 → TrustStore
    X509Certificate caCert = loadCertificate(caPath);
    KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
    trustStore.load(null, null);
    trustStore.setCertificateEntry("ca", caCert);

    TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
    tmf.init(trustStore);

    // 2. 디바이스 인증서 + 프라이빗 키 → KeyStore
    X509Certificate clientCert = loadCertificate(certPath);
    KeyPair keyPair = loadPrivateKey(keyPath); // Bouncy Castle PEM 파싱

    KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
    keyStore.load(null, null);
    keyStore.setCertificateEntry("cert", clientCert);
    keyStore.setKeyEntry("private-key", keyPair.getPrivate(), new char[0],
            new java.security.cert.Certificate[]{clientCert});

    KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
    kmf.init(keyStore, new char[0]);

    // 3. SSLContext 구성 (TLS 1.2 고정)
    SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
    sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

    return sslContext.getSocketFactory();
}Code language: Java (java)

여기서 Bouncy Castle이 필수인 이유가 나옵니다. AWS IoT Core에서 발급하는 인증서와 키는 PEM 형식(.pem.crt, .pem.key)인데, JDK의 KeyStore는 PEM을 직접 읽지 못합니다. Bouncy Castle의 PEMParser를 통해 PEM → Java KeyPair로 변환합니다:

// Bouncy Castle PEM 파싱 — 프라이빗 키 로드
private KeyPair loadPrivateKey(String path) throws Exception {
    try (InputStream is = getResourceStream(path);
         PEMParser parser = new PEMParser(new InputStreamReader(is))) {
        Object obj = parser.readObject();
        JcaPEMKeyConverter converter = new JcaPEMKeyConverter().setProvider("BC");
        if (obj instanceof PEMKeyPair) {
            return converter.getKeyPair((PEMKeyPair) obj);
        }
        throw new IllegalArgumentException("지원하지 않는 키 형식: " + obj.getClass().getName());
    }
}Code language: Java (java)

인증서 로드까지 완료했으니, 다음은 경로 해석입니다. 인증서 경로는 classpath와 절대 경로를 모두 지원하도록 했습니다. 개발 환경에서는 src/main/resources/certs/ 하위를 사용하고, GUI 도구에서는 사용자가 파일 선택기로 지정한 절대 경로를 사용합니다:

private InputStream getResourceStream(String path) throws IOException {
    // 절대 경로인 경우
    File file = new File(path);
    if (file.isAbsolute() && file.exists()) {
        return new FileInputStream(file);
    }
    // classpath 리소스
    ClassPathResource resource = new ClassPathResource(path);
    return resource.getInputStream();
}Code language: Java (java)

이렇게 만든 TLS 연결 설정을 외부에서 주입할 수 있도록, application.yml에서 환경변수 → 기본값 패턴으로 관리합니다:

# application.yml — MQTT 연결 설정
mqtt:
  aws-iot:
    endpoint: ${MQTT_ENDPOINT:a3azry3xgl1nir-ats.iot.ap-northeast-2.amazonaws.com}
    port: 8883
    client-id: ${MQTT_CLIENT_ID:mqttx_e198a166}
    cert-path: ${MQTT_CERT_PATH:certs/device-cert.pem.crt}
    key-path: ${MQTT_KEY_PATH:certs/private.pem.key}
    ca-path: ${MQTT_CA_PATH:certs/AmazonRootCA1.crt}
    qos: 1
    keep-alive: 60
    connection-timeout: 30
    clean-session: true
    topic-prefix: ${MQTT_TOPIC_PREFIX:test/mqtt}Code language: YAML (yaml)

☕ ${MQTT_ENDPOINT:기본값} 패턴 덕분에 CI 환경에서는 환경변수로, 로컬에서는 기본값으로, GUI에서는 System Property로 주입할 수 있어요. 하나의 설정 파일로 3가지 실행 환경을 커버합니다.

테스트 베이스 클래스 설계

35개 테스트 케이스가 공통으로 사용하는 기능을 Template Method 패턴의 베이스 클래스로 추출했습니다:

핵심 설계 포인트는 자동 리소스 정리입니다:

// MqttTestBase.java — 테스트 베이스 클래스
@SpringBootTest
@ActiveProfiles("test")
public abstract class MqttTestBase {

    protected final Logger log = LoggerFactory.getLogger(getClass());

    @Autowired
    protected MqttClientManager mqttClientManager;

    @Autowired
    protected AwsIotMqttConfig mqttConfig;

    @Autowired
    protected TestParameterConfig testParams;

    /** 테스트 중 생성된 클라이언트를 추적하여 자동 정리 */
    private final List<MqttClient> createdClients = new ArrayList<>();

    @BeforeEach
    void baseSetUp() {
        log.info("========== 테스트 시작: {} ==========", getClass().getSimpleName());
    }

    @AfterEach
    void baseTearDown() {
        // 💡 핵심: 테스트에서 생성한 모든 클라이언트를 자동 정리
        log.info("클라이언트 정리: {}개", createdClients.size());
        for (MqttClient client : createdClients) {
            mqttClientManager.safeDisconnect(client);
        }
        createdClients.clear();
    }

    /** 기본 설정으로 연결된 클라이언트 생성 (자동 정리 대상) */
    protected MqttClient connectClient() throws Exception {
        MqttClient client = mqttClientManager.createAndConnect();
        createdClients.add(client); // 자동 정리 대상에 등록
        return client;
    }

    /** 고유한 테스트용 Client ID 생성 */
    protected String uniqueClientId() {
        return "test-" + UUID.randomUUID().toString().substring(0, 8);
    }

    /** 구독 시작 — 비동기 메시지 수집 */
    protected List<MqttMessage> startCollecting(MqttClient client, String topic) throws Exception {
        CopyOnWriteArrayList<MqttMessage> messages = new CopyOnWriteArrayList<>();
        client.subscribe(topic, mqttConfig.getQos(), (t, msg) -> {
            messages.add(msg);
        });
        sleep(500); // 구독이 브로커에 전파될 시간
        return messages;
    }

    /** 지정 개수의 메시지가 수신될 때까지 대기 (polling) */
    protected boolean waitForMessages(List<MqttMessage> messages, int expectedCount, int timeoutSeconds) {
        long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L;
        while (messages.size() < expectedCount && System.currentTimeMillis() < deadline) {
            sleep(100);
        }
        return messages.size() >= expectedCount;
    }
}Code language: Java (java)

이 설계의 장점:

1. 리소스 누수 방지 — createdClients 리스트로 모든 클라이언트를 추적하고, @AfterEach에서 일괄 정리해요. 테스트가 중간에 실패해도 정리가 보장됩니다.
2. 보일러플레이트 제거 — 개별 테스트에서는 connectClient()만 호출하면 연결 + 정리 등록이 한 번에 끝납니다.
3. Publish-Subscribe 패턴 간소화 — startCollecting() → publish() → waitForMessages() 3단계로 메시지 흐름 검증이 완료됩니다.

Tag 기반 테스트 실행 시스템

JUnit 5의 @Tag를 활용해서 카테고리별·우선순위별 선택 실행을 구현했습니다:

// 테스트 클래스에 카테고리 태그
@Tag("payload")
@DisplayName("페이로드 크기 경계값 테스트")
class PayloadSizeBoundaryTest extends MqttTestBase {

    @Test
    @Tag("P0")  // 우선순위 태그
    @DisplayName("[P0] TC-PS-001-1: 2MB 초과 페이로드 발행 → 브로커 거부/연결 종료 확인")
    void testOversizedPayload2MB() {
        // ...
    }
}Code language: Java (java)

Gradle에서 카테고리별 태스크를 정의합니다:

// build.gradle — 카테고리별 테스트 태스크
tasks.register('testPayload', Test) {
    useJUnitPlatform { includeTags 'payload' }
    description = '페이로드(Payload) 테스트만 실행'
}

tasks.register('testTopic', Test) {
    useJUnitPlatform { includeTags 'topic' }
    description = '토픽(Topic) 테스트만 실행'
}

tasks.register('testConnection', Test) {
    useJUnitPlatform { includeTags 'connection' }
    description = '연결/세션(Connection) 테스트만 실행'
}

tasks.register('testMessageFlow', Test) {
    useJUnitPlatform { includeTags 'messageflow' }
    description = '메시지 흐름(Message Flow) 테스트만 실행'
}

tasks.register('testSecurity', Test) {
    useJUnitPlatform { includeTags 'security' }
    description = '보안(Security) 테스트만 실행'
}

// 우선순위별 실행
tasks.register('testP0', Test) {
    useJUnitPlatform { includeTags 'P0' }
    description = 'P0 (최우선) 테스트만 실행'
}

// GUI에서 동적 태그 선택
tasks.register('testSelected', Test) {
    useJUnitPlatform {
        if (project.hasProperty('includeTags')) {
            def tags = project.property('includeTags').toString().split(',')
            includeTags(*tags)
        }
    }
}Code language: Gradle (gradle)

실행 방법이 다양해집니다:

# 전체 테스트
./gradlew test

# 카테고리별
./gradlew testPayload
./gradlew testSecurity

# 최우선순위만
./gradlew testP0

# GUI에서 선택한 태그 조합
./gradlew testSelected -PincludeTags=payload,topicCode language: Bash (bash)

또한 23개의 테스트 파라미터를 환경변수로 주입할 수 있도록 했습니다:

// build.gradle — 환경변수 → 시스템 프로퍼티 매핑
tasks.withType(Test).configureEach {
    def testParamKeys = [
        'TEST_PARAM_JSON_NESTING_DEPTH'      : 'test.param.json-nesting-depth',
        'TEST_PARAM_OVERSIZED_PAYLOAD_KB'     : 'test.param.oversized-payload-kb',
        'TEST_PARAM_BURST_MESSAGE_COUNT'      : 'test.param.burst-message-count',
        'TEST_PARAM_SOAK_DURATION_SECONDS'    : 'test.param.soak-duration-seconds',
        'TEST_PARAM_SOAK_RATE_PER_SECOND'     : 'test.param.soak-rate-per-second',
        // ... 18개 더
    ]
    testParamKeys.each { envKey, propKey ->
        def val = System.getenv(envKey)
        if (val) {
            systemProperty propKey, val
        }
    }
}Code language: Gradle (gradle)

이렇게 하면 테스트 강도를 외부에서 조절할 수 있습니다:

# Burst 테스트를 500건으로, Soak 테스트를 120초로 변경
TEST_PARAM_BURST_MESSAGE_COUNT=500 \
TEST_PARAM_SOAK_DURATION_SECONDS=120 \
./gradlew testMessageFlowCode language: PHP (php)

프로젝트 전체 구조

최종적으로 완성된 프로젝트 구조는 다음과 같습니다:

mqtt-test/
├── src/main/java/com/mqtttest/
│   ├── client/
│   │   └── MqttClientManager.java      ← MQTT 연결 + TLS 설정
│   ├── config/
│   │   ├── AwsIotMqttConfig.java        ← application.yml 매핑
│   │   └── TestParameterConfig.java     ← 23개 테스트 파라미터
│   ├── gui/                             ← Part 3에서 상세 설명
│   │   ├── TestToolLauncher.java
│   │   ├── TestToolApp.java
│   │   ├── service/
│   │   └── view/
│   └── util/
│       └── TestDataGenerator.java       ← 테스트 데이터 팩토리
│
├── src/test/java/com/mqtttest/
│   ├── base/
│   │   └── MqttTestBase.java            ← 테스트 베이스 클래스
│   ├── payload/                         ← Part 2에서 상세 설명
│   │   ├── PayloadSizeBoundaryTest.java
│   │   └── PayloadFormatEncodingTest.java
│   ├── topic/
│   │   └── TopicAbnormalTest.java
│   ├── connection/
│   │   └── ConnectionSessionTest.java
│   ├── messageflow/
│   │   └── MessageFlowTest.java
│   ├── security/
│   │   └── SecurityTest.java      ← TLS·인증서 보안 검증 (<a href="/ko/lightsail-cryptominer-incident-response/" rel="noopener">크립토마이너 침해 대응기</a>에서 다룬 보안 위협도 참고)
│   └── report/
│       └── MarkdownReportListener.java  ← Part 3에서 상세 설명
│
├── src/main/resources/
│   ├── application.yml
│   └── certs/                           ← X.509 인증서 (gitignore)
│
└── build.gradle                         ← 카테고리 태스크 + Fat JAR + jpackageCode language: HTML, XML (xml)

마무리 — 핵심 3줄 요약

1. AWS IoT Core는 일반 MQTT 브로커와 다르다 — QoS 2 미지원, 128KB 페이로드 제한, X.509 인증 필수 등 제약사항을 사전에 정리하고 테스트에 반영해야 합니다.
2. MqttTestBase로 보일러플레이트를 제거한다 — 자동 리소스 정리, publish-subscribe 헬퍼, 유니크 ID 생성 등을 베이스 클래스에 집중시켜 개별 테스트를 깔끔하게 유지했습니다.
3. Tag + 환경변수로 유연한 실행 환경을 만든다 — 카테고리별(payload, security), 우선순위별(P0, P1) 선택 실행과 23개 파라미터의 외부 주입으로 다양한 시나리오에 대응합니다.

다음 편 예고: Part 2: 35개 테스트 케이스 구현기에서는 카테고리별 주요 테스트 케이스를 상세히 살펴봅니다. AWS IoT Core가 잘못된 JSON을 그대로 전달하는 이유, QoS 2 거부 시 연결이 끊기는 동작, 초당 100건 Burst 테스트 등 흥미로운 발견들을 공유하겠습니다.

조회수: 0