test_algorithms.py 8.81 KB
Newer Older
André Anjos's avatar
André Anjos committed
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
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :

###############################################################################
#                                                                             #
# Copyright (c) 2016 Idiap Research Institute, http://www.idiap.ch/           #
# Contact: beat.support@idiap.ch                                              #
#                                                                             #
# This file is part of the beat.cmdline module of the BEAT platform.          #
#                                                                             #
# Commercial License Usage                                                    #
# Licensees holding valid commercial BEAT licenses may use this file in       #
# accordance with the terms contained in a written agreement between you      #
# and Idiap. For further information contact tto@idiap.ch                     #
#                                                                             #
# Alternatively, this file may be used under the terms of the GNU Affero      #
# Public License version 3 as published by the Free Software and appearing    #
# in the file LICENSE.AGPL included in the packaging of this file.            #
# The BEAT platform is distributed in the hope that it will be useful, but    #
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY  #
# or FITNESS FOR A PARTICULAR PURPOSE.                                        #
#                                                                             #
# You should have received a copy of the GNU Affero Public License along      #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/.           #
#                                                                             #
###############################################################################


# Basic tests for the command line beat program: algorithms

import nose.tools
32
33
import os
import shutil
André Anjos's avatar
André Anjos committed
34
35
36
37
38
39
40
41
42
43
44
45
46

from . import platform, disconnected, prefix, tmp_prefix, user, token
from ..common import Selector
from ..scripts.beat import main
from beat.core.test.utils import slow, cleanup, skipif
from beat.core.algorithm import Storage


def call(*args, **kwargs):
  '''A central mechanism to call the main routine with the right parameters'''

  use_prefix = kwargs.get('prefix', prefix)
  use_platform = kwargs.get('platform', platform)
47
  use_cache = kwargs.get('cache', 'cache')
André Anjos's avatar
André Anjos committed
48
49
50
51
52
53

  return main((
      '--platform=%s' % use_platform,
      '--user=%s' % user,
      '--token=%s' % token,
      '--prefix=%s' % use_prefix,
54
      '--cache=%s' % use_cache,
André Anjos's avatar
André Anjos committed
55
56
      '--test-mode',
      'algorithm',
Philip ABBET's avatar
Philip ABBET committed
57
  ) + args)
André Anjos's avatar
André Anjos committed
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75


@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_remote_list():
  nose.tools.eq_(call('list', '--remote'), 0)


@nose.tools.with_setup(teardown=cleanup)
def test_local_list():
  nose.tools.eq_(call('list'), 0)


@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_one():
76
  obj = 'legacy/integers_add/1'
André Anjos's avatar
André Anjos committed
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
  nose.tools.eq_(call('pull', obj, prefix=tmp_prefix), 0)
  s = Storage(tmp_prefix, obj)
  assert s.exists()


@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_pull_all():
  nose.tools.eq_(call('pull', prefix=tmp_prefix), 0)


@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_diff():
93
  obj = 'legacy/integers_add/1'
André Anjos's avatar
André Anjos committed
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
  nose.tools.eq_(call('pull', obj, prefix=tmp_prefix), 0)

  # quickly modify the user algorithm by emptying it
  storage = Storage(tmp_prefix, obj)
  storage.code.save('class Algorithm:\n  pass')

  nose.tools.eq_(call('diff', obj, prefix=tmp_prefix), 0)


@slow
@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_status():
  test_diff()
  test_pull_one()
  nose.tools.eq_(call('status', prefix=tmp_prefix), 0)


def test_check_valid():
113
  obj = 'legacy/valid_algorithm/1'
André Anjos's avatar
André Anjos committed
114
115
116
117
  nose.tools.eq_(call('check', obj), 0)


def test_check_invalid():
118
  obj = 'legacy/no_inputs_declarations/1'
André Anjos's avatar
André Anjos committed
119
120
121
122
123
  nose.tools.eq_(call('check', obj), 1)


@nose.tools.with_setup(teardown=cleanup)
def test_create(obj=None):
124
  obj = obj or 'legacy/algorithm/1'
André Anjos's avatar
André Anjos committed
125
126
127
128
129
130
131
132
  nose.tools.eq_(call('create', obj, prefix=tmp_prefix), 0)
  s = Storage(tmp_prefix, obj)
  assert s.exists()
  return s


