comparison toolfactory/rgToolFactory2.py @ 98:67628c7dc9f3 draft

Uploaded

97:515be7318f2b

import time


from bioblend import toolshed

import galaxyxml.tool as gxt
import galaxyxml.tool.parameters as gxtp

import lxml

            self.args.tool_version,
            self.args.tool_desc,
            FAKEEXE,
        )
        self.newtarpath = "toolfactory_%s.tgz" % self.tool_name
        self.tooloutdir = "tfout"
        self.repdir = "TF_run_report_tempdir"
        self.testdir = os.path.join(self.tooloutdir, "test-data")
        if not os.path.exists(self.tooloutdir):
            os.mkdir(self.tooloutdir)
        if not os.path.exists(self.testdir):
            os.mkdir(self.testdir)  # make tests directory

                    sto = open(self.tlog, "w")
                sto.write(
                    "## Executing Toolfactory generated command line = %s\n" % scl
                )
            sto.flush()
            p = subprocess.run(self.cl, shell=False, stdout=sto, stderr=ste)
            sto.close()
            ste.close()
            retval = p.returncode
        else:  # work around special case - stdin and write to stdout
            if len(self.infiles) > 0:
                sti = open(self.infiles[0][IPATHPOS], "rb")
            else:
                sti = sys.stdin
            if len(self.outfiles) > 0:
                sto = open(self.outfiles[0][ONAMEPOS], "wb")
            else:
                sto = sys.stdout
            p = subprocess.run(self.cl, shell=False, stdout=sto, stdin=sti)
            sto.write("## Executing Toolfactory generated command line = %s\n" % scl)
            retval = p.returncode
            sto.close()
            sti.close()
        if os.path.isfile(self.tlog) and os.stat(self.tlog).st_size == 0:
            os.unlink(self.tlog)
        if os.path.isfile(self.elog) and os.stat(self.elog).st_size == 0:

        )
        repos = ts.repositories.get_repositories()
        rnames = [x.get("name", "?") for x in repos]
        rids = [x.get("id", "?") for x in repos]
        sto.write(f"############names={rnames} rids={rids}\n")
        cat = "ToolFactory generated tools"
        if self.args.tool_name not in rnames:
            tscat = ts.categories.get_categories()
            cnames = [x.get("name", "?") for x in tscat]
            cids = [x.get("id", "?") for x in tscat]
            catID = None
            if cat in cnames:
                ci = cnames.index(cat)
                catID = cids[ci]
            res = ts.repositories.create_repository(
                name=self.args.tool_name,
                synopsis="Synopsis:%s" % self.args.tool_desc,
                description=self.args.tool_desc,

            self.args.toolshed_url,
            "--section_label",
            "ToolFactory",
        ]
        tout.write("running\n%s\n" % " ".join(cll))
        p = subprocess.run(cll, shell=False, stderr=tout, stdout=tout)
        tout.write(
            "installed %s - got retcode %d\n" % (self.args.tool_name, p.returncode)
        )
        tout.close()
        return p.returncode

    def planemo_shedload(self):
        """
        planemo shed_create --shed_target testtoolshed
        planemo shed_init --name=<name>

                self.args.tool_name,
                "--shed_key",
                self.args.toolshed_api_key,
            ]
            try:
                p = subprocess.run(
                    cll, shell=False, cwd=self.tooloutdir, stdout=tout, stderr=tout
                )
            except:
                pass
            if p.returncode != 0:
                tout.write("Repository %s exists\n" % self.args.tool_name)
            else:
                tout.write("initiated %s\n" % self.args.tool_name)
        cll = [
            "planemo",

            "--shed_key",
            self.args.toolshed_api_key,
            "--tar",
            self.newtarpath,
        ]
        p = subprocess.run(cll, shell=False, stdout=tout, stderr=tout)
        tout.write("Ran %s got %d\n" % (" ".join(cll), p.returncode))
        tout.close()
        return p.returncode

    def eph_test(self, genoutputs=True):
        """problem getting jobid - ephemeris upload is the job before the one we want - but depends on how many inputs
        """
        if os.path.exists(self.tlog):

            "--owner",
            "fubar",
        ]
        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            p = subprocess.run(
               cll, shell=False, stderr=dummy, stdout=dummy
            )

            with open('tool_test_output.json','rb') as f:
                s = json.loads(f.read())

                        continue
                    dest = os.path.join(self.tooloutdir, entry.name)
                    src = os.path.join(ephouts, entry.name)
                    shutil.copyfile(src, dest)
        else:
            p = subprocess.run(
               cll, shell=False,  stderr=tout, stdout=tout)
            tout.write("eph_test Ran %s got %d" % (" ".join(cll), p.returncode))
        tout.close()
        return p.returncode

    def planemo_test_biocontainer(self, genoutputs=True):
        """planemo is a requirement so is available for testing but testing in a biocontainer
        requires some fiddling to use the hacked galaxy-central .venv

            dummy, tfile = tempfile.mkstemp()
            cll = [
                ".", os.path.join(self.args.galaxy_root,'.venv','bin','activate'),"&&",
                "planemo",
                "test",
                "--test_data", os.path.abspath(self.testdir),
                "--test_output", os.path.abspath(tool_test_path),
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                os.path.abspath(xreal),
            ]
            p = subprocess.run(
                cll,
                shell=False,
                cwd=self.tooloutdir,
                stderr=dummy,
                stdout=dummy,

        else:
            cll = [
                ".", os.path.join(self.args.galaxy_root,'.venv','bin','activate'),"&&",
                "planemo",
                "test",
                "--test_data", os.path.abspath(self.testdir),
                "--test_output", os.path.abspath(tool_test_path),
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                os.path.abspath(xreal),
            ]
            p = subprocess.run(
                cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
            )
        tout.close()
        return p.returncode

    def planemo_test(self, genoutputs=True):
        """planemo is a requirement so is available for testing but needs a different call if
        in the biocontainer - see above
        and for generating test outputs if command or test overrides are supplied

        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            cll = [
                "planemo",
                "test",
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                os.path.abspath(xreal),
            ]
            p = subprocess.run(
                cll,
                shell=False,
                cwd=self.testdir,
                stderr=dummy,
                stdout=dummy,

        else:
            cll = [
                "planemo",
                "test",
                "--test_data", os.path.abspath(self.testdir),
                "--test_output", os.path.abspath(tool_test_path),
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                os.path.abspath(xreal),
            ]
            p = subprocess.run(
                cll, shell=False, cwd=self.testdir, stderr=tout, stdout=tout
            )
        tout.close()
        return p.returncode


    def gal_test(self, genoutputs=True):
        """
export GALAXY_TEST_SAVE="./foo" && export GALAXY_TEST_NO_CLEANUP="1" \
&& export GALAXY_TEST_TMP_DIR=./foo && sh run_tests.sh --id rgtf2 --report_file tool_tests_tool_conf.html functional.test_toolbox

        """
        tool_test_rep_path = os.path.join(self.repdir,f"{self.tool_name}_galaxy_test_report.html")
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            cll = [
            "mkdir -p ./test","&&","export GALAXY_TEST_SAVE='./test'", "&&", "export GALAXY_TEST_NO_CLEANUP='1'", \
           "&&", "export GALAXY_TEST_TMP_DIR='./test'", "&&", f"sh {self.args.galaxy_root}/run_tests.sh --id {self.args.tool_id} --report_file {tool_test_rep_path} functional.test_toolbox",
            ]
            p = subprocess.run(
                cll,
                shell=False,
                cwd=self.testdir,
                stderr=dummy,
                stdout=dummy,
            )
            # if all went well, tgz is in ./test and down a nest of tmp directories lurk the output files
            outfiles = []
            for p in self.outfiles:
                oname = p[ONAMEPOS]
                outfiles.append(oname)
            paths = []
            for root, dirs, files in os.walk('./test'):
                for f in files:
                    if f in outfiles:
                        paths.append([root,f])
            for p in paths:
                src = os.path.join(p[0],p[1])
                dest = os.path.join(self.testdir,f"{p[1]}_sample")
                shutil.copyfile(src,dest)

        else:
            cll = [
            "mkdir -p ./test","&&","rm -rf ./test/*","&&","export GALAXY_TEST_SAVE='./test'", "&&", "export GALAXY_TEST_NO_CLEANUP=", \
           "&&", "export GALAXY_TEST_TMP_DIR='./test'", "&&", f"sh {self.args.galaxy_root}/run_tests.sh --id {self.args.tool_id} --report_file {tool_test_rep_path} functional.test_toolbox",
            ]
            p = subprocess.run(
                cll, shell=False, cwd=self.testdir, stderr=tout, stdout=tout
            )
        tout.close()
        return p.returncode


    def writeShedyml(self):
        """for planemo"""
        yuser = self.args.user_email.split("@")[0]

                )
        tf = tarfile.open(self.newtarpath, "w:gz")
        tf.add(name=self.tooloutdir, arcname=self.tool_name, filter=exclude_function)
        tf.close()
        shutil.copyfile(self.newtarpath, self.args.new_tool)

    def moveRunOutputs(self):
        """need to move planemo or run outputs into toolfactory collection"""
        with os.scandir(self.tooloutdir) as outs:
            for entry in outs:

    if args.make_Tool == "generate":
        retcode = r.run()
        r.moveRunOutputs()
        r.makeToolTar()
    else:
        r.makeToolTar()
        r.shedLoad()
        r.eph_galaxy_load()
        retcode = r.gal_test(genoutputs=True)  # this fails
        r.makeToolTar()
        retcode = r.gal_test(genoutputs=False)

98:67628c7dc9f3

import time


from bioblend import toolshed

# import docker

import galaxyxml.tool as gxt
import galaxyxml.tool.parameters as gxtp

import lxml

            self.args.tool_version,
            self.args.tool_desc,
            FAKEEXE,
        )
        self.newtarpath = "toolfactory_%s.tgz" % self.tool_name
        self.tooloutdir = "./tfout"
        self.repdir = "./TF_run_report_tempdir"
        self.testdir = os.path.join(self.tooloutdir, "test-data")
        if not os.path.exists(self.tooloutdir):
            os.mkdir(self.tooloutdir)
        if not os.path.exists(self.testdir):
            os.mkdir(self.testdir)  # make tests directory

                    sto = open(self.tlog, "w")
                sto.write(
                    "## Executing Toolfactory generated command line = %s\n" % scl
                )
            sto.flush()
            subp = subprocess.run(self.cl, shell=False, stdout=sto, stderr=ste)
            sto.close()
            ste.close()
            retval = subp.returncode
        else:  # work around special case - stdin and write to stdout
            if len(self.infiles) > 0:
                sti = open(self.infiles[0][IPATHPOS], "rb")
            else:
                sti = sys.stdin
            if len(self.outfiles) > 0:
                sto = open(self.outfiles[0][ONAMEPOS], "wb")
            else:
                sto = sys.stdout
            subp = subprocess.run(self.cl, shell=False, stdout=sto, stdin=sti)
            sto.write("## Executing Toolfactory generated command line = %s\n" % scl)
            retval = subp.returncode
            sto.close()
            sti.close()
        if os.path.isfile(self.tlog) and os.stat(self.tlog).st_size == 0:
            os.unlink(self.tlog)
        if os.path.isfile(self.elog) and os.stat(self.elog).st_size == 0:

        )
        repos = ts.repositories.get_repositories()
        rnames = [x.get("name", "?") for x in repos]
        rids = [x.get("id", "?") for x in repos]
        sto.write(f"############names={rnames} rids={rids}\n")
        tfcat = "ToolFactory generated tools"
        if self.args.tool_name not in rnames:
            tscat = ts.categories.get_categories()
            cnames = [x.get("name", "?").strip() for x in tscat]
            cids = [x.get("id", "?") for x in tscat]
            catID = None
            if tfcat.strip() in cnames:
                ci = cnames.index(tfcat)
                catID = cids[ci]
            res = ts.repositories.create_repository(
                name=self.args.tool_name,
                synopsis="Synopsis:%s" % self.args.tool_desc,
                description=self.args.tool_desc,

            self.args.toolshed_url,
            "--section_label",
            "ToolFactory",
        ]
        tout.write("running\n%s\n" % " ".join(cll))
        subp = subprocess.run(cll, shell=False, stderr=tout, stdout=tout)
        tout.write(
            "installed %s - got retcode %d\n" % (self.args.tool_name, subp.returncode)
        )
        tout.close()
        return subp.returncode

    def planemo_shedload(self):
        """
        planemo shed_create --shed_target testtoolshed
        planemo shed_init --name=<name>

                self.args.tool_name,
                "--shed_key",
                self.args.toolshed_api_key,
            ]
            try:
                subp = subprocess.run(
                    cll, shell=False, cwd=self.tooloutdir, stdout=tout, stderr=tout
                )
            except:
                pass
            if subp.returncode != 0:
                tout.write("Repository %s exists\n" % self.args.tool_name)
            else:
                tout.write("initiated %s\n" % self.args.tool_name)
        cll = [
            "planemo",

            "--shed_key",
            self.args.toolshed_api_key,
            "--tar",
            self.newtarpath,
        ]
        subp = subprocess.run(cll, shell=False, stdout=tout, stderr=tout)
        tout.write("Ran %s got %d\n" % (" ".join(cll),subp.returncode))
        tout.close()
        return subp.returncode

    def eph_test(self, genoutputs=True):
        """problem getting jobid - ephemeris upload is the job before the one we want - but depends on how many inputs
        """
        if os.path.exists(self.tlog):

            "--owner",
            "fubar",
        ]
        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            subp = subprocess.run(
               cll, shell=False, stderr=dummy, stdout=dummy
            )

            with open('tool_test_output.json','rb') as f:
                s = json.loads(f.read())

                        continue
                    dest = os.path.join(self.tooloutdir, entry.name)
                    src = os.path.join(ephouts, entry.name)
                    shutil.copyfile(src, dest)
        else:
            subp = subprocess.run(
               cll, shell=False,  stderr=tout, stdout=tout)
            tout.write("eph_test Ran %s got %d" % (" ".join(cll), subp.returncode))
        tout.close()
        return subp.returncode

    def planemo_test_biocontainer(self, genoutputs=True):
        """planemo is a requirement so is available for testing but testing in a biocontainer
        requires some fiddling to use the hacked galaxy-central .venv

            dummy, tfile = tempfile.mkstemp()
            cll = [
                ".", os.path.join(self.args.galaxy_root,'.venv','bin','activate'),"&&",
                "planemo",
                "test",
                "--test_data", self.testdir,
                "--test_output", tool_test_path,
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                xreal,
            ]
            subp = subprocess.run(
                cll,
                shell=False,
                cwd=self.tooloutdir,
                stderr=dummy,
                stdout=dummy,

        else:
            cll = [
                ".", os.path.join(self.args.galaxy_root,'.venv','bin','activate'),"&&",
                "planemo",
                "test",
                "--test_data", os.path.self.testdir,
                "--test_output", os.path.tool_test_path,
                "--skip_venv",
                "--galaxy_root",
                self.args.galaxy_root,
                xreal,
            ]
            subp = subprocess.run(
                cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
            )
        tout.close()
        return subp.returncode

    def planemo_test(self, genoutputs=True):
        """planemo is a requirement so is available for testing but needs a different call if
        in the biocontainer - see above
        and for generating test outputs if command or test overrides are supplied

        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            cll = [
                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                xreal,
            ]
            subp = subprocess.run(
                cll,
                shell=False,
                cwd=self.testdir,
                stderr=dummy,
                stdout=dummy,

        else:
            cll = [
                "planemo",
                "test",
                "--test_data", self.testdir,
                "--test_output",tool_test_path,
                "--galaxy_root",
                self.args.galaxy_root,
                xreal,
            ]
            subp = subprocess.run(
                cll, shell=False, cwd=self.testdir, stderr=tout, stdout=tout
            )
        tout.close()
        return subp.returncode


    def gal_test(self, genoutputs=True):
        """
export GALAXY_TEST_SAVE="./foo" && export GALAXY_TEST_NO_CLEANUP="1" \
&& export GALAXY_TEST_TMP_DIR=./foo && sh run_tests.sh --id rgtf2 --report_file tool_tests_tool_conf.html functional.test_toolbox

        """
        tool_test_rep_path = os.path.join(self.repdir,f"{self.tool_name}_galaxy_test_report_html.html")
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        testdir = '/tmp/testdir'
        ourenv = os.environ
        ourenv["GALAXY_TEST_SAVE"] = os.abspath(testdir)
        ourenv["GALAXY_TEST_NO_CLEANUP"] = "1"
        ourenv["GALAXY_TEST_TMP_DIR"] = os.abspath(testdir)
        if genoutputs:

            if not os.path.exists(testdir):
                os.mkdir(testdir)
            dummy = open('failedtest.log','w')
            cll = [
            "sh", f"{self.args.galaxy_root}/run_tests.sh", "--id", self.args.tool_name,
           "--report_file", tool_test_rep_path, "functional.test_toolbox",
           ]
            subp = subprocess.run(
                cll,
                shell=False,
                env = ourenv,
                cwd=self.testdir,
                stderr=dummy,
                stdout=dummy,
            )
            dummy.close()
            # if all went well, tgz is in ./test and down a nest of tmp directories lurk the output files
            outfiles = []
            for p in self.outfiles:
                oname = p[ONAMEPOS]
                outfiles.append(oname)
            paths = []
            for root, dirs, files in os.walk(testdir):
                for f in files:
                    tout.write(f"{root} {f}\n")
                    if f in outfiles:
                        paths.append([root,f])
                        tout.write(f"## found output {f}")

            if len(paths) != len(self.outfiles):
                tout.write(f"## problem - outfiles not all found")
            for path in paths:
                fname = path[1]
                src = os.path.join(path[0],fname)
                dest = os.path.join(self.testdir,f"{fname}_sample")
                shutil.copyfile(src,dest)
        else:
            cll = [
           "sh", f"{self.args.galaxy_root}/run_tests.sh", "--id", self.args.tool_name,
           "--report_file", tool_test_rep_path, "functional.test_toolbox",
            ]
            subp = subprocess.run(
                cll, env = ourenv,
                shell=False, cwd=self.testdir, stderr=tout, stdout=tout
            )
        tout.close()
        return subp.returncode



    def writeShedyml(self):
        """for planemo"""
        yuser = self.args.user_email.split("@")[0]

                )
        tf = tarfile.open(self.newtarpath, "w:gz")
        tf.add(name=self.tooloutdir, arcname=self.tool_name, filter=exclude_function)
        tf.close()
        shutil.copyfile(self.newtarpath, self.args.new_tool)

    def fakeToolTar(self):
        """move fake outputs into test-data and prepare the tarball"""
        excludeme = "tool_test_output"

        def exclude_function(tarinfo):
            filename = tarinfo.name
            return (
                None
                if filename.startswith(excludeme)
                else tarinfo
            )

        for p in self.outfiles:
            oname = p[ONAMEPOS]
            src = os.path.join(self.testdir,oname)
            if os.path.isfile(src):
                dest = os.path.join(self.testdir, "%s_sample" % oname)
                shutil.copyfile(src, dest)
                dest = os.path.join(self.repdir, "%s.sample" % (oname))
                shutil.copyfile(src, dest)
            else:
                with open(src,'w') as fayk:
                    fayk.write('fake!\n')
                dest = os.path.join(self.testdir, "%s_sample" % oname)
                shutil.copyfile(src, dest)
                print(
                    "### problem - output file %s not found in tooloutdir %s so faked"
                    % (src, self.tooloutdir)
                )
        tf = tarfile.open(self.newtarpath, "w:gz")
        tf.add(name=self.tooloutdir, arcname=self.tool_name, filter=exclude_function)
        tf.close()
        shutil.copyfile(self.newtarpath, self.args.new_tool)


    def moveRunOutputs(self):
        """need to move planemo or run outputs into toolfactory collection"""
        with os.scandir(self.tooloutdir) as outs:
            for entry in outs:

    if args.make_Tool == "generate":
        retcode = r.run()
        r.moveRunOutputs()
        r.makeToolTar()
    else:
        r.fakeToolTar()
        r.shedLoad()
        r.eph_galaxy_load()
        retcode = r.gal_test(genoutputs=True)  # this fails
        r.makeToolTar()
        retcode = r.gal_test(genoutputs=False)