# pylint: disable=C0103,E1101
import json
import os
import shutil
import socket
import sys
import tarfile
import tempfile
import unittest
import uuid
from ssl import SSLError
from typing import (
    Any,
    Callable,
    Collection,
    Dict,
    Iterable,
    List,
    Set,
    Tuple,
    Union,
)
from urllib.error import URLError
from urllib.request import urlopen

import pytest
from typing_extensions import Literal

import bioblend
from bioblend.galaxy import dataset_collections
from bioblend.galaxy.objects import (
    galaxy_instance,
    wrappers,
)
from . import test_util

bioblend.set_stream_logger("test", level="INFO")
socket.setdefaulttimeout(10.0)
SAMPLE_FN = test_util.get_abspath(os.path.join("data", "paste_columns.ga"))
SAMPLE_WF_COLL_FN = test_util.get_abspath(os.path.join("data", "paste_columns_collections.ga"))
SAMPLE_WF_PARAMETER_INPUT_FN = test_util.get_abspath(os.path.join("data", "workflow_with_parameter_input.ga"))
FOO_DATA = "foo\nbar\n"
FOO_DATA_2 = "foo2\nbar2\n"
SAMPLE_WF_DICT = {
    "deleted": False,
    "id": "9005c5112febe774",
    "inputs": {
        "571": {"label": "Input Dataset", "value": ""},
        "572": {"label": "Input Dataset", "value": ""},
    },
    "model_class": "StoredWorkflow",
    "name": "paste_columns",
    "owner": "user_foo",
    "published": False,
    "steps": {
        "571": {
            "id": 571,
            "input_steps": {},
            "tool_id": None,
            "tool_inputs": {"name": "Input Dataset"},
            "tool_version": None,
            "type": "data_input",
        },
        "572": {
            "id": 572,
            "input_steps": {},
            "tool_id": None,
            "tool_inputs": {"name": "Input Dataset"},
            "tool_version": None,
            "type": "data_input",
        },
        "573": {
            "id": 573,
            "input_steps": {
                "input1": {"source_step": 571, "step_output": "output"},
                "input2": {"source_step": 572, "step_output": "output"},
            },
            "tool_id": "Paste1",
            "tool_inputs": {
                "delimiter": '"T"',
                "input1": "null",
                "input2": "null",
            },
            "tool_version": "1.0.0",
            "type": "tool",
        },
    },
    "tags": [],
    "url": "/api/workflows/9005c5112febe774",
}
SAMPLE_INV_DICT: Dict[str, Any] = {
    "history_id": "2f94e8ae9edff68a",
    "id": "df7a1f0c02a5b08e",
    "inputs": {"0": {"id": "a7db2fac67043c7e", "src": "hda", "uuid": "7932ffe0-2340-4952-8857-dbaa50f1f46a"}},
    "model_class": "WorkflowInvocation",
    "state": "ready",
    "steps": [
        {
            "action": None,
            "id": "d413a19dec13d11e",
            "job_id": None,
            "model_class": "WorkflowInvocationStep",
            "order_index": 0,
            "state": None,
            "update_time": "2015-10-31T22:00:26",
            "workflow_step_id": "cbbbf59e8f08c98c",
            "workflow_step_label": None,
            "workflow_step_uuid": "b81250fd-3278-4e6a-b269-56a1f01ef485",
        },
        {
            "action": None,
            "id": "2f94e8ae9edff68a",
            "job_id": "e89067bb68bee7a0",
            "model_class": "WorkflowInvocationStep",
            "order_index": 1,
            "state": "new",
            "update_time": "2015-10-31T22:00:26",
            "workflow_step_id": "964b37715ec9bd22",
            "workflow_step_label": None,
            "workflow_step_uuid": "e62440b8-e911-408b-b124-e05435d3125e",
        },
    ],
    "update_time": "2015-10-31T22:00:26",
    "uuid": "c8aa2b1c-801a-11e5-a9e5-8ca98228593c",
    "workflow_id": "03501d7626bd192f",
}


def is_reachable(url: str) -> bool:
    res = None
    try:
        res = urlopen(url, timeout=5)
    except (SSLError, URLError, socket.timeout):
        return False
    if res is not None:
        res.close()
    return True


def upload_from_fs(
    lib: wrappers.Library, bnames: Iterable[str], **kwargs: Any
) -> Tuple[List[wrappers.LibraryDataset], List[str]]:
    tempdir = tempfile.mkdtemp(prefix="bioblend_test_")
    try:
        fnames = [os.path.join(tempdir, _) for _ in bnames]
        for fn in fnames:
            with open(fn, "w") as f:
                f.write(FOO_DATA)
        dss = lib.upload_from_galaxy_fs(fnames, **kwargs)
    finally:
        shutil.rmtree(tempdir)
    return dss, fnames


class MockWrapper(wrappers.Wrapper):
    BASE_ATTRS = ("a", "b")
    a: int
    b: List[int]

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)


