Move FakeStorletFile* from tests to stolets/tools
This is prep for the fundamentaly dry-run supports in ipython extension. The classes FakeStorletFile* in the unit test cases should be useful to run in local environment so move them under the packaging directry (i.e. storlets/tools/testtools.py), and then let's use them in our python storlet local environment. And add some unit tests to get them into the storlets package! Change-Id: Id6e8db79fe63031fd1d499f88ea832c5fb4a0363
This commit is contained in:
parent
e727b49507
commit
0200b993cc
62
storlets/tools/testtools.py
Normal file
62
storlets/tools/testtools.py
Normal file
@ -0,0 +1,62 @@
|
||||
# Copyright (c) 2015, 2016 OpenStack Foundation.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from cStringIO import StringIO
|
||||
|
||||
|
||||
class FakeStorletFile(object):
|
||||
def __init__(self):
|
||||
self._call_closed = False
|
||||
|
||||
def close(self):
|
||||
self._call_closed = True
|
||||
|
||||
@property
|
||||
def closed(self):
|
||||
return self._call_closed
|
||||
|
||||
|
||||
class FakeStorletFileIn(FakeStorletFile):
|
||||
def __init__(self, input_string, metadata):
|
||||
super(FakeStorletFileIn, self).__init__()
|
||||
self._input_string = StringIO(input_string)
|
||||
self._metadata = metadata
|
||||
self._pos = 0
|
||||
|
||||
def read(self, size=-1):
|
||||
return self._input_string.read(size)
|
||||
|
||||
def get_metadata(self):
|
||||
return self._metadata
|
||||
|
||||
|
||||
class FakeStorletFileOut(FakeStorletFile):
|
||||
def __init__(self):
|
||||
super(FakeStorletFileOut, self).__init__()
|
||||
self._output_string = []
|
||||
self._metadata = None
|
||||
|
||||
def write(self, data):
|
||||
self._output_string.append(data)
|
||||
|
||||
def set_metadata(self, metadata):
|
||||
if self._metadata is not None:
|
||||
raise IOError('Sending metadata twice is not allowed')
|
||||
self._metadata = {}
|
||||
# expect the incomming metadata should be dict
|
||||
self._metadata.update(metadata)
|
||||
|
||||
def read(self):
|
||||
return ''.join(self._output_string)
|
@ -14,54 +14,10 @@
|
||||
# limitations under the License.
|
||||
|
||||
from unittest import TestCase
|
||||
from cStringIO import StringIO
|
||||
from storlet_samples.simple.simple import SimpleStorlet
|
||||
import unittest
|
||||
from storlets.tools.testtools import FakeStorletFileIn, FakeStorletFileOut
|
||||
from tests.unit import FakeLogger
|
||||
|
||||
|
||||
class FakeStorletFile(object):
|
||||
def __init__(self):
|
||||
self._call_closed = False
|
||||
|
||||
def close(self):
|
||||
self._call_closed = True
|
||||
|
||||
@property
|
||||
def closed(self):
|
||||
return self._call_closed
|
||||
|
||||
|
||||
class FakeStorletFileIn(FakeStorletFile):
|
||||
def __init__(self, input_string, metadata):
|
||||
super(FakeStorletFileIn, self).__init__()
|
||||
self._input_string = StringIO(input_string)
|
||||
self._metadata = metadata
|
||||
self._pos = 0
|
||||
|
||||
def read(self, size=-1):
|
||||
return self._input_string.read(size)
|
||||
|
||||
def get_metadata(self):
|
||||
return self._metadata
|
||||
|
||||
|
||||
class FakeStorletFileOut(FakeStorletFile):
|
||||
def __init__(self):
|
||||
super(FakeStorletFileOut, self).__init__()
|
||||
self._output_string = []
|
||||
self._metadata = None
|
||||
|
||||
def write(self, data):
|
||||
self._output_string.append(data)
|
||||
|
||||
def set_metadata(self, metadata):
|
||||
if self._metadata is None:
|
||||
self._metadata = {}
|
||||
self._metadata.update(metadata)
|
||||
|
||||
def read(self):
|
||||
return ''.join(self._output_string)
|
||||
import unittest
|
||||
|
||||
|
||||
class TestSimpleStorlet(TestCase):
|
||||
|
56
tests/unit/tools/test_testtools.py
Normal file
56
tests/unit/tools/test_testtools.py
Normal file
@ -0,0 +1,56 @@
|
||||
# Copyright (c) 2015, 2016 OpenStack Foundation.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest import TestCase
|
||||
|
||||
from storlets.tools.testtools import FakeStorletFileIn, FakeStorletFileOut
|
||||
|
||||
|
||||
class TestTestTools(TestCase):
|
||||
def test_fake_storlet_file_in(self):
|
||||
input_string = 'abcdefghijklmonp'
|
||||
input_metadata = {'foo': 'bar'}
|
||||
fake_storlet_in = FakeStorletFileIn(input_string, input_metadata)
|
||||
self.assertEqual(input_string, fake_storlet_in.read())
|
||||
self.assertEqual(input_metadata, fake_storlet_in.get_metadata())
|
||||
|
||||
fake_storlet_in.close()
|
||||
self.assertTrue(fake_storlet_in.closed)
|
||||
|
||||
def test_fake_storlet_file_out(self):
|
||||
out_string = 'abcdefghijklmonp'
|
||||
fake_storlet_out = FakeStorletFileOut()
|
||||
fake_storlet_out.write(out_string)
|
||||
self.assertEqual(out_string, fake_storlet_out.read())
|
||||
fake_storlet_out.write(out_string)
|
||||
self.assertEqual(out_string * 2, fake_storlet_out.read())
|
||||
|
||||
self.assertEqual(None, fake_storlet_out._metadata)
|
||||
fake_storlet_out.set_metadata({'test': 'tester'})
|
||||
self.assertEqual({'test': 'tester'}, fake_storlet_out._metadata)
|
||||
|
||||
fake_storlet_out.close()
|
||||
self.assertTrue(fake_storlet_out.closed)
|
||||
|
||||
def test_fake_storlet_file_out_set_metadata_twice_will_cause_error(self):
|
||||
fake_storlet_out = FakeStorletFileOut()
|
||||
# first call, it's ok
|
||||
fake_storlet_out.set_metadata({})
|
||||
# second call will raise an exception
|
||||
with self.assertRaises(IOError) as cm:
|
||||
fake_storlet_out.set_metadata({})
|
||||
self.assertEqual(
|
||||
"Sending metadata twice is not allowed",
|
||||
cm.exception.message)
|
Loading…
Reference in New Issue
Block a user