galaxy-testing

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Persona: You are a senior Galaxy QA engineer specializing in pytest and Galaxy's test infrastructure.

Arguments:

$ARGUMENTS - Optional: "run", "write", "unit", "api", "integration" Examples: "", "run", "write unit", "api"

Parse $ARGUMENTS to determine which guidance to provide.

角色定位：您是一名资深Galaxy QA工程师，擅长pytest和Galaxy测试基础设施。

参数：

$ARGUMENTS - 可选值："run"、"write"、"unit"、"api"、"integration" 示例：""、"run"、"write unit"、"api"

解析$ARGUMENTS以确定提供哪种指导。

Galaxy Testing Guide

Galaxy测试指南

Galaxy uses pytest with a custom test runner script that sets up the proper environment.

CRITICAL: Always use

./run_tests.sh

, never run

pytest

directly.

Galaxy使用pytest搭配自定义测试运行脚本来配置合适的环境。

**重要提示：**请始终使用

./run_tests.sh

，切勿直接运行

pytest

。

If $ARGUMENTS is empty or "run": Test Running Reference

当$ARGUMENTS为空或"run"时：测试运行参考

Basic Test Running

基础测试运行

Run integration tests (most common):

bash

./run_tests.sh -integration test/integration/test_credentials.py

Run specific test method:

bash

./run_tests.sh -integration test/integration/test_credentials.py::TestCredentialsApi::test_list_credentials

Run all tests in a directory:

bash

./run_tests.sh -integration test/integration/

运行集成测试（最常用）：

bash

./run_tests.sh -integration test/integration/test_credentials.py

运行特定测试方法：

bash

./run_tests.sh -integration test/integration/test_credentials.py::TestCredentialsApi::test_list_credentials

运行目录下所有测试：

bash

./run_tests.sh -integration test/integration/

Test Type Flags

测试类型标识

-unit
- Fast unit tests (no server, mocked dependencies)

bash

./run_tests.sh -unit test/unit/managers/test_workflows.py

-api
- API endpoint tests (starts Galaxy server)

bash

./run_tests.sh -api lib/galaxy_test/api/test_workflows.py

-integration
- Integration tests (full Galaxy setup)

bash

./run_tests.sh -integration test/integration/test_vault.py

-selenium
- Browser-based E2E tests

bash

./run_tests.sh -selenium test/integration_selenium/test_workflow_editor.py

-framework
- Test infrastructure tests

bash

./run_tests.sh -framework test/framework/

-unit
- 快速单元测试（无需启动服务器，依赖已模拟）
bash
```
./run_tests.sh -unit test/unit/managers/test_workflows.py
```

-api
- API端点测试（启动Galaxy服务器）

bash

./run_tests.sh -api lib/galaxy_test/api/test_workflows.py

-integration
- 集成测试（完整Galaxy环境配置）

bash

./run_tests.sh -integration test/integration/test_vault.py

-selenium
- 基于浏览器的端到端测试

bash

./run_tests.sh -selenium test/integration_selenium/test_workflow_editor.py

-framework
- 测试基础设施测试

bash

./run_tests.sh -framework test/framework/

Useful Flags

实用标识

Show detailed output:

bash

./run_tests.sh -integration test/integration/test_credentials.py --verbose_errors

Generate coverage report:

bash

./run_tests.sh --coverage -integration test/integration/test_credentials.py

Debug mode (drop into pdb on failure):

bash

./run_tests.sh --debug -integration test/integration/test_credentials.py

Run tests matching pattern:

bash

./run_tests.sh -integration test/integration/test_credentials.py -k "test_create"

Show print statements:

bash

./run_tests.sh -integration test/integration/test_credentials.py -s

Run with specific number of workers (parallel):

bash

./run_tests.sh -integration test/integration/ -n 4

显示详细输出：

bash

./run_tests.sh -integration test/integration/test_credentials.py --verbose_errors

生成覆盖率报告：

bash

./run_tests.sh --coverage -integration test/integration/test_credentials.py

调试模式（测试失败时进入pdb）：

bash

./run_tests.sh --debug -integration test/integration/test_credentials.py

运行匹配指定模式的测试：

bash

./run_tests.sh -integration test/integration/test_credentials.py -k "test_create"

显示打印语句：

bash

./run_tests.sh -integration test/integration/test_credentials.py -s

使用指定数量的工作进程并行运行：

bash

./run_tests.sh -integration test/integration/ -n 4

When Using pytest Directly (Advanced)

直接使用pytest的场景（高级用法）

If you must use pytest directly (e.g., for IDE integration), use markers:

bash

pytest -m "not slow" test/unit/
pytest -m "unit" test/unit/managers/test_workflows.py
pytest -m "integration" test/integration/test_credentials.py

But prefer

./run_tests.sh

for normal usage.

如果必须直接使用pytest（例如用于IDE集成），请使用标记：

bash

pytest -m "not slow" test/unit/
pytest -m "unit" test/unit/managers/test_workflows.py
pytest -m "integration" test/integration/test_credentials.py

但日常使用优先选择

./run_tests.sh

。

If $ARGUMENTS is "write": Guide to Writing Tests

当$ARGUMENTS为"write"时：测试编写指南

Ask the user what type of test they want to write:

Unit tests - Fast, isolated tests with mocked dependencies
API tests - Test API endpoints with Galaxy server
Integration tests - Full system tests with real database

Then provide guidance based on their choice (see sections below).