class TestWrapper(unittest.TestCase):
    def setUp(self):
        self.d: Dict[str, Any] = {"a": 1, "b": [2, 3], "c": {"x": 4}}
        with pytest.raises(TypeError):
            wrappers.Wrapper(self.d)
        self.w = MockWrapper(self.d)

    def test_initialize(self):
        for k in MockWrapper.BASE_ATTRS:
            assert getattr(self.w, k) == self.d[k]
        self.w.a = 222
        self.w.b[0] = 222
        assert self.w.a == 222
        assert self.w.b[0] == 222
        assert self.d["a"] == 1
        assert self.d["b"][0] == 2
        with pytest.raises(AttributeError):
            _ = self.w.foo  # type: ignore[attr-defined]
        with pytest.raises(AttributeError):
            self.w.foo = 0  # type: ignore[assignment]

    def test_taint(self):
        assert not self.w.is_modified
        self.w.a = 111  # pylint: disable=W0201
        assert self.w.is_modified

    def test_serialize(self):
        w = MockWrapper.from_json(self.w.to_json())
        assert w.wrapped == self.w.wrapped

    def test_clone(self):
        w = self.w.clone()
        assert w.wrapped == self.w.wrapped
        w.b[0] = 111
        assert self.w.b[0] == 2

    def test_kwargs(self):
        parent = MockWrapper({"a": 10})
        w = MockWrapper(self.d, parent=parent)
        assert w.parent is parent
        with pytest.raises(AttributeError):
            w.parent = 0  # type: ignore[assignment,misc]


@test_util.skip_unless_galaxy()
class GalaxyObjectsTestBase(unittest.TestCase):
    gi: galaxy_instance.GalaxyInstance

    @classmethod
    def setUpClass(cls) -> None:
        galaxy_key = os.environ["BIOBLEND_GALAXY_API_KEY"]
        galaxy_url = os.environ["BIOBLEND_GALAXY_URL"]
        cls.gi = galaxy_instance.GalaxyInstance(galaxy_url, galaxy_key)


class TestWorkflow(GalaxyObjectsTestBase):
    def setUp(self):
        self.wf = wrappers.Workflow(SAMPLE_WF_DICT)

    def test_initialize(self):
        assert self.wf.id == "9005c5112febe774"
        assert self.wf.name == "paste_columns"
        assert not self.wf.deleted
        assert self.wf.owner == "user_foo"
        assert not self.wf.published
        assert self.wf.tags == []
        assert self.wf.input_labels_to_ids == {"Input Dataset": {"571", "572"}}
        assert self.wf.tool_labels_to_ids == {"Paste1": {"573"}}
        assert self.wf.data_input_ids == {"571", "572"}
        assert self.wf.source_ids == {"571", "572"}
        assert self.wf.sink_ids == {"573"}

    def test_dag(self):
        inv_dag: Dict[str, Set[str]] = {}
        for h, tails in self.wf.dag.items():
            for t in tails:
                inv_dag.setdefault(str(t), set()).add(h)
        assert self.wf.inv_dag == inv_dag
        heads = set(self.wf.dag)
        assert heads == set.union(*self.wf.inv_dag.values())
        tails = set(self.wf.inv_dag)
        assert tails == set.union(*self.wf.dag.values())
        ids = self.wf.sorted_step_ids()
        assert set(ids) == heads | tails
        for h, tails in self.wf.dag.items():
            for t in tails:
                assert ids.index(h) < ids.index(t)

    def test_steps(self):
        steps = SAMPLE_WF_DICT["steps"]
        assert isinstance(steps, dict)
        for sid, s in self.wf.steps.items():
            assert isinstance(s, wrappers.Step)
            assert s.id == sid
            assert sid in steps
            assert s.parent is self.wf
        assert self.wf.data_input_ids == {"571", "572"}
        assert self.wf.tool_ids == {"573"}

    def test_taint(self):
        assert not self.wf.is_modified
        self.wf.steps["571"].tool_id = "foo"
        assert self.wf.is_modified

    def test_input_map(self):
        history = wrappers.History({}, gi=self.gi)
        library = wrappers.Library({}, gi=self.gi)
        hda = wrappers.HistoryDatasetAssociation({"id": "hda_id"}, container=history, gi=self.gi)
        ldda = wrappers.LibraryDatasetDatasetAssociation({"id": "ldda_id"}, container=library, gi=self.gi)
        input_map = self.wf._convert_input_map({"0": hda, "1": ldda, "2": {"id": "hda2_id", "src": "hda"}})
        assert input_map == {
            "0": {"id": "hda_id", "src": "hda"},
            "1": {"id": "ldda_id", "src": "ldda"},
            "2": {"id": "hda2_id", "src": "hda"},
        }


