Coverage for test / test_archive_reports.py: 100%
74 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-21 14:31 +0000
1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import json
6import os
7import shutil
8import tempfile
9import unittest
10from unittest.mock import MagicMock, patch
12import yaml
13from crowdsourcing.archive_manager import ArchiveManager
14from crowdsourcing.archive_reports import check_and_archive_reports
17class TestArchiveReports(unittest.TestCase):
18 def setUp(self):
19 """Set up test environment before each test."""
20 # Create a temporary directory for test files
21 self.test_dir = tempfile.mkdtemp()
22 self.reports_dir = os.path.join(self.test_dir, "docs/validation_reports")
23 os.makedirs(self.reports_dir)
25 # Create test config file
26 self.config = {
27 "validation_reports": {
28 "max_reports_before_archive": 3,
29 "reports_dir": self.reports_dir,
30 "index_file": os.path.join(self.reports_dir, "index.json"),
31 },
32 "zenodo": {
33 "metadata_template": {
34 "title": "Test Archive",
35 "description": "Test Description",
36 "creators": [{"name": "Test Creator"}],
37 "access_right": "open",
38 "upload_type": "dataset",
39 "license": "CC0-1.0",
40 }
41 },
42 }
43 self.config_path = os.path.join(self.test_dir, "test_config.yaml")
44 with open(self.config_path, "w") as f:
45 yaml.dump(self.config, f)
47 # Set environment variables for testing
48 self.env_patcher = patch.dict(
49 os.environ,
50 {
51 "ENVIRONMENT": "development",
52 "ZENODO_SANDBOX": "fake-token",
53 },
54 )
55 self.env_patcher.start()
57 # Patch ArchiveManager to use our test config
58 self.archive_manager_patcher = patch(
59 "crowdsourcing.archive_reports.ArchiveManager",
60 return_value=ArchiveManager(config_path=self.config_path),
61 )
62 self.archive_manager_patcher.start()
64 def tearDown(self):
65 """Clean up after each test."""
66 self.archive_manager_patcher.stop()
67 self.env_patcher.stop()
68 shutil.rmtree(self.test_dir)
70 @patch("crowdsourcing.archive_reports.logger")
71 def test_check_and_archive_reports_no_reports(self, mock_logger):
72 """Test check_and_archive_reports when there are no reports to archive."""
73 # Initialize archive manager and run check
74 check_and_archive_reports()
76 # Verify logging
77 mock_logger.info.assert_any_call("Starting report archival check")
78 mock_logger.info.assert_any_call(
79 "Archival threshold not reached, no action needed"
80 )
82 @patch("crowdsourcing.archive_reports.logger")
83 def test_check_and_archive_reports_below_threshold(self, mock_logger):
84 """Test check_and_archive_reports when number of reports is below threshold."""
85 # Initialize archive manager
86 archive_manager = ArchiveManager(config_path=self.config_path)
88 # Add some reports (less than threshold)
89 for i in range(2): # threshold is 3
90 report_name = f"validation_issue_{i}.html"
91 report_path = os.path.join(self.reports_dir, report_name)
92 with open(report_path, "w") as f:
93 f.write(f"Test report {i}")
94 archive_manager.add_report(
95 report_name, f"https://example.com/{report_name}"
96 )
98 # Run check
99 check_and_archive_reports()
101 # Verify logging
102 mock_logger.info.assert_any_call("Starting report archival check")
103 mock_logger.info.assert_any_call(
104 "Archival threshold not reached, no action needed"
105 )
107 @patch("crowdsourcing.archive_reports.logger")
108 @patch("crowdsourcing.zenodo_utils.create_deposition_resource")
109 @patch("crowdsourcing.zenodo_utils.get_zenodo_token")
110 @patch("requests.put")
111 @patch("requests.post")
112 def test_check_and_archive_reports_above_threshold(
113 self, mock_post, mock_put, mock_get_token, mock_create_deposition, mock_logger
114 ):
115 """Test check_and_archive_reports when number of reports is above threshold."""
116 # Setup mocks
117 mock_response = {"id": "123", "links": {"bucket": "http://bucket-url"}}
118 mock_deposition_response = MagicMock()
119 mock_deposition_response.json.return_value = mock_response
120 mock_deposition_response.raise_for_status = MagicMock()
122 mock_publish_response = MagicMock()
123 mock_publish_response.json.return_value = {"doi": "10.5281/zenodo.123"}
124 mock_publish_response.raise_for_status = MagicMock()
125 mock_post.side_effect = [mock_deposition_response, mock_publish_response]
127 mock_get_token.return_value = "fake-token"
128 mock_put.return_value.raise_for_status = MagicMock()
130 # Mock ArchiveManager to use our test config
131 with patch(
132 "crowdsourcing.archive_reports.ArchiveManager"
133 ) as mock_archive_manager:
134 mock_instance = mock_archive_manager.return_value
135 mock_instance.needs_archival.return_value = True
136 mock_instance.archive_reports.return_value = "10.5281/zenodo.123"
138 # Run check
139 check_and_archive_reports()
141 # Verify logging
142 mock_logger.info.assert_has_calls(
143 [
144 unittest.mock.call("Starting report archival check"),
145 unittest.mock.call(
146 "Archival threshold reached, starting archival process"
147 ),
148 unittest.mock.call(
149 "Successfully archived reports. DOI: 10.5281/zenodo.123"
150 ),
151 ],
152 any_order=False,
153 )
155 @patch("crowdsourcing.archive_reports.logger")
156 def test_check_and_archive_reports_error(self, mock_logger):
157 """Test check_and_archive_reports when an error occurs."""
158 # Create a mock ArchiveManager that raises an exception
159 with patch("crowdsourcing.archive_reports.ArchiveManager") as mock_manager:
160 mock_instance = mock_manager.return_value
161 mock_instance.needs_archival.side_effect = Exception("Test error")
163 # Run check
164 with self.assertRaises(Exception):
165 check_and_archive_reports()
167 # Verify logging
168 mock_logger.info.assert_called_with("Starting report archival check")
169 mock_logger.error.assert_called_with(
170 "Error during report archival: Test error"
171 )
174if __name__ == "__main__": # pragma: no cover
175 unittest.main()