4주차 과제 - 김현수 (작성중)

컨트리뷰션
,
1

구현 과제

test_task / openstack image task list [ list | show ]: @김현수

설명

태스크 목록 조회

openstack image task list는 이미지 관련 비동기 작업들의 목록을 표시합니다. 이미지 가져오기, 내보내기, 복제 등의 작업 진행상황을 추적할 수 있다.

태스크 상세 조회

openstack image task show <task-id>는 특정 태스크의 상세 정보를 표시

핵심 옵션

필터링 옵션

가장 유용한 기능으로, 대량의 태스크에서 원하는 것만 선별할 수 있다.

  • --type <type>: 태스크 타입별 필터링 (import, export, clone)

  • --status <status>: 상태별 필터링 (pending, processing, success, failure)

두 옵션을 조합하여 사용할 수 있어, 특정 타입의 실패한 작업만 찾는 것이 가능

정렬 옵션

  • --sort-key <key>[:direction]: 생성일시, 상태, 타입 등으로 정렬

  • 기본값은 created_at이며, :desc를 붙이면 최신순으로 정렬

페이징 옵션

  • --limit <number>: 표시할 태스크 수 제한

  • --marker <task-id>: 특정 태스크 이후부터 조회

대량의 태스크가 있는 환경에서는 성능상 limit을 사용하는 것이 좋다

테스트 코드

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import fixtures

from openstackclient.tests.functional.image import base


class TaskTests(base.BaseImageTests):
    """Functional tests for Task commands"""

    def setUp(self):
        super().setUp()

        ver_fixture = fixtures.EnvironmentVariable('OS_IMAGE_API_VERSION', '2')
        self.useFixture(ver_fixture)

    def tearDown(self):
        super().tearDown()

    def test_task_list(self):
        output = self.openstack('task list', parse_output=True)
        self.assertIsInstance(output, list)
        
        if output:
            task = output[0]
            self.assertIn('ID', task)
            self.assertIn('Type', task)
            self.assertIn('Status', task)
            self.assertIn('Owner', task)

    def test_task_list_with_type_filter(self):
        output = self.openstack('task list --type import', parse_output=True)
        self.assertIsInstance(output, list)
        
        for task in output:
            self.assertEqual('import', task.get('Type'))

    def test_task_list_with_status_filter(self):
        output = self.openstack('task list --status pending', parse_output=True)
        self.assertIsInstance(output, list)
        
        for task in output:
            self.assertEqual('pending', task.get('Status'))

    def test_task_list_with_limit(self):
        output = self.openstack('task list --limit 1', parse_output=True)
        self.assertIsInstance(output, list)
        self.assertLessEqual(len(output), 1)

    def test_task_list_with_sort_key(self):
        output = self.openstack('task list --sort-key created_at', parse_output=True)
        self.assertIsInstance(output, list)

    def test_task_list_with_sort_dir(self):
        output = self.openstack('task list --sort-dir asc', parse_output=True)
        self.assertIsInstance(output, list)

    def test_task_show_with_existing_task(self):
        # First get a task ID from the list
        task_list = self.openstack('task list --limit 1', parse_output=True)
        
        if task_list:
            task_id = task_list[0]['ID']
            output = self.openstack(f'task show {task_id}', parse_output=True)
            
            # Check that the expected fields are present
            expected_fields = [
                'id',
                'type',
                'status',
                'owner_id',
                'created_at',
                'updated_at'
            ]
            
            for field in expected_fields:
                self.assertIn(field, output)
            
            self.assertEqual(task_id, output['id'])
        else:
            self.skipTest("No tasks available for testing task show")

    def test_task_show_nonexistent_task(self):
        # Test with a non-existent task ID
        fake_task_id = 'nonexistent-task-id'
        
        try:
            self.openstack(f'task show {fake_task_id}')
            self.fail("Expected command to fail with non-existent task ID")
        except Exception:
            # Expected to fail
            pass