@test_util.skip_unless_galaxy("release_19.09")
class TestInvocation(GalaxyObjectsTestBase):
    dataset: wrappers.HistoryDatasetAssociation
    history: wrappers.History
    inv: wrappers.Invocation
    workflow: wrappers.Workflow
    workflow_pause: wrappers.Workflow

    @classmethod
    def setUpClass(cls) -> None:
        super().setUpClass()
        cls.inv = wrappers.Invocation(SAMPLE_INV_DICT, gi=cls.gi)
        with open(SAMPLE_FN) as f:
            cls.workflow = cls.gi.workflows.import_new(f.read())
        path_pause = test_util.get_abspath(os.path.join("data", "test_workflow_pause.ga"))
        with open(path_pause) as f:
            cls.workflow_pause = cls.gi.workflows.import_new(f.read())
        cls.history = cls.gi.histories.create(name="TestInvocation")
        cls.dataset = cls.history.paste_content("1\t2\t3")

    @classmethod
    def tearDownClass(cls):
        cls.history.delete(purge=True)

    def test_initialize(self):
        assert self.inv.workflow_id == "03501d7626bd192f"
        assert self.inv.history_id == "2f94e8ae9edff68a"
        assert self.inv.id == "df7a1f0c02a5b08e"
        assert self.inv.state == "ready"
        assert self.inv.update_time == "2015-10-31T22:00:26"
        assert self.inv.uuid == "c8aa2b1c-801a-11e5-a9e5-8ca98228593c"

    def test_initialize_steps(self):
        for step, step_dict in zip(self.inv.steps, SAMPLE_INV_DICT["steps"]):
            assert isinstance(step_dict, dict)
            assert isinstance(step, wrappers.InvocationStep)
            assert step.parent is self.inv
            assert step.id == step_dict["id"]
            assert step.job_id == step_dict["job_id"]
            assert step.order_index == step_dict["order_index"]
            assert step.state == step_dict["state"]
            assert step.update_time == step_dict["update_time"]
            assert step.workflow_step_id == step_dict["workflow_step_id"]
            assert step.workflow_step_label == step_dict["workflow_step_label"]
            assert step.workflow_step_uuid == step_dict["workflow_step_uuid"]

    def test_initialize_inputs(self):
        for i, input in enumerate(self.inv.inputs):
            assert input == {**SAMPLE_INV_DICT["inputs"][str(i)], "label": str(i)}

    def test_sorted_step_ids(self):
        assert self.inv.sorted_step_ids() == ["d413a19dec13d11e", "2f94e8ae9edff68a"]

    def test_step_states(self):
        assert self.inv.step_states() == {None, "new"}

    def test_number_of_steps(self):
        assert self.inv.number_of_steps() == 2

    def test_sorted_steps_by(self):
        assert len(self.inv.sorted_steps_by()) == 2
        steps = self.inv.sorted_steps_by(step_ids={"2f94e8ae9edff68a"})
        assert len(steps) == 1
        assert steps[0].id == "2f94e8ae9edff68a"
        assert self.inv.sorted_steps_by(step_ids={"unmatched_id"}) == []
        steps = self.inv.sorted_steps_by(states={"new"})
        assert len(steps) == 1
        assert steps[0].state == "new"
        assert self.inv.sorted_steps_by(states={"unmatched_state"}) == []
        steps = self.inv.sorted_steps_by(indices={0}, states={None, "new"})
        assert len(steps) == 1
        assert steps[0].order_index == 0
        assert self.inv.sorted_steps_by(indices={2}) == []

    def test_cancel(self):
        inv = self._obj_invoke_workflow()
        inv.cancel()
        assert inv.state == "cancelled"

    def test_wait(self):
        inv = self._obj_invoke_workflow()
        inv.wait()
        assert inv.state == "scheduled"

    def test_refresh(self):
        inv = self._obj_invoke_workflow()
        inv.state = "placeholder"
        # use wait_for_invocation() directly, because inv.wait() will update inv automatically
        self.gi.gi.invocations.wait_for_invocation(inv.id)
        inv.refresh()
        assert inv.state == "scheduled"

    def test_run_step_actions(self):
        inv = self.workflow_pause.invoke(
            inputs={"0": self.dataset},
            history=self.history,
        )
        for _ in range(20):
            with pytest.raises(bioblend.TimeoutException):
                inv.wait(maxwait=0.5, interval=0.5)
            inv.refresh()
            if len(inv.steps) >= 3:
                break
        assert inv.steps[2].action is None
        inv.run_step_actions([inv.steps[2]], [True])
        assert inv.steps[2].action is True

    def test_summary(self):
        inv = self._obj_invoke_workflow()
        inv.wait()
        summary = inv.summary()
        assert summary["populated_state"] == "ok"

    def test_step_jobs_summary(self):
        inv = self._obj_invoke_workflow()
        inv.wait()
        step_jobs_summary = inv.step_jobs_summary()
        assert len(step_jobs_summary) == 1
        assert step_jobs_summary[0]["populated_state"] == "ok"

    def test_report(self):
        inv = self._obj_invoke_workflow()
        report = inv.report()
        assert report["workflows"] == {self.workflow.id: {"name": "paste_columns"}}

    @test_util.skip_unless_galaxy("release_20.09")
    def test_biocompute_object(self):
        inv = self._obj_invoke_workflow()
        inv.wait()
        biocompute_object = inv.biocompute_object()
        assert len(biocompute_object["description_domain"]["pipeline_steps"]) == 1

    def _obj_invoke_workflow(self) -> wrappers.Invocation:
        return self.workflow.invoke(
            inputs={"Input 1": self.dataset, "Input 2": self.dataset},
            history=self.history,
            inputs_by="name",
        )


