Coverage for test/file_manager_test.py: 100%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2022 Arcangelo Massari <arcangelo.massari@unibo.it> 

4# 

5# Permission to use, copy, modify, and/or distribute this software for any purpose 

6# with or without fee is hereby granted, provided that the above copyright notice 

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED 'AS IS' AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

15# SOFTWARE. 

16 

17 

18import json 

19import os 

20import unittest 

21from shutil import rmtree 

22 

23from oc_meta.lib.file_manager import (read_zipped_json, rm_tmp_csv_files, 

24 unzip_files_in_dir, zip_files_in_dir) 

25 

26BASE = os.path.join('test', 'file_manager') 

27UNZIPPED_DIR = os.path.join(BASE, 'unzipped_dir') 

28OUTPUT_DIR = os.path.join(BASE, 'output') 

29OUTPUT_DIR_1 = os.path.join(BASE, 'output_1') 

30 

31 

32class test_JsonArchiveManager(unittest.TestCase): 

33 def test_zip_jsons_files_in_dir(self): 

34 zip_files_in_dir(UNZIPPED_DIR, OUTPUT_DIR) 

35 for dirpath, _, filenames in os.walk(OUTPUT_DIR): 

36 for filename in filenames: 

37 json_data = read_zipped_json(os.path.join(dirpath, filename)) 

38 original_path = os.path.join( 

39 UNZIPPED_DIR, 

40 dirpath.replace(f'{OUTPUT_DIR}{os.sep}', '')) 

41 with open(os.path.join(original_path, filename.replace('.zip', '.json')), 'r', encoding='utf-8') as original_f: 

42 original_json = json.load(original_f) 

43 self.assertEqual(json_data, original_json) 

44 rmtree(OUTPUT_DIR) 

45 

46 def test_unzip_files_in_dir(self): 

47 zip_files_in_dir(UNZIPPED_DIR, OUTPUT_DIR_1) 

48 unzip_files_in_dir(OUTPUT_DIR_1, OUTPUT_DIR_1) 

49 for dirpath, _, filenames in os.walk(OUTPUT_DIR_1): 

50 for filename in filenames: 

51 if os.path.splitext(filename)[1] == ".json": 

52 with open(os.path.join(dirpath, filename), encoding='utf-8') as f: 

53 json_data = json.load(f) 

54 original_zip = read_zipped_json(os.path.join(dirpath, filename.replace('.json', '.zip'))) 

55 self.assertEqual(json_data, original_zip) 

56 rmtree(OUTPUT_DIR_1) 

57 

58 # def test_rm_tmp_csv_files(self): 

59 # csv_dir = os.path.join(BASE, 'csv') 

60 # os.mkdir(csv_dir) 

61 # files = ['0_2022-09-29T02-35-31', '0_2022-09-29T08-03-27', '2_2022-09-29T08-03-27', '4_2022-09-29T08-03-27', '4_2022-09-29T02-35-31'] 

62 # for file in files: 

63 # fp = open(os.path.join(csv_dir, f'{file}.csv'), 'w') 

64 # fp.close() 

65 # rm_tmp_csv_files(csv_dir) 

66 # output = os.listdir(csv_dir) 

67 # expected_output = ['0_2022-09-29T08-03-27.csv', '2_2022-09-29T08-03-27.csv', '4_2022-09-29T08-03-27.csv'] 

68 # rmtree(csv_dir) 

69 # self.assertEqual(output, expected_output) 

70 

71 

72if __name__ == '__main__': # pragma: no cover 

73 unittest.main()