Pytest로 InfluxDB 병렬 테스트 시 fixture와 테스트 코드 예시

InfluxDB를 사용한 코드에 대한 테스트 코드를 작성

PostgreSQL이나 MySQL을 테스트할 때 테스트 데이터베이스를 만들어서 하는 것처럼 실제로 InfluxDB에 데이터를 저장하고 읽어서 테스트를 하고 싶었다. 그래야 코드가 정상 작동하는지 확실하게 테스트할 수 있기 때문이다.

 

InfluxDB 관련 테스트 팩키지 X

InfluxDB관련 테스트 팩키지가 따로 없었다. 그래서 fixture로 추가해서 사용하기로 했다.

 

fixture 함수

conftest.py 파일에 아래 함수를 추가했다.

def use_influxdb(func):
    if "INFLUXDB_TEST_TOKEN" not in os.environ:
        return pytest.mark.skip(reason="INFLUXDB_TEST_TOKEN not set")(func)
    return func


@pytest.fixture
def influxdb_test_client():
    unique_id = uuid.uuid4()
    test_org_name = f"test_org_{unique_id}"
    test_bucket_name = f"test_bucket_{unique_id}"
    client = InfluxDBClient(url="http://localhost:8086", token=settings.INFLUXDB_TEST_TOKEN)
    test_org = client.organizations_api().create_organization(name=test_org_name)
    client.buckets_api().create_bucket(bucket_name=test_bucket_name, org=test_org)
    setattr(client, "test_org", test_org_name)
    setattr(client, "test_bucket", test_bucket_name)

    yield client

    client.organizations_api().delete_organization(test_org.id)
    client.close()

 

병렬 테스트 대응을 위해서 위처럼 만들게 되었다.

개별 테스트 때 organization와 bucket의 이름을 uuid 기반으로 만들고 테스트가 끝나면 삭제하는 방식이다.

이렇게 해야 병렬 테스트 때 이슈가 없다. test_org, test_bucket 식으로 고정해서 만들고 삭제하면, 병렬 테스트 때 간헐적으로, 테스트 버킷이 삭제되어서 오류가 발생한다.

 

테스트 코드 예시

테스트 코드에서는 아래 예처럼 사용하면 된다.

    @use_influxdb
    def test_get_mapping_distance_meter(self, influxdb_test_client):
        
        write_api = influxdb_test_client.write_api(write_options=SYNCHRONOUS)
        p0 = (
            Point("test")
            .tag("name", "test")
            .field("id", 1)
            .time(timezone.make_aware(datetime.datetime(2024, 1, 1, 0, 1)))
        )
        write_api.write(org=influxdb_test_client.test_org, bucket=influxdb_test_client.test_bucket, record=p0)
        
        with patch("influxdb.InfluxDb._get_influxdb_org") as mock_org, patch(
            "influxdb.InfluxDb._get_influxdb_bucket"
        ) as mock_bucket:
            mock_org.return_value = influxdb_test_client.test_org
            mock_bucket.return_value = influxdb_test_client.test_bucket
            ...

 

아래처럼 InfluxDb 클래스에 메소드가 있다고 할 때 이 두 메소드를 mocking해줘야 한다. 그래야 정상적으로 테스트된다.

class InfluxDb:

    def _get_influxdb_org(self):
        return "taptorestart"

    def _get_influxdb_bucket(self):
        return "data"