@test_util.skip_unless_galaxy("release_19.09")
class TestObjInvocationClient(GalaxyObjectsTestBase):
    history: wrappers.History
    inv: wrappers.Invocation
    workflow: wrappers.Workflow

    @classmethod
    def setUpClass(cls) -> None:
        super().setUpClass()
        with open(SAMPLE_FN) as f:
            cls.workflow = cls.gi.workflows.import_new(f.read())
        cls.history = cls.gi.histories.create(name="TestGalaxyObjInvocationClient")
        dataset = cls.history.paste_content("1\t2\t3")
        cls.inv = cls.workflow.invoke(
            inputs={"Input 1": dataset, "Input 2": dataset},
            history=cls.history,
            inputs_by="name",
        )
        cls.inv.wait()

    @classmethod
    def tearDownClass(cls):
        cls.history.delete(purge=True)

    def test_get(self):
        inv = self.gi.invocations.get(self.inv.id)
        assert inv.id == self.inv.id
        assert inv.workflow_id == self.workflow.id
        assert inv.history_id == self.history.id
        assert inv.state == "scheduled"
        assert inv.update_time == self.inv.update_time
        assert inv.uuid == self.inv.uuid

    def test_get_previews(self):
        previews = self.gi.invocations.get_previews()
        assert {type(preview) for preview in previews} == {wrappers.InvocationPreview}
        inv_preview = next(p for p in previews if p.id == self.inv.id)
        assert inv_preview.id == self.inv.id
        assert inv_preview.workflow_id == self.workflow.id
        assert inv_preview.history_id == self.history.id
        assert inv_preview.state == "scheduled"
        assert inv_preview.update_time == self.inv.update_time
        assert inv_preview.uuid == self.inv.uuid

    def test_list(self):
        invs = self.gi.invocations.list()
        inv = next(i for i in invs if i.id == self.inv.id)
        assert inv.id == self.inv.id
        assert inv.workflow_id == self.workflow.id
        assert inv.history_id == self.history.id
        assert inv.state == "scheduled"
        assert inv.update_time == self.inv.update_time
        assert inv.uuid == self.inv.uuid
        assert len(self.inv.steps) > 0
        history = self.gi.histories.create(name="TestGalaxyObjInvocationClientList")
        assert self.gi.invocations.list(history=history) == []
        history.delete(purge=True)


