Skip to content

Commit

Permalink
remove uneeded tests
Browse files Browse the repository at this point in the history
  • Loading branch information
Robert Armstrong committed Jun 3, 2019
1 parent 1a65267 commit ffb675e
Showing 1 changed file with 4 additions and 95 deletions.
99 changes: 4 additions & 95 deletions tests/test_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from unittest.mock import patch, MagicMock
from tests.test_band_mapper import product_layer
from datetime import datetime

import numpy as np

Expand Down Expand Up @@ -35,87 +36,8 @@ def test_s3_browser_uris(s3_url_datasets):
assert "http://test-bucket.s3-website-ap-southeast-2.amazonaws.com/?prefix=hello_world" in uris
assert "http://test-bucket.s3-website-ap-southeast-2.amazonaws.com/?prefix=hello.word/foo.bar" in uris

from datacube_wms.data import _make_destination

def test_make_destination():
dest = _make_destination((256, 256), -1, "int16")

assert dest.shape == (256, 256)
assert (dest == -1).all()
assert dest.dtype == np.dtype("int16")

def test_get_measurement():

class fakegeobox:
def __init__(self, shape):
self.shape = shape

class fakesource:
def __init__(self, band=1):
self.band = band
def get_bandnumber(self):
return self.band

def fake_delayed(some_callable):
return some_callable

def fake_as_delayed():
class fakeda:
def from_delayed(self, destination, shape, dtype):
return destination
return fakeda()

def fake_read_file(source, geobox, band, no_data, resampling):
return np.zeros(geobox.shape).astype("int16")

gb = fakegeobox((256, 256))
sources = [ fakesource() ]

# test basic
with patch('datacube_wms.data._read_file', new_callable=lambda: fake_read_file) as rf, patch('datacube_wms.data.da', new_callable=fake_as_delayed) as da, patch('datacube_wms.data.delayed', new_callable=lambda: fake_delayed) as delayed:
result = datacube_wms.data._get_measurement(sources, gb, None, -1, "int16", fuse_func=None)

assert (result == 0).all()
assert result.shape == (256, 256)

# Test fuse funcing
sources = [ fakesource(1), fakesource(2) ]
gb = fakegeobox((16,16))

def fake_read_file2(source, geobox, band, no_data, resampling):
if source.get_bandnumber() == 1:
return np.full((16,16), np.nan)
else:
return np.full((16,16), 1., "float64")

with patch('datacube_wms.data._read_file', new_callable=lambda: fake_read_file2) as rf, patch('datacube_wms.data.da', new_callable=fake_as_delayed) as da, patch('datacube_wms.data.delayed', new_callable=lambda: fake_delayed) as delayed:
result = datacube_wms.data._get_measurement(sources, gb, None, np.nan, "float64", fuse_func=None)

assert (result == 1.).all()
assert result.shape == (16,16)

# Test custom fuse func
sources = [ fakesource(1), fakesource(2) ]

def fake_read_file3(source, geobox, band, no_data, resampling):
if source.get_bandnumber() == 1:
return np.full((16,16), 56)
else:
return np.full((16,16), 20)

def fake_fuse(dest, src):
where_src_56 = (src == 56)
np.copyto(dest, src, where=where_src_56)

with patch('datacube_wms.data._read_file', new_callable=lambda: fake_read_file3) as rf, patch('datacube_wms.data.da', new_callable=fake_as_delayed) as da, patch('datacube_wms.data.delayed', new_callable=lambda: fake_delayed) as delayed:
result = datacube_wms.data._get_measurement(sources, gb, None, -1, "int16", fuse_func=None)

assert (result == 56).all()
assert result.shape == (16,16)

@patch('datacube_wms.data.new_datasource')
@patch('xarray.Dataset')
def test_read_data(dataset, new_datasource):
def test_read_data(dataset):

class fake_coords:
def __init__(self):
Expand Down Expand Up @@ -144,28 +66,15 @@ def __getitem__(self, item):

class fake_dataset:
def __init__(self):
self.center_time = datetime.utcnow()
self.id = 1

class fake_datasource:
def __init__(self):
self._dataset = fake_dataset()

# Test overviews / no overviews

datasets = [ fake_dataset() ]
measurements = [ fake_measurement("test", -1, "int16") ]
geobox = fakegeobox()
with patch('datacube.Datacube.load_data') as load_data, patch('datacube_wms.data._get_measurement') as get_measurement:
with patch('datacube.Datacube.load_data') as load_data:
datacube_wms.data.read_data(datasets, measurements, geobox)

assert load_data.called
assert not get_measurement.called

with patch('datacube.Datacube.load_data') as load_data, patch('datacube_wms.data._get_measurement') as get_measurement:
datacube_wms.data.read_data(datasets, measurements, geobox)

assert not load_data.called
assert get_measurement.called

def test_make_derived_band_dict_nan():
class fake_data:
Expand Down

0 comments on commit ffb675e

Please sign in to comment.