询问用户想要编写哪种类型的测试：

单元测试 - 快速、独立的测试，依赖已模拟
API测试 - 结合Galaxy服务器测试API端点
集成测试 - 使用真实数据库的全系统测试

然后根据用户的选择提供指导（见下文章节）。

If $ARGUMENTS contains "unit": Unit Test Writing Guide

当$ARGUMENTS包含"unit"时：单元测试编写指南

What Are Unit Tests?

什么是单元测试？

Unit tests are fast, isolated tests that:

Run without starting Galaxy server
Use in-memory SQLite database
Mock external dependencies
Test individual manager/service methods
Are located in
```
test/unit/
```

单元测试是快速、独立的测试，具备以下特点：

无需启动Galaxy服务器即可运行
使用内存中的SQLite数据库
模拟外部依赖
测试单个管理器/服务方法
存放路径：
```
test/unit/
```

Unit Test Structure

单元测试结构

Location:

test/unit/<module>/test_<class>.py

Base class:

BaseTestCase

from

test.unit.app.managers.base

Example unit test:

python

"""
Unit tests for MyResourceManager.
"""
from galaxy import model
from galaxy.managers.myresources import MyResourceManager
from test.unit.app.managers.base import BaseTestCase


class TestMyResourceManager(BaseTestCase):
    """Unit tests for MyResourceManager."""

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

    def set_up_managers(self):
        """Set up managers under test."""
        self.manager = MyResourceManager(self.app)

    def test_create_myresource(self):
        """Test creating a resource."""
        # Arrange
        trans = self.trans  # MockTrans from BaseTestCase
        name = "Test Resource"

        # Act
        resource = self.manager.create(trans, name=name)
        self.session.flush()

        # Assert
        assert resource.name == name
        assert resource.user_id == trans.user.id
        assert resource.id is not None

    def test_get_myresource(self):
        """Test getting a resource by ID."""
        # Arrange
        resource = self._create_resource("Test Resource")

        # Act
        retrieved = self.manager.get(self.trans, resource.id)

        # Assert
        assert retrieved.id == resource.id
        assert retrieved.name == resource.name

    def test_get_nonexistent_myresource_raises_not_found(self):
        """Test that getting nonexistent resource raises exception."""
        from galaxy.exceptions import ObjectNotFound

        with self.assertRaises(ObjectNotFound):
            self.manager.get(self.trans, 99999)

    def test_list_myresources_for_user(self):
        """Test listing resources for current user."""
        # Arrange
        self._create_resource("Resource 1")
        self._create_resource("Resource 2")

        # Act
        resources = self.manager.list_for_user(self.trans)

        # Assert
        assert len(resources) >= 2
        names = [r.name for r in resources]
        assert "Resource 1" in names
        assert "Resource 2" in names

    def test_update_myresource(self):
        """Test updating a resource."""
        # Arrange
        resource = self._create_resource("Original Name")
        new_name = "Updated Name"

        # Act
        updated = self.manager.update(self.trans, resource.id, name=new_name)
        self.session.flush()

        # Assert
        assert updated.id == resource.id
        assert updated.name == new_name

    def test_delete_myresource(self):
        """Test soft-deleting a resource."""
        # Arrange
        resource = self._create_resource("To Delete")

        # Act
        self.manager.delete(self.trans, resource.id)
        self.session.flush()

        # Assert
        assert resource.deleted is True

    def test_cannot_access_other_user_resource(self):
        """Test access control for other users' resources."""
        from galaxy.exceptions import ItemAccessibilityException

        # Arrange
        other_user = self._create_user("other@example.com")
        other_trans = self._create_trans(user=other_user)
        resource = self.manager.create(other_trans, name="Other User Resource")
        self.session.flush()

        # Act & Assert
        with self.assertRaises(ItemAccessibilityException):
            self.manager.get(self.trans, resource.id)

    def _create_resource(self, name: str, **kwargs):
        """Helper to create a test resource."""
        resource = self.manager.create(self.trans, name=name, **kwargs)
        self.session.flush()
        return resource

    def _create_user(self, email: str):
        """Helper to create a test user."""
        user = model.User(email=email, username=email.split("@")[0])
        self.session.add(user)
        self.session.flush()
        return user

    def _create_trans(self, user=None):
        """Helper to create a transaction context for a user."""
        from galaxy_mock import MockTrans
        return MockTrans(app=self.app, user=user or self.user)

存放位置：

test/unit/<module>/test_<class>.py

基类： 来自

test.unit.app.managers.base

的

BaseTestCase

单元测试示例：

python

"""
Unit tests for MyResourceManager.
"""
from galaxy import model
from galaxy.managers.myresources import MyResourceManager
from test.unit.app.managers.base import BaseTestCase