class TestGalaxyInstance(GalaxyObjectsTestBase):
    def test_library(self):
        name = f"test_{uuid.uuid4().hex}"
        description, synopsis = "D", "S"
        lib = self.gi.libraries.create(name, description=description, synopsis=synopsis)
        assert lib.name == name
        assert lib.description == description
        assert lib.synopsis == synopsis
        assert len(lib.content_infos) == 1  # root folder
        assert len(lib.folder_ids) == 1
        assert len(lib.dataset_ids) == 0
        assert lib.id in [_.id for _ in self.gi.libraries.list()]
        lib.delete()
        assert not lib.is_mapped

    def test_workflow_from_str(self):
        with open(SAMPLE_FN) as f:
            wf = self.gi.workflows.import_new(f.read())
        self._check_and_del_workflow(wf)

    def test_workflow_collections_from_str(self):
        with open(SAMPLE_WF_COLL_FN) as f:
            wf = self.gi.workflows.import_new(f.read())
        self._check_and_del_workflow(wf)

    def test_workflow_parameter_input(self):
        with open(SAMPLE_WF_PARAMETER_INPUT_FN) as f:
            self.gi.workflows.import_new(f.read())

    def test_workflow_from_dict(self):
        with open(SAMPLE_FN) as f:
            wf = self.gi.workflows.import_new(json.load(f))
        self._check_and_del_workflow(wf)

    def test_workflow_publish_from_dict(self):
        with open(SAMPLE_FN) as f:
            wf = self.gi.workflows.import_new(json.load(f), publish=True)
        self._check_and_del_workflow(wf, check_is_public=True)

    def test_workflow_missing_tools(self):
        with open(SAMPLE_FN) as f:
            wf_dump = json.load(f)
        wf_info = self.gi.gi.workflows.import_workflow_dict(wf_dump)
        wf_dict = self.gi.gi.workflows.show_workflow(wf_info["id"])
        for id_, step in wf_dict["steps"].items():
            if step["type"] == "tool":
                for k in "tool_inputs", "tool_version":
                    wf_dict["steps"][id_][k] = None
        wf = wrappers.Workflow(wf_dict, gi=self.gi)
        assert not wf.is_runnable
        with pytest.raises(RuntimeError):
            wf.invoke()
        wf.delete()

    def test_workflow_export(self):
        with open(SAMPLE_FN) as f:
            wf1 = self.gi.workflows.import_new(f.read())
        wf2 = self.gi.workflows.import_new(wf1.export())
        assert wf1.id != wf2.id
        for wf in wf1, wf2:
            self._check_and_del_workflow(wf)

    def _check_and_del_workflow(self, wf: wrappers.Workflow, check_is_public: bool = False) -> None:
        # Galaxy appends additional text to imported workflow names
        assert wf.name.startswith("paste_columns")
        assert len(wf.steps) == 3
        for step_id, step in wf.steps.items():
            assert isinstance(step, wrappers.Step)
            assert step_id == step.id
            assert isinstance(step.tool_inputs, dict)
            if step.type == "tool":
                assert step.tool_id is not None
                assert step.tool_version is not None
                assert isinstance(step.input_steps, dict)
            elif step.type in ("data_collection_input", "data_input"):
                assert step.tool_id is None
                assert step.tool_version is None
                assert step.input_steps == {}
        wf_ids = {_.id for _ in self.gi.workflows.list()}
        assert wf.id in wf_ids
        if check_is_public:
            assert wf.published
        wf.delete()

    # not very accurate:
    #   * we can't publish a wf from the API
    #   * we can't directly get another user's wf
    def test_workflow_from_shared(self):
        all_prevs = {_.id: _ for _ in self.gi.workflows.get_previews(published=True)}
        pub_only_ids = set(all_prevs).difference(_.id for _ in self.gi.workflows.get_previews())
        if pub_only_ids:
            wf_id = pub_only_ids.pop()
            imported = self.gi.workflows.import_shared(wf_id)
            assert isinstance(imported, wrappers.Workflow)
            imported.delete()
        else:
            self.skipTest("no published workflows, manually publish a workflow to run this test")

    def test_get_libraries(self):
        self._test_multi_get("libraries")

    def test_get_histories(self):
        self._test_multi_get("histories")

    def test_get_workflows(self):
        self._test_multi_get("workflows")

    def _normalized_functions(
        self, obj_type: Literal["histories", "libraries", "workflows"]
    ) -> Tuple[Callable, Dict[str, Any]]:
        if obj_type == "libraries":
            create: Callable = self.gi.libraries.create
            del_kwargs = {}
        elif obj_type == "histories":
            create = self.gi.histories.create
            del_kwargs = {"purge": True}
        elif obj_type == "workflows":

            def create(name):
                with open(SAMPLE_FN) as f:
                    d = json.load(f)
                d["name"] = name
                return self.gi.workflows.import_new(d)

            del_kwargs = {}
        return create, del_kwargs

    def _test_multi_get(self, obj_type: Literal["histories", "libraries", "workflows"]) -> None:
        obj_gi_client = getattr(self.gi, obj_type)
        create, del_kwargs = self._normalized_functions(obj_type)

        def ids(seq: Iterable[wrappers.Wrapper]) -> Set[str]:
            return {_.id for _ in seq}

        names = [f"test_{uuid.uuid4().hex}" for _ in range(2)]
        objs = []
        try:
            objs = [create(_) for _ in names]
            assert ids(objs) <= ids(obj_gi_client.list())
            if obj_type != "workflows":
                filtered = obj_gi_client.list(name=names[0])
                assert len(filtered) == 1
                assert filtered[0].id == objs[0].id
                del_id = objs[-1].id
                objs.pop().delete(**del_kwargs)
                assert del_id in ids(obj_gi_client.get_previews(deleted=True))
            else:
                # Galaxy appends info strings to imported workflow names
                prev = obj_gi_client.get_previews()[0]
                filtered = obj_gi_client.list(name=prev.name)
                assert len(filtered) == 1
                assert filtered[0].id == prev.id
        finally:
            for o in objs:
                o.delete(**del_kwargs)

    def test_delete_libraries_by_name(self):
        self._test_delete_by_name("libraries")
        self._test_delete_by_ambiguous_name("libraries")

    def test_delete_histories_by_name(self):
        self._test_delete_by_name("histories")
        self._test_delete_by_ambiguous_name("histories")

    def test_delete_workflows_by_name(self):
        self._test_delete_by_name("workflows")
        self._test_delete_by_ambiguous_name("workflows")

    def _test_delete_by_name(self, obj_type: Literal["histories", "libraries", "workflows"]) -> None:
        obj_gi_client = getattr(self.gi, obj_type)
        create, del_kwargs = self._normalized_functions(obj_type)
        name = f"test_{uuid.uuid4().hex}"
        create(name)
        prevs = [_ for _ in obj_gi_client.get_previews(name=name) if not _.deleted]
        assert len(prevs) == 1
        del_kwargs["name"] = name
        obj_gi_client.delete(**del_kwargs)
        prevs = [_ for _ in obj_gi_client.get_previews(name=name) if not _.deleted]
        assert len(prevs) == 0

    def _test_delete_by_ambiguous_name(self, obj_type: Literal["histories", "libraries", "workflows"]) -> None:
        obj_gi_client = getattr(self.gi, obj_type)
        create, del_kwargs = self._normalized_functions(obj_type)
        name = f"test_{uuid.uuid4().hex}"
        objs = [create(name) for _ in range(2)]
        prevs = [_ for _ in obj_gi_client.get_previews(name=name) if not _.deleted]
        assert len(prevs) == len(objs)
        del_kwargs["name"] = name
        with pytest.raises(ValueError):
            obj_gi_client.delete(**del_kwargs)
        # Cleanup
        del del_kwargs["name"]
        for prev in prevs:
            del_kwargs["id_"] = prev.id
            obj_gi_client.delete(**del_kwargs)


