galaxy-testing

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

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

./run_tests.sh -integration test/integration/

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

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

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

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

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

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

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

"""
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)

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

# 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

"""
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"]

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

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

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

# 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

"""
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}")

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

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

# 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

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

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()

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

# 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

# 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/