mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-19 02:21:32 +08:00
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""Integration tests for workflow syncing."""
|
|
|
|
import json
|
|
import os
|
|
|
|
import yaml
|
|
|
|
from base import integration_util # noqa: I100,I202
|
|
from base.populators import ( # noqa: I100
|
|
DatasetPopulator,
|
|
WorkflowPopulator,
|
|
)
|
|
from base.workflow_fixtures import WORKFLOW_SIMPLE_CAT_TWICE # noqa: I100
|
|
|
|
|
|
class WorkflowSyncTestCase(integration_util.IntegrationTestCase):
|
|
|
|
framework_tool_and_types = True
|
|
require_admin_user = True
|
|
|
|
def setUp(self):
|
|
super(WorkflowSyncTestCase, self).setUp()
|
|
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
|
|
self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
|
|
|
|
@classmethod
|
|
def handle_galaxy_config_kwds(cls, config):
|
|
cls.workflow_directory = cls._test_driver.mkdtemp()
|
|
|
|
def test_sync_format2(self):
|
|
workflow_path = self._write_workflow_content("workflow.yml", WORKFLOW_SIMPLE_CAT_TWICE)
|
|
workflow_id = self.workflow_populator.import_workflow_from_path(workflow_path)
|
|
with self.workflow_populator.export_for_update(workflow_id) as workflow_object:
|
|
workflow_object["annotation"] = "new annotation"
|
|
with open(workflow_path, "r") as f:
|
|
data = yaml.safe_load(f)
|
|
assert data["doc"] == "new annotation"
|
|
|
|
def test_sync_ga(self):
|
|
workflow_json = self.workflow_populator.load_workflow("synctest")
|
|
workflow_path = self._write_workflow_content("workflow.ga", json.dumps(workflow_json))
|
|
workflow_id = self.workflow_populator.import_workflow_from_path(workflow_path)
|
|
with self.workflow_populator.export_for_update(workflow_id) as workflow_object:
|
|
workflow_object["annotation"] = "new annotation"
|
|
with open(workflow_path, "r") as f:
|
|
data = json.load(f)
|
|
assert data["annotation"] == "new annotation"
|
|
|
|
def _write_workflow_content(self, filename, content):
|
|
workflow_path = os.path.join(self.workflow_directory, filename)
|
|
with open(workflow_path, "w") as f:
|
|
f.write(content)
|
|
return workflow_path
|