Coverage for test/compact_merge_csv_test.py: 98%

1#!/usr/bin/env python 

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

3# Copyright (c) 2025 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 

17import csv 

18import os 

19import tempfile 

20import unittest 

21from pathlib import Path 

22 

23from oc_meta.run.merge.compact_output_csv import ( 

24 process_merge_directory, 

25 process_merge_file, 

26) 

27 

28 

29class TestProcessMergeCSV(unittest.TestCase): 

30 def setUp(self): 

31 # Create a temporary directory for test files 

32 self.test_dir = tempfile.mkdtemp() # Using default system temp directory 

33 

34 # Create a test CSV file 

35 self.test_csv_content = [ 

36 ["surviving_entity", "merged_entities", "Done"], 

37 ["https://w3id.org/oc/meta/id/1", "https://w3id.org/oc/meta/id/2", "True"], 

38 ["https://w3id.org/oc/meta/id/3", "https://w3id.org/oc/meta/id/4", "False"], 

39 [ 

40 "https://w3id.org/oc/meta/id/5", 

41 "https://w3id.org/oc/meta/id/6; https://w3id.org/oc/meta/id/7", 

42 "True", 

43 ], 

44 ] 

45 

46 self.test_file = os.path.join(self.test_dir, "test.csv") 

47 with open(self.test_file, "w", newline="", encoding="utf-8") as f: 

48 writer = csv.writer(f) 

49 writer.writerows(self.test_csv_content) 

50 

51 def tearDown(self): 

52 # Clean up temporary files 

53 for file in Path(self.test_dir).glob("*.csv"): 

54 file.unlink() 

55 os.rmdir(self.test_dir) 

56 

57 def test_process_merge_file(self): 

58 results = process_merge_file(self.test_file) 

59 

60 # Should only include rows with Done=True 

61 self.assertEqual(len(results["valid_entries"]), 2) 

62 self.assertEqual(results["total_rows"], 3) # 3 data rows (excluding header) 

63 

64 # Check first result 

65 self.assertEqual( 

66 results["valid_entries"][0][0], "https://w3id.org/oc/meta/id/1" 

67 ) 

68 self.assertEqual( 

69 results["valid_entries"][0][1], "https://w3id.org/oc/meta/id/2" 

70 ) 

71 

72 # Check second result with multiple merged entities 

73 self.assertEqual( 

74 results["valid_entries"][1][0], "https://w3id.org/oc/meta/id/5" 

75 ) 

76 self.assertEqual( 

77 results["valid_entries"][1][1], 

78 "https://w3id.org/oc/meta/id/6; https://w3id.org/oc/meta/id/7", 

79 ) 

80 

81 def test_process_merge_directory(self): 

82 # Create a second test file 

83 second_file = os.path.join(self.test_dir, "test2.csv") 

84 with open(second_file, "w", newline="", encoding="utf-8") as f: 

85 writer = csv.writer(f) 

86 writer.writerows( 

87 [ 

88 ["surviving_entity", "merged_entities", "Done"], 

89 [ 

90 "https://w3id.org/oc/meta/id/8", 

91 "https://w3id.org/oc/meta/id/9", 

92 "True", 

93 ], 

94 ] 

95 ) 

96 

97 output_file = os.path.join(self.test_dir, "output.csv") 

98 process_merge_directory(self.test_dir, output_file) 

99 

100 # Check output file 

101 with open(output_file, "r", encoding="utf-8") as f: 

102 reader = csv.reader(f) 

103 rows = list(reader) 

104 

105 # Check header 

106 self.assertEqual(rows[0], ["surviving_entity", "merged_entities"]) 

107 

108 # Should have 3 rows total (2 from first file + 1 from second file) 

109 self.assertEqual(len(rows), 4) 

110 

111 

112if __name__ == "__main__": 

113 unittest.main()