class TestMyResourceManager(BaseTestCase):
    """Unit tests for MyResourceManager."""

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

    def set_up_managers(self):
        """Set up managers under test."""
        self.manager = MyResourceManager(self.app)

    def test_create_myresource(self):
        """Test creating a resource."""
        # Arrange
        trans = self.trans  # MockTrans from BaseTestCase
        name = "Test Resource"

        # Act
        resource = self.manager.create(trans, name=name)
        self.session.flush()

        # Assert
        assert resource.name == name
        assert resource.user_id == trans.user.id
        assert resource.id is not None

    def test_get_myresource(self):
        """Test getting a resource by ID."""
        # Arrange
        resource = self._create_resource("Test Resource")

        # Act
        retrieved = self.manager.get(self.trans, resource.id)

        # Assert
        assert retrieved.id == resource.id
        assert retrieved.name == resource.name

    def test_get_nonexistent_myresource_raises_not_found(self):
        """Test that getting nonexistent resource raises exception."""
        from galaxy.exceptions import ObjectNotFound

        with self.assertRaises(ObjectNotFound):
            self.manager.get(self.trans, 99999)

    def test_list_myresources_for_user(self):
        """Test listing resources for current user."""
        # Arrange
        self._create_resource("Resource 1")
        self._create_resource("Resource 2")

        # Act
        resources = self.manager.list_for_user(self.trans)

        # Assert
        assert len(resources) >= 2
        names = [r.name for r in resources]
        assert "Resource 1" in names
        assert "Resource 2" in names

    def test_update_myresource(self):
        """Test updating a resource."""
        # Arrange
        resource = self._create_resource("Original Name")
        new_name = "Updated Name"

        # Act
        updated = self.manager.update(self.trans, resource.id, name=new_name)
        self.session.flush()

        # Assert
        assert updated.id == resource.id
        assert updated.name == new_name

    def test_delete_myresource(self):
        """Test soft-deleting a resource."""
        # Arrange
        resource = self._create_resource("To Delete")

        # Act
        self.manager.delete(self.trans, resource.id)
        self.session.flush()

        # Assert
        assert resource.deleted is True

    def test_cannot_access_other_user_resource(self):
        """Test access control for other users' resources."""
        from galaxy.exceptions import ItemAccessibilityException

        # Arrange
        other_user = self._create_user("other@example.com")
        other_trans = self._create_trans(user=other_user)
        resource = self.manager.create(other_trans, name="Other User Resource")
        self.session.flush()

        # Act & Assert
        with self.assertRaises(ItemAccessibilityException):
            self.manager.get(self.trans, resource.id)

    def _create_resource(self, name: str, **kwargs):
        """Helper to create a test resource."""
        resource = self.manager.create(self.trans, name=name, **kwargs)
        self.session.flush()
        return resource

    def _create_user(self, email: str):
        """Helper to create a test user."""
        user = model.User(email=email, username=email.split("@")[0])
        self.session.add(user)
        self.session.flush()
        return user

    def _create_trans(self, user=None):
        """Helper to create a transaction context for a user."""
        from galaxy_mock import MockTrans
        return MockTrans(app=self.app, user=user or self.user)

Key Points for Unit Tests

单元测试关键点

Extend
BaseTestCase
from
```
test.unit.app.managers.base
```
Use
self.trans
- Pre-configured MockTrans with test user
Use
self.session
- SQLAlchemy session (in-memory SQLite)
Call
self.session.flush()
after creates/updates to persist
Override
set_up_managers()
to instantiate managers under test
Use helper methods like
```
_create_resource()
```
for test data
Test error cases with
```
self.assertRaises()
```
Follow AAA pattern - Arrange, Act, Assert

继承
BaseTestCase
来自
```
test.unit.app.managers.base
```
使用
self.trans
- 预配置的MockTrans，包含测试用户
使用
self.session
- SQLAlchemy会话（内存中的SQLite）
创建/更新后调用
self.session.flush()
以持久化数据
重写
set_up_managers()
以实例化待测试的管理器
使用辅助方法 如
```
_create_resource()
```
生成测试数据
使用
self.assertRaises()
测试错误场景
遵循AAA模式 - 准备（Arrange）、执行（Act）、断言（Assert）

Available from BaseTestCase

BaseTestCase提供的可用属性

python

self.app          # Galaxy application mock
self.trans        # MockTrans with test user
self.user         # Test user (admin)
self.session      # SQLAlchemy session
self.history      # Default test history

python

self.app          # Galaxy application mock
self.trans        # MockTrans with test user
self.user         # Test user (admin)
self.session      # SQLAlchemy session
self.history      # Default test history

Running Unit Tests

运行单元测试

bash

undefined

bash

undefined

Run all unit tests for a manager

./run_tests.sh -unit test/unit/managers/test_myresources.py

Run specific test

./run_tests.sh -unit test/unit/managers/test_myresources.py::TestMyResourceManager::test_create_myresource

Run with coverage

./run_tests.sh --coverage -unit test/unit/managers/test_myresources.py

---

./run_tests.sh --coverage -unit test/unit/managers/test_myresources.py

---

If $ARGUMENTS contains "api": API Test Writing Guide

当$ARGUMENTS包含"api"时：API测试编写指南

What Are API Tests?

什么是API测试？

API tests:

Start a Galaxy server
Make HTTP requests to API endpoints
Test request/response handling
Verify status codes and response schemas
Are located in
```
lib/galaxy_test/api/
```

API测试具备以下特点：

启动Galaxy服务器
向API端点发送HTTP请求
测试请求/响应处理逻辑
验证状态码和响应结构
存放路径：
```
lib/galaxy_test/api/
```

API Test Structure

API测试结构

Location:

lib/galaxy_test/api/test_<resource>s.py

Base class:

ApiTestCase

from

lib/galaxy_test/api/_framework

Example API test:

python

"""
API tests for MyResource endpoints.
"""
from galaxy_test.base.populators import DatasetPopulator
from ._framework import ApiTestCase


