-
Notifications
You must be signed in to change notification settings - Fork 13
/
test_export.py
95 lines (83 loc) · 3.74 KB
/
test_export.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
import os
import unittest
from pyiron_base import Project
from pyiron_base.project.archiving.export_archive import export_database
import pandas as pd
from pandas._testing import assert_frame_equal
from filecmp import dircmp
from shutil import rmtree
from pyiron_base._tests import PyironTestCase, ToyJob
class TestPack(PyironTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
# this is used to create a folder/a compressed file, are not path
cls.arch_dir = 'archive_folder'
# this is used to create a folder/a compressed file, are not path
cls.arch_dir_comp = cls.arch_dir + '_comp'
cls.pr = Project('test')
cls.pr.remove_jobs(recursive=True, silently=True)
cls.job = cls.pr.create_job(job_type=ToyJob, job_name='toy')
cls.job.run()
cls.pr.pack(destination_path=cls.arch_dir, compress=False)
cls.file_location = os.path.dirname(os.path.abspath(__file__)).replace(
"\\", "/"
)
@classmethod
def tearDownClass(cls) -> None:
super().tearDownClass()
cls.pr.remove(enable=True)
uncompressed_pr = Project(cls.arch_dir)
uncompressed_pr.remove(enable=True, enforce=True)
os.remove('export.csv')
def test_exportedCSV(self):
# in the first test, the csv file from the packing function is read
# and is compared with the return dataframe from export_database
directory_to_transfer = os.path.basename(self.pr.path[:-1])
self.pr.pack(destination_path=self.arch_dir, compress=False)
df_read = pd.read_csv('export.csv')
df_read.drop(df_read.keys()[0], inplace=True, axis=1)
# this removes the "None/NaN/empty" cells as well as the unnamed column
df_read.dropna(inplace=True, axis=1)
df_read['timestart'] = pd.to_datetime(df_read['timestart'])
df_read["hamversion"] = float(df_read["hamversion"])
df_exp = export_database(
self.pr, directory_to_transfer, 'archive_folder'
).dropna(axis=1)
df_exp["hamversion"] = float(df_exp["hamversion"])
assert_frame_equal(df_exp, df_read)
def test_HDF5(self):
# first we check whether the toy.h5 file exists
# in the exported directory
h5_file_path = self.arch_dir + '/' + self.pr.name + "/toy.h5"
self.assertTrue(os.path.exists(h5_file_path))
def test_compress(self):
# here we check whether the packing function
# does the compressibility right
self.pr.pack(destination_path=self.arch_dir_comp, compress=True)
file_path = self.arch_dir_comp + ".tar.gz"
self.assertTrue(os.path.exists(file_path))
os.remove(file_path)
def test_content(self):
# here we test the content of the archive_folder and
# compare it with the content of the project directory
path_to_compare = self.arch_dir + "/" + self.pr.name
compare_obj = dircmp(path_to_compare, self.pr.path)
self.assertEqual(len(compare_obj.diff_files), 0)
def test_export_with_targz_extension(self):
os.makedirs(os.path.join(os.curdir, 'tmp'))
tmp_path = os.path.join(os.curdir, 'tmp')
tar_arch = self.arch_dir_comp + '.tar.gz'
self.pr.pack(destination_path=os.path.join(tmp_path, tar_arch),
csv_file_name=os.path.join(tmp_path, 'exported.csv'), compress=True)
desirable_lst = [tar_arch, 'exported.csv']
desirable_lst.sort()
content_tmp = os.listdir(tmp_path)
content_tmp.sort()
try:
rmtree(tmp_path)
except Exception as err_msg:
print(f"deleting unsuccessful: {err_msg}")
self.assertListEqual(desirable_lst, content_tmp)
if __name__ == "__main__":
unittest.main()