class TestLibrary(GalaxyObjectsTestBase):
    # just something that can be expected to be always up
    DS_URL = "https://tools.ietf.org/rfc/rfc1866.txt"

    def setUp(self):
        super().setUp()
        self.lib = self.gi.libraries.create(f"test_{uuid.uuid4().hex}")

    def tearDown(self):
        self.lib.delete()

    def test_root_folder(self):
        r = self.lib.root_folder
        assert r.parent is None

    def test_folder(self):
        name = f"test_{uuid.uuid4().hex}"
        desc = "D"
        folder = self.lib.create_folder(name, description=desc)
        assert folder.name == name
        assert folder.description == desc
        assert folder.container is self.lib
        assert folder.parent is not None
        assert folder.parent.id == self.lib.root_folder.id
        assert len(self.lib.content_infos) == 2
        assert len(self.lib.folder_ids) == 2
        assert folder.id in self.lib.folder_ids
        retrieved = self.lib.get_folder(folder.id)
        assert folder.id == retrieved.id

    def _check_datasets(self, dss: Collection[wrappers.LibraryDataset]) -> None:
        assert len(dss) == len(self.lib.dataset_ids)
        assert {_.id for _ in dss} == set(self.lib.dataset_ids)
        for ds in dss:
            assert ds.container is self.lib

    def test_dataset(self):
        folder = self.lib.create_folder(f"test_{uuid.uuid4().hex}")
        ds = self.lib.upload_data(FOO_DATA, folder=folder)
        assert len(self.lib.content_infos) == 3
        assert len(self.lib.folder_ids) == 2
        self._check_datasets([ds])

    def test_dataset_from_url(self):
        if is_reachable(self.DS_URL):
            ds = self.lib.upload_from_url(self.DS_URL)
            self._check_datasets([ds])
        else:
            self.skipTest(f"{self.DS_URL} not reachable")

    def test_dataset_from_local(self):
        with tempfile.NamedTemporaryFile(mode="w", prefix="bioblend_test_") as f:
            f.write(FOO_DATA)
            f.flush()
            ds = self.lib.upload_from_local(f.name)
        self._check_datasets([ds])

    def test_datasets_from_fs(self):
        bnames = [f"f{i}.txt" for i in range(2)]
        dss, fnames = upload_from_fs(self.lib, bnames)
        self._check_datasets(dss)
        dss, fnames = upload_from_fs(self.lib, bnames, link_data_only="link_to_files")
        for ds, fn in zip(dss, fnames):
            assert ds.file_name == fn

    def test_copy_from_dataset(self):
        hist = self.gi.histories.create(f"test_{uuid.uuid4().hex}")
        try:
            hda = hist.paste_content(FOO_DATA)
            ds = self.lib.copy_from_dataset(hda)
        finally:
            hist.delete(purge=True)
        self._check_datasets([ds])

    def test_get_dataset(self):
        ds = self.lib.upload_data(FOO_DATA)
        retrieved = self.lib.get_dataset(ds.id)
        assert ds.id == retrieved.id

    def test_get_datasets(self):
        bnames = [f"f{i}.txt" for i in range(2)]
        dss, _ = upload_from_fs(self.lib, bnames)
        retrieved = self.lib.get_datasets()
        assert len(dss) == len(retrieved)
        assert {_.id for _ in dss} == {_.id for _ in retrieved}
        name = f"/{bnames[0]}"
        selected = self.lib.get_datasets(name=name)
        assert len(selected) == 1
        assert selected[0].name == bnames[0]


class TestLDContents(GalaxyObjectsTestBase):
    def setUp(self):
        super().setUp()
        self.lib = self.gi.libraries.create(f"test_{uuid.uuid4().hex}")
        self.ds = self.lib.upload_data(FOO_DATA)
        self.ds.wait()

    def tearDown(self):
        self.lib.delete()

    def test_dataset_get_stream(self):
        for idx, c in enumerate(self.ds.get_stream(chunk_size=1)):
            assert FOO_DATA[idx].encode() == c

    def test_dataset_peek(self):
        fetched_data = self.ds.peek(chunk_size=4)
        assert FOO_DATA[0:4].encode() == fetched_data

    def test_dataset_download(self):
        with tempfile.TemporaryFile() as f:
            self.ds.download(f)
            f.seek(0)
            assert FOO_DATA.encode() == f.read()

    def test_dataset_get_contents(self):
        assert FOO_DATA.encode() == self.ds.get_contents()

    def test_dataset_delete(self):
        self.ds.delete()
        # Cannot test this yet because the 'deleted' attribute is not exported
        # by the API at the moment
        # assert self.ds.deleted

    def test_dataset_update(self):
        new_name = f"test_{uuid.uuid4().hex}"
        new_misc_info = f"Annotation for {new_name}"
        new_genome_build = "hg19"
        updated_ldda = self.ds.update(name=new_name, misc_info=new_misc_info, genome_build=new_genome_build)
        assert self.ds.id == updated_ldda.id
        assert self.ds.name == new_name
        assert self.ds.misc_info == new_misc_info
        assert self.ds.genome_build == new_genome_build


