Skip to content

Commit 6f01ca3

Browse files
devsetgodevsetgo
committed
file mover tests
1 parent bdfff4f commit 6f01ca3

File tree

1 file changed

+194
-0
lines changed

1 file changed

+194
-0
lines changed

‎tests/test_common_functions/test_file_mover/test_file_mover.py

Lines changed: 194 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,194 @@
1+
import tempfile
2+
import unittest
3+
from pathlib import Path
4+
from unittest.mock import patch
5+
from watchfiles import Change # Import Change from watchfiles
6+
import logging
7+
8+
from dsg_lib.common_functions.file_mover import process_files_flow
9+
10+
logger = logging.getLogger(__name__)
11+
logging.basicConfig(level=logging.DEBUG)
12+
13+
14+
class TestFileMover(unittest.TestCase):
15+
def setUp(self):
16+
self.test_dir = tempfile.TemporaryDirectory()
17+
self.src_folder = Path(self.test_dir.name) / "src"
18+
self.temp_folder = Path(self.test_dir.name) / "temp"
19+
self.dest_folder = Path(self.test_dir.name) / "dest"
20+
self.src_folder.mkdir()
21+
self.temp_folder.mkdir()
22+
self.dest_folder.mkdir()
23+
self.test_file = self.src_folder / "test.txt"
24+
self.test_file.write_text("This is a test file.")
25+
26+
def tearDown(self):
27+
self.test_dir.cleanup()
28+
29+
def test_process_files_flow_move_only(self):
30+
# Test moving a file without compression
31+
with patch(
32+
"dsg_lib.common_functions.file_mover.watch",
33+
return_value=[[(Change.added, str(self.test_file))]],
34+
):
35+
process_files_flow(
36+
source_dir=str(self.src_folder),
37+
temp_dir=str(self.temp_folder),
38+
final_dir=str(self.dest_folder),
39+
file_pattern="*.txt",
40+
compress=False,
41+
max_iterations=1, # Limit iterations for testing
42+
)
43+
self.assertFalse(self.test_file.exists())
44+
self.assertTrue((self.dest_folder / "test.txt").exists())
45+
46+
def test_process_files_flow_with_compression(self):
47+
# Test moving and compressing a file
48+
with patch(
49+
"dsg_lib.common_functions.file_mover.watch",
50+
return_value=iter([[(Change.added, str(self.test_file))]]),
51+
):
52+
process_files_flow(
53+
source_dir=str(self.src_folder),
54+
temp_dir=str(self.temp_folder),
55+
final_dir=str(self.dest_folder),
56+
file_pattern="*.txt",
57+
compress=True,
58+
max_iterations=1, # Limit iterations for testing
59+
)
60+
self.assertFalse(self.test_file.exists())
61+
compressed_file = next(self.dest_folder.glob("*.zip"), None)
62+
self.assertIsNotNone(
63+
compressed_file
64+
) # Ensure a compressed file exists in the destination folder
65+
66+
def test_process_files_flow_invalid_pattern(self):
67+
# Test with a file that does not match the pattern
68+
with patch(
69+
"dsg_lib.common_functions.file_mover.watch",
70+
return_value=[[(Change.added, str(self.test_file))]],
71+
):
72+
process_files_flow(
73+
source_dir=str(self.src_folder),
74+
temp_dir=str(self.temp_folder),
75+
final_dir=str(self.dest_folder),
76+
file_pattern="*.log",
77+
compress=False,
78+
max_iterations=1, # Limit iterations for testing
79+
)
80+
self.assertTrue(self.test_file.exists())
81+
self.assertFalse((self.dest_folder / "test.txt").exists())
82+
83+
def test_process_files_flow_error_handling(self):
84+
# Test error handling during file processing
85+
with patch(
86+
"dsg_lib.common_functions.file_mover.shutil.move",
87+
side_effect=Exception("Mocked error"),
88+
):
89+
with patch(
90+
"dsg_lib.common_functions.file_mover.watch",
91+
return_value=[[(Change.added, str(self.test_file))]],
92+
):
93+
with self.assertRaises(Exception): # Ensure exception is raised
94+
process_files_flow(
95+
source_dir=str(self.src_folder),
96+
temp_dir=str(self.temp_folder),
97+
final_dir=str(self.dest_folder),
98+
file_pattern="*.txt",
99+
compress=False,
100+
max_iterations=1, # Limit iterations for testing
101+
)
102+
self.assertTrue(self.test_file.exists())
103+
104+
def test_process_files_flow_compression_error(self):
105+
# Test error handling during compression
106+
with patch(
107+
"dsg_lib.common_functions.file_mover.shutil.make_archive",
108+
side_effect=Exception("Mocked compression error"),
109+
):
110+
with patch(
111+
"dsg_lib.common_functions.file_mover.watch",
112+
return_value=[[(Change.added, str(self.test_file))]],
113+
):
114+
with self.assertRaises(Exception): # Ensure exception is raised
115+
process_files_flow(
116+
source_dir=str(self.src_folder),
117+
temp_dir=str(self.temp_folder),
118+
final_dir=str(self.dest_folder),
119+
file_pattern="*.txt",
120+
compress=True,
121+
max_iterations=1, # Limit iterations for testing
122+
)
123+
# Verify that the original file remains in the temporary folder
124+
self.assertTrue((self.temp_folder / "test.txt").exists())
125+
# Verify that the compressed file does not exist
126+
self.assertFalse((self.temp_folder / "test.zip").exists())
127+
128+
def test_process_files_flow_cleanup_error(self):
129+
# Test error handling during cleanup of uncompressed files
130+
with patch(
131+
"dsg_lib.common_functions.file_mover.Path.unlink",
132+
side_effect=Exception("Mocked cleanup error"),
133+
):
134+
with patch(
135+
"dsg_lib.common_functions.file_mover.watch",
136+
return_value=[[(Change.added, str(self.test_file))]],
137+
):
138+
process_files_flow(
139+
source_dir=str(self.src_folder),
140+
temp_dir=str(self.temp_folder),
141+
final_dir=str(self.dest_folder),
142+
file_pattern="*.txt",
143+
compress=True,
144+
max_iterations=1, # Limit iterations for testing
145+
)
146+
# Verify that the compressed file exists in the destination folder
147+
compressed_file = next(self.dest_folder.glob("*.zip"), None)
148+
self.assertIsNotNone(compressed_file) # Ensure a compressed file exists
149+
150+
def test_process_files_flow_existing_files(self):
151+
logger.debug("Starting test_process_files_flow_existing_files")
152+
with patch(
153+
"dsg_lib.common_functions.file_mover.watch", return_value=iter([[]])
154+
):
155+
process_files_flow(
156+
source_dir=str(self.src_folder),
157+
temp_dir=str(self.temp_folder),
158+
final_dir=str(self.dest_folder),
159+
file_pattern="*.txt", # Ensure the pattern matches the test file
160+
compress=False,
161+
max_iterations=1, # Limit iterations for testing
162+
)
163+
logger.debug(
164+
"Finished process_files_flow in test_process_files_flow_existing_files"
165+
)
166+
# Verify that the file is moved to the destination directory
167+
self.assertFalse(self.test_file.exists())
168+
self.assertTrue((self.dest_folder / "test.txt").exists())
169+
170+
def test_watch_processing_error_exception(self):
171+
# Force _process_file to throw an exception during watch processing
172+
with patch("dsg_lib.common_functions.file_mover.Path.glob", return_value=[]):
173+
with patch(
174+
"dsg_lib.common_functions.file_mover._process_file",
175+
side_effect=Exception("Mocked processing error"),
176+
):
177+
with patch(
178+
"dsg_lib.common_functions.file_mover.watch",
179+
return_value=iter([[(Change.added, str(self.test_file))]]),
180+
):
181+
with self.assertRaises(Exception) as context:
182+
process_files_flow(
183+
source_dir=str(self.src_folder),
184+
temp_dir=str(self.temp_folder),
185+
final_dir=str(self.dest_folder),
186+
file_pattern="*.txt",
187+
compress=False,
188+
max_iterations=1, # Limit iterations for testing
189+
)
190+
self.assertIn("Mocked processing error", str(context.exception))
191+
192+
193+
if __name__ == "__main__":
194+
unittest.main()

0 commit comments

Comments
 (0)