I am writing some unit tests to test reading in raster data into an xr.Dataset data structure via the rioxarray.open_rasterio() method. The code creates a test raster to read into rioxarray.open_rasterio() and then removes the file once the test finishes. The code works perfectly fine when I just load the raster as an xr.DataArray. It's able to create the raster and remove it upon completion. However, when I pass in the band_as_variable=True parameter into rioxarray.open_rasterio() to make the raster a xr.Dataset, the code doesn't remove the temporary raster. I believe the context manager being used here is not able to close the file so when it tries to remove it, it cannot. But why then would this code work when I don't pass in a value for band_as_variable? You should be able to copy and run this code on your own machine, assuming you have installed all of the packages I use in your own environment.
from abc import ABC, abstractmethod
import xarray as xr
import rasterio
import rioxarray
import os
import numpy as np
import unittest
import traceback
class RasterDataReaderInterface(ABC):
@abstractmethod
def read_data(self):
"""
Abstract method to read raster data from the source.
This method should be implemented in the derived classes.
"""
def test_read_data_not_implemented_error(self):
# Test that instantiating RasterDataReaderInterface directly raises a TypeError
with self.assertRaises(TypeError):
reader = RasterDataReaderInterface() # This should raise a TypeError
class TestLocalRasterReader(unittest.TestCase):
'''
Test cases for the LocalRasterReader class.
'''
def setUp(self) -> None:
''' Set up the unit tests for testing all of my methods.'''
# Create a temporary raster file for testing
self.temp_raster_file = "temp_raster.tif"
# Create a small temporary raster for testing
with rasterio.open(self.temp_raster_file, 'w', driver='GTiff', width=10, height=10, count=1, dtype='uint8') as dst:
dst.write_band(1, np.zeros((10, 10), dtype='uint8'))
self.temp_invalid_file = "invalid_file.txt" # A non-raster file
with open(self.temp_invalid_file, "w") as file:
pass
def tearDown(self) -> None:
'''
Clean up the temporary files used for unit testing.
'''
try:
# Remove the temporary raster file
if os.path.exists(self.temp_raster_file):
os.remove(self.temp_raster_file)
except Exception as e:
print(f"Error occurred while removing {self.temp_raster_file}: {e}")
try:
# Remove the temporary invalid file
if os.path.exists(self.temp_invalid_file):
os.remove(self.temp_invalid_file)
except Exception as e:
print(f"Error occurred while removing {self.temp_invalid_file}: {e}")
def test_read_data(self) -> None:
'''
Test reading in a raster file from a local machine.
- Assertions:
- assertIsNotNone if xr.DataArray is not None
- assertIsInstance if object is of instance xr.DataArray
'''
# Test reading raster data successfully
reader = LocalRasterReader(self.temp_raster_file)
xarray_data = reader.read_data()
# Assert that xarray_data is not None and is an instance of xarray DataArray. What about xarray DataSet?
self.assertIsNotNone(xarray_data)
self.assertIsInstance(xarray_data, xr.Dataset)
# Add more specific assertions based on your requirements
# For example, you could assert the shape of the xarray, or specific values
def no_test_read_data_file_not_found(self) -> None:
'''
Test if the raster file is not found when running read_data()
- Exceptions:
- rasterio.errors.RasterioIOError if the raster file is not found.
- Assertions:
- assertRaises if rasterio.errors.RasterioIOError is raised in read_data()
- assertTrue if the string "No such file or directory" appears in the rasterio.errors.RasterioIOError exception.
'''
non_existing_file = "non_existing_file.tif"
reader = LocalRasterReader(non_existing_file)
with self.assertRaises(rasterio.errors.RasterioIOError) as context:
reader.read_data()
self.assertTrue("No such file or directory" in str(context.exception))
def no_test_read_data_invalid_file(self) -> None:
'''
Test if the raster file found is invalid.
- Exceptions:
- rasterio.errors.RasterioIOError if the raster file is not recognized as a supported file format.
- Assertions:
- assertRaises if rasterio.errors.RasterioIOError is raised in read_data()
- assertTrue if the string "not recognized as a supported file format" appears in the rasterio.errors.RasterioIOError exception.
'''
# Test behavior when trying to read an invalid raster file
reader = LocalRasterReader(self.temp_invalid_file)
with self.assertRaises(rasterio.errors.RasterioIOError) as context:
reader.read_data()
self.assertTrue("not recognized as a supported file format" in str(context.exception))
class LocalRasterReader(RasterDataReaderInterface):
def __init__(self, file_path: str) -> None:
'''
Initialize a LocalRasterReader instance.
Parameters:
- file_path (str): The absolute path to the raster file.
Attributes:
- file_path (str): The absolute path to the raster file being read.
'''
self.file_path = file_path
def read_data(self) -> xr.Dataset:
try:
with rioxarray.open_rasterio(self.file_path, band_as_variable=True) as xarray_data:
return xarray_data
except rasterio.errors.RasterioIOError as e:
print(f"Error reading raster data from {e}:")
raise
class EarthEngineRasterReader(RasterDataReaderInterface):
def read_data(self):
# Implement code to stream raster data from Google Earth Engine
pass
if __name__ == '__main__':
unittest.main()