class TestMyResourcesApi(ApiTestCase):
    """Tests for /api/myresources endpoints."""

    def setUp(self):
        super().setUp()
        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)

    def test_create_myresource(self):
        """Test POST /api/myresources."""
        payload = {
            "name": "Test Resource",
            "description": "Test description",
        }
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)

        resource = response.json()
        self._assert_has_keys(resource, "id", "name", "description", "create_time")
        assert resource["name"] == "Test Resource"
        assert resource["description"] == "Test description"

    def test_list_myresources(self):
        """Test GET /api/myresources."""
        # Create test data
        self._create_myresource("Resource 1")
        self._create_myresource("Resource 2")

        # List
        response = self._get("myresources")
        self._assert_status_code_is_ok(response)

        data = response.json()
        assert "items" in data
        assert "total_count" in data
        assert data["total_count"] >= 2
        assert len(data["items"]) >= 2

    def test_get_myresource(self):
        """Test GET /api/myresources/{id}."""
        resource_id = self._create_myresource("Test Resource")

        response = self._get(f"myresources/{resource_id}")
        self._assert_status_code_is_ok(response)

        resource = response.json()
        assert resource["id"] == resource_id
        assert resource["name"] == "Test Resource"

    def test_update_myresource(self):
        """Test PUT /api/myresources/{id}."""
        resource_id = self._create_myresource("Original Name")

        payload = {"name": "Updated Name"}
        response = self._put(f"myresources/{resource_id}", data=payload, json=True)
        self._assert_status_code_is_ok(response)

        updated = response.json()
        assert updated["name"] == "Updated Name"

    def test_delete_myresource(self):
        """Test DELETE /api/myresources/{id}."""
        resource_id = self._create_myresource("To Delete")

        response = self._delete(f"myresources/{resource_id}")
        self._assert_status_code_is(response, 204)

        # Verify deletion
        response = self._get(f"myresources/{resource_id}")
        self._assert_status_code_is(response, 404)

    def test_get_nonexistent_myresource_returns_404(self):
        """Test that getting nonexistent resource returns 404."""
        response = self._get("myresources/invalid_id")
        self._assert_status_code_is(response, 404)

    def test_create_with_invalid_data_returns_422(self):
        """Test validation error handling."""
        payload = {}  # Missing required 'name'
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 422)

    def test_access_control_prevents_viewing_other_user_resource(self):
        """Test that users cannot access other users' resources."""
        # Create as first user
        resource_id = self._create_myresource("User 1 Resource")

        # Switch to different user
        with self._different_user():
            response = self._get(f"myresources/{resource_id}")
            self._assert_status_code_is(response, 403)

    def test_admin_can_access_all_resources(self):
        """Test that admin users have broader access."""
        # Create as regular user
        resource_id = self._create_myresource("User Resource")

        # Access as admin
        response = self._get(f"myresources/{resource_id}", admin=True)
        self._assert_status_code_is_ok(response)

    def _create_myresource(self, name: str, **kwargs) -> str:
        """Helper to create a resource and return its ID."""
        payload = {
            "name": name,
            "description": kwargs.get("description", f"Description for {name}"),
        }
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)
        return response.json()["id"]

存放位置：

lib/galaxy_test/api/test_<resource>s.py

基类： 来自

lib/galaxy_test/api/_framework

的

ApiTestCase

API测试示例：

python

"""
API tests for MyResource endpoints.
"""
from galaxy_test.base.populators import DatasetPopulator
from ._framework import ApiTestCase


class TestMyResourcesApi(ApiTestCase):
    """Tests for /api/myresources endpoints."""

    def setUp(self):
        super().setUp()
        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)

    def test_create_myresource(self):
        """Test POST /api/myresources."""
        payload = {
            "name": "Test Resource",
            "description": "Test description",
        }
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)

        resource = response.json()
        self._assert_has_keys(resource, "id", "name", "description", "create_time")
        assert resource["name"] == "Test Resource"
        assert resource["description"] == "Test description"

    def test_list_myresources(self):
        """Test GET /api/myresources."""
        # Create test data
        self._create_myresource("Resource 1")
        self._create_myresource("Resource 2")

        # List
        response = self._get("myresources")
        self._assert_status_code_is_ok(response)

        data = response.json()
        assert "items" in data
        assert "total_count" in data
        assert data["total_count"] >= 2
        assert len(data["items"]) >= 2

    def test_get_myresource(self):
        """Test GET /api/myresources/{id}."""
        resource_id = self._create_myresource("Test Resource")

        response = self._get(f"myresources/{resource_id}")
        self._assert_status_code_is_ok(response)

        resource = response.json()
        assert resource["id"] == resource_id
        assert resource["name"] == "Test Resource"

    def test_update_myresource(self):
        """Test PUT /api/myresources/{id}."""
        resource_id = self._create_myresource("Original Name")

        payload = {"name": "Updated Name"}
        response = self._put(f"myresources/{resource_id}", data=payload, json=True)
        self._assert_status_code_is_ok(response)

        updated = response.json()
        assert updated["name"] == "Updated Name"

    def test_delete_myresource(self):
        """Test DELETE /api/myresources/{id}."""
        resource_id = self._create_myresource("To Delete")

        response = self._delete(f"myresources/{resource_id}")
        self._assert_status_code_is(response, 204)

        # Verify deletion
        response = self._get(f"myresources/{resource_id}")
        self._assert_status_code_is(response, 404)

    def test_get_nonexistent_myresource_returns_404(self):
        """Test that getting nonexistent resource returns 404."""
        response = self._get("myresources/invalid_id")
        self._assert_status_code_is(response, 404)

    def test_create_with_invalid_data_returns_422(self):
        """Test validation error handling."""
        payload = {}  # Missing required 'name'
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 422)

    def test_access_control_prevents_viewing_other_user_resource(self):
        """Test that users cannot access other users' resources."""
        # Create as first user
        resource_id = self._create_myresource("User 1 Resource")

        # Switch to different user
        with self._different_user():
            response = self._get(f"myresources/{resource_id}")
            self._assert_status_code_is(response, 403)

    def test_admin_can_access_all_resources(self):
        """Test that admin users have broader access."""
        # Create as regular user
        resource_id = self._create_myresource("User Resource")

        # Access as admin
        response = self._get(f"myresources/{resource_id}", admin=True)
        self._assert_status_code_is_ok(response)

    def _create_myresource(self, name: str, **kwargs) -> str:
        """Helper to create a resource and return its ID."""
        payload = {
            "name": name,
            "description": kwargs.get("description", f"Description for {name}"),
        }
        response = self._post("myresources", data=payload, json=True)
        self._assert_status_code_is(response, 201)
        return response.json()["id"]

