Integration Tests¶
Integration tests validate that different components work together correctly, focusing on end-to-end workflows and component interactions.
Overview¶
The integration test suite covers:
- End-to-End Workflows - Complete processing pipelines from data input to final output
- Data Processing Pipelines - Multi-step geospatial data processing workflows
- Parallel Processing - Testing concurrent operations and thread safety
- Project Flow Integration - Testing the ProjectFlow framework with real workflows
End-to-End Workflow Testing¶
Comprehensive tests that validate complete processing workflows from start to finish.
Key End-to-End Test Cases: - ✅ Complete geospatial processing workflows - ✅ Data input through final output validation - ✅ Multi-component integration testing
Consolidated Integration Tests for End-to-End Workflows
This file consolidates tests from: - end_to_end_workflows/test_dummy_raster_integration.py - end_to_end_workflows/test_end_to_end_workflow.py
Covers comprehensive end-to-end workflow testing including:
- Complete pipeline: test file → analysis → quality assessment → QMD generation
- Multiple test file processing workflows
- Template system integration validation
- Plugin system end-to-end execution
- Configuration system integration
- Error handling across complete workflows
- Dummy raster generation and tiling validation
- Raster tiling operations with sum verification
- Integration reproducibility and consistency testing
- Complete workflow validation from creation through verification
QMD_COMPONENTS_AVAILABLE
¶
TestFileMetadata
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
__init__(self, file_path, category = None)
special
¶
ProcessingResult
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
__init__(self, success = True)
special
¶
TestAnalysisEngine
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
class TestAnalysisEngine:
def analyze_test_file(self, file_path):
metadata = TestFileMetadata(file_path)
# Basic analysis - count test functions
with open(file_path, 'r') as f:
content = f.read()
metadata.content = content
import re
test_functions = re.findall(r'def (test_\w+)', content)
metadata.test_functions = test_functions
return metadata
analyze_test_file(self, file_path)
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def analyze_test_file(self, file_path):
metadata = TestFileMetadata(file_path)
# Basic analysis - count test functions
with open(file_path, 'r') as f:
content = f.read()
metadata.content = content
import re
test_functions = re.findall(r'def (test_\w+)', content)
metadata.test_functions = test_functions
return metadata
TestCategory
¶
QualityCategory
¶
QualityAssessment
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
__init__(self)
special
¶
QualityAssessmentEngine
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
assess_quality(self, metadata)
¶
PluginManager
¶
TemplateSystem
¶
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
render_qmd(self, metadata, template_name = 'default')
¶
TestDummyRasterGeneration (TestCase)
¶
Test dummy raster generation utilities for integration testing (from test_dummy_raster_integration.py)
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
class TestDummyRasterGeneration(unittest.TestCase):
"""Test dummy raster generation utilities for integration testing (from test_dummy_raster_integration.py)"""
@pytest.fixture(autouse=True)
def setup_temp_dir(self):
"""Setup temporary directory for tests"""
self.temp_dir = tempfile.mkdtemp()
yield
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_create_dummy_raster_basic(self):
"""Test basic dummy raster creation with known values"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_basic.tif")
width, height = 100, 100
cell_size = 0.1
# Create dummy raster using utility function
hb.create_dummy_raster(
output_path=output_path,
width=width,
height=height,
cell_size=cell_size,
data_type=gdal.GDT_Float32,
fill_value=42.0,
nodata_value=-999.0
)
# Verify file exists
assert os.path.exists(output_path), "Dummy raster file should be created"
# Verify raster properties
dataset = gdal.Open(output_path)
assert dataset is not None, "Raster should be readable"
assert dataset.RasterXSize == width, f"Width should be {width}"
assert dataset.RasterYSize == height, f"Height should be {height}"
# Verify geotransform
geotransform = dataset.GetGeoTransform()
assert abs(geotransform[1] - cell_size) < 1e-6, f"Pixel size should be {cell_size}"
assert abs(geotransform[5] + cell_size) < 1e-6, f"Pixel size should be {cell_size}"
# Verify data values
band = dataset.GetRasterBand(1)
assert band.GetNoDataValue() == -999.0, "NoData value should be set correctly"
array = band.ReadAsArray()
assert np.all(array == 42.0), "All pixels should have fill value"
dataset = None # Close dataset
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
def test_create_dummy_raster_with_pattern(self):
"""Test dummy raster creation with mathematical pattern"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_pattern.tif")
width, height = 50, 50
# Create raster with gradient pattern for known sum calculation
hb.create_dummy_raster_with_pattern(
output_path=output_path,
width=width,
height=height,
pattern_type="gradient",
cell_size=1.0
)
# Verify file and basic properties
assert os.path.exists(output_path), "Pattern raster should be created"
dataset = gdal.Open(output_path)
band = dataset.GetRasterBand(1)
array = band.ReadAsArray()
# Verify gradient pattern (each row should have incrementing values)
expected_sum = sum(i * width for i in range(height)) # 0*50 + 1*50 + 2*50 + ... + 49*50
actual_sum = np.sum(array)
assert abs(actual_sum - expected_sum) < 1e-6, "Gradient pattern sum should be calculable"
assert array[0, 0] == 0, "Top-left should be 0 in gradient"
assert array[-1, -1] == height - 1, "Bottom-right should be height-1"
dataset = None
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
def test_create_dummy_raster_with_known_sum(self):
"""Test dummy raster creation with predetermined sum for validation"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_known_sum.tif")
width, height = 10, 10
target_sum = 1000.0
# Create raster with known total sum
hb.create_dummy_raster_with_known_sum(
output_path=output_path,
width=width,
height=height,
target_sum=target_sum,
cell_size=0.5
)
# Verify file creation
assert os.path.exists(output_path), "Known sum raster should be created"
# Verify sum calculation
dataset = gdal.Open(output_path)
band = dataset.GetRasterBand(1)
array = band.ReadAsArray()
actual_sum = np.sum(array)
assert abs(actual_sum - target_sum) < 1e-6, f"Sum should be exactly {target_sum}"
# Verify reasonable value distribution
assert np.all(array > 0), "All values should be positive for sum verification"
assert array.shape == (height, width), "Array shape should match dimensions"
dataset = None
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
setup_temp_dir(self)
¶
test_create_dummy_raster_basic(self)
¶
Test basic dummy raster creation with known values
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_create_dummy_raster_basic(self):
"""Test basic dummy raster creation with known values"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_basic.tif")
width, height = 100, 100
cell_size = 0.1
# Create dummy raster using utility function
hb.create_dummy_raster(
output_path=output_path,
width=width,
height=height,
cell_size=cell_size,
data_type=gdal.GDT_Float32,
fill_value=42.0,
nodata_value=-999.0
)
# Verify file exists
assert os.path.exists(output_path), "Dummy raster file should be created"
# Verify raster properties
dataset = gdal.Open(output_path)
assert dataset is not None, "Raster should be readable"
assert dataset.RasterXSize == width, f"Width should be {width}"
assert dataset.RasterYSize == height, f"Height should be {height}"
# Verify geotransform
geotransform = dataset.GetGeoTransform()
assert abs(geotransform[1] - cell_size) < 1e-6, f"Pixel size should be {cell_size}"
assert abs(geotransform[5] + cell_size) < 1e-6, f"Pixel size should be {cell_size}"
# Verify data values
band = dataset.GetRasterBand(1)
assert band.GetNoDataValue() == -999.0, "NoData value should be set correctly"
array = band.ReadAsArray()
assert np.all(array == 42.0), "All pixels should have fill value"
dataset = None # Close dataset
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_create_dummy_raster_with_pattern(self)
¶
Test dummy raster creation with mathematical pattern
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_create_dummy_raster_with_pattern(self):
"""Test dummy raster creation with mathematical pattern"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_pattern.tif")
width, height = 50, 50
# Create raster with gradient pattern for known sum calculation
hb.create_dummy_raster_with_pattern(
output_path=output_path,
width=width,
height=height,
pattern_type="gradient",
cell_size=1.0
)
# Verify file and basic properties
assert os.path.exists(output_path), "Pattern raster should be created"
dataset = gdal.Open(output_path)
band = dataset.GetRasterBand(1)
array = band.ReadAsArray()
# Verify gradient pattern (each row should have incrementing values)
expected_sum = sum(i * width for i in range(height)) # 0*50 + 1*50 + 2*50 + ... + 49*50
actual_sum = np.sum(array)
assert abs(actual_sum - expected_sum) < 1e-6, "Gradient pattern sum should be calculable"
assert array[0, 0] == 0, "Top-left should be 0 in gradient"
assert array[-1, -1] == height - 1, "Bottom-right should be height-1"
dataset = None
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_create_dummy_raster_with_known_sum(self)
¶
Test dummy raster creation with predetermined sum for validation
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_create_dummy_raster_with_known_sum(self):
"""Test dummy raster creation with predetermined sum for validation"""
temp_dir = tempfile.mkdtemp()
try:
output_path = os.path.join(temp_dir, "dummy_known_sum.tif")
width, height = 10, 10
target_sum = 1000.0
# Create raster with known total sum
hb.create_dummy_raster_with_known_sum(
output_path=output_path,
width=width,
height=height,
target_sum=target_sum,
cell_size=0.5
)
# Verify file creation
assert os.path.exists(output_path), "Known sum raster should be created"
# Verify sum calculation
dataset = gdal.Open(output_path)
band = dataset.GetRasterBand(1)
array = band.ReadAsArray()
actual_sum = np.sum(array)
assert abs(actual_sum - target_sum) < 1e-6, f"Sum should be exactly {target_sum}"
# Verify reasonable value distribution
assert np.all(array > 0), "All values should be positive for sum verification"
assert array.shape == (height, width), "Array shape should match dimensions"
dataset = None
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
TestRasterTilingIntegration (TestCase)
¶
Test raster tiling operations with sum verification
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
class TestRasterTilingIntegration(unittest.TestCase):
"""Test raster tiling operations with sum verification"""
def test_tile_dummy_raster_sum_conservation(self):
"""Test that tiling preserves total sum of values"""
temp_dir = tempfile.mkdtemp()
try:
# Create dummy raster with known sum
base_raster = os.path.join(temp_dir, "base_for_tiling.tif")
width, height = 100, 100
target_sum = 5000.0
hb.create_dummy_raster_with_known_sum(
output_path=base_raster,
width=width,
height=height,
target_sum=target_sum,
cell_size=1.0
)
# Get original sum
original_array = hb.as_array(base_raster)
original_sum = np.sum(original_array)
# Tile the raster
tile_size = 25 # Create 4x4 = 16 tiles
tiles_dir = os.path.join(temp_dir, "tiles")
os.makedirs(tiles_dir, exist_ok=True)
tile_paths = hb.tile_raster_into_grid(
input_raster_path=base_raster,
output_dir=tiles_dir,
tile_size=tile_size,
overlap=0
)
# Verify tiles were created
assert len(tile_paths) > 0, "Tiles should be created"
# Sum all tile values
total_tiles_sum = 0.0
valid_tiles = 0
for tile_path in tile_paths:
if os.path.exists(tile_path):
tile_array = hb.as_array(tile_path)
tile_sum = np.sum(tile_array[tile_array != hb.get_ndv_from_path(tile_path)])
total_tiles_sum += tile_sum
valid_tiles += 1
# Verify sum conservation
assert valid_tiles > 0, "At least one valid tile should exist"
sum_difference = abs(total_tiles_sum - original_sum)
tolerance = original_sum * 0.001 # 0.1% tolerance for floating point precision
assert sum_difference < tolerance, (
f"Sum should be conserved: original={original_sum}, "
f"tiles_total={total_tiles_sum}, difference={sum_difference}"
)
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_tile_dummy_raster_sum_conservation(self)
¶
Test that tiling preserves total sum of values
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_tile_dummy_raster_sum_conservation(self):
"""Test that tiling preserves total sum of values"""
temp_dir = tempfile.mkdtemp()
try:
# Create dummy raster with known sum
base_raster = os.path.join(temp_dir, "base_for_tiling.tif")
width, height = 100, 100
target_sum = 5000.0
hb.create_dummy_raster_with_known_sum(
output_path=base_raster,
width=width,
height=height,
target_sum=target_sum,
cell_size=1.0
)
# Get original sum
original_array = hb.as_array(base_raster)
original_sum = np.sum(original_array)
# Tile the raster
tile_size = 25 # Create 4x4 = 16 tiles
tiles_dir = os.path.join(temp_dir, "tiles")
os.makedirs(tiles_dir, exist_ok=True)
tile_paths = hb.tile_raster_into_grid(
input_raster_path=base_raster,
output_dir=tiles_dir,
tile_size=tile_size,
overlap=0
)
# Verify tiles were created
assert len(tile_paths) > 0, "Tiles should be created"
# Sum all tile values
total_tiles_sum = 0.0
valid_tiles = 0
for tile_path in tile_paths:
if os.path.exists(tile_path):
tile_array = hb.as_array(tile_path)
tile_sum = np.sum(tile_array[tile_array != hb.get_ndv_from_path(tile_path)])
total_tiles_sum += tile_sum
valid_tiles += 1
# Verify sum conservation
assert valid_tiles > 0, "At least one valid tile should exist"
sum_difference = abs(total_tiles_sum - original_sum)
tolerance = original_sum * 0.001 # 0.1% tolerance for floating point precision
assert sum_difference < tolerance, (
f"Sum should be conserved: original={original_sum}, "
f"tiles_total={total_tiles_sum}, difference={sum_difference}"
)
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
TestIntegrationReproducibility (TestCase)
¶
Test reproducibility and consistency of integration operations
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
class TestIntegrationReproducibility(unittest.TestCase):
"""Test reproducibility and consistency of integration operations"""
def test_dummy_raster_reproducibility(self):
"""Test that dummy raster generation is reproducible with same parameters"""
temp_dir = tempfile.mkdtemp()
try:
# Create two identical rasters with same parameters
params = {
'width': 50,
'height': 50,
'cell_size': 0.5,
'pattern_type': 'gradient'
}
raster1_path = os.path.join(temp_dir, "reproducible1.tif")
raster2_path = os.path.join(temp_dir, "reproducible2.tif")
# Create identical rasters
hb.create_dummy_raster_with_pattern(
output_path=raster1_path,
**params
)
hb.create_dummy_raster_with_pattern(
output_path=raster2_path,
**params
)
# Compare arrays
array1 = hb.as_array(raster1_path)
array2 = hb.as_array(raster2_path)
assert np.array_equal(array1, array2), "Identical parameters should produce identical rasters"
assert np.sum(array1) == np.sum(array2), "Sums should be identical"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_dummy_raster_reproducibility(self)
¶
Test that dummy raster generation is reproducible with same parameters
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_dummy_raster_reproducibility(self):
"""Test that dummy raster generation is reproducible with same parameters"""
temp_dir = tempfile.mkdtemp()
try:
# Create two identical rasters with same parameters
params = {
'width': 50,
'height': 50,
'cell_size': 0.5,
'pattern_type': 'gradient'
}
raster1_path = os.path.join(temp_dir, "reproducible1.tif")
raster2_path = os.path.join(temp_dir, "reproducible2.tif")
# Create identical rasters
hb.create_dummy_raster_with_pattern(
output_path=raster1_path,
**params
)
hb.create_dummy_raster_with_pattern(
output_path=raster2_path,
**params
)
# Compare arrays
array1 = hb.as_array(raster1_path)
array2 = hb.as_array(raster2_path)
assert np.array_equal(array1, array2), "Identical parameters should produce identical rasters"
assert np.sum(array1) == np.sum(array2), "Sums should be identical"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
TestEndToEndWorkflowValidation (TestCase)
¶
Test complete end-to-end workflows from dummy raster creation through tiling and verification
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
@pytest.mark.integration
@pytest.mark.slow
class TestEndToEndWorkflowValidation(unittest.TestCase):
"""Test complete end-to-end workflows from dummy raster creation through tiling and verification"""
def test_complete_integration_workflow(self):
"""Test complete workflow: create dummy raster -> tile -> verify -> aggregate results"""
temp_dir = tempfile.mkdtemp()
try:
# Step 1: Create test data with known properties
workflow_dir = os.path.join(temp_dir, "complete_workflow")
os.makedirs(workflow_dir, exist_ok=True)
base_raster = os.path.join(workflow_dir, "workflow_base.tif")
target_sum = 10000.0
hb.create_dummy_raster_with_known_sum(
output_path=base_raster,
width=80,
height=60,
target_sum=target_sum,
cell_size=0.25
)
# Step 2: Tile the raster
tiles_dir = os.path.join(workflow_dir, "tiles")
os.makedirs(tiles_dir, exist_ok=True)
tile_paths = hb.tile_raster_into_grid(
input_raster_path=base_raster,
output_dir=tiles_dir,
tile_size=20,
overlap=0
)
# Step 3: Verify each tile individually
tile_metadata = []
total_from_tiles = 0.0
for i, tile_path in enumerate(tile_paths):
if os.path.exists(tile_path):
tile_array = hb.as_array(tile_path)
nodata_value = hb.get_ndv_from_path(tile_path)
if nodata_value is not None:
valid_data = tile_array[tile_array != nodata_value]
else:
valid_data = tile_array.flatten()
tile_sum = float(np.sum(valid_data))
tile_mean = float(np.mean(valid_data)) if len(valid_data) > 0 else 0.0
tile_metadata.append({
'tile_id': i,
'path': tile_path,
'sum': tile_sum,
'mean': tile_mean,
'valid_pixels': len(valid_data),
'shape': list(tile_array.shape)
})
total_from_tiles += tile_sum
# Step 4: Validate workflow results
assert len(tile_metadata) > 0, "Should have created valid tiles"
# Sum conservation check
sum_difference = abs(total_from_tiles - target_sum)
tolerance = target_sum * 0.001 # 0.1% tolerance
assert sum_difference < tolerance, (
f"End-to-end sum conservation failed: "
f"expected={target_sum}, actual={total_from_tiles}, diff={sum_difference}"
)
# Verify tile coverage (all pixels accounted for)
total_valid_pixels = sum(meta['valid_pixels'] for meta in tile_metadata)
expected_pixels = 80 * 60 # width * height
assert total_valid_pixels == expected_pixels, (
f"Pixel count mismatch: expected={expected_pixels}, actual={total_valid_pixels}"
)
# Step 5: Generate workflow report
workflow_report = {
'timestamp': datetime.now().isoformat(),
'base_raster': {
'path': base_raster,
'target_sum': target_sum,
'dimensions': (80, 60),
'cell_size': 0.25
},
'tiling_results': {
'tiles_created': len(tile_metadata),
'total_sum_from_tiles': float(total_from_tiles),
'sum_conservation_error': float(sum_difference),
'coverage_pixels': total_valid_pixels
},
'validation_status': 'PASSED' if sum_difference < tolerance else 'FAILED',
'tile_details': tile_metadata
}
# Save report for analysis
report_path = os.path.join(workflow_dir, "integration_workflow_report.json")
with open(report_path, 'w') as f:
json.dump(workflow_report, f, indent=2)
# Final assertion: workflow completed successfully
assert workflow_report['validation_status'] == 'PASSED', "Complete workflow should pass validation"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_complete_integration_workflow(self)
¶
Test complete workflow: create dummy raster -> tile -> verify -> aggregate results
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_complete_integration_workflow(self):
"""Test complete workflow: create dummy raster -> tile -> verify -> aggregate results"""
temp_dir = tempfile.mkdtemp()
try:
# Step 1: Create test data with known properties
workflow_dir = os.path.join(temp_dir, "complete_workflow")
os.makedirs(workflow_dir, exist_ok=True)
base_raster = os.path.join(workflow_dir, "workflow_base.tif")
target_sum = 10000.0
hb.create_dummy_raster_with_known_sum(
output_path=base_raster,
width=80,
height=60,
target_sum=target_sum,
cell_size=0.25
)
# Step 2: Tile the raster
tiles_dir = os.path.join(workflow_dir, "tiles")
os.makedirs(tiles_dir, exist_ok=True)
tile_paths = hb.tile_raster_into_grid(
input_raster_path=base_raster,
output_dir=tiles_dir,
tile_size=20,
overlap=0
)
# Step 3: Verify each tile individually
tile_metadata = []
total_from_tiles = 0.0
for i, tile_path in enumerate(tile_paths):
if os.path.exists(tile_path):
tile_array = hb.as_array(tile_path)
nodata_value = hb.get_ndv_from_path(tile_path)
if nodata_value is not None:
valid_data = tile_array[tile_array != nodata_value]
else:
valid_data = tile_array.flatten()
tile_sum = float(np.sum(valid_data))
tile_mean = float(np.mean(valid_data)) if len(valid_data) > 0 else 0.0
tile_metadata.append({
'tile_id': i,
'path': tile_path,
'sum': tile_sum,
'mean': tile_mean,
'valid_pixels': len(valid_data),
'shape': list(tile_array.shape)
})
total_from_tiles += tile_sum
# Step 4: Validate workflow results
assert len(tile_metadata) > 0, "Should have created valid tiles"
# Sum conservation check
sum_difference = abs(total_from_tiles - target_sum)
tolerance = target_sum * 0.001 # 0.1% tolerance
assert sum_difference < tolerance, (
f"End-to-end sum conservation failed: "
f"expected={target_sum}, actual={total_from_tiles}, diff={sum_difference}"
)
# Verify tile coverage (all pixels accounted for)
total_valid_pixels = sum(meta['valid_pixels'] for meta in tile_metadata)
expected_pixels = 80 * 60 # width * height
assert total_valid_pixels == expected_pixels, (
f"Pixel count mismatch: expected={expected_pixels}, actual={total_valid_pixels}"
)
# Step 5: Generate workflow report
workflow_report = {
'timestamp': datetime.now().isoformat(),
'base_raster': {
'path': base_raster,
'target_sum': target_sum,
'dimensions': (80, 60),
'cell_size': 0.25
},
'tiling_results': {
'tiles_created': len(tile_metadata),
'total_sum_from_tiles': float(total_from_tiles),
'sum_conservation_error': float(sum_difference),
'coverage_pixels': total_valid_pixels
},
'validation_status': 'PASSED' if sum_difference < tolerance else 'FAILED',
'tile_details': tile_metadata
}
# Save report for analysis
report_path = os.path.join(workflow_dir, "integration_workflow_report.json")
with open(report_path, 'w') as f:
json.dump(workflow_report, f, indent=2)
# Final assertion: workflow completed successfully
assert workflow_report['validation_status'] == 'PASSED', "Complete workflow should pass validation"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
TestQMDAutomationEndToEnd (TestCase)
¶
Test complete QMD automation workflows from test files to QMD output (from test_end_to_end_workflow.py)
This test suite validates complete workflows from test files to QMD output:
- Complete pipeline: test file → analysis → quality assessment → QMD generation
- Multiple test file processing workflows
- Template system integration validation
- Plugin system end-to-end execution
- Configuration system integration
- Error handling across complete workflows
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
class TestQMDAutomationEndToEnd(unittest.TestCase):
"""
Test complete QMD automation workflows from test files to QMD output
(from test_end_to_end_workflow.py)
This test suite validates complete workflows from test files to QMD output:
- Complete pipeline: test file → analysis → quality assessment → QMD generation
- Multiple test file processing workflows
- Template system integration validation
- Plugin system end-to-end execution
- Configuration system integration
- Error handling across complete workflows
"""
def setUp(self):
"""Set up test environment for QMD automation testing"""
self.temp_dir = tempfile.mkdtemp()
self.test_files_dir = os.path.join(self.temp_dir, "test_files")
self.output_dir = os.path.join(self.temp_dir, "qmd_output")
os.makedirs(self.test_files_dir)
os.makedirs(self.output_dir)
# Create sample test files for processing
self.create_sample_test_files()
# Initialize engines
self.analysis_engine = TestAnalysisEngine()
self.quality_engine = QualityAssessmentEngine()
self.plugin_manager = PluginManager()
self.template_system = TemplateSystem()
def tearDown(self):
"""Clean up test environment"""
shutil.rmtree(self.temp_dir, ignore_errors=True)
def create_sample_test_files(self):
"""Create various sample test files for end-to-end testing"""
# High-quality test file with comprehensive tests
high_quality_content = '''
"""
Comprehensive unit tests for mathematical operations module.
This module provides extensive testing coverage for mathematical operations
including basic arithmetic, advanced functions, and edge cases.
"""
import unittest
import math
import pytest
from typing import List, Union
class TestMathematicalOperations(unittest.TestCase):
"""Test suite for mathematical operations."""
def setUp(self):
"""Set up test fixtures."""
self.test_numbers = [1, 2, 3, 4, 5]
self.zero_value = 0
self.negative_numbers = [-1, -2, -3]
def test_addition_positive_numbers(self):
"""Test addition with positive numbers."""
result = 2 + 3
self.assertEqual(result, 5)
def test_addition_negative_numbers(self):
"""Test addition with negative numbers."""
result = -2 + (-3)
self.assertEqual(result, -5)
def test_addition_mixed_numbers(self):
"""Test addition with mixed positive and negative numbers."""
result = 5 + (-3)
self.assertEqual(result, 2)
def test_division_by_zero_raises_exception(self):
"""Test that division by zero raises ZeroDivisionError."""
with self.assertRaises(ZeroDivisionError):
result = 10 / 0
def test_square_root_positive_number(self):
"""Test square root of positive numbers."""
result = math.sqrt(16)
self.assertEqual(result, 4.0)
def test_power_operations(self):
"""Test various power operations."""
self.assertEqual(2**3, 8)
self.assertEqual(5**0, 1)
self.assertEqual(4**0.5, 2.0)
@pytest.mark.parametrize("x,y,expected", [
(2, 3, 5),
(0, 5, 5),
(-1, 1, 0),
(10, -5, 5)
])
def test_parametrized_addition(self, x, y, expected):
"""Parametrized test for addition operations."""
assert x + y == expected
if __name__ == '__main__':
unittest.main()
'''
# Medium-quality test file with some tests but limited documentation
medium_quality_content = '''
import unittest
class TestStringOperations(unittest.TestCase):
def test_string_upper(self):
result = "hello".upper()
self.assertEqual(result, "HELLO")
def test_string_lower(self):
result = "WORLD".lower()
self.assertEqual(result, "world")
def test_string_length(self):
text = "test"
self.assertEqual(len(text), 4)
if __name__ == '__main__':
unittest.main()
'''
# Low-quality test file (stub with minimal content)
low_quality_content = '''
def test_something():
pass
'''
# Integration test file
integration_content = '''
"""
Integration tests for database operations.
Tests the integration between the application and database layer.
"""
import unittest
import tempfile
import os
from unittest.mock import patch, MagicMock
class TestDatabaseIntegration(unittest.TestCase):
"""Integration tests for database operations."""
def setUp(self):
"""Set up test database."""
self.temp_db = tempfile.mktemp()
def tearDown(self):
"""Clean up test database."""
if os.path.exists(self.temp_db):
os.remove(self.temp_db)
def test_database_connection_integration(self):
"""Test database connection establishment."""
# Mock database connection
with patch('database.connect') as mock_connect:
mock_connect.return_value = MagicMock()
# Test integration logic here
assert True
def test_data_persistence_integration(self):
"""Test data persistence across operations."""
# Test data persistence
assert True
if __name__ == '__main__':
unittest.main()
'''
# Performance test file
performance_content = '''
"""
Performance benchmarks for critical operations.
Measures and validates performance requirements for key system functions.
"""
import time
import unittest
import pytest
class TestPerformanceBenchmarks(unittest.TestCase):
"""Performance benchmark test suite."""
@pytest.mark.benchmark
def test_operation_performance(self):
"""Benchmark critical operation performance."""
start_time = time.time()
# Simulate operation
for i in range(1000):
result = i ** 2
end_time = time.time()
duration = end_time - start_time
# Should complete in less than 1 second
self.assertLess(duration, 1.0)
def test_memory_usage_benchmark(self):
"""Test memory usage stays within bounds."""
# Mock memory usage test
assert True
if __name__ == '__main__':
unittest.main()
'''
# Write test files
test_files = [
("test_math_high_quality.py", high_quality_content),
("test_strings_medium_quality.py", medium_quality_content),
("test_stub_low_quality.py", low_quality_content),
("test_database_integration.py", integration_content),
("test_performance_benchmarks.py", performance_content)
]
for filename, content in test_files:
file_path = os.path.join(self.test_files_dir, filename)
with open(file_path, 'w') as f:
f.write(content)
def test_single_file_complete_pipeline(self):
"""Test complete pipeline processing for a single file."""
test_file = os.path.join(self.test_files_dir, "test_math_high_quality.py")
# Step 1: Analyze test file
metadata = self.analysis_engine.analyze_test_file(test_file)
# Verify analysis results
self.assertEqual(metadata.file_path, test_file)
self.assertGreater(len(metadata.test_functions), 0)
self.assertIn("test_addition_positive_numbers", metadata.test_functions)
# Step 2: Quality assessment
quality_assessment = self.quality_engine.assess_quality(metadata)
# High-quality file should score well
self.assertGreater(quality_assessment.quality_score, 80)
# Step 3: Plugin processing
processing_results = self.plugin_manager.process_file(metadata)
# Verify processing completed successfully
self.assertTrue(all(result.success for result in processing_results))
# Step 4: QMD generation
qmd_content = self.template_system.render_qmd(metadata)
# Verify QMD content was generated
self.assertIsInstance(qmd_content, str)
self.assertGreater(len(qmd_content), 0)
self.assertIn(os.path.basename(test_file), qmd_content)
# Step 5: Save QMD output
output_path = os.path.join(self.output_dir, "test_math_high_quality.qmd")
with open(output_path, 'w') as f:
f.write(qmd_content)
# Verify output file exists
self.assertTrue(os.path.exists(output_path))
def test_multiple_files_batch_processing(self):
"""Test batch processing of multiple test files."""
test_files = glob.glob(os.path.join(self.test_files_dir, "*.py"))
processing_results = []
for test_file in test_files:
# Process each file through complete pipeline
try:
# Analysis
metadata = self.analysis_engine.analyze_test_file(test_file)
# Quality assessment
quality = self.quality_engine.assess_quality(metadata)
# Plugin processing
plugin_results = self.plugin_manager.process_file(metadata)
# QMD generation
qmd_content = self.template_system.render_qmd(metadata)
# Save output
output_filename = os.path.basename(test_file).replace('.py', '.qmd')
output_path = os.path.join(self.output_dir, output_filename)
with open(output_path, 'w') as f:
f.write(qmd_content)
processing_results.append({
'file_path': test_file,
'success': True,
'quality_score': quality.quality_score,
'output_path': output_path
})
except Exception as e:
processing_results.append({
'file_path': test_file,
'success': False,
'error': str(e)
})
# Verify all files were processed
self.assertEqual(len(processing_results), len(test_files))
# Verify most files processed successfully (allow for some failures in edge cases)
successful_processing = [r for r in processing_results if r['success']]
success_rate = len(successful_processing) / len(processing_results)
self.assertGreater(success_rate, 0.8) # At least 80% success rate
# Verify output files exist for successful processing
for result in successful_processing:
if 'output_path' in result:
self.assertTrue(os.path.exists(result['output_path']))
def test_template_system_integration(self):
"""Test integration with template system for various file types."""
test_cases = [
("test_math_high_quality.py", "default"),
("test_database_integration.py", "integration"),
("test_performance_benchmarks.py", "performance")
]
for filename, template_type in test_cases:
test_file = os.path.join(self.test_files_dir, filename)
# Analyze file
metadata = self.analysis_engine.analyze_test_file(test_file)
# Generate QMD with specific template
qmd_content = self.template_system.render_qmd(metadata, template_type)
# Verify template-specific content
self.assertIn(os.path.basename(test_file), qmd_content)
self.assertGreater(len(qmd_content), 0)
def test_error_handling_across_pipeline(self):
"""Test error handling across complete workflow pipeline."""
# Create invalid test file
invalid_file = os.path.join(self.test_files_dir, "invalid_syntax.py")
with open(invalid_file, 'w') as f:
f.write("def invalid_syntax(\n") # Intentionally invalid Python
# Test graceful handling of invalid file
try:
metadata = self.analysis_engine.analyze_test_file(invalid_file)
# Should handle gracefully or raise appropriate exception
except SyntaxError:
# Expected behavior for invalid syntax
pass
except Exception as e:
# Should not crash with unhandled exception
self.fail(f"Unhandled exception in pipeline: {e}")
def test_configuration_system_integration(self):
"""Test integration with configuration system."""
# Test would verify configuration loading and application
# For now, just verify components can be configured
self.assertIsInstance(self.analysis_engine, TestAnalysisEngine)
self.assertIsInstance(self.quality_engine, QualityAssessmentEngine)
self.assertIsInstance(self.plugin_manager, PluginManager)
def test_performance_requirements_validation(self):
"""Test that end-to-end pipeline meets performance requirements."""
test_file = os.path.join(self.test_files_dir, "test_math_high_quality.py")
# Measure complete pipeline performance
start_time = time.time()
# Run complete pipeline
metadata = self.analysis_engine.analyze_test_file(test_file)
quality = self.quality_engine.assess_quality(metadata)
results = self.plugin_manager.process_file(metadata)
qmd_content = self.template_system.render_qmd(metadata)
end_time = time.time()
total_time = end_time - start_time
# Should complete single file processing in reasonable time
self.assertLess(total_time, 10.0) # Less than 10 seconds per file
@pytest.mark.slow
def test_stress_testing_multiple_files(self):
"""Test system behavior under stress with many files."""
# Create additional test files for stress testing
stress_test_dir = os.path.join(self.temp_dir, "stress_test")
os.makedirs(stress_test_dir)
# Generate multiple test files
for i in range(20):
test_file = os.path.join(stress_test_dir, f"test_stress_{i:03d}.py")
with open(test_file, 'w') as f:
f.write(f'''
def test_function_{i}():
"""Test function {i}"""
assert {i} == {i}
def test_another_{i}():
"""Another test function {i}"""
result = {i} * 2
assert result == {i * 2}
''')
# Process all stress test files
stress_files = glob.glob(os.path.join(stress_test_dir, "*.py"))
successful_processing = 0
start_time = time.time()
for test_file in stress_files:
try:
metadata = self.analysis_engine.analyze_test_file(test_file)
quality = self.quality_engine.assess_quality(metadata)
results = self.plugin_manager.process_file(metadata)
qmd_content = self.template_system.render_qmd(metadata)
successful_processing += 1
except Exception as e:
print(f"Failed to process {test_file}: {e}")
end_time = time.time()
total_time = end_time - start_time
# Verify high success rate even under stress
success_rate = successful_processing / len(stress_files)
self.assertGreater(success_rate, 0.9) # 90% success rate minimum
# Verify reasonable performance even with many files
average_time_per_file = total_time / len(stress_files)
self.assertLess(average_time_per_file, 5.0) # Less than 5 seconds average per file
create_sample_test_files(self)
¶
Create various sample test files for end-to-end testing
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def create_sample_test_files(self):
"""Create various sample test files for end-to-end testing"""
# High-quality test file with comprehensive tests
high_quality_content = '''
"""
Comprehensive unit tests for mathematical operations module.
This module provides extensive testing coverage for mathematical operations
including basic arithmetic, advanced functions, and edge cases.
"""
import unittest
import math
import pytest
from typing import List, Union
class TestMathematicalOperations(unittest.TestCase):
"""Test suite for mathematical operations."""
def setUp(self):
"""Set up test fixtures."""
self.test_numbers = [1, 2, 3, 4, 5]
self.zero_value = 0
self.negative_numbers = [-1, -2, -3]
def test_addition_positive_numbers(self):
"""Test addition with positive numbers."""
result = 2 + 3
self.assertEqual(result, 5)
def test_addition_negative_numbers(self):
"""Test addition with negative numbers."""
result = -2 + (-3)
self.assertEqual(result, -5)
def test_addition_mixed_numbers(self):
"""Test addition with mixed positive and negative numbers."""
result = 5 + (-3)
self.assertEqual(result, 2)
def test_division_by_zero_raises_exception(self):
"""Test that division by zero raises ZeroDivisionError."""
with self.assertRaises(ZeroDivisionError):
result = 10 / 0
def test_square_root_positive_number(self):
"""Test square root of positive numbers."""
result = math.sqrt(16)
self.assertEqual(result, 4.0)
def test_power_operations(self):
"""Test various power operations."""
self.assertEqual(2**3, 8)
self.assertEqual(5**0, 1)
self.assertEqual(4**0.5, 2.0)
@pytest.mark.parametrize("x,y,expected", [
(2, 3, 5),
(0, 5, 5),
(-1, 1, 0),
(10, -5, 5)
])
def test_parametrized_addition(self, x, y, expected):
"""Parametrized test for addition operations."""
assert x + y == expected
if __name__ == '__main__':
unittest.main()
'''
# Medium-quality test file with some tests but limited documentation
medium_quality_content = '''
import unittest
class TestStringOperations(unittest.TestCase):
def test_string_upper(self):
result = "hello".upper()
self.assertEqual(result, "HELLO")
def test_string_lower(self):
result = "WORLD".lower()
self.assertEqual(result, "world")
def test_string_length(self):
text = "test"
self.assertEqual(len(text), 4)
if __name__ == '__main__':
unittest.main()
'''
# Low-quality test file (stub with minimal content)
low_quality_content = '''
def test_something():
pass
'''
# Integration test file
integration_content = '''
"""
Integration tests for database operations.
Tests the integration between the application and database layer.
"""
import unittest
import tempfile
import os
from unittest.mock import patch, MagicMock
class TestDatabaseIntegration(unittest.TestCase):
"""Integration tests for database operations."""
def setUp(self):
"""Set up test database."""
self.temp_db = tempfile.mktemp()
def tearDown(self):
"""Clean up test database."""
if os.path.exists(self.temp_db):
os.remove(self.temp_db)
def test_database_connection_integration(self):
"""Test database connection establishment."""
# Mock database connection
with patch('database.connect') as mock_connect:
mock_connect.return_value = MagicMock()
# Test integration logic here
assert True
def test_data_persistence_integration(self):
"""Test data persistence across operations."""
# Test data persistence
assert True
if __name__ == '__main__':
unittest.main()
'''
# Performance test file
performance_content = '''
"""
Performance benchmarks for critical operations.
Measures and validates performance requirements for key system functions.
"""
import time
import unittest
import pytest
class TestPerformanceBenchmarks(unittest.TestCase):
"""Performance benchmark test suite."""
@pytest.mark.benchmark
def test_operation_performance(self):
"""Benchmark critical operation performance."""
start_time = time.time()
# Simulate operation
for i in range(1000):
result = i ** 2
end_time = time.time()
duration = end_time - start_time
# Should complete in less than 1 second
self.assertLess(duration, 1.0)
def test_memory_usage_benchmark(self):
"""Test memory usage stays within bounds."""
# Mock memory usage test
assert True
if __name__ == '__main__':
unittest.main()
'''
# Write test files
test_files = [
("test_math_high_quality.py", high_quality_content),
("test_strings_medium_quality.py", medium_quality_content),
("test_stub_low_quality.py", low_quality_content),
("test_database_integration.py", integration_content),
("test_performance_benchmarks.py", performance_content)
]
for filename, content in test_files:
file_path = os.path.join(self.test_files_dir, filename)
with open(file_path, 'w') as f:
f.write(content)
test_single_file_complete_pipeline(self)
¶
Test complete pipeline processing for a single file.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_single_file_complete_pipeline(self):
"""Test complete pipeline processing for a single file."""
test_file = os.path.join(self.test_files_dir, "test_math_high_quality.py")
# Step 1: Analyze test file
metadata = self.analysis_engine.analyze_test_file(test_file)
# Verify analysis results
self.assertEqual(metadata.file_path, test_file)
self.assertGreater(len(metadata.test_functions), 0)
self.assertIn("test_addition_positive_numbers", metadata.test_functions)
# Step 2: Quality assessment
quality_assessment = self.quality_engine.assess_quality(metadata)
# High-quality file should score well
self.assertGreater(quality_assessment.quality_score, 80)
# Step 3: Plugin processing
processing_results = self.plugin_manager.process_file(metadata)
# Verify processing completed successfully
self.assertTrue(all(result.success for result in processing_results))
# Step 4: QMD generation
qmd_content = self.template_system.render_qmd(metadata)
# Verify QMD content was generated
self.assertIsInstance(qmd_content, str)
self.assertGreater(len(qmd_content), 0)
self.assertIn(os.path.basename(test_file), qmd_content)
# Step 5: Save QMD output
output_path = os.path.join(self.output_dir, "test_math_high_quality.qmd")
with open(output_path, 'w') as f:
f.write(qmd_content)
# Verify output file exists
self.assertTrue(os.path.exists(output_path))
test_multiple_files_batch_processing(self)
¶
Test batch processing of multiple test files.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_multiple_files_batch_processing(self):
"""Test batch processing of multiple test files."""
test_files = glob.glob(os.path.join(self.test_files_dir, "*.py"))
processing_results = []
for test_file in test_files:
# Process each file through complete pipeline
try:
# Analysis
metadata = self.analysis_engine.analyze_test_file(test_file)
# Quality assessment
quality = self.quality_engine.assess_quality(metadata)
# Plugin processing
plugin_results = self.plugin_manager.process_file(metadata)
# QMD generation
qmd_content = self.template_system.render_qmd(metadata)
# Save output
output_filename = os.path.basename(test_file).replace('.py', '.qmd')
output_path = os.path.join(self.output_dir, output_filename)
with open(output_path, 'w') as f:
f.write(qmd_content)
processing_results.append({
'file_path': test_file,
'success': True,
'quality_score': quality.quality_score,
'output_path': output_path
})
except Exception as e:
processing_results.append({
'file_path': test_file,
'success': False,
'error': str(e)
})
# Verify all files were processed
self.assertEqual(len(processing_results), len(test_files))
# Verify most files processed successfully (allow for some failures in edge cases)
successful_processing = [r for r in processing_results if r['success']]
success_rate = len(successful_processing) / len(processing_results)
self.assertGreater(success_rate, 0.8) # At least 80% success rate
# Verify output files exist for successful processing
for result in successful_processing:
if 'output_path' in result:
self.assertTrue(os.path.exists(result['output_path']))
test_template_system_integration(self)
¶
Test integration with template system for various file types.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_template_system_integration(self):
"""Test integration with template system for various file types."""
test_cases = [
("test_math_high_quality.py", "default"),
("test_database_integration.py", "integration"),
("test_performance_benchmarks.py", "performance")
]
for filename, template_type in test_cases:
test_file = os.path.join(self.test_files_dir, filename)
# Analyze file
metadata = self.analysis_engine.analyze_test_file(test_file)
# Generate QMD with specific template
qmd_content = self.template_system.render_qmd(metadata, template_type)
# Verify template-specific content
self.assertIn(os.path.basename(test_file), qmd_content)
self.assertGreater(len(qmd_content), 0)
test_error_handling_across_pipeline(self)
¶
Test error handling across complete workflow pipeline.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_error_handling_across_pipeline(self):
"""Test error handling across complete workflow pipeline."""
# Create invalid test file
invalid_file = os.path.join(self.test_files_dir, "invalid_syntax.py")
with open(invalid_file, 'w') as f:
f.write("def invalid_syntax(\n") # Intentionally invalid Python
# Test graceful handling of invalid file
try:
metadata = self.analysis_engine.analyze_test_file(invalid_file)
# Should handle gracefully or raise appropriate exception
except SyntaxError:
# Expected behavior for invalid syntax
pass
except Exception as e:
# Should not crash with unhandled exception
self.fail(f"Unhandled exception in pipeline: {e}")
test_configuration_system_integration(self)
¶
Test integration with configuration system.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_configuration_system_integration(self):
"""Test integration with configuration system."""
# Test would verify configuration loading and application
# For now, just verify components can be configured
self.assertIsInstance(self.analysis_engine, TestAnalysisEngine)
self.assertIsInstance(self.quality_engine, QualityAssessmentEngine)
self.assertIsInstance(self.plugin_manager, PluginManager)
test_performance_requirements_validation(self)
¶
Test that end-to-end pipeline meets performance requirements.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
def test_performance_requirements_validation(self):
"""Test that end-to-end pipeline meets performance requirements."""
test_file = os.path.join(self.test_files_dir, "test_math_high_quality.py")
# Measure complete pipeline performance
start_time = time.time()
# Run complete pipeline
metadata = self.analysis_engine.analyze_test_file(test_file)
quality = self.quality_engine.assess_quality(metadata)
results = self.plugin_manager.process_file(metadata)
qmd_content = self.template_system.render_qmd(metadata)
end_time = time.time()
total_time = end_time - start_time
# Should complete single file processing in reasonable time
self.assertLess(total_time, 10.0) # Less than 10 seconds per file
test_stress_testing_multiple_files(self)
¶
Test system behavior under stress with many files.
Source code in hazelbean_tests/integration/test_end_to_end_workflow.py
@pytest.mark.slow
def test_stress_testing_multiple_files(self):
"""Test system behavior under stress with many files."""
# Create additional test files for stress testing
stress_test_dir = os.path.join(self.temp_dir, "stress_test")
os.makedirs(stress_test_dir)
# Generate multiple test files
for i in range(20):
test_file = os.path.join(stress_test_dir, f"test_stress_{i:03d}.py")
with open(test_file, 'w') as f:
f.write(f'''
def test_function_{i}():
"""Test function {i}"""
assert {i} == {i}
def test_another_{i}():
"""Another test function {i}"""
result = {i} * 2
assert result == {i * 2}
''')
# Process all stress test files
stress_files = glob.glob(os.path.join(stress_test_dir, "*.py"))
successful_processing = 0
start_time = time.time()
for test_file in stress_files:
try:
metadata = self.analysis_engine.analyze_test_file(test_file)
quality = self.quality_engine.assess_quality(metadata)
results = self.plugin_manager.process_file(metadata)
qmd_content = self.template_system.render_qmd(metadata)
successful_processing += 1
except Exception as e:
print(f"Failed to process {test_file}: {e}")
end_time = time.time()
total_time = end_time - start_time
# Verify high success rate even under stress
success_rate = successful_processing / len(stress_files)
self.assertGreater(success_rate, 0.9) # 90% success rate minimum
# Verify reasonable performance even with many files
average_time_per_file = total_time / len(stress_files)
self.assertLess(average_time_per_file, 5.0) # Less than 5 seconds average per file
Data Processing Pipeline Testing¶
Tests for multi-step data processing workflows that integrate multiple hazelbean components.
Key Integration Test Cases Covered:
- ✅ test_reclassify_raster_hb() - Raster value reclassification workflows
- ✅ test_reclassify_raster_with_negatives_hb() - Handle negative values in reclassification
- ✅ test_reclassify_raster_arrayframe() - ArrayFrame-based reclassification
- ✅ Raster resampling and alignment operations
- ✅ Multi-step geospatial processing pipelines
- ✅ Data format conversion and validation
Consolidated Integration Tests for Data Processing Workflows
This file consolidates tests from:
- data_processing_workflows/test_align.py
- data_processing_workflows/test_describe.py
- data_processing_workflows/test_get_path_integration.py
- data_processing_workflows/test_pyramids_original.py
- data_processing_workflows/test_pyramids.py
- data_processing_workflows/test_raster_vector_interface.py
- data_processing_workflows/test_spatial_projection.py
- data_processing_workflows/test_spatial_utils.py
Covers comprehensive data processing integration testing including: - Raster resampling and alignment operations - Array and data structure operations - Path resolution and cloud storage integration - Pyramid processing and COG validation - Raster-vector interface operations - Spatial projection and transformation - Spatial utilities and analysis functions
L
¶
BaseDataProcessingTest (TestCase)
¶
Base class for data processing integration tests with shared setup
Source code in hazelbean_tests/integration/test_data_processing.py
class BaseDataProcessingTest(TestCase):
"""Base class for data processing integration tests with shared setup"""
def setUp(self):
self.data_dir = os.path.join(os.path.dirname(__file__), "../../data")
self.test_data_dir = os.path.join(self.data_dir, "tests")
self.cartographic_data_dir = os.path.join(self.data_dir, "cartographic/ee")
self.pyramid_data_dir = os.path.join(self.data_dir, "pyramids")
self.crops_data_dir = os.path.join(self.data_dir, "crops/johnson")
# Common test paths
self.ee_r264_ids_900sec_path = os.path.join(self.cartographic_data_dir, "ee_r264_ids_900sec.tif")
self.global_1deg_raster_path = os.path.join(self.pyramid_data_dir, "ha_per_cell_3600sec.tif")
self.ee_r264_correspondence_vector_path = os.path.join(self.cartographic_data_dir, "ee_r264_simplified900sec.gpkg")
self.ee_r264_correspondence_csv_path = os.path.join(self.cartographic_data_dir, "ee_r264_correspondence.csv")
self.maize_calories_path = os.path.join(self.data_dir, "crops/johnson/crop_calories/maize_calories_per_ha_masked.tif")
self.ha_per_cell_column_900sec_path = hb.get_path(hb.ha_per_cell_column_ref_paths[900])
self.ha_per_cell_900sec_path = hb.get_path(hb.ha_per_cell_ref_paths[900])
self.pyramid_match_900sec_path = hb.get_path(hb.pyramid_match_ref_paths[900])
# Pyramid-specific paths
self.ha_per_cell_path = os.path.join(self.pyramid_data_dir, "ha_per_cell_300sec.tif")
self.valid_cog_path = os.path.join(self.test_data_dir, "valid_cog_example.tif")
self.invalid_cog_path = os.path.join(self.test_data_dir, "invalid_cog_example.tif")
self.valid_pog_path = os.path.join(self.cartographic_data_dir, "ee_r264_ids_900sec.tif")
# Spatial utils specific setup
user_dir = os.path.expanduser("~")
self.output_dir = os.path.join(user_dir, "temp")
def tearDown(self):
pass
TestAlignmentOperations (BaseDataProcessingTest)
¶
Tests for raster alignment and resampling operations (from test_align.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestAlignmentOperations(BaseDataProcessingTest):
"""Tests for raster alignment and resampling operations (from test_align.py)"""
def test_resample_to_match(self):
"""Test basic raster resampling to match reference raster"""
output_dir = 'data'
output_path = hb.temp('.tif', 'resampled', delete_on_finish, output_dir)
hb.resample_to_match(self.ee_r264_ids_900sec_path,
self.global_1deg_raster_path,
output_path,
resample_method='near',
output_data_type=6,
src_ndv=None,
ndv=None,
compress=True,
calc_raster_stats=False,
add_overviews=False,
pixel_size_override=None)
output2_path = hb.temp('.tif', 'mask', delete_on_finish, output_dir)
hb.create_valid_mask_from_vector_path(self.ee_r264_correspondence_vector_path, self.global_1deg_raster_path, output2_path,
all_touched=True)
def test_misc_operations(self):
"""Test miscellaneous array and data structure operations"""
output_dir = 'data'
# Test comma linebreak string to array conversion
input_string = '''0,1,1
3,2,2
1,4,1'''
a = hb.comma_linebreak_string_to_2d_array(input_string)
a = hb.comma_linebreak_string_to_2d_array(input_string, dtype=np.int8)
# Test numpy array save/load operations
a = np.random.rand(5, 5)
temp_path = hb.temp('.npy', 'npytest', delete_on_finish, output_dir)
hb.save_array_as_npy(a, temp_path)
r = hb.describe(temp_path, surpress_print=True, surpress_logger=True)
# Test directory operations
folder_list = ['asdf', 'asdf/qwer']
hb.create_directories(folder_list)
hb.remove_dirs(folder_list, safety_check='delete')
# Test dict/dataframe conversion
input_dict = {
'row_1': {'col_1': 1, 'col_2': 2},
'row_2': {'col_1': 3, 'col_2': 4}
}
df = hb.dict_to_df(input_dict)
generated_dict = hb.df_to_dict(df)
assert(input_dict == generated_dict)
test_resample_to_match(self)
¶
Test basic raster resampling to match reference raster
Source code in hazelbean_tests/integration/test_data_processing.py
def test_resample_to_match(self):
"""Test basic raster resampling to match reference raster"""
output_dir = 'data'
output_path = hb.temp('.tif', 'resampled', delete_on_finish, output_dir)
hb.resample_to_match(self.ee_r264_ids_900sec_path,
self.global_1deg_raster_path,
output_path,
resample_method='near',
output_data_type=6,
src_ndv=None,
ndv=None,
compress=True,
calc_raster_stats=False,
add_overviews=False,
pixel_size_override=None)
output2_path = hb.temp('.tif', 'mask', delete_on_finish, output_dir)
hb.create_valid_mask_from_vector_path(self.ee_r264_correspondence_vector_path, self.global_1deg_raster_path, output2_path,
all_touched=True)
test_misc_operations(self)
¶
Test miscellaneous array and data structure operations
Source code in hazelbean_tests/integration/test_data_processing.py
def test_misc_operations(self):
"""Test miscellaneous array and data structure operations"""
output_dir = 'data'
# Test comma linebreak string to array conversion
input_string = '''0,1,1
3,2,2
1,4,1'''
a = hb.comma_linebreak_string_to_2d_array(input_string)
a = hb.comma_linebreak_string_to_2d_array(input_string, dtype=np.int8)
# Test numpy array save/load operations
a = np.random.rand(5, 5)
temp_path = hb.temp('.npy', 'npytest', delete_on_finish, output_dir)
hb.save_array_as_npy(a, temp_path)
r = hb.describe(temp_path, surpress_print=True, surpress_logger=True)
# Test directory operations
folder_list = ['asdf', 'asdf/qwer']
hb.create_directories(folder_list)
hb.remove_dirs(folder_list, safety_check='delete')
# Test dict/dataframe conversion
input_dict = {
'row_1': {'col_1': 1, 'col_2': 2},
'row_2': {'col_1': 3, 'col_2': 4}
}
df = hb.dict_to_df(input_dict)
generated_dict = hb.df_to_dict(df)
assert(input_dict == generated_dict)
TestDescribeOperations (BaseDataProcessingTest)
¶
Tests for array description and analysis (from test_describe.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestDescribeOperations(BaseDataProcessingTest):
"""Tests for array description and analysis (from test_describe.py)"""
def test_describe(self):
"""Test describe functionality for arrays"""
a = np.random.rand(5, 5)
tmp_path = hb.temp('.npy', remove_at_exit=True)
hb.save_array_as_npy(a, tmp_path)
hb.describe(tmp_path, surpress_print=True, surpress_logger=True)
test_describe(self)
¶
Test describe functionality for arrays
Source code in hazelbean_tests/integration/test_data_processing.py
TestGetPathIntegration (BaseDataProcessingTest)
¶
Tests for ProjectFlow.get_path() integration functionality (from test_get_path_integration.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestGetPathIntegration(BaseDataProcessingTest):
"""Tests for ProjectFlow.get_path() integration functionality (from test_get_path_integration.py)"""
def setUp(self):
super().setUp()
# Additional setup for get_path tests
self.test_dir = tempfile.mkdtemp()
# Create ProjectFlow instance
self.p = hb.ProjectFlow(self.test_dir)
# Create test directory structure
os.makedirs(os.path.join(self.test_dir, "intermediate"), exist_ok=True)
os.makedirs(os.path.join(self.test_dir, "input"), exist_ok=True)
# Create test files in project directories
self.create_test_files()
def tearDown(self):
super().tearDown()
"""Clean up test directories"""
shutil.rmtree(self.test_dir, ignore_errors=True)
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
@pytest.mark.integration
def test_google_cloud_bucket_integration(self):
"""Test Google Cloud bucket integration (without actual cloud calls)"""
# Arrange
self.p.input_bucket_name = "test-hazelbean-bucket"
test_file = "cloud_test_file.tif"
# Act
resolved_path = self.p.get_path(test_file)
# Assert
# Should return a valid path (either local or constructed cloud path)
self.assertIsInstance(resolved_path, str)
self.assertIn(test_file, resolved_path)
@pytest.mark.integration
def test_bucket_name_assignment(self):
"""Test bucket name assignment"""
# Arrange & Act
self.p.input_bucket_name = "test-bucket"
# Assert
self.assertEqual(self.p.input_bucket_name, "test-bucket")
@pytest.mark.integration
def test_cloud_path_fallback(self):
"""Test cloud path fallback when local file not found"""
# Arrange
self.p.input_bucket_name = "test-bucket"
test_file = "only_in_cloud.tif"
# Act
resolved_path = self.p.get_path(test_file)
# Assert
# Should return a constructed path even if file doesn't exist locally
self.assertIsInstance(resolved_path, str)
self.assertIn(test_file, resolved_path)
@pytest.mark.integration
def test_existing_cartographic_data_access(self):
"""Test access to existing cartographic data"""
# Arrange
cartographic_files = [
"cartographic/ee/ee_r264_ids_900sec.tif",
"cartographic/ee/ee_r264_simplified900sec.gpkg",
"cartographic/ee/ee_r264_correspondence.csv"
]
# Act & Assert
for file_path in cartographic_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
@pytest.mark.integration
def test_existing_pyramid_data_access(self):
"""Test access to existing pyramid data"""
# Arrange
pyramid_files = [
"pyramids/ha_per_cell_900sec.tif",
"pyramids/ha_per_cell_3600sec.tif",
"pyramids/match_900sec.tif"
]
# Act & Assert
for file_path in pyramid_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
@pytest.mark.integration
def test_existing_crops_data_access(self):
"""Test access to existing crops data"""
# Arrange
crops_path = "crops/johnson/crop_calories/maize_calories_per_ha_masked.tif"
# Act
resolved_path = self.p.get_path(crops_path)
# Assert
self.assertIsInstance(resolved_path, str)
self.assertIn("maize_calories_per_ha_masked.tif", resolved_path)
@pytest.mark.integration
def test_existing_test_data_access(self):
"""Test access to existing test data"""
# Arrange
test_files = [
"tests/valid_cog_example.tif",
"tests/invalid_cog_example.tif"
]
# Act & Assert
for file_path in test_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
create_test_files(self)
¶
Create test files in project directories for testing
Source code in hazelbean_tests/integration/test_data_processing.py
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
test_google_cloud_bucket_integration(self)
¶
Test Google Cloud bucket integration (without actual cloud calls)
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_google_cloud_bucket_integration(self):
"""Test Google Cloud bucket integration (without actual cloud calls)"""
# Arrange
self.p.input_bucket_name = "test-hazelbean-bucket"
test_file = "cloud_test_file.tif"
# Act
resolved_path = self.p.get_path(test_file)
# Assert
# Should return a valid path (either local or constructed cloud path)
self.assertIsInstance(resolved_path, str)
self.assertIn(test_file, resolved_path)
test_bucket_name_assignment(self)
¶
Test bucket name assignment
test_cloud_path_fallback(self)
¶
Test cloud path fallback when local file not found
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_cloud_path_fallback(self):
"""Test cloud path fallback when local file not found"""
# Arrange
self.p.input_bucket_name = "test-bucket"
test_file = "only_in_cloud.tif"
# Act
resolved_path = self.p.get_path(test_file)
# Assert
# Should return a constructed path even if file doesn't exist locally
self.assertIsInstance(resolved_path, str)
self.assertIn(test_file, resolved_path)
test_existing_cartographic_data_access(self)
¶
Test access to existing cartographic data
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_existing_cartographic_data_access(self):
"""Test access to existing cartographic data"""
# Arrange
cartographic_files = [
"cartographic/ee/ee_r264_ids_900sec.tif",
"cartographic/ee/ee_r264_simplified900sec.gpkg",
"cartographic/ee/ee_r264_correspondence.csv"
]
# Act & Assert
for file_path in cartographic_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
test_existing_pyramid_data_access(self)
¶
Test access to existing pyramid data
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_existing_pyramid_data_access(self):
"""Test access to existing pyramid data"""
# Arrange
pyramid_files = [
"pyramids/ha_per_cell_900sec.tif",
"pyramids/ha_per_cell_3600sec.tif",
"pyramids/match_900sec.tif"
]
# Act & Assert
for file_path in pyramid_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
test_existing_crops_data_access(self)
¶
Test access to existing crops data
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_existing_crops_data_access(self):
"""Test access to existing crops data"""
# Arrange
crops_path = "crops/johnson/crop_calories/maize_calories_per_ha_masked.tif"
# Act
resolved_path = self.p.get_path(crops_path)
# Assert
self.assertIsInstance(resolved_path, str)
self.assertIn("maize_calories_per_ha_masked.tif", resolved_path)
test_existing_test_data_access(self)
¶
Test access to existing test data
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
def test_existing_test_data_access(self):
"""Test access to existing test data"""
# Arrange
test_files = [
"tests/valid_cog_example.tif",
"tests/invalid_cog_example.tif"
]
# Act & Assert
for file_path in test_files:
resolved_path = self.p.get_path(file_path)
self.assertIsInstance(resolved_path, str)
self.assertIn(os.path.basename(file_path), resolved_path)
TestPyramidOperations (BaseDataProcessingTest)
¶
Tests for pyramid processing operations (from test_pyramids_original.py and test_pyramids.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestPyramidOperations(BaseDataProcessingTest):
"""Tests for pyramid processing operations (from test_pyramids_original.py and test_pyramids.py)"""
def test_load_geotiff_chunk_by_cr(self):
"""Test loading GeoTIFF chunks by column-row coordinates"""
hb.load_geotiff_chunk_by_cr_size(self.global_1deg_raster_path, (1, 2, 5, 5))
def test_load_geotiff_chunk_by_bb(self):
"""Test loading GeoTIFF chunks by bounding box"""
input_path = self.maize_calories_path
left_lat = -40
bottom_lon = -25
lat_size = .2
lon_size = 1
bb = [left_lat,
bottom_lon,
left_lat + lat_size,
bottom_lon + lon_size]
hb.load_geotiff_chunk_by_bb(input_path, bb)
def test_add_rows_or_cols_to_geotiff(self):
"""Test adding rows or columns to GeoTIFF"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
r_above, r_below, c_above, c_below = 10, 20, 0, 0
hb.add_rows_or_cols_to_geotiff(temp_path, r_above, r_below, c_above, c_below, remove_temporary_files=True)
def test_fill_to_match_extent(self):
"""Test filling raster to match extent"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'expand_to_bounding_box', True)
hb.fill_to_match_extent(temp_path, self.global_1deg_raster_path, temp2_path)
def test_fill_to_match_extent_manual(self):
"""Test manual fill to match extent"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'expand_to_bounding_box', True)
hb.fill_to_match_extent(temp_path, self.global_1deg_raster_path, temp2_path)
def test_convert_ndv_to_alpha_band(self):
"""Test converting no-data values to alpha band"""
output_path = hb.temp(folder=os.path.dirname(self.maize_calories_path), remove_at_exit=True)
hb.convert_ndv_to_alpha_band(self.maize_calories_path, output_path)
@pytest.mark.integration
@pytest.mark.slow
def test_raster_to_area_raster(self):
"""Check if TIFF files are valid Cloud-Optimized GeoTIFFs (COGs)."""
temp_path = hb.temp('.tif', filename_start='test_raster_to_area_raster', remove_at_exit=True, tag_along_file_extensions=['.aux.xml'])
with self.subTest(file=self.ha_per_cell_path):
raster_to_area_raster(self.ha_per_cell_path, temp_path)
result = hb.path_exists(temp_path)
self.assertTrue(result)
# Make it a pog
temp_pog_path = hb.temp('.tif', filename_start='test_area_raster_as_pog', remove_at_exit=True, tag_along_file_extensions=['.aux.xml'])
hb.make_path_pog(temp_path, temp_pog_path, output_data_type=7, verbose=True)
result = hb.is_path_pog(temp_pog_path, check_tiled=True, full_check=True, raise_exceptions=False, verbose=True)
self.assertTrue(result)
test_load_geotiff_chunk_by_cr(self)
¶
test_load_geotiff_chunk_by_bb(self)
¶
Test loading GeoTIFF chunks by bounding box
Source code in hazelbean_tests/integration/test_data_processing.py
def test_load_geotiff_chunk_by_bb(self):
"""Test loading GeoTIFF chunks by bounding box"""
input_path = self.maize_calories_path
left_lat = -40
bottom_lon = -25
lat_size = .2
lon_size = 1
bb = [left_lat,
bottom_lon,
left_lat + lat_size,
bottom_lon + lon_size]
hb.load_geotiff_chunk_by_bb(input_path, bb)
test_add_rows_or_cols_to_geotiff(self)
¶
Test adding rows or columns to GeoTIFF
Source code in hazelbean_tests/integration/test_data_processing.py
def test_add_rows_or_cols_to_geotiff(self):
"""Test adding rows or columns to GeoTIFF"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
r_above, r_below, c_above, c_below = 10, 20, 0, 0
hb.add_rows_or_cols_to_geotiff(temp_path, r_above, r_below, c_above, c_below, remove_temporary_files=True)
test_fill_to_match_extent(self)
¶
Test filling raster to match extent
Source code in hazelbean_tests/integration/test_data_processing.py
def test_fill_to_match_extent(self):
"""Test filling raster to match extent"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'expand_to_bounding_box', True)
hb.fill_to_match_extent(temp_path, self.global_1deg_raster_path, temp2_path)
test_fill_to_match_extent_manual(self)
¶
Test manual fill to match extent
Source code in hazelbean_tests/integration/test_data_processing.py
def test_fill_to_match_extent_manual(self):
"""Test manual fill to match extent"""
incomplete_array = hb.load_geotiff_chunk_by_bb(self.global_1deg_raster_path, [-180, -80, 180, 70])
temp_path = hb.temp('.tif', 'test_add_rows_or_cols_to_geotiff', True)
geotransform_override = hb.get_raster_info_hb(self.global_1deg_raster_path)['geotransform']
geotransform_override = [-180, 1, 0, 80, 0, -1]
n_rows_override = 150
hb.save_array_as_geotiff(incomplete_array, temp_path, self.global_1deg_raster_path, geotransform_override=geotransform_override, n_rows_override=n_rows_override)
temp2_path = hb.temp('.tif', 'expand_to_bounding_box', True)
hb.fill_to_match_extent(temp_path, self.global_1deg_raster_path, temp2_path)
test_convert_ndv_to_alpha_band(self)
¶
Test converting no-data values to alpha band
Source code in hazelbean_tests/integration/test_data_processing.py
test_raster_to_area_raster(self)
¶
Check if TIFF files are valid Cloud-Optimized GeoTIFFs (COGs).
Source code in hazelbean_tests/integration/test_data_processing.py
@pytest.mark.integration
@pytest.mark.slow
def test_raster_to_area_raster(self):
"""Check if TIFF files are valid Cloud-Optimized GeoTIFFs (COGs)."""
temp_path = hb.temp('.tif', filename_start='test_raster_to_area_raster', remove_at_exit=True, tag_along_file_extensions=['.aux.xml'])
with self.subTest(file=self.ha_per_cell_path):
raster_to_area_raster(self.ha_per_cell_path, temp_path)
result = hb.path_exists(temp_path)
self.assertTrue(result)
# Make it a pog
temp_pog_path = hb.temp('.tif', filename_start='test_area_raster_as_pog', remove_at_exit=True, tag_along_file_extensions=['.aux.xml'])
hb.make_path_pog(temp_path, temp_pog_path, output_data_type=7, verbose=True)
result = hb.is_path_pog(temp_pog_path, check_tiled=True, full_check=True, raise_exceptions=False, verbose=True)
self.assertTrue(result)
TestRasterVectorInterface (BaseDataProcessingTest)
¶
Tests for raster-vector interface operations (from test_raster_vector_interface.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestRasterVectorInterface(BaseDataProcessingTest):
"""Tests for raster-vector interface operations (from test_raster_vector_interface.py)"""
def test_raster_calculator_hb(self):
"""Test hazelbean raster calculator"""
t1 = hb.temp(remove_at_exit=True)
hb.raster_calculator_hb([(self.ee_r264_ids_900sec_path, 1), (self.ee_r264_ids_900sec_path, 1)], lambda x, y: x + y, t1, 7, -9999)
# LEARNING POINT, I had to be very careful here with type casting to ensure the summation methods yielded the same.
a = np.sum(hb.as_array(t1))
b = np.sum(hb.as_array(self.ee_r264_ids_900sec_path).astype(np.float64)) * np.float64(2.0)
assert a == b
def test_assert_gdal_paths_in_same_projection(self):
"""Test assertion of GDAL paths in same projection"""
self.assertTrue(
hb.assert_gdal_paths_in_same_projection([
self.ee_r264_correspondence_vector_path,
self.ee_r264_ids_900sec_path,
self.maize_calories_path,
], return_result=True)
)
self.assertTrue(
hb.assert_gdal_paths_in_same_projection([
self.ee_r264_correspondence_vector_path,
self.ee_r264_ids_900sec_path,
self.maize_calories_path,
], return_result=True)
)
def test_zonal_statistics_faster(self):
"""Test fast zonal statistics implementation"""
test_results = []
zone_ids_raster_path = hb.temp('.tif', remove_at_exit=True)
# Test using the pregenereated
start = time.time()
results_dict = hb.zonal_statistics_flex(self.maize_calories_path, self.ee_r264_correspondence_vector_path,
zone_ids_raster_path=zone_ids_raster_path, verbose=False)
print('results_dict', results_dict)
def test_zonal_statistics_enumeration(self):
"""Test zonal statistics enumeration"""
test_results = []
zone_ids_raster_path = hb.temp('.tif', remove_at_exit=True)
# Test using the pregenereated
start = time.time()
results_dict = hb.zonal_statistics_flex(self.ee_r264_ids_900sec_path, self.ee_r264_correspondence_vector_path,
zone_ids_raster_path=zone_ids_raster_path, verbose=False)
print('results_dict', results_dict)
def test_super_simplify(self):
"""Test vector super simplification"""
input_vector_path = self.ee_r264_correspondence_vector_path
id_column_label = 'ee_r264_id'
blur_size = 300.0
output_path = 'simplified_vector.gpkg'
raster_vector_interface.vector_super_simplify(input_vector_path, id_column_label, blur_size, output_path, remove_temp_files=True)
test_raster_calculator_hb(self)
¶
Test hazelbean raster calculator
Source code in hazelbean_tests/integration/test_data_processing.py
def test_raster_calculator_hb(self):
"""Test hazelbean raster calculator"""
t1 = hb.temp(remove_at_exit=True)
hb.raster_calculator_hb([(self.ee_r264_ids_900sec_path, 1), (self.ee_r264_ids_900sec_path, 1)], lambda x, y: x + y, t1, 7, -9999)
# LEARNING POINT, I had to be very careful here with type casting to ensure the summation methods yielded the same.
a = np.sum(hb.as_array(t1))
b = np.sum(hb.as_array(self.ee_r264_ids_900sec_path).astype(np.float64)) * np.float64(2.0)
assert a == b
test_assert_gdal_paths_in_same_projection(self)
¶
Test assertion of GDAL paths in same projection
Source code in hazelbean_tests/integration/test_data_processing.py
def test_assert_gdal_paths_in_same_projection(self):
"""Test assertion of GDAL paths in same projection"""
self.assertTrue(
hb.assert_gdal_paths_in_same_projection([
self.ee_r264_correspondence_vector_path,
self.ee_r264_ids_900sec_path,
self.maize_calories_path,
], return_result=True)
)
self.assertTrue(
hb.assert_gdal_paths_in_same_projection([
self.ee_r264_correspondence_vector_path,
self.ee_r264_ids_900sec_path,
self.maize_calories_path,
], return_result=True)
)
test_zonal_statistics_faster(self)
¶
Test fast zonal statistics implementation
Source code in hazelbean_tests/integration/test_data_processing.py
def test_zonal_statistics_faster(self):
"""Test fast zonal statistics implementation"""
test_results = []
zone_ids_raster_path = hb.temp('.tif', remove_at_exit=True)
# Test using the pregenereated
start = time.time()
results_dict = hb.zonal_statistics_flex(self.maize_calories_path, self.ee_r264_correspondence_vector_path,
zone_ids_raster_path=zone_ids_raster_path, verbose=False)
print('results_dict', results_dict)
test_zonal_statistics_enumeration(self)
¶
Test zonal statistics enumeration
Source code in hazelbean_tests/integration/test_data_processing.py
def test_zonal_statistics_enumeration(self):
"""Test zonal statistics enumeration"""
test_results = []
zone_ids_raster_path = hb.temp('.tif', remove_at_exit=True)
# Test using the pregenereated
start = time.time()
results_dict = hb.zonal_statistics_flex(self.ee_r264_ids_900sec_path, self.ee_r264_correspondence_vector_path,
zone_ids_raster_path=zone_ids_raster_path, verbose=False)
print('results_dict', results_dict)
test_super_simplify(self)
¶
Test vector super simplification
Source code in hazelbean_tests/integration/test_data_processing.py
def test_super_simplify(self):
"""Test vector super simplification"""
input_vector_path = self.ee_r264_correspondence_vector_path
id_column_label = 'ee_r264_id'
blur_size = 300.0
output_path = 'simplified_vector.gpkg'
raster_vector_interface.vector_super_simplify(input_vector_path, id_column_label, blur_size, output_path, remove_temp_files=True)
TestSpatialProjection (BaseDataProcessingTest)
¶
Tests for spatial projection operations (from test_spatial_projection.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestSpatialProjection(BaseDataProcessingTest):
"""Tests for spatial projection operations (from test_spatial_projection.py)"""
def test_resample_to_cell_size(self):
"""Test resampling to specific cell size"""
output_path = hb.temp('.tif', 'test_resample_to_match', True)
pixel_size_override = 1.0
hb.resample_to_match(self.maize_calories_path, self.ee_r264_ids_900sec_path, output_path, resample_method='near',
output_data_type=6, src_ndv=None, ndv=None, compress=True,
calc_raster_stats=False,
add_overviews=False,
pixel_size_override=pixel_size_override)
def test_resample_to_match(self):
"""Test resampling to match reference raster"""
output_path = hb.temp('.tif', 'test_resample_to_match', True)
hb.resample_to_match(self.maize_calories_path, self.ee_r264_ids_900sec_path, output_path, resample_method='near',
output_data_type=6, src_ndv=None, ndv=None, compress=True,
calc_raster_stats=False,
add_overviews=False,)
output2_path = hb.temp('.tif', 'mask', True)
hb.create_valid_mask_from_vector_path(self.ee_r264_correspondence_vector_path, self.ee_r264_ids_900sec_path, output2_path,
all_touched=True)
output3_path = hb.temp('.tif', 'masked', True)
hb.set_ndv_by_mask_path(output_path, output2_path, output_path=output3_path, ndv=-9999.)
test_resample_to_cell_size(self)
¶
Test resampling to specific cell size
Source code in hazelbean_tests/integration/test_data_processing.py
def test_resample_to_cell_size(self):
"""Test resampling to specific cell size"""
output_path = hb.temp('.tif', 'test_resample_to_match', True)
pixel_size_override = 1.0
hb.resample_to_match(self.maize_calories_path, self.ee_r264_ids_900sec_path, output_path, resample_method='near',
output_data_type=6, src_ndv=None, ndv=None, compress=True,
calc_raster_stats=False,
add_overviews=False,
pixel_size_override=pixel_size_override)
test_resample_to_match(self)
¶
Test resampling to match reference raster
Source code in hazelbean_tests/integration/test_data_processing.py
def test_resample_to_match(self):
"""Test resampling to match reference raster"""
output_path = hb.temp('.tif', 'test_resample_to_match', True)
hb.resample_to_match(self.maize_calories_path, self.ee_r264_ids_900sec_path, output_path, resample_method='near',
output_data_type=6, src_ndv=None, ndv=None, compress=True,
calc_raster_stats=False,
add_overviews=False,)
output2_path = hb.temp('.tif', 'mask', True)
hb.create_valid_mask_from_vector_path(self.ee_r264_correspondence_vector_path, self.ee_r264_ids_900sec_path, output2_path,
all_touched=True)
output3_path = hb.temp('.tif', 'masked', True)
hb.set_ndv_by_mask_path(output_path, output2_path, output_path=output3_path, ndv=-9999.)
TestSpatialUtils (BaseDataProcessingTest)
¶
Tests for spatial utilities and analysis functions (from test_spatial_utils.py)
Source code in hazelbean_tests/integration/test_data_processing.py
class TestSpatialUtils(BaseDataProcessingTest):
"""Tests for spatial utilities and analysis functions (from test_spatial_utils.py)"""
def test_get_wkt_from_epsg_code(self):
"""Test WKT generation from EPSG codes"""
hb.get_wkt_from_epsg_code(hb.common_epsg_codes_by_name['wgs84'])
def test_rank_array(self):
"""Test array ranking functionality"""
array = np.random.rand(6, 6)
nan_mask = np.zeros((6, 6))
nan_mask[1:3, 2:5] = 1
ranked_array, ranked_pared_keys = hb.get_rank_array_and_keys(array, nan_mask=nan_mask)
assert (ranked_array[1, 2] == -9999)
assert (len(ranked_pared_keys[0] == 30))
def test_create_vector_from_raster_extents(self):
"""Test creating vector from raster extents"""
extent_path = hb.temp('.shp', remove_at_exit=True)
hb.create_vector_from_raster_extents(self.pyramid_match_900sec_path, extent_path)
self.assertTrue(os.path.exists(extent_path))
def test_read_1d_npy_chunk(self):
"""Test reading 1D numpy array chunks"""
r = np.random.randint(2,9,200)
temp_path = hb.temp('.npy', remove_at_exit=True)
hb.save_array_as_npy(r, temp_path)
output = hb.read_1d_npy_chunk(temp_path, 3, 8)
self.assertTrue(sum(r[3:3+8])==sum(output))
def test_get_attribute_table_columns_from_shapefile(self):
"""Test extracting attribute table columns from shapefiles"""
r = hb.get_attribute_table_columns_from_shapefile(self.ee_r264_correspondence_vector_path, cols='ee_r264_id')
self.assertIsNotNone(r)
def test_extract_features_in_shapefile_by_attribute(self):
"""Test feature extraction by attribute"""
output_gpkg_path = hb.temp('.gpkg', remove_at_exit=True)
column_name = 'ee_r264_id'
column_filter = 77
hb.extract_features_in_shapefile_by_attribute(self.ee_r264_correspondence_vector_path, output_gpkg_path, column_name, column_filter)
def test_get_bounding_box(self):
"""Test bounding box extraction from various data types"""
zones_vector_path = self.ee_r264_correspondence_vector_path
zone_ids_raster_path = self.ee_r264_ids_900sec_path
zone_values_path = self.ha_per_cell_900sec_path
run_all = 0
remove_temporary_files = 1
output_dir = self.test_data_dir
# Test getting a Bounding Box of a raster
bb = hb.get_bounding_box(self.global_1deg_raster_path)
print(bb)
# Test getting a Bounding Box of a vector
bb = hb.get_bounding_box(zones_vector_path)
print(bb)
# Create a new GPKG for just the country of RWA
rwa_vector_path = hb.temp('.gpkg', 'rwa', remove_temporary_files, output_dir)
hb.extract_features_in_shapefile_by_attribute(zones_vector_path, rwa_vector_path, "ee_r264_id", 70)
# Get the bounding box of that new vector
bb = hb.get_bounding_box(rwa_vector_path)
print(bb)
def test_reading_csvs(self):
"""Test auto downloading of files via get_path"""
# Test that it does find a path that exists
p = hb.ProjectFlow(self.output_dir)
p.base_data_dir = '../../../base_data'
# You can put the api credentials anywhere in the folder structure. Preferred is at the root of base data.
p.data_credentials_path = None
p.input_bucket_name = 'gtap_invest_seals_2023_04_21'
test_path = p.get_path('cartographic/gadm/gadm_410_adm0_labels_test.csv', verbose=True)
df = pd.read_csv(test_path)
assert len(df) > 0
hb.remove_path(test_path)
# Now try it WITH credentials
p.data_credentials_path = p.get_path('api_key_credentials.json')
test_path = p.get_path('cartographic/gadm/gadm_410_adm0_labels_test.csv', verbose=True)
df = pd.read_csv(test_path)
assert len(df) > 0
hb.remove_path(test_path)
def test_get_reclassification_dict_from_df(self):
"""Test reclassification dictionary generation from DataFrame"""
# Test that it does find a path that exists
p = hb.ProjectFlow(self.output_dir)
p.base_data_dir = '../../../base_data'
correspondence_path = p.get_path(os.path.join(self.data_dir, 'cartographic', 'ee', 'ee_r264_correspondence.csv'))
from hazelbean import utils
# TODO This should be extended to cover classification dicts from correspondences but also structured and unstructured mappings.
r = utils.get_reclassification_dict_from_df(correspondence_path, 'gtapv7_r160_id', 'gtapv7_r50_id', 'gtapv7_r160_label', 'gtapv7_r50_label')
hb.print_iterable(r)
def test_clipping_simple(self):
"""Test simple raster clipping operations"""
output_path = hb.temp('.tif', 'clipped', delete_on_finish, self.output_dir)
hb.clip_raster_by_vector_simple(self.ee_r264_ids_900sec_path,
output_path,
self.ee_r264_correspondence_vector_path,
output_data_type=6,
gtiff_creation_options=hb.DEFAULT_GTIFF_CREATION_OPTIONS)
print('Created', output_path)
output_dir = 'data'
output_path = hb.temp('.tif', 'clipped_attr', delete_on_finish, output_dir)
hb.clip_raster_by_vector_simple(self.ee_r264_ids_900sec_path,
output_path,
self.ee_r264_correspondence_vector_path,
output_data_type=6,
clip_vector_filter='ee_r264_id="120"',
gtiff_creation_options=hb.DEFAULT_GTIFF_CREATION_OPTIONS)
print('Created', output_path)
def test_reclassify_raster_hb(self):
"""Test raster reclassification with hazelbean"""
rules = {235: 34}
output_path = hb.temp('.tif', 'reclassify', True, self.output_dir)
hb.reclassify_raster_hb(self.ee_r264_ids_900sec_path,
rules,
output_path)
def test_reclassify_raster_with_negatives_hb(self):
"""Test raster reclassification with negative values"""
rules = {235: -555}
output_path = hb.temp('.tif', 'reclassify', False, self.output_dir)
hb.reclassify_raster_hb(self.ee_r264_ids_900sec_path,
rules,
output_path,
output_data_type=5)
print(hb.enumerate_raster_path(output_path))
output_with_neg_path = hb.temp('.tif', 'reclassify_with_neg', False, self.output_dir)
rules = {
235: -444,
241: -9999,
-555: -888,
} # Adding a rule for 241 to be reclassified to -9999
hb.reclassify_raster_hb(output_path,
rules,
output_with_neg_path,
output_data_type=5)
print(hb.enumerate_raster_path(output_with_neg_path))
def test_reclassify_raster_arrayframe(self):
"""Test raster reclassification with arrayframe"""
rules = {235: 34}
output_path = hb.temp('.tif', 'reclassify', True, self.output_dir)
hb.reclassify_raster_arrayframe(self.ee_r264_ids_900sec_path,
rules,
output_path)
test_get_wkt_from_epsg_code(self)
¶
test_rank_array(self)
¶
Test array ranking functionality
Source code in hazelbean_tests/integration/test_data_processing.py
def test_rank_array(self):
"""Test array ranking functionality"""
array = np.random.rand(6, 6)
nan_mask = np.zeros((6, 6))
nan_mask[1:3, 2:5] = 1
ranked_array, ranked_pared_keys = hb.get_rank_array_and_keys(array, nan_mask=nan_mask)
assert (ranked_array[1, 2] == -9999)
assert (len(ranked_pared_keys[0] == 30))
test_create_vector_from_raster_extents(self)
¶
Test creating vector from raster extents
Source code in hazelbean_tests/integration/test_data_processing.py
test_read_1d_npy_chunk(self)
¶
Test reading 1D numpy array chunks
Source code in hazelbean_tests/integration/test_data_processing.py
test_get_attribute_table_columns_from_shapefile(self)
¶
Test extracting attribute table columns from shapefiles
Source code in hazelbean_tests/integration/test_data_processing.py
test_extract_features_in_shapefile_by_attribute(self)
¶
Test feature extraction by attribute
Source code in hazelbean_tests/integration/test_data_processing.py
def test_extract_features_in_shapefile_by_attribute(self):
"""Test feature extraction by attribute"""
output_gpkg_path = hb.temp('.gpkg', remove_at_exit=True)
column_name = 'ee_r264_id'
column_filter = 77
hb.extract_features_in_shapefile_by_attribute(self.ee_r264_correspondence_vector_path, output_gpkg_path, column_name, column_filter)
test_get_bounding_box(self)
¶
Test bounding box extraction from various data types
Source code in hazelbean_tests/integration/test_data_processing.py
def test_get_bounding_box(self):
"""Test bounding box extraction from various data types"""
zones_vector_path = self.ee_r264_correspondence_vector_path
zone_ids_raster_path = self.ee_r264_ids_900sec_path
zone_values_path = self.ha_per_cell_900sec_path
run_all = 0
remove_temporary_files = 1
output_dir = self.test_data_dir
# Test getting a Bounding Box of a raster
bb = hb.get_bounding_box(self.global_1deg_raster_path)
print(bb)
# Test getting a Bounding Box of a vector
bb = hb.get_bounding_box(zones_vector_path)
print(bb)
# Create a new GPKG for just the country of RWA
rwa_vector_path = hb.temp('.gpkg', 'rwa', remove_temporary_files, output_dir)
hb.extract_features_in_shapefile_by_attribute(zones_vector_path, rwa_vector_path, "ee_r264_id", 70)
# Get the bounding box of that new vector
bb = hb.get_bounding_box(rwa_vector_path)
print(bb)
test_reading_csvs(self)
¶
Test auto downloading of files via get_path
Source code in hazelbean_tests/integration/test_data_processing.py
def test_reading_csvs(self):
"""Test auto downloading of files via get_path"""
# Test that it does find a path that exists
p = hb.ProjectFlow(self.output_dir)
p.base_data_dir = '../../../base_data'
# You can put the api credentials anywhere in the folder structure. Preferred is at the root of base data.
p.data_credentials_path = None
p.input_bucket_name = 'gtap_invest_seals_2023_04_21'
test_path = p.get_path('cartographic/gadm/gadm_410_adm0_labels_test.csv', verbose=True)
df = pd.read_csv(test_path)
assert len(df) > 0
hb.remove_path(test_path)
# Now try it WITH credentials
p.data_credentials_path = p.get_path('api_key_credentials.json')
test_path = p.get_path('cartographic/gadm/gadm_410_adm0_labels_test.csv', verbose=True)
df = pd.read_csv(test_path)
assert len(df) > 0
hb.remove_path(test_path)
test_get_reclassification_dict_from_df(self)
¶
Test reclassification dictionary generation from DataFrame
Source code in hazelbean_tests/integration/test_data_processing.py
def test_get_reclassification_dict_from_df(self):
"""Test reclassification dictionary generation from DataFrame"""
# Test that it does find a path that exists
p = hb.ProjectFlow(self.output_dir)
p.base_data_dir = '../../../base_data'
correspondence_path = p.get_path(os.path.join(self.data_dir, 'cartographic', 'ee', 'ee_r264_correspondence.csv'))
from hazelbean import utils
# TODO This should be extended to cover classification dicts from correspondences but also structured and unstructured mappings.
r = utils.get_reclassification_dict_from_df(correspondence_path, 'gtapv7_r160_id', 'gtapv7_r50_id', 'gtapv7_r160_label', 'gtapv7_r50_label')
hb.print_iterable(r)
test_clipping_simple(self)
¶
Test simple raster clipping operations
Source code in hazelbean_tests/integration/test_data_processing.py
def test_clipping_simple(self):
"""Test simple raster clipping operations"""
output_path = hb.temp('.tif', 'clipped', delete_on_finish, self.output_dir)
hb.clip_raster_by_vector_simple(self.ee_r264_ids_900sec_path,
output_path,
self.ee_r264_correspondence_vector_path,
output_data_type=6,
gtiff_creation_options=hb.DEFAULT_GTIFF_CREATION_OPTIONS)
print('Created', output_path)
output_dir = 'data'
output_path = hb.temp('.tif', 'clipped_attr', delete_on_finish, output_dir)
hb.clip_raster_by_vector_simple(self.ee_r264_ids_900sec_path,
output_path,
self.ee_r264_correspondence_vector_path,
output_data_type=6,
clip_vector_filter='ee_r264_id="120"',
gtiff_creation_options=hb.DEFAULT_GTIFF_CREATION_OPTIONS)
print('Created', output_path)
test_reclassify_raster_hb(self)
¶
Test raster reclassification with hazelbean
Source code in hazelbean_tests/integration/test_data_processing.py
test_reclassify_raster_with_negatives_hb(self)
¶
Test raster reclassification with negative values
Source code in hazelbean_tests/integration/test_data_processing.py
def test_reclassify_raster_with_negatives_hb(self):
"""Test raster reclassification with negative values"""
rules = {235: -555}
output_path = hb.temp('.tif', 'reclassify', False, self.output_dir)
hb.reclassify_raster_hb(self.ee_r264_ids_900sec_path,
rules,
output_path,
output_data_type=5)
print(hb.enumerate_raster_path(output_path))
output_with_neg_path = hb.temp('.tif', 'reclassify_with_neg', False, self.output_dir)
rules = {
235: -444,
241: -9999,
-555: -888,
} # Adding a rule for 241 to be reclassified to -9999
hb.reclassify_raster_hb(output_path,
rules,
output_with_neg_path,
output_data_type=5)
print(hb.enumerate_raster_path(output_with_neg_path))
test_reclassify_raster_arrayframe(self)
¶
Test raster reclassification with arrayframe
Source code in hazelbean_tests/integration/test_data_processing.py
ProjectFlow Integration Testing¶
Tests for the ProjectFlow framework, ensuring that project management and task execution work correctly together.
Key ProjectFlow Integration Tests: - ✅ Project initialization and setup workflows - ✅ Task dependency management and execution - ✅ Multi-step project processing pipelines
Parallel Processing Integration Testing¶
Tests for concurrent operations, thread safety, and parallel processing workflows.
Key Parallel Processing Tests: - ✅ Concurrent raster processing operations - ✅ Thread safety validation for shared resources - ✅ Parallel workflow performance testing
Running Integration Tests¶
To run the complete integration test suite:
# Activate the hazelbean environment
conda activate hazelbean_env
# Run all integration tests
pytest hazelbean_tests/integration/ -v
# Run specific integration test
pytest hazelbean_tests/integration/test_project_flow.py -v
# Run with detailed output
pytest hazelbean_tests/integration/ -v -s
# Run with timeout for long-running tests
pytest hazelbean_tests/integration/ --timeout=300
Test Characteristics¶
Integration tests typically:
- Take longer to run - Process real data and complete workflows
- Use realistic data - Work with actual geospatial datasets
- Test component interactions - Verify that modules work together correctly
- Validate workflows - Ensure end-to-end processing produces expected results
- Check resource usage - Monitor memory and computational requirements
Test Data¶
Integration tests use test data located in:
hazelbean_tests/data/- Sample datasets for testinghazelbean_tests/temp_test_data/- Temporary files created during testing- External data sources when testing real-world scenarios
Troubleshooting¶
Common integration test issues:
- Data availability - Ensure test datasets are present
- Environment setup - Verify all dependencies are installed
- Resource limits - Some tests may require significant memory or processing time
- Network access - Some tests may download test data
Related Test Categories¶
- Unit Tests → Understand individual components in Unit Tests
- Performance Tests → Measure workflow efficiency in Performance Tests
- System Tests → Validate complete system in System Tests
Test Dependencies¶
Integration tests build upon unit tests:
| Integration Test | Related Unit Tests | Purpose |
|---|---|---|
| test_project_flow.py | test_utils.py, test_os_funcs.py | End-to-end project workflows |
| test_data_processing.py | test_arrayframe.py, test_cog.py | Multi-step data processing |
| test_parallel_processing.py | test_utils.py | Concurrent operations |
| test_end_to_end_workflow.py | Multiple unit modules | Complete workflows |