-
Notifications
You must be signed in to change notification settings - Fork 2
/
benchtests.py
executable file
·133 lines (115 loc) · 5 KB
/
benchtests.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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# test hellox worlt dfgjl
"""
:Description: Unit tests for the modular benchmark (Python Clustering Algorithms BenchMarking Framework).
:Authors: (c) Artem Lutov <artem@exascale.info>
:Organizations: eXascale Infolab <http://exascale.info/>, Lumais <http://www.lumais.com/>
:Date: 2018-06
"""
from __future__ import print_function, division # Required for stderr output, must be the first import
import unittest
import os
import glob
import tempfile
import shutil
import tarfile
import time
from multiprocessing import Value
from benchutils import nameVersion, tobackup, syncedTime, ORIGDIR, _BCKDIR
# from benchapps import preparePath
class TestUtils(unittest.TestCase):
"""Tests for the Benchmark utilities"""
# __bdir = None # Base directory for the tests
# @classmethod
# def setUpClass(cls):
# cls.__bdir = tempfile.mkdtemp(prefix='tmp_bmtests')
# @classmethod
# def tearDownClass(cls):
# if cls.__bdir is not None:
# shutil.rmtree(cls.__bdir)
def test_nameVersion(self):
"""nameVersion() tests"""
# Test for the non-existent name
randname = 's;35>.ds8u9'
stval0 = None
synctime = syncedTime(stval0, lock=False)
suffix = 'suf' # Versioning suffix
self.assertFalse(os.path.exists(randname))
self.assertEqual(nameVersion(randname, False), randname)
self.assertEqual(nameVersion(randname, False, suffix='suf'), '_'.join((randname, suffix)))
# Consider path expansion with for the non-existent path
self.assertRaises(StopIteration, next, glob.iglob(randname + '*'))
self.assertEqual(nameVersion(randname, True), randname)
# Check with Synctime
self.assertEqual(nameVersion(randname, True, synctime), randname)
self.assertFalse(synctime.value
, 'synctime.value should not be changed for the non-existent path')
# Check path expansion to the existent path
path = next(glob.iglob('*'))
self.assertNotEqual(nameVersion(path, True), path
, 'Unexistent versioned name is expected for the existent path')
# Check with Synctime
# None value
self.assertNotEqual(nameVersion(path, True, synctime), path)
self.assertTrue(synctime.value
, 'synctime.value should be initialized for the existent path')
self.assertIn('_' + suffix, nameVersion(path, True, synctime, suffix=suffix))
# Non None value
stval = synctime.value # Non None stval should be permanent
self.assertNotEqual(nameVersion(path, True, synctime), path)
self.assertEqual(synctime.value, stval, 'synctime.value should be permanent')
# $ python -m unittest benchtests.TestUtils.test_tobackup
def test_tobackup(self):
"""tobackup() tests"""
# Create tmp dir to test backuping
bdir = tempfile.mkdtemp(prefix='tmp_bmtests')
try:
clspref = 'cls1' # Prefix of the items being backed up
bcksuf = 'k123' # Backup name suffix
clsdir = tempfile.mkdtemp(prefix=clspref, dir=bdir)
clsf1 = tempfile.mkstemp(suffix='.cls', prefix=clspref, dir=clsdir, text=True)
os.write(clsf1[0], 'Some content\n')
os.close(clsf1[0])
clslog = tempfile.mkstemp(suffix='.log', prefix=clspref, dir=bdir)
bckarch = tobackup(clsdir, expand=False, xsuffix=bcksuf, move=False)
# print('bckarch: ' + bckarch)
self.assertTrue(bckarch.startswith(bdir) and os.path.exists(bckarch)
, 'The backup archive should exist')
self.assertTrue(os.path.exists(clsdir) and os.path.exists(clslog[1]))
# Move paths to the origdir and create symlinks instead of the former paths
# Note: relative path are used otherwise orig files overwrite symlinks
origdir = '/'.join((bdir, ORIGDIR))
os.mkdir(origdir)
curdir = os.getcwd() # Original current dir
os.chdir(bdir) # Base dir of the archiving items
try:
for p in glob.iglob(clspref + '*'):
shutil.move(p, ORIGDIR)
# Create RELATIVE symlink to be able to extract the archive anywhere
pname = os.path.split(p)[1]
opath = ORIGDIR + pname # Path of the file in the orig dir
# opath = os.path.relpath(opath, bdir)
os.symlink(opath, pname)
# Back up target symlinks with their origins
# print('> bck src: ', bdir + '/' + clspref)
# print('> bckdir content:', os.listdir(bdir))
# print('> bckdir orig (', origdir, ') content:', os.listdir(origdir))
# relpath is False because clspref is already relative
bckarch = tobackup(clspref, expand=True, xsuffix=bcksuf, move=True, relpath=False)
# print('> bckarch ({}): {}'.format(type(bckarch).__name__, bckarch))
self.assertIn('_' + bcksuf, bckarch)
self.assertTrue(_BCKDIR in bckarch and os.path.exists(bckarch)
, 'The backup archive should exist')
self.assertFalse(os.path.exists(clsdir) or os.path.exists(clslog[1]))
with tarfile.open(bckarch, 'r') as baf:
# print('> arch content: ', baf.getnames())
self.assertNotEqual(len([name for name in baf.getnames() if ORIGDIR in name]), 0)
finally:
os.chdir(curdir)
finally:
shutil.rmtree(bdir)
if __name__ == '__main__':
unittest.main()
# if unittest.main().result: # verbosity=2
# print('Try to re-execute the tests (hot run) or set x2-3 larger TEST_LATENCY')