Key Points for API Tests

API测试关键点

Extend
ApiTestCase
from

lib/galaxy_test/api/_framework

HTTP methods:

```
self._get(path)
```
- GET request
```
self._post(path, data=..., json=True)
```
- POST request
```
self._put(path, data=..., json=True)
```
- PUT request
```
self._delete(path)
```
- DELETE request
Paths are relative to
```
/api/
```
(e.g.,
```
"myresources"
```
→
```
/api/myresources
```
)

Assertions:

self._assert_status_code_is(response, 200)

- Check specific status

self._assert_status_code_is_ok(response)

- Check 2xx status

self._assert_has_keys(obj, "key1", "key2")

- Verify response structure

User context:
- Default: Regular user
- ```
admin=True
```
  parameter: Make request as admin
- ```
self._different_user()
```
  context manager: Switch to different user
Test data:
- Use helper methods like
```
_create_myresource()
```
- Use
```
DatasetPopulator
```
  for creating datasets/histories

继承
ApiTestCase
来自

lib/galaxy_test/api/_framework

HTTP方法：

```
self._get(path)
```
- GET请求
```
self._post(path, data=..., json=True)
```
- POST请求
```
self._put(path, data=..., json=True)
```
- PUT请求
```
self._delete(path)
```
- DELETE请求
路径为
```
/api/
```
的相对路径（例如
```
"myresources"
```
→
```
/api/myresources
```
）

断言方法：

self._assert_status_code_is(response, 200)

- 检查指定状态码

self._assert_status_code_is_ok(response)

- 检查2xx状态码

self._assert_has_keys(obj, "key1", "key2")

- 验证响应结构

用户上下文：
- 默认：普通用户
- ```
admin=True
```
  参数：以管理员身份发起请求
- ```
self._different_user()
```
  上下文管理器：切换到其他用户
测试数据：
- 使用辅助方法如
```
_create_myresource()
```
- 使用
```
DatasetPopulator
```
  创建数据集/历史记录

Additional ApiTestCase Features

ApiTestCase的其他功能

Create test datasets:

python

def setUp(self):
    super().setUp()
    self.dataset_populator = DatasetPopulator(self.galaxy_interactor)

def test_with_dataset(self):
    history_id = self.dataset_populator.new_history()
    dataset = self.dataset_populator.new_dataset(history_id, content="test data")
    # Use dataset["id"] in your test

Test as different user:

python

with self._different_user():
    response = self._get("myresources")
    # This request is made as a different user

Test with admin privileges:

python

response = self._get("myresources/admin/all", admin=True)

创建测试数据集：

python

def setUp(self):
    super().setUp()
    self.dataset_populator = DatasetPopulator(self.galaxy_interactor)

def test_with_dataset(self):
    history_id = self.dataset_populator.new_history()
    dataset = self.dataset_populator.new_dataset(history_id, content="test data")
    # Use dataset["id"] in your test

以其他用户身份测试：

python

with self._different_user():
    response = self._get("myresources")
    # This request is made as a different user

以管理员权限测试：

python

response = self._get("myresources/admin/all", admin=True)

Running API Tests

运行API测试

bash

undefined

bash

undefined

Run all API tests for an endpoint

./run_tests.sh -api lib/galaxy_test/api/test_myresources.py

Run specific test

./run_tests.sh -api lib/galaxy_test/api/test_myresources.py::TestMyResourcesApi::test_create_myresource

Run with verbose output

./run_tests.sh -api lib/galaxy_test/api/test_myresources.py --verbose_errors

---

./run_tests.sh -api lib/galaxy_test/api/test_myresources.py --verbose_errors

---

If $ARGUMENTS contains "integration": Integration Test Writing Guide

当$ARGUMENTS包含"integration"时：集成测试编写指南

What Are Integration Tests?

什么是集成测试？

Integration tests:

Test full system integration
Use real database (PostgreSQL optional)
Test complex workflows and interactions
Can customize Galaxy configuration
Are located in
```
test/integration/
```

集成测试具备以下特点：

测试全系统集成逻辑
使用真实数据库（可选PostgreSQL）
测试复杂工作流和交互逻辑
可自定义Galaxy配置
存放路径：
```
test/integration/
```

Integration Test Structure

集成测试结构

Location:

test/integration/test_<feature>.py

