|
| 1 | +import subprocess |
| 2 | +import time |
| 3 | +from unittest.mock import patch |
| 4 | +import sys |
| 5 | +import os |
| 6 | +import json |
| 7 | +import requests |
| 8 | +from datetime import datetime |
| 9 | +import pytest |
| 10 | +import importlib.util |
| 11 | +import traceback |
| 12 | + |
| 13 | + |
| 14 | +class CaltechDataTester: |
| 15 | + def __init__(self): |
| 16 | + # Use GitHub Actions environment or create a local test directory |
| 17 | + self.test_dir = os.environ.get( |
| 18 | + "GITHUB_WORKSPACE", os.path.join(os.getcwd(), "caltech_test_data") |
| 19 | + ) |
| 20 | + self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| 21 | + |
| 22 | + # Ensure test directory exists |
| 23 | + os.makedirs(self.test_dir, exist_ok=True) |
| 24 | + |
| 25 | + # Create test run directory |
| 26 | + self.test_run_dir = os.path.join(self.test_dir, f"test_run_{self.timestamp}") |
| 27 | + os.makedirs(self.test_run_dir, exist_ok=True) |
| 28 | + |
| 29 | + # Initialize logging |
| 30 | + self.log_file = os.path.join(self.test_run_dir, "test_log.txt") |
| 31 | + |
| 32 | + def log(self, message): |
| 33 | + """Log message to both console and file""" |
| 34 | + print(message) |
| 35 | + with open(self.log_file, "a") as f: |
| 36 | + f.write(f"{datetime.now()}: {message}\n") |
| 37 | + |
| 38 | + def create_test_files(self): |
| 39 | + """Create necessary test files""" |
| 40 | + csv_path = os.path.join(self.test_run_dir, "test_data.csv") |
| 41 | + with open(csv_path, "w") as f: |
| 42 | + f.write("date,temperature,humidity\n") |
| 43 | + f.write("2023-01-01,25.5,60\n") |
| 44 | + f.write("2023-01-02,26.0,62\n") |
| 45 | + f.write("2023-01-03,24.8,65\n") |
| 46 | + |
| 47 | + self.log(f"Created test CSV file: {csv_path}") |
| 48 | + return csv_path |
| 49 | + |
| 50 | + def import_cli_module(self): |
| 51 | + """Dynamically import cli module from the correct path""" |
| 52 | + cli_path = os.path.join( |
| 53 | + os.environ.get("GITHUB_WORKSPACE", os.getcwd()), "caltechdata_api", "cli.py" |
| 54 | + ) |
| 55 | + spec = importlib.util.spec_from_file_location("cli", cli_path) |
| 56 | + cli_module = importlib.util.module_from_spec(spec) |
| 57 | + spec.loader.exec_module(cli_module) |
| 58 | + return cli_module |
| 59 | + |
| 60 | + def generate_test_responses(self): |
| 61 | + """Generate test responses for CLI prompts""" |
| 62 | + return { |
| 63 | + "Do you want to create or edit a CaltechDATA record? (create/edit): ": "create", |
| 64 | + "Do you want to use metadata from an existing file or create new metadata? (existing/create): ": "create", |
| 65 | + "Enter the title of the dataset: ": f"Test Dataset {self.timestamp}", |
| 66 | + "Enter the abstract or description of the dataset: ": "This is an automated test dataset containing sample climate data for validation purposes.", |
| 67 | + "Enter the number corresponding to the desired license: ": "1", |
| 68 | + "Enter your ORCID identifier: ": os.environ.get( |
| 69 | + "TEST_ORCID", "0000-0002-1825-0097" |
| 70 | + ), |
| 71 | + "How many funding entries do you want to provide? ": "1", |
| 72 | + "Enter the award number for funding: ": "NSF-1234567", |
| 73 | + "Enter the award title for funding: ": "Automated Testing Grant", |
| 74 | + "Enter the funder ROR (https://ror.org): ": "021nxhr62", |
| 75 | + "Do you want to upload or link data files? (upload/link/n): ": "upload", |
| 76 | + "Enter the filename to upload as a supporting file (or 'n' to finish): ": "test_data.csv", |
| 77 | + "Do you want to add more files? (y/n): ": "n", |
| 78 | + "Do you want to send this record to CaltechDATA? (y/n): ": "y", |
| 79 | + } |
| 80 | + |
| 81 | + def run_test_submission(self): |
| 82 | + """Run the complete test submission process""" |
| 83 | + try: |
| 84 | + self.log("Starting test submission process...") |
| 85 | + |
| 86 | + # Create test files |
| 87 | + test_csv = self.create_test_files() |
| 88 | + |
| 89 | + # Dynamically import cli module |
| 90 | + cli_module = self.import_cli_module() |
| 91 | + |
| 92 | + # Generate responses |
| 93 | + responses = self.generate_test_responses() |
| 94 | + |
| 95 | + # Setup output capture |
| 96 | + class OutputCapture: |
| 97 | + def __init__(self): |
| 98 | + self.output = [] |
| 99 | + |
| 100 | + def write(self, text): |
| 101 | + self.output.append(text) |
| 102 | + sys.__stdout__.write(text) |
| 103 | + |
| 104 | + def flush(self): |
| 105 | + pass |
| 106 | + |
| 107 | + def get_output(self): |
| 108 | + return "".join(self.output) |
| 109 | + |
| 110 | + output_capture = OutputCapture() |
| 111 | + sys.stdout = output_capture |
| 112 | + |
| 113 | + # Mock input and run CLI |
| 114 | + def mock_input(prompt): |
| 115 | + self.log(f"Prompt: {prompt}") |
| 116 | + if prompt in responses: |
| 117 | + response = responses[prompt] |
| 118 | + self.log(f"Response: {response}") |
| 119 | + return response |
| 120 | + return "" |
| 121 | + |
| 122 | + with patch("builtins.input", side_effect=mock_input): |
| 123 | + # Use -test flag to use test mode |
| 124 | + sys.argv = [sys.argv[0], "-test"] |
| 125 | + cli_module.main() |
| 126 | + |
| 127 | + # Restore stdout |
| 128 | + sys.stdout = sys.__stdout__ |
| 129 | + |
| 130 | + return True |
| 131 | + |
| 132 | + except Exception as e: |
| 133 | + self.log(f"Error in test submission: {e}") |
| 134 | + traceback.print_exc() |
| 135 | + return False |
| 136 | + finally: |
| 137 | + # Cleanup |
| 138 | + if "test_csv" in locals() and os.path.exists(test_csv): |
| 139 | + os.remove(test_csv) |
| 140 | + self.log("Test files cleaned up") |
| 141 | + |
| 142 | + |
| 143 | +def main(): |
| 144 | + tester = CaltechDataTester() |
| 145 | + |
| 146 | + success = tester.run_test_submission() |
| 147 | + |
| 148 | + if success: |
| 149 | + tester.log("\n🎉 Test submission completed successfully!") |
| 150 | + sys.exit(0) |
| 151 | + else: |
| 152 | + tester.log("\n❌ Test submission failed - check logs for details") |
| 153 | + sys.exit(1) |
| 154 | + |
| 155 | + |
| 156 | +if __name__ == "__main__": |
| 157 | + main() |
0 commit comments