class TestHistory(GalaxyObjectsTestBase):
    def setUp(self):
        super().setUp()
        self.hist = self.gi.histories.create(f"test_{uuid.uuid4().hex}")

    def tearDown(self):
        self.hist.delete(purge=True)

    def test_create_delete(self):
        name = f"test_{uuid.uuid4().hex}"
        hist = self.gi.histories.create(name)
        assert hist.name == name
        hist_id = hist.id
        assert hist_id in [_.id for _ in self.gi.histories.list()]
        hist.delete(purge=True)
        assert not hist.is_mapped
        h = self.gi.histories.get(hist_id)
        assert h.deleted

    def _check_dataset(self, hda: wrappers.HistoryDatasetAssociation) -> None:
        assert isinstance(hda, wrappers.HistoryDatasetAssociation)
        assert hda.container is self.hist
        assert len(self.hist.dataset_ids) == 1
        assert self.hist.dataset_ids[0] == hda.id

    def test_import_dataset(self):
        lib = self.gi.libraries.create(f"test_{uuid.uuid4().hex}")
        lds = lib.upload_data(FOO_DATA)
        assert len(self.hist.dataset_ids) == 0
        hda = self.hist.import_dataset(lds)
        lib.delete()
        self._check_dataset(hda)

    def test_upload_file(self):
        with tempfile.NamedTemporaryFile(mode="w", prefix="bioblend_test_") as f:
            f.write(FOO_DATA)
            f.flush()
            hda = self.hist.upload_file(f.name)
        self._check_dataset(hda)

    def test_paste_content(self):
        hda = self.hist.paste_content(FOO_DATA)
        self._check_dataset(hda)

    def test_get_dataset(self):
        hda = self.hist.paste_content(FOO_DATA)
        retrieved = self.hist.get_dataset(hda.id)
        assert hda.id == retrieved.id

    def test_get_datasets(self):
        bnames = [f"f{i}.txt" for i in range(2)]
        lib = self.gi.libraries.create(f"test_{uuid.uuid4().hex}")
        lds = upload_from_fs(lib, bnames)[0]
        hdas = [self.hist.import_dataset(_) for _ in lds]
        lib.delete()
        retrieved = self.hist.get_datasets()
        assert len(hdas) == len(retrieved)
        assert {_.id for _ in hdas} == {_.id for _ in retrieved}
        selected = self.hist.get_datasets(name=bnames[0])
        assert len(selected) == 1
        assert selected[0].name == bnames[0]

    def test_export_and_download(self):
        jeha_id = self.hist.export(wait=True, maxwait=60)
        assert jeha_id
        tempdir = tempfile.mkdtemp(prefix="bioblend_test_")
        temp_fn = os.path.join(tempdir, "export.tar.gz")
        try:
            with open(temp_fn, "wb") as fo:
                self.hist.download(jeha_id, fo)
            assert tarfile.is_tarfile(temp_fn)
        finally:
            shutil.rmtree(tempdir)

    def test_update(self):
        new_name = f"test_{uuid.uuid4().hex}"
        new_annotation = f"Annotation for {new_name}"
        new_tags = ["tag1", "tag2"]
        updated_hist = self.hist.update(name=new_name, annotation=new_annotation, tags=new_tags)
        assert self.hist.id == updated_hist.id
        assert self.hist.name == new_name
        assert self.hist.annotation == new_annotation
        assert self.hist.tags == new_tags
        updated_hist = self.hist.update(published=True)
        assert self.hist.id == updated_hist.id
        assert self.hist.published

    def test_create_dataset_collection(self):
        self._create_collection_description()
        hdca = self.hist.create_dataset_collection(self.collection_description)
        assert isinstance(hdca, wrappers.HistoryDatasetCollectionAssociation)
        assert hdca.collection_type == "list"
        assert hdca.container is self.hist
        assert len(hdca.elements) == 2
        assert self.dataset1.id == hdca.elements[0]["object"]["id"]
        assert self.dataset2.id == hdca.elements[1]["object"]["id"]

    def test_delete_dataset_collection(self):
        self._create_collection_description()
        hdca = self.hist.create_dataset_collection(self.collection_description)
        hdca.delete()
        assert hdca.deleted

    def _create_collection_description(self) -> None:
        self.dataset1 = self.hist.paste_content(FOO_DATA)
        self.dataset2 = self.hist.paste_content(FOO_DATA_2)
        self.collection_description = dataset_collections.CollectionDescription(
            name="MyDatasetList",
            elements=[
                dataset_collections.HistoryDatasetElement(name="sample1", id=self.dataset1.id),
                dataset_collections.HistoryDatasetElement(name="sample2", id=self.dataset2.id),
            ],
        )


