Buckets:
| #!/usr/bin/env python3 | |
| import sys | |
| import os | |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src")) | |
| def test_dataset(): | |
| print("🧪 Testing dataset construction...") | |
| from dataset.tta_dataset import ( | |
| construct_demonstrations, | |
| apply_task_rule, | |
| AbstractTaskRule, | |
| ) | |
| demos = construct_demonstrations() | |
| print(f" Generated {len(demos['demonstrations'])} demonstrations") | |
| result = apply_task_rule("hello", AbstractTaskRule.ReverseAppendLastChar) | |
| assert result == "olleho", f"Expected 'olleho', got '{result}'" | |
| print(" ✅ Task rule works correctly") | |
| return True | |
| def test_settings(): | |
| print("⚙️ Testing configuration...") | |
| from utils.settings import ModelArgs, DataArgs, EvaluatorArgs | |
| m_args = ModelArgs() | |
| d_args = DataArgs() | |
| e_args = EvaluatorArgs() | |
| assert m_args.model_name_or_path == "gpt2" | |
| assert len(d_args.demonstrations) == 4 | |
| print(" ✅ Configuration classes work") | |
| return True | |
| def test_utils(): | |
| print("🔧 Testing utilities...") | |
| from utils.io import ensure_dir, SafeDictAccessReadOnly | |
| from utils.logging import ExperimentLogger | |
| ensure_dir("test_dir") | |
| assert os.path.exists("test_dir") | |
| os.rmdir("test_dir") | |
| test_dict = {"a": {"b": "c"}} | |
| safe_dict = SafeDictAccessReadOnly(test_dict) | |
| assert safe_dict.get("a.b") == "c" | |
| logger = ExperimentLogger.setup("test_logs", "test_run") | |
| logger.log({"test": "value"}) | |
| print(" ✅ Utilities work correctly") | |
| return True | |
| def test_viral(): | |
| print("🌟 Testing viral examples...") | |
| from viral.viral_examples import integrate_viral_examples | |
| base_examples = ["test", "example"] | |
| viral_examples = integrate_viral_examples(base_examples, max_more=2) | |
| assert len(viral_examples) >= len(base_examples) | |
| print(f" ✅ Generated {len(viral_examples)} examples") | |
| return True | |
| def main(): | |
| print("🚀 Running TTT Pipeline Component Tests") | |
| print("=" * 50) | |
| tests = [test_dataset, test_settings, test_utils, test_viral] | |
| passed = 0 | |
| for test in tests: | |
| try: | |
| if test(): | |
| passed += 1 | |
| except Exception as e: | |
| print(f" ❌ {test.__name__} failed: {e}") | |
| print("\n" + "=" * 50) | |
| print(f"📊 Test Results: {passed}/{len(tests)} passed") | |
| if passed == len(tests): | |
| print("🎉 All components working! Ready to run TTT pipeline.") | |
| return True | |
| else: | |
| print("⚠️ Some components need fixing.") | |
| return False | |
| if __name__ == "__main__": | |
| success = main() | |
| sys.exit(0 if success else 1) | |
Xet Storage Details
- Size:
- 2.64 kB
- Xet hash:
- e0cb49867340fa2fd35cb8f67156dcc86de06573a56173f0e2127023f163f09e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.