Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/bioblend/_tests/test_util.py @ 0:d30785e31577 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:18:57 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d30785e31577
1 """ General support infrastructure not tied to any particular test.
2 """
3 import os
4 import random
5 import string
6 import unittest
7
8 import bioblend
9
10 NO_CLOUDMAN_MESSAGE = "CloudMan required and no CloudMan AMI configured."
11 NO_GALAXY_MESSAGE = "Externally configured Galaxy required, but not found. Set BIOBLEND_GALAXY_URL and BIOBLEND_GALAXY_API_KEY to run this test."
12 OLD_GALAXY_RELEASE = "Testing on Galaxy %s, but need %s to run this test."
13 MISSING_TOOL_MESSAGE = "Externally configured Galaxy instance requires tool %s to run test."
14
15
16 def skip_unless_cloudman():
17 """ Decorate tests with this to skip the test if CloudMan is not
18 configured.
19 """
20 if 'BIOBLEND_AMI_ID' not in os.environ:
21 return unittest.skip(NO_CLOUDMAN_MESSAGE)
22 else:
23 return lambda f: f
24
25
26 def skip_unless_galaxy(min_release=None):
27 """ Decorate tests with this to skip the test if Galaxy is not
28 configured.
29 """
30 if min_release is not None:
31 galaxy_release = os.environ.get('GALAXY_VERSION', None)
32 if galaxy_release is not None and galaxy_release != 'dev':
33 if not galaxy_release.startswith('release_'):
34 raise ValueError("The value of GALAXY_VERSION environment variable should start with 'release_'")
35 if not min_release.startswith('release_'):
36 raise Exception("min_release should start with 'release_'")
37 if galaxy_release[8:] < min_release[8:]:
38 return unittest.skip(OLD_GALAXY_RELEASE % (galaxy_release, min_release))
39
40 if 'BIOBLEND_GALAXY_URL' not in os.environ:
41 return unittest.skip(NO_GALAXY_MESSAGE)
42
43 if 'BIOBLEND_GALAXY_API_KEY' not in os.environ and 'BIOBLEND_GALAXY_MASTER_API_KEY' in os.environ:
44 galaxy_url = os.environ['BIOBLEND_GALAXY_URL']
45 galaxy_master_api_key = os.environ['BIOBLEND_GALAXY_MASTER_API_KEY']
46 gi = bioblend.galaxy.GalaxyInstance(galaxy_url, galaxy_master_api_key)
47
48 if 'BIOBLEND_GALAXY_USER_EMAIL' in os.environ:
49 galaxy_user_email = os.environ['BIOBLEND_GALAXY_USER_EMAIL']
50 else:
51 galaxy_user_email = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(5)) + "@localhost.localdomain"
52
53 galaxy_user_id = None
54 for user in gi.users.get_users():
55 if user["email"] == galaxy_user_email:
56 galaxy_user_id = user["id"]
57 break
58
59 if galaxy_user_id is None:
60 try:
61 config = gi.config.get_config()
62 except Exception:
63 # If older Galaxy for instance just assume use_remote_user is False.
64 config = {}
65
66 if config.get("use_remote_user", False):
67 new_user = gi.users.create_remote_user(galaxy_user_email)
68 else:
69 galaxy_user = galaxy_user_email.split("@", 1)[0]
70 galaxy_password = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(20))
71
72 # Create a new user and get a new API key for her
73 new_user = gi.users.create_local_user(galaxy_user, galaxy_user_email, galaxy_password)
74 galaxy_user_id = new_user["id"]
75
76 api_key = gi.users.create_user_apikey(galaxy_user_id)
77 os.environ["BIOBLEND_GALAXY_API_KEY"] = api_key
78
79 if 'BIOBLEND_GALAXY_API_KEY' not in os.environ:
80 return unittest.skip(NO_GALAXY_MESSAGE)
81
82 return lambda f: f
83
84
85 def skip_unless_tool(tool_id):
86 """ Decorate a Galaxy test method as requiring a specific tool,
87 skip the test case if the tool is unavailable.
88 """
89
90 def method_wrapper(method):
91
92 def wrapped_method(has_gi, *args, **kwargs):
93 tools = has_gi.gi.tools.get_tools()
94 # In panels by default, so flatten out sections...
95 tool_ids = [_['id'] for _ in tools]
96 if tool_id not in tool_ids:
97 raise unittest.SkipTest(MISSING_TOOL_MESSAGE % tool_id)
98
99 return method(has_gi, *args, **kwargs)
100
101 # Must preserve method name so nose can detect and report tests by
102 # name.
103 wrapped_method.__name__ = method.__name__
104 return wrapped_method
105
106 return method_wrapper
107
108
109 def get_abspath(path):
110 return os.path.abspath(os.path.join(os.path.dirname(__file__), path))