class TestHDAContents(GalaxyObjectsTestBase):
    def setUp(self):
        super().setUp()
        self.hist = self.gi.histories.create(f"test_{uuid.uuid4().hex}")
        self.ds = self.hist.paste_content(FOO_DATA)

    def tearDown(self):
        self.hist.delete(purge=True)

    def test_dataset_get_stream(self):
        for idx, c in enumerate(self.ds.get_stream(chunk_size=1)):
            assert FOO_DATA[idx].encode() == c

    def test_dataset_peek(self):
        fetched_data = self.ds.peek(chunk_size=4)
        assert FOO_DATA[0:4].encode() == fetched_data

    def test_dataset_download(self):
        with tempfile.TemporaryFile() as f:
            self.ds.download(f)
            f.seek(0)
            assert FOO_DATA.encode() == f.read()

    def test_dataset_get_contents(self):
        assert FOO_DATA.encode() == self.ds.get_contents()

    def test_dataset_update(self):
        new_name = f"test_{uuid.uuid4().hex}"
        new_annotation = f"Annotation for {new_name}"
        new_genome_build = "hg19"
        updated_hda = self.ds.update(name=new_name, annotation=new_annotation, genome_build=new_genome_build)
        assert self.ds.id == updated_hda.id
        assert self.ds.name == new_name
        assert self.ds.annotation == new_annotation
        assert self.ds.genome_build == new_genome_build

    def test_dataset_delete(self):
        self.ds.delete()
        assert self.ds.deleted
        assert not self.ds.purged

    def test_dataset_purge(self):
        self.ds.delete(purge=True)
        assert self.ds.deleted
        assert self.ds.purged


@test_util.skip_unless_galaxy("release_19.09")
class TestRunWorkflow(GalaxyObjectsTestBase):
    def setUp(self):
        super().setUp()
        self.lib = self.gi.libraries.create(f"test_{uuid.uuid4().hex}")
        with open(SAMPLE_FN) as f:
            self.wf = self.gi.workflows.import_new(f.read())
        self.contents = ["one\ntwo\n", "1\n2\n"]
        self.inputs = [self.lib.upload_data(_) for _ in self.contents]

    def tearDown(self):
        self.wf.delete()
        self.lib.delete()

    def _test(self, existing_hist: bool = False, pass_params: bool = False) -> None:
        hist_name = f"test_{uuid.uuid4().hex}"
        if existing_hist:
            hist: Union[str, wrappers.History] = self.gi.histories.create(hist_name)
        else:
            hist = hist_name
        if pass_params:
            params = {"Paste1": {"delimiter": "U"}}
            sep = "_"  # 'U' maps to '_' in the paste tool
        else:
            params = None
            sep = "\t"  # default
        input_map = {"Input 1": self.inputs[0], "Input 2": self.inputs[1]}
        sys.stderr.write(os.linesep)
        inv = self.wf.invoke(inputs=input_map, params=params, history=hist, inputs_by="name")
        out_hist = self.gi.histories.get(inv.history_id)
        inv.wait()
        last_step = inv.sorted_steps_by()[-1]
        out_ds = last_step.get_outputs()["out_file1"]
        assert out_ds.container.id == out_hist.id
        res = out_ds.get_contents()
        exp_rows = zip(*(_.splitlines() for _ in self.contents))
        exp_res = ("\n".join(sep.join(t) for t in exp_rows) + "\n").encode()
        assert res == exp_res
        if isinstance(hist, wrappers.History):  # i.e. existing_hist == True
            assert out_hist.id == hist.id
        out_hist.delete(purge=True)

    def test_existing_history(self) -> None:
        self._test(existing_hist=True)

    def test_new_history(self) -> None:
        self._test(existing_hist=False)

    def test_params(self) -> None:
        self._test(pass_params=True)


@test_util.skip_unless_galaxy("release_19.09")
class TestRunDatasetCollectionWorkflow(GalaxyObjectsTestBase):
    def setUp(self):
        super().setUp()
        with open(SAMPLE_WF_COLL_FN) as f:
            self.wf = self.gi.workflows.import_new(f.read())
        self.hist = self.gi.histories.create(f"test_{uuid.uuid4().hex}")

    def tearDown(self):
        self.wf.delete()
        self.hist.delete(purge=True)

    def test_run_workflow_with_dataset_collection(self):
        dataset1 = self.hist.paste_content(FOO_DATA)
        dataset2 = self.hist.paste_content(FOO_DATA_2)
        collection_description = dataset_collections.CollectionDescription(
            name="MyDatasetList",
            elements=[
                dataset_collections.HistoryDatasetElement(name="sample1", id=dataset1.id),
                dataset_collections.HistoryDatasetElement(name="sample2", id=dataset2.id),
            ],
        )
        dataset_collection = self.hist.create_dataset_collection(collection_description)
        assert len(self.hist.content_infos) == 3
        input_map = {"0": dataset_collection, "1": dataset1}
        inv = self.wf.invoke(input_map, history=self.hist)
        inv.wait()
        self.hist.refresh()
        assert len(self.hist.content_infos) == 6
        last_step = inv.sorted_steps_by()[-1]
        out_hdca = last_step.get_output_collections()["out_file1"]
        assert out_hdca.collection_type == "list"
        assert len(out_hdca.elements) == 2
        assert out_hdca.container.id == self.hist.id


class TestJob(GalaxyObjectsTestBase):
    def test_get(self):
        job_prevs = self.gi.jobs.get_previews()
        if len(job_prevs) > 0:
            job_prev = job_prevs[0]
            assert isinstance(job_prev, wrappers.JobPreview)
            job = self.gi.jobs.get(job_prev.id)
            assert isinstance(job, wrappers.Job)
            assert job.id == job_prev.id
        for job in self.gi.jobs.list():
            assert isinstance(job, wrappers.Job)