Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion codecarbon/emissions_tracker.py
Original file line number Diff line number Diff line change
Expand Up @@ -843,7 +843,7 @@ def _measure_power_and_energy(self) -> None:
raise e

warning_duration = self._measure_power_secs * 3
if last_duration > warning_duration:
if last_duration > warning_duration and not self._scheduler._stopped:
warn_msg = (
"Background scheduler didn't run for a long period"
+ " (%ds), results might be inaccurate"
Expand Down
88 changes: 88 additions & 0 deletions tests/test_emissions_tracker.py
Original file line number Diff line number Diff line change
Expand Up @@ -520,3 +520,91 @@ def test_carbon_tracker_offline_context_manager(
self.assertEqual("United States", emissions_df["country_name"].values[0])
self.assertEqual("USA", emissions_df["country_iso_code"].values[0])
self.assertIsInstance(tracker.final_emissions, float)

@mock.patch("codecarbon.emissions_tracker.logger")
def test_scheduler_warning_suppressed_when_stopped(
self,
mock_logger,
mock_setup_intel_cli,
mock_log_values,
mocked_get_gpu_details,
mocked_env_cloud_details,
mocked_is_gpu_details_available,
):
"""Test that scheduler warning is suppressed when scheduler is stopped."""
with EmissionsTracker(
output_dir=self.temp_path,
measure_power_secs=1, # Short interval for testing
) as tracker:
# Stop the scheduler to simulate task mode or manual stopping
tracker._scheduler.stop()

# Artificially set last measured time to simulate long delay
import time

tracker._last_measured_time = time.perf_counter() - 10 # 10 seconds ago

# Reset mock to clear any previous warning calls
mock_logger.warning.reset_mock()

# Call _measure_power_and_energy directly - this would normally trigger warning
tracker._measure_power_and_energy()

# Verify that if warning was called, it wasn't the scheduler warning
if mock_logger.warning.called:
for call in mock_logger.warning.call_args_list:
args, kwargs = call
if (
args
and "Background scheduler didn't run for a long period"
in str(args[0])
):
self.fail(
"Scheduler warning was called when it should have been suppressed"
)

@mock.patch("codecarbon.emissions_tracker.logger")
def test_scheduler_warning_shown_when_running(
self,
mock_logger,
mock_setup_intel_cli,
mock_log_values,
mocked_get_gpu_details,
mocked_env_cloud_details,
mocked_is_gpu_details_available,
):
"""Test that scheduler warning is shown when scheduler is running but delayed."""
with EmissionsTracker(
output_dir=self.temp_path,
measure_power_secs=1, # Short interval for testing
) as tracker:
# Ensure scheduler is running (default state)
self.assertFalse(tracker._scheduler._stopped)

# Artificially set last measured time to simulate long delay
import time

tracker._last_measured_time = time.perf_counter() - 10 # 10 seconds ago

# Reset mock to clear any previous warning calls
mock_logger.warning.reset_mock()

# Call _measure_power_and_energy directly - this should trigger warning
tracker._measure_power_and_energy()

# Verify warning was logged since scheduler should be running
scheduler_warning_found = False
if mock_logger.warning.called:
for call in mock_logger.warning.call_args_list:
args, kwargs = call
if (
args
and "Background scheduler didn't run for a long period"
in str(args[0])
):
scheduler_warning_found = True
break

self.assertTrue(
scheduler_warning_found, "Expected scheduler warning was not found"
)
Loading