-
Notifications
You must be signed in to change notification settings - Fork 2
/
test_submit_functions.py
98 lines (73 loc) · 3.43 KB
/
test_submit_functions.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
from egasub.ega.services import login,logout,prepare_submission, query_by_id, api_url, query_by_type, _obj_type_to_endpoint, object_submission, register_obj,validate_obj, _validate_submit_obj, update_obj, delete_obj, submit_submission
import pytest
import requests
from egasub.ega.entities.submission_subset_data import SubmissionSubsetData
from egasub.ega.entities.submission import Submission
from egasub.exceptions import CredentialsError
from egasub.submission.submittable.unaligned import Unaligned
import os
def test_login_function(ctx, mock_server):
with pytest.raises(CredentialsError):
login(ctx)
ctx.obj['SETTINGS']['ega_submitter_account'] = 'test_account'
with pytest.raises(CredentialsError):
login(ctx)
ctx.obj['SETTINGS']['ega_submitter_password'] = 'test_password'
login(ctx)
assert not ctx.obj['SUBMISSION']['sessionToken'] == None
assert ctx.obj['SUBMISSION']['sessionToken'] == "abcdefg"
def test_prepare_submission(ctx):
requests.post("%ssubmissions" % (ctx.obj['SETTINGS']['apiUrl']))
subset = SubmissionSubsetData([2,3],[5,2],[4,34],[54,1],[88,7],[1,3],[44,11],[2,11])
submission = Submission('a title', 'a description', subset)
prepare_submission(ctx,submission)
assert ctx.obj['SUBMISSION']['id'] == "12345"
def test_api_url(ctx):
assert api_url(ctx) == "http://example.com/"
del ctx.obj['SETTINGS']['apiUrl']
assert api_url(ctx) == "https://ega.crg.eu/submitterportal/v1/"
ctx.obj['SETTINGS']['apiUrl'] = "http://example.com/"
def test_query_by_id(ctx, mock_server):
result = query_by_id(ctx,'sample','sample_alias','ALIAS')
assert result[0].get('id') == '12345'
result = query_by_id(ctx,'study','test_alias','ALIAS')
assert result[0].get('id') == '12345'
result = query_by_id(ctx,'dataset','dataset_alias','ALIAS')
assert result[0].get('id') == '12345'
result = query_by_id(ctx,'policy','policy_alias','ALIAS')
assert result[0].get('id') == '12345'
def test_query_by_type(ctx, mock_server):
result = query_by_type(ctx, 'sample', "SUBMITTED")
assert result[0].get('id') == '12345'
result = query_by_type(ctx, 'study', "SUBMITTED")
assert result[0].get('id') == '12345'
result = query_by_type(ctx, 'dataset', "SUBMITTED")
assert result[0].get('id') == '12345'
result = query_by_type(ctx, 'policy', "SUBMITTED")
assert result[0].get('id') == '12345'
def test_obj_type_to_endpoint():
assert _obj_type_to_endpoint("dataset") == "datasets"
assert _obj_type_to_endpoint("sample") == "samples"
assert _obj_type_to_endpoint("policy") == "policies"
with pytest.raises(Exception):
_obj_type_to_endpoint("test")
def test_logout_function(ctx):
logout(ctx)
with pytest.raises(KeyError):
logout(ctx)
assert not 'sessionToken' in ctx.obj['SUBMISSION']
def test_register_obj(ctx):
current = os.getcwd()
os.chdir(os.path.join(current, 'tests/data/workspace/submittable'))
ctx.obj['SUBMISSION']['id'] = "12345"
ctx.obj['SUBMISSION']['sessionToken'] = "sdfsd"
register_obj(ctx, Unaligned('test_u').sample, 'sample')
os.chdir(current)
def test_submit_submission(ctx):
current = os.getcwd()
os.chdir(os.path.join(current,'tests/data/workspace/submittable'))
ctx.obj['SUBMISSION']['id'] = "12345"
ctx.obj['SUBMISSION']['sessionToken'] = "sdfsd"
submit_submission(ctx, Unaligned('test_u').sample)
#ctx.obj['SUBMISSION']['id'] = None
os.chdir(current)