Source code for galaxy_test.api.test_visualizations

import json
import uuid

from requests import put

from galaxy_test.api.sharable import SharingApiTests
from galaxy_test.base.api_asserts import assert_has_keys
from ._framework import ApiTestCase

INDEX_KEYS = ["id", "title", "type", "dbkey"]
SHOW_KEYS = INDEX_KEYS + ["user_id", "model_class", "revisions", "latest_revision", "annotation"]
REVISION_KEYS = ["id", "title", "visualization_id", "dbkey", "model_class", "config"]


[docs]class TestVisualizationsApi(ApiTestCase, SharingApiTests):
    api_name = "visualizations"

[docs]    def create(self, _: str) -> str:
        viz_id, _ = self._create_viz()
        return viz_id

[docs]    def test_index_and_show(self):
        self._create_viz()  # to ensure on exists to index
        index = self._get("visualizations").json()
        assert len(index) >= 1
        for viz in index:
            self._verify_viz_object(viz, show=False)
        first_viz = index[0]
        self._show_viz(first_viz["id"])

[docs]    def test_index_ordering(self):
        ids = []
        for x in range(3):
            v = self._new_viz(title=f"visualization-order-{x}", slug=f"slug-order-{x}").json()
            ids.append(v["id"])
        index_ids = self._index_ids()
        for x in range(2):
            assert index_ids.index(ids[x]) > index_ids.index(ids[x + 1])
        index_ids = self._index_ids(dict(sort_desc=False))
        for x in range(2):
            assert index_ids.index(ids[x]) < index_ids.index(ids[x + 1])

[docs]    def test_index_filtering(self):
        ids = []
        for x in range(3):
            v = self._new_viz(title=f"visualization-filter-{x}", slug=f"slug-filter-{x}").json()
            ids.append(v["id"])
        index_ids = self._index_ids(dict(show_own=False))
        for x in range(3):
            assert ids[x] not in index_ids
        self._publish_viz(ids[0])
        index_ids = self._index_ids(dict(show_own=False))
        assert ids[0] in index_ids
        index_ids = self._index_ids(dict(show_own=False, show_published=False))
        for x in range(3):
            assert ids[x] in index_ids
        index_ids = self._index_ids(dict(search="visualization-filter-1"))
        assert ids[1] in index_ids
        self._update_viz(ids[1], dict(deleted=True))
        index_ids = self._index_ids(dict(search="visualization-filter-1"))
        assert ids[1] not in index_ids
        index_ids = self._index_ids(dict(search="is:deleted", show_published=False))
        assert ids[1] in index_ids

[docs]    def test_create(self):
        viz_id, viz_request = self._create_viz()
        self._show_viz(viz_id, assert_ok=True)

[docs]    def test_create_fails_without_title(self):
        response = self._new_viz(title="")
        self._assert_status_code_is(response, 400)

[docs]    def test_create_fails_with_bad_slug(self):
        response = self._new_viz(slug="123_()")
        self._assert_status_code_is(response, 400)

[docs]    def test_create_fails_with_invalid_config(self):
        response = self._new_viz(config="3 = nime")
        self._assert_status_code_is(response, 400)

[docs]    def test_sharing(self):
        viz_id, _ = self._create_viz()
        sharing_response = self._get(f"visualizations/{viz_id}/sharing")
        self._assert_status_code_is(sharing_response, 200)
        self._assert_has_keys(
            sharing_response.json(), "title", "importable", "id", "username_and_slug", "published", "users_shared_with"
        )

[docs]    def test_update_title(self):
        viz_id, viz = self._create_viz()
        update_url = self._api_url(f"visualizations/{viz_id}", use_key=True)
        response = put(update_url, {"title": "New Name"})
        self._assert_status_code_is(response, 200)
        updated_viz = self._show_viz(viz_id)
        assert updated_viz["title"] == "New Name"

    def _create_viz(self, **kwds):
        response = self._new_viz(**kwds)
        self._assert_status_code_is(response, 200)
        viz = response.json()
        return viz["id"], viz

    def _index(self, params):
        index_response = self._get("visualizations", data=params or {})
        return index_response.json()

    def _index_ids(self, params=None):
        return [entry["id"] for entry in self._index(params)]

    def _new_viz(self, title=None, slug=None, config=None):
        uuid_str = str(uuid.uuid4())
        title = title if title is not None else "Test Visualization"
        slug = slug if slug is not None else f"test-visualization-{uuid_str}"
        config = (
            config
            if config is not None
            else json.dumps(
                {
                    "x": 10,
                    "y": 12,
                }
            )
        )
        create_payload = {
            "title": title,
            "slug": slug,
            "type": "example",
            "dbkey": "hg17",
            "annotation": "this is a test of the emergency visualization system",
            "config": config,
        }
        response = self._post("visualizations", data=create_payload)
        return response

    def _publish_viz(self, id):
        sharing_response = self._put(f"visualizations/{id}/publish")
        assert sharing_response.status_code == 200
        return sharing_response.json()

    def _show_viz(self, viz_id, assert_ok=True):
        show_response = self._get(f"visualizations/{viz_id}")
        if assert_ok:
            self._assert_status_code_is(show_response, 200)
        viz = show_response.json()
        if assert_ok:
            self._verify_viz_object(viz, show=True)
        return viz

    def _update_viz(self, viz_id, params=None):
        response = self._put(f"visualizations/{viz_id}", params or {}, json=True)
        self._assert_status_code_is(response, 200)
        return response

    def _verify_viz_object(self, obj, show=False):
        assert_has_keys(obj, *(SHOW_KEYS if show else INDEX_KEYS))
        if show:
            assert "revisions" in obj
            revisions = obj["revisions"]
            assert len(revisions) >= 1
            assert "latest_revision" in obj
            latest_revision = obj["latest_revision"]
            assert_has_keys(latest_revision, *REVISION_KEYS)
            assert latest_revision["model_class"] == "VisualizationRevision"
            assert latest_revision["id"] in revisions