@nose.tools.with_setup(teardown=cleanup)
def test_new_version():
133
  obj = 'legacy/algorithm/1'
André Anjos's avatar
André Anjos committed
134
  test_create(obj)
135
  obj2 = 'legacy/algorithm/2'
André Anjos's avatar
André Anjos committed
136
137
138
139
140
141
142
143
144
145
146
  nose.tools.eq_(call('version', obj, prefix=tmp_prefix), 0)
  s = Storage(tmp_prefix, obj2)
  assert s.exists()

  # check version status
  with Selector(tmp_prefix) as selector:
    assert selector.version_of('algorithm', obj2) == obj


@nose.tools.with_setup(teardown=cleanup)
def test_fork():
147
  obj = 'legacy/algorithm/1'
André Anjos's avatar
André Anjos committed
148
  test_create(obj)
149
  obj2 = 'legacy/different/1'
André Anjos's avatar
André Anjos committed
150
151
152
153
154
155
156
157
158
159
160
  nose.tools.eq_(call('fork', obj, obj2, prefix=tmp_prefix), 0)
  s = Storage(tmp_prefix, obj2)
  assert s.exists()

  # check fork status
  with Selector(tmp_prefix) as selector:
    assert selector.forked_from('algorithm', obj2) == obj


@nose.tools.with_setup(teardown=cleanup)
def test_delete_local():
161
  obj = 'legacy/algorithm/1'
André Anjos's avatar
André Anjos committed
162
163
164
165
166
167
168
169
170
171
172
173
174
  storage = test_create(obj)

  # quickly make sure it exists
  storage = Storage(tmp_prefix, obj)
  assert storage.exists()

  nose.tools.eq_(call('rm', obj, prefix=tmp_prefix), 0)
  assert not storage.exists()


@nose.tools.with_setup(teardown=cleanup)
@skipif(disconnected, "missing test platform (%s)" % platform)
def test_push_and_delete():
175
  obj = 'legacy/newobject/1'
André Anjos's avatar
André Anjos committed
176
177
178
179
180
  test_create(obj)

  # now push the new object and then delete it remotely
  nose.tools.eq_(call('push', obj, prefix=tmp_prefix), 0)
  nose.tools.eq_(call('rm', '--remote', obj, prefix=tmp_prefix), 0)
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206


@slow
@nose.tools.with_setup(teardown=cleanup)
def test_execute_algorithm_using_database():
  instructions = os.path.join(os.path.dirname(__file__), 'instructions/algo_using_database.json')
  nose.tools.eq_(call('execute', instructions, cache=tmp_prefix), 0)


@slow
@nose.tools.with_setup(teardown=cleanup)
def test_execute_algorithm_using_cached_files():
  instructions_dir = os.path.join(os.path.dirname(__file__), 'instructions')
  cache_dir = os.path.join(tmp_prefix, 'ab', 'cd', 'ef')

  os.makedirs(cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data.checksum'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index.checksum'), cache_dir)

  instructions = os.path.join(instructions_dir, 'algo_using_cached_files.json')

  nose.tools.eq_(call('execute', instructions, cache=tmp_prefix), 0)


207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@slow
@nose.tools.with_setup(teardown=cleanup)
def test_execute_algorithm_using_database_and_cached_files():
  instructions_dir = os.path.join(os.path.dirname(__file__), 'instructions')
  cache_dir = os.path.join(tmp_prefix, 'ab', 'cd', 'ef')

  os.makedirs(cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data.checksum'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index.checksum'), cache_dir)

  instructions = os.path.join(instructions_dir, 'algo_using_database_and_cached_files.json')

  nose.tools.eq_(call('execute', instructions, cache=tmp_prefix), 0)


224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@slow
@nose.tools.with_setup(teardown=cleanup)
def test_execute_analyzer():
  instructions_dir = os.path.join(os.path.dirname(__file__), 'instructions')
  cache_dir = os.path.join(tmp_prefix, 'ab', 'cd', 'ef')

  os.makedirs(cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.data.checksum'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index'), cache_dir)
  shutil.copy(os.path.join(instructions_dir, '0123456789AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA.0.9.index.checksum'), cache_dir)

  instructions = os.path.join(instructions_dir, 'analyzer.json')

  nose.tools.eq_(call('execute', instructions, cache=tmp_prefix), 0)