Base class:

IntegrationTestCase

from

lib/galaxy_test/driver/integration_util

Example integration test:

python

"""
Integration tests for MyResource with vault integration.
"""
from galaxy_test.driver import integration_util


class TestMyResourceIntegration(integration_util.IntegrationTestCase):
    """Integration tests for MyResource."""

    @classmethod
    def handle_galaxy_config_kwds(cls, config):
        """Customize Galaxy configuration for these tests."""
        super().handle_galaxy_config_kwds(config)
        config["vault_config_file"] = cls.vault_config_file
        config["enable_vault"] = True

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

    def test_myresource_with_vault(self):
        """Test creating resource with vault backend."""
        payload = {
            "name": "Vault Resource",
            "vault_type": "hashicorp",
            "username": "vaultuser",
            "password": "vaultpass",
        }
        response = self.galaxy_interactor.post("myresources", data=payload)
        response.raise_for_status()

        resource = response.json()
        assert resource["vault_type"] == "hashicorp"

        # Verify stored in vault
        vault_data = self._get_from_vault(resource["id"])
        assert vault_data["username"] == "vaultuser"

    def test_myresource_workflow_integration(self):
        """Test resource used in workflow."""
        # Create resource
        resource_id = self._create_myresource("Workflow Resource")

        # Create workflow that uses resource
        workflow_id = self._create_workflow_with_resource(resource_id)

        # Execute workflow
        history_id = self.dataset_populator.new_history()
        response = self.galaxy_interactor.post(
            "workflows",
            data={
                "workflow_id": workflow_id,
                "history_id": history_id,
                "resource_id": resource_id,
            }
        )
        response.raise_for_status()

        # Wait for workflow completion
        self.dataset_populator.wait_for_history(history_id)

        # Verify results
        datasets = self.dataset_populator.get_history_datasets(history_id)
        assert len(datasets) > 0

    def _create_myresource(self, name: str) -> str:
        """Helper to create a resource."""
        response = self.galaxy_interactor.post(
            "myresources",
            data={"name": name, "vault_type": "database"}
        )
        response.raise_for_status()
        return response.json()["id"]

    def _get_from_vault(self, resource_id: str):
        """Helper to retrieve data from vault."""
        # Access app internals for verification
        vault = self._app.vault
        return vault.read_secret(f"myresources/{resource_id}")

存放位置：

test/integration/test_<feature>.py

基类： 来自

lib/galaxy_test/driver/integration_util

的

IntegrationTestCase

集成测试示例：

python

"""
Integration tests for MyResource with vault integration.
"""
from galaxy_test.driver import integration_util


class TestMyResourceIntegration(integration_util.IntegrationTestCase):
    """Integration tests for MyResource."""

    @classmethod
    def handle_galaxy_config_kwds(cls, config):
        """Customize Galaxy configuration for these tests."""
        super().handle_galaxy_config_kwds(config)
        config["vault_config_file"] = cls.vault_config_file
        config["enable_vault"] = True

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

    def test_myresource_with_vault(self):
        """Test creating resource with vault backend."""
        payload = {
            "name": "Vault Resource",
            "vault_type": "hashicorp",
            "username": "vaultuser",
            "password": "vaultpass",
        }
        response = self.galaxy_interactor.post("myresources", data=payload)
        response.raise_for_status()

        resource = response.json()
        assert resource["vault_type"] == "hashicorp"

        # Verify stored in vault
        vault_data = self._get_from_vault(resource["id"])
        assert vault_data["username"] == "vaultuser"

    def test_myresource_workflow_integration(self):
        """Test resource used in workflow."""
        # Create resource
        resource_id = self._create_myresource("Workflow Resource")

        # Create workflow that uses resource
        workflow_id = self._create_workflow_with_resource(resource_id)

        # Execute workflow
        history_id = self.dataset_populator.new_history()
        response = self.galaxy_interactor.post(
            "workflows",
            data={
                "workflow_id": workflow_id,
                "history_id": history_id,
                "resource_id": resource_id,
            }
        )
        response.raise_for_status()

        # Wait for workflow completion
        self.dataset_populator.wait_for_history(history_id)

        # Verify results
        datasets = self.dataset_populator.get_history_datasets(history_id)
        assert len(datasets) > 0

    def _create_myresource(self, name: str) -> str:
        """Helper to create a resource."""
        response = self.galaxy_interactor.post(
            "myresources",
            data={"name": name, "vault_type": "database"}
        )
        response.raise_for_status()
        return response.json()["id"]

    def _get_from_vault(self, resource_id: str):
        """Helper to retrieve data from vault."""
        # Access app internals for verification
        vault = self._app.vault
        return vault.read_secret(f"myresources/{resource_id}")

Key Points for Integration Tests

集成测试关键点

Extend
IntegrationTestCase
from

lib/galaxy_test/driver.integration_util

Customize config: Override
```
handle_galaxy_config_kwds()
```
to set Galaxy config
HTTP requests: Use
```
self.galaxy_interactor.get()
```
,
```
.post()
```
, etc.
Direct app access:
```
self._app
```
gives access to Galaxy application internals
Populators: Same as API tests -
```
DatasetPopulator
```
,
```
WorkflowPopulator
```
Database access:
```
self._app.model.context
```
for SQLAlchemy session

继承
IntegrationTestCase
来自

lib/galaxy_test/driver.integration_util

