Clean up test resources.
(self)
| 40 | self.scheduler = None |
| 41 | |
| 42 | def tearDown(self) -> None: |
| 43 | """Clean up test resources.""" |
| 44 | |
| 45 | def cleanup(scheduler): |
| 46 | try: |
| 47 | scheduler.stop() |
| 48 | remove_atexit_callback(scheduler) |
| 49 | except Exception as e: |
| 50 | print( |
| 51 | f'Warning: Could not clean up scheduler {scheduler.repo_id}: {e}' |
| 52 | ) |
| 53 | |
| 54 | def remove_atexit_callback(scheduler): |
| 55 | try: |
| 56 | atexit.unregister(scheduler.commit_scheduled_changes) |
| 57 | except Exception as e: |
| 58 | print( |
| 59 | f'Warning: Could not remove atexit callback for scheduler {scheduler.repo_id}: {e}' |
| 60 | ) |
| 61 | |
| 62 | # Stop scheduler if it exists |
| 63 | if self.scheduler is not None: |
| 64 | try: |
| 65 | cleanup(self.scheduler) |
| 66 | except Exception: |
| 67 | pass |
| 68 | |
| 69 | # Try to delete test repo (may not exist for mocked tests) |
| 70 | try: |
| 71 | if hasattr(self, 'api') and TEST_ACCESS_TOKEN1: |
| 72 | self.api.login(TEST_ACCESS_TOKEN1) |
| 73 | self.api.delete_repo(repo_id=self.repo_id, repo_type='dataset') |
| 74 | except Exception as e: |
| 75 | logger.warning(f'Failed to delete test repo {self.repo_id}: {e}') |
| 76 | |
| 77 | # Clean up temporary directories |
| 78 | if self.cache_dir.exists(): |
| 79 | shutil.rmtree(self.cache_dir, ignore_errors=True) |
| 80 | |
| 81 | delete_credential() |
| 82 | |
| 83 | @unittest.skipUnless(test_level() >= 2, 'skip test in current test level') |
| 84 | def test_invalid_folder_path_nonexistent(self) -> None: |
nothing calls this directly
no test coverage detected