OpenStack Endpoint [CRUD] Functional Test

들어가며

OpenStack SDK의 Identity Endpoint에 대한 functional test가 없어, 테스트 코드를 작성해보았습니다 😃 (with 오픈소스 컨트리뷰션!)

OpenStack Endpoint란?

OpenStack에서 Endpoint는 각 서비스의 접속 지점을 정의하는 핵심 구성 요소입니다. 클라이언트가 특정 서비스에 접근할 때, 어떤 URL로 요청을 보내야 하는지를 알려주는 역할을 합니다.

CLI 명령어

# Endpoint 생성
openstack endpoint create compute public https://nova.example.com/v2.1 --region RegionOne

# Endpoint 목록 조회
openstack endpoint list --service compute

# Endpoint 수정 (비활성화)
openstack endpoint set <endpoint-id> --disable

# Endpoint 삭제
openstack endpoint delete <endpoint-id>

SDK에서 CRUD 매핑

🔗 OpenStack Endpoint CRUD 동작 구조

# Create
endpoint = identity.create_endpoint(...)  # POST /v3/endpoints

# Read  
endpoint = identity.get_endpoint(id)      # GET /v3/endpoints/{id}
endpoints = identity.endpoints()          # GET /v3/endpoints

# Update
endpoint = identity.update_endpoint(...)  # PATCH /v3/endpoints/{id}

# Delete
identity.delete_endpoint(endpoint)       # DELETE /v3/endpoints/{id}

Functional Test란?

Functional Test는 Unit Test와 달리 실제 환경에서 전체 기능을 검증하는 테스트입니다.

• 실제 OpenStack 환경에서 실행
• 실제 HTTP 요청을 서버에 전송
• 실제 데이터베이스에 리소스 생성/수정/삭제
• End-to-End 시나리오 검증

테스트 코드 작성 과정

🔗 전체 코드 PR

1) 테스트 구조 설계

먼저 기존 functional test들의 패턴을 분석했습니다. OpenStack SDK의 functional test는 다음과 같은 구조를 가집니다.

class TestEndpoint(base.BaseFunctionalTest):
    def setUp(self):
        super().setUp()
        # 테스트에 필요한 리소스 준비
        
    def test_endpoint(self):
        # 메인 CRUD 테스트
        
    def test_endpoint_list_filters(self):
        # 필터링 기능 테스트

2) CRUD 테스트 구현

def test_endpoint(self):
    # CREATE: Endpoint 생성
    endpoint = self.operator_cloud.identity.create_endpoint(
        service_id=self.service.id,
        interface='public',
        url=self.test_url,
        region_id=self.region.id,
        is_enabled=True,
    )
    self.addCleanup(self._delete_endpoint, endpoint)
    
    # 생성 결과 검증
    self.assertIsInstance(endpoint, _endpoint.Endpoint)
    self.assertEqual(self.service.id, endpoint.service_id)
    self.assertEqual('public', endpoint.interface)
    self.assertEqual(self.test_url, endpoint.url)
    self.assertEqual(self.region.id, endpoint.region_id)
    self.assertTrue(endpoint.is_enabled)
    
    # UPDATE: Endpoint 수정
    endpoint = self.operator_cloud.identity.update_endpoint(
        endpoint, url=self.updated_url, is_enabled=False
    )
    
    # 수정 결과 검증
    self.assertEqual(self.updated_url, endpoint.url)
    self.assertFalse(endpoint.is_enabled)
    
    # READ: 단일 조회
    fetched_endpoint = self.operator_cloud.identity.get_endpoint(endpoint.id)
    self.assertEqual(endpoint.id, fetched_endpoint.id)
    self.assertEqual(self.updated_url, fetched_endpoint.url)
    
    # READ: 검색으로 조회
    found_endpoint = self.operator_cloud.identity.find_endpoint(endpoint.id)
    self.assertEqual(endpoint.id, found_endpoint.id)
    
    # READ: 목록 조회
    endpoints = list(self.operator_cloud.identity.endpoints())
    endpoint_ids = {ep.id for ep in endpoints}
    self.assertIn(endpoint.id, endpoint_ids)
    
    # DELETE는 cleanup에서 자동으로 실행

3) 필터링 테스트 구현

def test_endpoint_list_filters(self):
    # 테스트용 Endpoint 생성
    endpoint = self.operator_cloud.identity.create_endpoint(
        service_id=self.service.id,
        interface='internal',
        url=self.test_url,
        region_id=self.region.id,
        is_enabled=True,
    )
    self.addCleanup(self._delete_endpoint, endpoint)
    
    # Service ID로 필터링
    service_endpoints = list(
        self.operator_cloud.identity.endpoints(service_id=self.service.id)
    )
    service_endpoint_ids = {ep.id for ep in service_endpoints}
    self.assertIn(endpoint.id, service_endpoint_ids)
    
    # Interface로 필터링
    internal_endpoints = list(
        self.operator_cloud.identity.endpoints(interface='internal')
    )
    internal_endpoint_ids = {ep.id for ep in internal_endpoints}
    self.assertIn(endpoint.id, internal_endpoint_ids)
    
    # Region ID로 필터링
    region_endpoints = list(
        self.operator_cloud.identity.endpoints(region_id=self.region.id)
    )
    region_endpoint_ids = {ep.id for ep in region_endpoints}
    self.assertIn(endpoint.id, region_endpoint_ids)
    
    # 존재하지 않는 Service로 필터링
    empty_endpoints = list(
        self.operator_cloud.identity.endpoints(service_id='nonexistent')
    )
    self.assertEqual([], empty_endpoints)

환경 설정 트러블슈팅

1) Cloud 설정 오류

$ tox -e functional -- openstack.tests.functional.identity.v3.test_endpoint
Cloud devstack-admin was not found.

‼️ Functional Test는 여러 cloud 설정을 필요로 했습니다.

# ~/.config/openstack/clouds.yaml
clouds:
  devstack:           # 기본 cloud
    auth_type: password
    auth:
      auth_url: http://<IP>/identity
      username: admin
      password: admin
      project_name: admin
      user_domain_name: default
      project_domain_name: default
    region_name: RegionOne
    interface: public
    identity_api_version: 3

  devstack-alt:       # 대체 cloud (동일한 설정)
    auth_type: password
    # ... 동일한 설정
    
  devstack-admin:     # 관리자 cloud (동일한 설정)
    auth_type: password
    # ... 동일한 설정
    
  devstack-system-admin:  # 시스템 관리자 cloud (동일한 설정)
    auth_type: password
    # ... 동일한 설정

2) HTTP 401 인증 오류

CLI에서는 인증이 성공하는데 tox 환경에서는 계속 401 오류가 발생했습니다.

$ openstack --os-cloud devstack-admin token issue  # 성공
$ tox -e functional -- test_endpoint  # HTTP 401

‼️ 여러 시도 끝에 도메인 설정 방식이 문제였음을 발견했습니다.

# 실패하던 설정
auth:
  user_domain_id: default
  project_domain_id: default

# 성공한 설정  
auth:
  user_domain_name: default
  project_domain_name: default

테스트 실행

tox -e functional -- openstack.tests.functional.identity.v3.test_endpoint -v

결과

{0} openstack.tests.functional.identity.v3.test_endpoint.TestEndpoint.test_endpoint [1.619s] ... ok
{0} openstack.tests.functional.identity.v3.test_endpoint.TestEndpoint.test_endpoint_list_filters [1.191s] ... ok

======
Totals
======
Ran: 2 tests in 2.8094 sec.
 - Passed: 2
 - Skipped: 0
 - Failed: 0