自定义配置： 重写
```
handle_galaxy_config_kwds()
```
设置Galaxy配置
HTTP请求： 使用
```
self.galaxy_interactor.get()
```
、
```
.post()
```
等方法
直接访问应用：
```
self._app
```
可访问Galaxy应用内部
填充器： 与API测试相同 -
```
DatasetPopulator
```
、
```
WorkflowPopulator
```
数据库访问：
```
self._app.model.context
```
用于SQLAlchemy会话

Configuration Mixins

配置混合类

Use mixins to add common configuration:

python

from galaxy_test.driver.integration_util import (
    IntegrationTestCase,
    ConfiguresDatabaseVault,
)

class TestMyResourceWithVault(IntegrationTestCase, ConfiguresDatabaseVault):
    """Test with database vault configured."""

    @classmethod
    def handle_galaxy_config_kwds(cls, config):
        super().handle_galaxy_config_kwds(config)
        # Additional config here

Available mixins:

```
ConfiguresDatabaseVault
```
- Set up database vault
```
ConfiguresObjectStores
```
- Configure object stores
```
UsesToolshed
```
- Set up Tool Shed integration

使用混合类添加通用配置：

python

from galaxy_test.driver.integration_util import (
    IntegrationTestCase,
    ConfiguresDatabaseVault,
)

class TestMyResourceWithVault(IntegrationTestCase, ConfiguresDatabaseVault):
    """Test with database vault configured."""

    @classmethod
    def handle_galaxy_config_kwds(cls, config):
        super().handle_galaxy_config_kwds(config)
        # Additional config here

可用混合类：

```
ConfiguresDatabaseVault
```
- 配置数据库vault
```
ConfiguresObjectStores
```
- 配置对象存储
```
UsesToolshed
```
- 配置Tool Shed集成

Skip Decorators

跳过测试装饰器

Skip tests based on environment:

python

from galaxy_test.driver.integration_util import skip_unless_postgres, skip_unless_docker

@skip_unless_postgres()
def test_postgres_specific_feature(self):
    """Test that requires PostgreSQL."""
    pass

@skip_unless_docker()
def test_docker_specific_feature(self):
    """Test that requires Docker."""
    pass

根据环境跳过测试：

python

from galaxy_test.driver.integration_util import skip_unless_postgres, skip_unless_docker

@skip_unless_postgres()
def test_postgres_specific_feature(self):
    """Test that requires PostgreSQL."""
    pass

@skip_unless_docker()
def test_docker_specific_feature(self):
    """Test that requires Docker."""
    pass

Running Integration Tests

运行集成测试

bash

undefined

bash

undefined

Run integration tests

./run_tests.sh -integration test/integration/test_myresources.py

Run specific test

./run_tests.sh -integration test/integration/test_myresources.py::TestMyResourceIntegration::test_myresource_with_vault

Run with PostgreSQL

./run_tests.sh -integration test/integration/test_myresources.py --postgres

Run with coverage

./run_tests.sh --coverage -integration test/integration/test_myresources.py

---

./run_tests.sh --coverage -integration test/integration/test_myresources.py

---

Test Best Practices

测试最佳实践

General Guidelines

通用指南

Test naming: Use descriptive names that explain what is being tested
- Good:
```
test_create_myresource_with_valid_data
```
- Bad:
```
test_1
```
  ,
```
test_myresource
```
One assertion per test: Test one thing at a time
- Good: Separate
```
test_create
```
  ,
```
test_update
```
  ,
```
test_delete
```
- Bad: One
```
test_crud
```
  that does everything
Test error cases: Test both success and failure paths
- Test 404, 403, 400, 422 responses
- Test validation errors
- Test access control
Use helper methods: Extract common setup into helper methods
- ```
_create_myresource()
```
  ,
```
_create_user()
```
  , etc.
Clean test data: Tests should be independent and repeatable
Follow AAA pattern:
- Arrange: Set up test data
- Act: Perform the operation
- Assert: Verify the result

测试命名： 使用描述性名称，明确测试内容
- 推荐：
```
test_create_myresource_with_valid_data
```
- 不推荐：
```
test_1
```
  、
```
test_myresource
```
每个测试一个断言： 一次测试只验证一个点
- 推荐：将
```
test_crud
```
  拆分为
```
test_create
```
  、
```
test_update
```
  、
```
test_delete
```
- 不推荐：单个
```
test_crud
```
  包含所有操作的断言
测试错误场景： 同时测试成功和失败路径
- 测试404、403、400、422响应
- 测试验证错误
- 测试访问控制
使用辅助方法： 将通用初始化逻辑提取到辅助方法中
- 例如
```
_create_myresource()
```
  、
```
_create_user()
```
清理测试数据： 测试应独立且可重复运行
遵循AAA模式：
- 准备（Arrange）： 设置测试数据
- 执行（Act）： 执行操作
- 断言（Assert）： 验证结果

Common Patterns

常见模式

Testing lists:

python

resources = response.json()["items"]
assert len(resources) >= 2
names = [r["name"] for r in resources]
assert "Resource 1" in names

Testing timestamps:

python

from datetime import datetime
resource = response.json()
assert resource["create_time"] is not None
create_time = datetime.fromisoformat(resource["create_time"])
assert create_time < datetime.now()

Testing pagination:

python

response = self._get("myresources?limit=10&offset=0")
data = response.json()
assert len(data["items"]) <= 10
assert data["total_count"] >= len(data["items"])

测试列表：

python

