Skip to content
Closed
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
60 changes: 60 additions & 0 deletions sagemaker-core/tests/unit/workflow/test_utilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,35 @@ def test_get_processing_code_hash_s3_uri(self):
assert result is None

def test_get_processing_code_hash_with_dependencies(self):

def test_get_processing_code_hash_with_none_dependencies(self):
"""Test get_processing_code_hash with None dependencies does not raise TypeError"""
with tempfile.TemporaryDirectory() as temp_dir:
code_file = Path(temp_dir, "script.py")
code_file.write_text("print('hello')")

# Should not raise TypeError when dependencies is None
result = get_processing_code_hash(
code=str(code_file), source_dir=temp_dir, dependencies=None
)

assert result is not None
assert len(result) == 64

def test_get_processing_code_hash_code_only_with_none_dependencies(self):
"""Test get_processing_code_hash with code only and None dependencies"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write("print('hello')")
temp_file = f.name

try:
result = get_processing_code_hash(code=temp_file, source_dir=None, dependencies=None)

assert result is not None
assert len(result) == 64
finally:
os.unlink(temp_file)

"""Test get_processing_code_hash with dependencies"""
with tempfile.TemporaryDirectory() as temp_dir:
code_file = Path(temp_dir, "script.py")
Expand Down Expand Up @@ -317,6 +346,37 @@ def test_get_training_code_hash_s3_uri(self):
assert result is None

def test_get_training_code_hash_pipeline_variable(self):

def test_get_training_code_hash_with_none_dependencies_source_dir(self):
"""Test get_training_code_hash with None dependencies and source_dir"""
with tempfile.TemporaryDirectory() as temp_dir:
entry_file = Path(temp_dir, "train.py")
entry_file.write_text("print('training')")

# Should not raise TypeError when dependencies is None
result = get_training_code_hash(
entry_point=str(entry_file), source_dir=temp_dir, dependencies=None
)

assert result is not None
assert len(result) == 64

def test_get_training_code_hash_with_none_dependencies_entry_point_only(self):
"""Test get_training_code_hash with None dependencies and entry_point only"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write("print('training')")
temp_file = f.name

try:
result = get_training_code_hash(
entry_point=temp_file, source_dir=None, dependencies=None
)

assert result is not None
assert len(result) == 64
finally:
os.unlink(temp_file)

"""Test get_training_code_hash with pipeline variable returns None"""
with patch("sagemaker.core.workflow.is_pipeline_variable", return_value=True):
result = get_training_code_hash(
Expand Down
Loading