comparison toolfactory/rgToolFactory2.py @ 75:25dd9364a32a draft

Uploaded

74:3d4324df911b

# one to run the executable with the supplied test data and settings
# Be simpler to write the tool, then run it with planemo and soak up the test outputs.


import argparse
import copy
import logging
import os
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile
import time

from bioblend import galaxy
from bioblend import toolshed

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

# need this until a PR/version bump to fix galaxyxml prepending the exe even
# with override.


def timenow():
    """return current time as a string
    """
    return time.strftime("%d/%m/%Y %H:%M:%S", time.localtime(time.time()))


def quote_non_numeric(s):
    """return a prequoted string for non-numerics

    t = t.replace("\\$", "$")
    return t


def parse_citations(citations_text):
    """
    """
    citations = [c for c in citations_text.split("**ENTRY**") if c.strip()]
    citation_tuples = []
    for citation in citations:
        if citation.startswith("doi"):
            citation_tuples.append(("doi", citation[len("doi") :].strip()))

    def cleanuppar(self):
        """ positional parameters are complicated by their numeric ordinal"""
        for i, p in enumerate(self.infiles):
            if self.args.parampass == "positional":
                assert p[ICLPOS].isdigit(), (
                    "Positional parameters must be ordinal integers - got %s for %s"
                    % (p[ICLPOS], p[ILABPOS])
                )
            p.append(p[ICLPOS])
            if p[ICLPOS].isdigit() or self.args.parampass == "0":
                scl = "input%d" % (i + 1)
                p[ICLPOS] = scl
            self.infiles[i] = p
        for i, p in enumerate(
            self.outfiles
        ):  # trying to automagically gather using extensions
            if self.args.parampass == "positional" and p[OCLPOS] != "STDOUT":
                assert p[OCLPOS].isdigit(), (
                    "Positional parameters must be ordinal integers - got %s for %s"
                    % (p[OCLPOS], p[ONAMEPOS])
                )
            p.append(p[OCLPOS])
            if p[OCLPOS].isdigit() or p[OCLPOS] == "STDOUT":
                scl = p[ONAMEPOS]
                p[OCLPOS] = scl
            self.outfiles[i] = p
        for i, p in enumerate(self.addpar):
            if self.args.parampass == "positional":
                assert p[ACLPOS].isdigit(), (
                    "Positional parameters must be ordinal integers - got %s for %s"
                    % (p[ACLPOS], p[ANAMEPOS])
                )
            p.append(p[ACLPOS])
            if p[ACLPOS].isdigit():
                scl = "input%s" % p[ACLPOS]
                p[ACLPOS] = scl
            self.addpar[i] = p

    def clsimple(self):
        """ no parameters - uses < and > for i/o
        """
        aCL = self.cl.append
        aXCL = self.xmlcl.append

        if len(self.infiles) > 0:
            aCL("<")

        if self.lastxclredirect:
            aXCL(self.lastxclredirect[0])
            aXCL(self.lastxclredirect[1])

    def clargparse(self):
        """ argparse style
        """
        aCL = self.cl.append
        aXCL = self.xmlcl.append
        # inputs then params in argparse named form
        for (o_v, k, v, koverride) in self.xclsuffix:
            if koverride > "":

            newfmt = self.outfiles[0][OFMTPOS]
            anout = gxtp.OutputData(newname, format=newfmt, num_dashes=0)
            anout.command_line_override = "> $%s" % newname
            anout.positional = self.is_positional
            self.toutputs.append(anout)
            tp = gxtp.TestOutput(name=newname, value="%s_sample" % newname, format=newfmt)
            self.testparam.append(tp)

    def makeXML(self):
        """
        Create a Galaxy xml tool wrapper for the new script

            self.newtool.command_override = self.xmlcl
        if self.args.help_text:
            helptext = open(self.args.help_text, "r").readlines()
            safertext = [html_escape(x) for x in helptext]
            if False and self.args.script_path:
                scrp = self.script.split('\n')
                scrpt = ['   %s' % x for x in scrp] # try to stop templating
                scrpt.insert(0,"```\n")
                if len(scrpt) > 300:
                    safertext = safertext + scrpt[:100] + ['>500 lines - stuff deleted','......'] + scrpt[-100:]
                else:
                    safertext = safertext + scrpt
                safertext.append("\n```")
            self.newtool.help = "\n".join([x for x in safertext])
        else:

                if os.path.exists(self.tlog):
                    sto = open(self.tlog, "a")
                else:
                    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()

            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 retval != 0 and err:  # problem
            sys.stderr.write(err)
        logging.debug("run done")
        return retval


    def shedLoad(self):
        """
        {'deleted': False,
              'description': 'Tools for manipulating data',
              'id': '175812cd7caaf439',

        if os.path.exists(self.tlog):
            sto = open(self.tlog, "a")
        else:
            sto = open(self.tlog, "w")

        ts = toolshed.ToolShedInstance(url=self.args.toolshed_url,key=self.args.toolshed_api_key,verify=False)
        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}')
        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,
                          type='unrestricted', remote_repository_url=self.args.toolshed_url,
                          homepage_url=None, category_ids=catID)
            tid = res.get('id',None)
            sto.write(f'##########create res={res}')
        else:
             i = rnames.index(self.args.tool_name)
             tid = rids[i]
        res = ts.repositories.update_repository(id=tid, tar_ball_path=self.newtarpath, commit_message=None)
        sto.write(f'#####update res={res}')
        sto.close()


    def eph_galaxy_load(self):
        """load the new tool from the local toolshed after planemo uploads it
        """
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        cll = [

            self.args.tool_name,
            "--owner",
            "fubar",
            "--toolshed",
            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>

        """
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        ts = toolshed.ToolShedInstance(url=self.args.toolshed_url,key=self.args.toolshed_api_key,verify=False)
        repos = ts.repositories.get_repositories()
        rnames = [x.get('name','?') for x in repos]
        rids = [x.get('id','?') for x in repos]
        tout.write(f'############names={rnames} rids={rids}')
        cat = 'ToolFactory generated tools'
        if self.args.tool_name not in rnames:
            cll = ["planemo", "shed_create", "--shed_target", "local",
            "--owner","fubar","--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" % self.args.tool_name)
            else:
                tout.write("initiated %s" % self.args.tool_name)
        cll = [
            "planemo",
            "shed_upload",

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


    def eph_test(self):
            """
            """
            if os.path.exists(self.tlog):
                tout = open(self.tlog, "a")
            else:
                tout = open(self.tlog, "w")
            cll = [
                    "shed-tools",
                    "test",
                 "-g",
                self.args.galaxy_url,
                "-a",
                self.args.galaxy_api_key,
                "--name",
                self.args.tool_name,
                "--owner",
                "fubar",
                   ]
            p = subprocess.run(
                    cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
                )
            tout.write("eph_test Ran %s got %d" % (" ".join(cll), p.returncode))
            tout.close()
            return p.returncode

    def planemo_test(self, genoutputs=True):
        """planemo is a requirement so is available for testing
        and for generating test outputs if command or test overrides are supplied
        test outputs are sent to repdir for display

        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        if genoutputs:
            dummy,tfile = tempfile.mkstemp()
            cll = [
                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                xreal,
            ]
            p = subprocess.run(
                    cll, shell=False, cwd=self.tooloutdir, stderr=dummy, stdout=dummy,
                )
        else:
            cll = ["planemo", "test", "--galaxy_root",
                self.args.galaxy_root,
                xreal,]
            p = subprocess.run(
                    cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
                )
        tout.close()
        return p.returncode



    def writeShedyml(self):
        """for planemo
        """
        yuser = self.args.user_email.split("@")[0]
        yfname = os.path.join(self.tooloutdir, ".shed.yml")
        yamlf = open(yfname, "w")
        odict = {
            "name": self.tool_name,

        }
        yaml.dump(odict, yamlf, allow_unicode=True)
        yamlf.close()

    def makeTool(self):
        """write xmls and input samples into place
        """
        self.makeXML()
        if self.args.script_path:
            stname = os.path.join(self.tooloutdir, "%s" % (self.sfile))
            if not os.path.exists(stname):
                shutil.copyfile(self.sfile, stname)

            shutil.copyfile(pth, dest)
            dest = os.path.join(self.repdir, "%s.%s" % (p[ICLPOS], p[IFMTPOS]))
            shutil.copyfile(pth, dest)

    def makeToolTar(self):
        """ move 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)  or os.path.splitext(filename)[1].startswith(excludeme) else tarinfo

        for p in self.outfiles:
            src = p[ONAMEPOS]
            if os.path.isfile(src):
                dest = os.path.join(self.testdir, "%s_sample" % src)

        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 not entry.is_file() or entry.name.startswith("."):
                    continue
                if "." in entry.name:

    a("--edit_additional_parameters", action="store_true", default=False)
    a("--parampass", default="positional")
    a("--tfout", default="./tfout")
    a("--new_tool", default="new_tool")
    a("--galaxy_url", default="http://localhost:8080")
    a("--toolshed_url", default="http://localhost:9009") # make sure this is NOT 127.0.0.1 - it won't work if tool_sheds_conf.xml has localhost
    #a("--galaxy_api_key", default="1e62ddad74fe9bf112859f4e9efea48b")
    #a("--toolshed_api_key", default="9154c91f2a162bf12fda15764f43846c")

    a("--toolshed_api_key", default="fakekey")
    a("--galaxy_api_key", default="fakekey")
    a("--galaxy_root", default="/galaxy-central")


    args = parser.parse_args()
    assert not args.bad_user, (
        'UNAUTHORISED: %s is NOT authorized to use this tool until Galaxy admin adds %s to "admin_users" in the Galaxy configuration file'
        % (args.bad_user, args.bad_user)

        r.moveRunOutputs()
        r.makeToolTar()
        retcode = r.planemo_test(genoutputs=False)
        r.moveRunOutputs()
        if args.make_Tool == "gentestinstall":
            retcode = r.planemo_shedload() #r.shedLoad()
            print(f'planemo_shedload returned {retcode}')
            r.eph_galaxy_load()



if __name__ == "__main__":
    main()

75:25dd9364a32a

# one to run the executable with the supplied test data and settings
# Be simpler to write the tool, then run it with planemo and soak up the test outputs.


import argparse
import logging
import os
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile
import time


from bioblend import toolshed

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

# need this until a PR/version bump to fix galaxyxml prepending the exe even
# with override.


def timenow():
    """return current time as a string"""
    return time.strftime("%d/%m/%Y %H:%M:%S", time.localtime(time.time()))


def quote_non_numeric(s):
    """return a prequoted string for non-numerics

    t = t.replace("\\$", "$")
    return t


def parse_citations(citations_text):
    """"""
    citations = [c for c in citations_text.split("**ENTRY**") if c.strip()]
    citation_tuples = []
    for citation in citations:
        if citation.startswith("doi"):
            citation_tuples.append(("doi", citation[len("doi") :].strip()))

    def cleanuppar(self):
        """ positional parameters are complicated by their numeric ordinal"""
        for i, p in enumerate(self.infiles):
            if self.args.parampass == "positional":
                assert p[
                    ICLPOS
                ].isdigit(), "Positional parameters must be ordinal integers - got %s for %s" % (
                    p[ICLPOS],
                    p[ILABPOS],
                )
            p.append(p[ICLPOS])
            if p[ICLPOS].isdigit() or self.args.parampass == "0":
                scl = "input%d" % (i + 1)
                p[ICLPOS] = scl
            self.infiles[i] = p
        for i, p in enumerate(
            self.outfiles
        ):  # trying to automagically gather using extensions
            if self.args.parampass == "positional" and p[OCLPOS] != "STDOUT":
                assert p[
                    OCLPOS
                ].isdigit(), "Positional parameters must be ordinal integers - got %s for %s" % (
                    p[OCLPOS],
                    p[ONAMEPOS],
                )
            p.append(p[OCLPOS])
            if p[OCLPOS].isdigit() or p[OCLPOS] == "STDOUT":
                scl = p[ONAMEPOS]
                p[OCLPOS] = scl
            self.outfiles[i] = p
        for i, p in enumerate(self.addpar):
            if self.args.parampass == "positional":
                assert p[
                    ACLPOS
                ].isdigit(), "Positional parameters must be ordinal integers - got %s for %s" % (
                    p[ACLPOS],
                    p[ANAMEPOS],
                )
            p.append(p[ACLPOS])
            if p[ACLPOS].isdigit():
                scl = "input%s" % p[ACLPOS]
                p[ACLPOS] = scl
            self.addpar[i] = p

    def clsimple(self):
        """no parameters - uses < and > for i/o"""
        aCL = self.cl.append
        aXCL = self.xmlcl.append

        if len(self.infiles) > 0:
            aCL("<")

        if self.lastxclredirect:
            aXCL(self.lastxclredirect[0])
            aXCL(self.lastxclredirect[1])

    def clargparse(self):
        """argparse style"""
        aCL = self.cl.append
        aXCL = self.xmlcl.append
        # inputs then params in argparse named form
        for (o_v, k, v, koverride) in self.xclsuffix:
            if koverride > "":

            newfmt = self.outfiles[0][OFMTPOS]
            anout = gxtp.OutputData(newname, format=newfmt, num_dashes=0)
            anout.command_line_override = "> $%s" % newname
            anout.positional = self.is_positional
            self.toutputs.append(anout)
            tp = gxtp.TestOutput(
                name=newname, value="%s_sample" % newname, format=newfmt
            )
            self.testparam.append(tp)

    def makeXML(self):
        """
        Create a Galaxy xml tool wrapper for the new script

            self.newtool.command_override = self.xmlcl
        if self.args.help_text:
            helptext = open(self.args.help_text, "r").readlines()
            safertext = [html_escape(x) for x in helptext]
            if False and self.args.script_path:
                scrp = self.script.split("\n")
                scrpt = ["   %s" % x for x in scrp]  # try to stop templating
                scrpt.insert(0, "```\n")
                if len(scrpt) > 300:
                    safertext = (
                        safertext + scrpt[:100] + \
                        [">500 lines - stuff deleted", "......"] + scrpt[-100:]
                    )
                else:
                    safertext = safertext + scrpt
                safertext.append("\n```")
            self.newtool.help = "\n".join([x for x in safertext])
        else:

                if os.path.exists(self.tlog):
                    sto = open(self.tlog, "a")
                else:
                    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()

            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 retval != 0 and err:  # problem
            sys.stderr.write(err)
        logging.debug("run done")
        return retval

    def shedLoad(self):
        """
        {'deleted': False,
              'description': 'Tools for manipulating data',
              'id': '175812cd7caaf439',

        if os.path.exists(self.tlog):
            sto = open(self.tlog, "a")
        else:
            sto = open(self.tlog, "w")

        ts = toolshed.ToolShedInstance(
            url=self.args.toolshed_url, key=self.args.toolshed_api_key, verify=False
        )
        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}")
        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,
                type="unrestricted",
                remote_repository_url=self.args.toolshed_url,
                homepage_url=None,
                category_ids=catID,
            )
            tid = res.get("id", None)
            sto.write(f"##########create res={res}")
        else:
            i = rnames.index(self.args.tool_name)
            tid = rids[i]
        res = ts.repositories.update_repository(
            id=tid, tar_ball_path=self.newtarpath, commit_message=None
        )
        sto.write(f"#####update res={res}")
        sto.close()

    def eph_galaxy_load(self):
        """load the new tool from the local toolshed after planemo uploads it"""
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        cll = [

            self.args.tool_name,
            "--owner",
            "fubar",
            "--toolshed",
            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>

        """
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        ts = toolshed.ToolShedInstance(
            url=self.args.toolshed_url, key=self.args.toolshed_api_key, verify=False
        )
        repos = ts.repositories.get_repositories()
        rnames = [x.get("name", "?") for x in repos]
        rids = [x.get("id", "?") for x in repos]
        tout.write(f"############names={rnames} rids={rids}")
        #cat = "ToolFactory generated tools"
        if self.args.tool_name not in rnames:
            cll = [
                "planemo",
                "shed_create",
                "--shed_target",
                "local",
                "--owner",
                "fubar",
                "--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" % self.args.tool_name)
            else:
                tout.write("initiated %s" % self.args.tool_name)
        cll = [
            "planemo",
            "shed_upload",

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

    def eph_test(self):
        """"""
        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        cll = [
            "shed-tools",
            "test",
            "-g",
            self.args.galaxy_url,
            "-a",
            self.args.galaxy_api_key,
            "--name",
            self.args.tool_name,
            "--owner",
            "fubar",
        ]
        p = subprocess.run(
            cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
        )
        tout.write("eph_test Ran %s got %d" % (" ".join(cll), p.returncode))
        tout.close()
        return p.returncode

    def planemo_test(self, genoutputs=True):
        """planemo is a requirement so is available for testing
        and for generating test outputs if command or test overrides are supplied
        test outputs are sent to repdir for display

        if os.path.exists(self.tlog):
            tout = open(self.tlog, "a")
        else:
            tout = open(self.tlog, "w")
        if genoutputs:
            dummy, tfile = tempfile.mkstemp()
            cll = [
                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                "--update_test_data",
                xreal,
            ]
            p = subprocess.run(
                cll,
                shell=False,
                cwd=self.tooloutdir,
                stderr=dummy,
                stdout=dummy,
            )
        else:
            cll = [
                "planemo",
                "test",
                "--galaxy_root",
                self.args.galaxy_root,
                xreal,
            ]
            p = subprocess.run(
                cll, shell=False, cwd=self.tooloutdir, stderr=tout, stdout=tout
            )
        tout.close()
        return p.returncode

    def writeShedyml(self):
        """for planemo"""
        yuser = self.args.user_email.split("@")[0]
        yfname = os.path.join(self.tooloutdir, ".shed.yml")
        yamlf = open(yfname, "w")
        odict = {
            "name": self.tool_name,

        }
        yaml.dump(odict, yamlf, allow_unicode=True)
        yamlf.close()

    def makeTool(self):
        """write xmls and input samples into place"""
        self.makeXML()
        if self.args.script_path:
            stname = os.path.join(self.tooloutdir, "%s" % (self.sfile))
            if not os.path.exists(stname):
                shutil.copyfile(self.sfile, stname)

            shutil.copyfile(pth, dest)
            dest = os.path.join(self.repdir, "%s.%s" % (p[ICLPOS], p[IFMTPOS]))
            shutil.copyfile(pth, dest)

    def makeToolTar(self):
        """move 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) or \
                os.path.splitext(filename)[1].startswith(excludeme)
                else tarinfo
            )

        for p in self.outfiles:
            src = p[ONAMEPOS]
            if os.path.isfile(src):
                dest = os.path.join(self.testdir, "%s_sample" % src)

        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 not entry.is_file() or entry.name.startswith("."):
                    continue
                if "." in entry.name:

    a("--edit_additional_parameters", action="store_true", default=False)
    a("--parampass", default="positional")
    a("--tfout", default="./tfout")
    a("--new_tool", default="new_tool")
    a("--galaxy_url", default="http://localhost:8080")
    a(
        "--toolshed_url", default="http://localhost:9009"
    )  # make sure this is NOT 127.0.0.1 - it won't work if tool_sheds_conf.xml has localhost
    # a("--galaxy_api_key", default="1e62ddad74fe9bf112859f4e9efea48b")
    # a("--toolshed_api_key", default="9154c91f2a162bf12fda15764f43846c")

    a("--toolshed_api_key", default="fakekey")
    a("--galaxy_api_key", default="fakekey")
    a("--galaxy_root", default="/galaxy-central")

    args = parser.parse_args()
    assert not args.bad_user, (
        'UNAUTHORISED: %s is NOT authorized to use this tool until Galaxy admin adds %s to "admin_users" in the Galaxy configuration file'
        % (args.bad_user, args.bad_user)

        r.moveRunOutputs()
        r.makeToolTar()
        retcode = r.planemo_test(genoutputs=False)
        r.moveRunOutputs()
        if args.make_Tool == "gentestinstall":
            retcode = r.planemo_shedload()  # r.shedLoad()
            print(f"planemo_shedload returned {retcode}")
            r.eph_galaxy_load()


if __name__ == "__main__":
    main()