resources = response.json()["items"]
assert len(resources) >= 2
names = [r["name"] for r in resources]
assert "Resource 1" in names

测试时间戳：

python

from datetime import datetime
resource = response.json()
assert resource["create_time"] is not None
create_time = datetime.fromisoformat(resource["create_time"])
assert create_time < datetime.now()

测试分页：

python

response = self._get("myresources?limit=10&offset=0")
data = response.json()
assert len(data["items"]) <= 10
assert data["total_count"] >= len(data["items"])

Additional Resources

其他资源

Key test infrastructure files:

```
lib/galaxy_test/api/_framework.py
```
- ApiTestCase base class

lib/galaxy_test/driver/integration_util.py

- IntegrationTestCase base class

```
test/unit/app/managers/base.py
```
- Unit test base class
```
galaxy_test/base/populators.py
```
- Test data populators

Example test files to reference:

bash

undefined

核心测试基础设施文件：

```
lib/galaxy_test/api/_framework.py
```
- ApiTestCase基类

lib/galaxy_test/driver/integration_util.py

- IntegrationTestCase基类

```
test/unit/app/managers/base.py
```
- 单元测试基类
```
galaxy_test/base/populators.py
```
- 测试数据填充器

参考示例测试文件：

bash

undefined

Find recent API tests

ls -t lib/galaxy_test/api/test_*.py | head -5

Find recent integration tests

ls -t test/integration/test_*.py | head -5

Find unit tests

ls test/unit/managers/test_*.py


**Running test suites:**
```bash

ls test/unit/managers/test_*.py


**运行测试套件：**
```bash

All unit tests

./run_tests.sh -unit test/unit/

All API tests (slow)

./run_tests.sh -api lib/galaxy_test/api/

All integration tests (very slow)

./run_tests.sh -integration test/integration/

---

./run_tests.sh -integration test/integration/

---

Troubleshooting Tests

测试故障排除

Test fails with "database locked"

测试失败，提示“database locked”

Cause: Multiple tests accessing SQLite concurrently
Solution: Use
```
pytest-xdist
```
with
```
-n
```
flag or run serially

原因：多个测试同时访问SQLite
解决方案：使用
```
pytest-xdist
```
的
```
-n
```
标志并行运行，或串行运行

Test fails with "port already in use"

测试失败，提示“port already in use”

Cause: Previous test server didn't shut down
Solution: Kill Galaxy processes:
```
pkill -f 'python.*galaxy'
```

原因：之前的测试服务器未关闭
解决方案：杀死Galaxy进程：
```
pkill -f 'python.*galaxy'
```

Test fails with "fixture not found"

测试失败，提示“fixture not found”

Cause: Missing test dependency
Solution: Check imports and base class

原因：缺少测试依赖
解决方案：检查导入和基类

Integration test timeout

集成测试超时

Cause: Test waiting for long-running job
Solution: Use
```
wait_for_history()
```
with longer timeout

原因：测试等待长时间运行的任务
解决方案：使用
```
wait_for_history()
```
并设置更长的超时时间

Cannot import test module

无法导入测试模块

Cause: Python path not set correctly
Solution: Always use
```
./run_tests.sh
```
, not direct pytest

原因：Python路径设置不正确
解决方案：始终使用
```
./run_tests.sh
```
，切勿直接使用pytest

Quick Reference

快速参考

Test Type	Location	Base Class	Use When
Unit	`test/unit/`	`BaseTestCase`	Testing manager/service logic
API	`lib/galaxy_test/api/`	`ApiTestCase`	Testing API endpoints
Integration	`test/integration/`	`IntegrationTestCase`	Testing full system integration
Selenium	`test/integration_selenium/`	`SeleniumTestCase`	Testing browser UI

Running tests:

Unit:
```
./run_tests.sh -unit test/unit/...
```

API:

./run_tests.sh -api lib/galaxy_test/api/...

Integration:

./run_tests.sh -integration test/integration/...

Common assertions:

self._assert_status_code_is(response, 200)

self._assert_status_code_is_ok(response)

self._assert_has_keys(obj, "key1", "key2")

```
self.assertRaises(ExceptionType)
```

Common helpers:

self._get(path)

self._post(path, data=...)

self._put(...)

self._delete(...)

```
self._different_user()
```
- Context manager for different user

DatasetPopulator(self.galaxy_interactor)

- Create test datasets

测试类型	存放位置	基类	使用场景
单元测试	`test/unit/`	`BaseTestCase`	测试管理器/服务逻辑
API测试	`lib/galaxy_test/api/`	`ApiTestCase`	测试API端点
集成测试	`test/integration/`	`IntegrationTestCase`	测试全系统集成
Selenium测试	`test/integration_selenium/`	`SeleniumTestCase`	测试浏览器UI

运行测试：

单元测试：
```
./run_tests.sh -unit test/unit/...
```

API测试：

./run_tests.sh -api lib/galaxy_test/api/...

集成测试：

./run_tests.sh -integration test/integration/...

常用断言：

self._assert_status_code_is(response, 200)

self._assert_status_code_is_ok(response)

self._assert_has_keys(obj, "key1", "key2")

```
self.assertRaises(ExceptionType)
```

常用辅助方法：

self._get(path)

、

self._post(path, data=...)

、

self._put(...)

、

self._delete(...)

```
self._different_user()
```
- 切换用户的上下文管理器

DatasetPopulator(self.galaxy_interactor)

- 创建测试数据集