|
| 1 | +import pytest |
| 2 | +from humanloop import Humanloop, FileType |
| 3 | +from pathlib import Path |
| 4 | +from typing import List, NamedTuple |
| 5 | + |
| 6 | + |
| 7 | +class SyncableFile(NamedTuple): |
| 8 | + path: str |
| 9 | + type: FileType |
| 10 | + model: str |
| 11 | + id: str = "" |
| 12 | + |
| 13 | + |
| 14 | +@pytest.fixture |
| 15 | +def test_file_structure(humanloop_client: Humanloop, get_test_path) -> List[SyncableFile]: |
| 16 | + """Creates a predefined structure of files in Humanloop for testing sync""" |
| 17 | + files: List[SyncableFile] = [ |
| 18 | + SyncableFile( |
| 19 | + path="prompts/gpt-4", |
| 20 | + type="prompt", |
| 21 | + model="gpt-4", |
| 22 | + ), |
| 23 | + SyncableFile( |
| 24 | + path="prompts/gpt-4o", |
| 25 | + type="prompt", |
| 26 | + model="gpt-4o", |
| 27 | + ), |
| 28 | + SyncableFile( |
| 29 | + path="prompts/nested/complex/gpt-4o", |
| 30 | + type="prompt", |
| 31 | + model="gpt-4o", |
| 32 | + ), |
| 33 | + SyncableFile( |
| 34 | + path="agents/gpt-4", |
| 35 | + type="agent", |
| 36 | + model="gpt-4", |
| 37 | + ), |
| 38 | + SyncableFile( |
| 39 | + path="agents/gpt-4o", |
| 40 | + type="agent", |
| 41 | + model="gpt-4o", |
| 42 | + ), |
| 43 | + ] |
| 44 | + |
| 45 | + # Create the files in Humanloop |
| 46 | + created_files = [] |
| 47 | + for file in files: |
| 48 | + full_path = get_test_path(file.path) |
| 49 | + if file.type == "prompt": |
| 50 | + response = humanloop_client.prompts.upsert( |
| 51 | + path=full_path, |
| 52 | + model=file.model, |
| 53 | + ) |
| 54 | + elif file.type == "agent": |
| 55 | + response = humanloop_client.agents.upsert( |
| 56 | + path=full_path, |
| 57 | + model=file.model, |
| 58 | + ) |
| 59 | + created_files.append(SyncableFile(path=full_path, type=file.type, model=file.model, id=response.id)) |
| 60 | + |
| 61 | + return created_files |
| 62 | + |
| 63 | + |
| 64 | +@pytest.fixture |
| 65 | +def cleanup_local_files(): |
| 66 | + """Cleanup any locally synced files after tests""" |
| 67 | + yield |
| 68 | + # Clean up the local humanloop directory after tests |
| 69 | + local_dir = Path("humanloop") |
| 70 | + if local_dir.exists(): |
| 71 | + import shutil |
| 72 | + |
| 73 | + shutil.rmtree(local_dir) |
| 74 | + |
| 75 | + |
| 76 | +def test_sync_basic(humanloop_client: Humanloop, test_file_structure: List[SyncableFile], cleanup_local_files): |
| 77 | + """Test that humanloop.sync() correctly syncs remote files to local filesystem""" |
| 78 | + # Run the sync |
| 79 | + successful_files = humanloop_client.sync() |
| 80 | + |
| 81 | + # Verify each file was synced correctly |
| 82 | + for file in test_file_structure: |
| 83 | + # Get the extension based on file type: .prompt, .agent |
| 84 | + extension = f".{file.type}" |
| 85 | + |
| 86 | + # The local path should mirror the remote path structure |
| 87 | + local_path = Path("humanloop") / f"{file.path}{extension}" |
| 88 | + |
| 89 | + # Basic assertions |
| 90 | + assert local_path.exists(), f"Expected synced file at {local_path}" |
| 91 | + assert local_path.parent.exists(), f"Expected directory at {local_path.parent}" |
| 92 | + |
| 93 | + # Verify it's not empty |
| 94 | + content = local_path.read_text() |
| 95 | + assert content, f"File at {local_path} should not be empty" |
0 commit comments