-
Notifications
You must be signed in to change notification settings - Fork 2.2k
/
test_hub_build.py
88 lines (67 loc) · 3.13 KB
/
test_hub_build.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
import os
import pytest
from jina.docker.hubio import HubIO
from jina.parsers.hub import set_hub_build_parser, set_hub_pushpull_parser
cur_dir = os.path.dirname(os.path.abspath(__file__))
@pytest.mark.timeout(360)
def test_hub_build_pull():
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu'), '--push', '--test-uses', '--raise-error'])
HubIO(args).build()
args = set_hub_pushpull_parser().parse_args(['jinahub/pod.dummy_mwu_encoder'])
HubIO(args).pull()
args = set_hub_pushpull_parser().parse_args(['jinahub/pod.dummy_mwu_encoder:0.0.6'])
HubIO(args).pull()
@pytest.mark.timeout(360)
def test_hub_build_uses():
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu'), '--test-uses', '--raise-error'])
HubIO(args).build()
# build again it shall not fail
HubIO(args).build()
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu'), '--test-uses', '--daemon', '--raise-error'])
HubIO(args).build()
# build again it shall not fail
HubIO(args).build()
def test_hub_build_failures():
for j in ['bad-dockerfile', 'bad-pythonfile', 'missing-dockerfile', 'missing-manifest']:
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu-bad', j)])
assert not HubIO(args).build()['is_build_success']
def test_hub_build_no_pymodules():
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu-bad', 'fail-to-start')])
assert HubIO(args).build()['is_build_success']
args = set_hub_build_parser().parse_args(
[os.path.join(cur_dir, 'hub-mwu-bad', 'fail-to-start'), '--test-uses'])
assert not HubIO(args).build()['is_build_success']
@pytest.fixture()
def requirements(request, tmpdir):
requirements_file = os.path.join(tmpdir, 'requirements.txt')
with open(requirements_file, 'w') as fp:
fp.write(request.param)
@pytest.mark.parametrize('requirements', ['jina\ntorch>=2', 'jina>=0.2\ntoch==3'], indirect=True)
def test_jina_version_freeze(requirements, tmpdir):
import pkg_resources
from jina import __version__
args = set_hub_build_parser().parse_args([str(tmpdir)])
hubio = HubIO(args)
hubio._freeze_jina_version()
requirements_file = os.path.join(tmpdir, 'requirements.txt')
with open(requirements_file, 'r') as fp:
requirements = pkg_resources.parse_requirements(fp)
assert len(list(filter(lambda x: 'jina' in str(x), requirements))) == 1
for req in requirements:
if 'jina' in str(req):
assert str(req) == f'jina=={__version__}'
@pytest.mark.parametrize('requirements', ['torch'], indirect=True)
def test_jina_version_freeze_no_jina_dependency(requirements, tmpdir):
import pkg_resources
args = set_hub_build_parser().parse_args([str(tmpdir)])
hubio = HubIO(args)
hubio._freeze_jina_version()
requirements_file = os.path.join(tmpdir, 'requirements.txt')
with open(requirements_file, 'r') as fp:
requirements = pkg_resources.parse_requirements(fp)
assert len(list(filter(lambda x: 'jina' in str(x), requirements))) == 0