Commit 40357852 authored by Daniele Venzano's avatar Daniele Venzano
Browse files

Fix tests

parent 88a8e3c4
import json
import pytest
from zoe_scheduler.state import init as state_init, Base, AlchemySession
from zoe_client.state.application import ApplicationState
from zoe_scheduler.application_description import ZoeApplication
from zoe_scheduler.configuration import init as conf_init, scheduler_conf
def pytest_addoption(parser):
parser.addoption("--test-environment", default="local", help="Test environment: 'local' or 'travis'")
@pytest.fixture(scope='session')
def environment(request):
return request.config.getoption("--test-environment")
@pytest.fixture(scope='session')
def configuration(environment):
if environment == 'local':
conf_init('tests/resources/zoe-local.conf')
else:
conf_init('tests/resources/zoe-travis.conf')
@pytest.fixture(scope='session')
def state_connection(request, configuration):
engine = state_init(scheduler_conf().db_url)
connection = engine.connect()
trans = connection.begin()
Base.metadata.create_all(connection)
def fin():
trans.rollback()
connection.close()
engine.dispose()
request.addfinalizer(fin)
return connection
@pytest.fixture(scope='function')
def state_session(state_connection, request):
inner_trans = state_connection.begin_nested()
session = AlchemySession(bind=state_connection)
def fin():
session.close()
inner_trans.rollback()
request.addfinalizer(fin)
return session
@pytest.fixture(scope='function')
def application(state_session, notebook_test):
app = ApplicationState()
app.user_id = 1
app.description = ZoeApplication.from_dict(notebook_test)
state_session.add(app)
state_session.flush()
return app
@pytest.fixture(scope='session')
def notebook_test():
jsondata = open("tests/resources/spark-notebook-test.json", "r")
dictdata = json.load(jsondata)
return dictdata
......@@ -4,9 +4,6 @@ docker_private_registry = 10.1.0.1:5000
status_refresh_interval = 10
check_terminated_interval = 30
db_connect = mysql+mysqlconnector://root@localhost/zoe
storage_path = /tmp/zoe
http_listen_address = 192.168.45.25
http_listen_port = 4390
ipc_listen_address = 127.0.0.1
ipc_listen_port = 8723
object_storage_url = http://localhost:4390
......@@ -23,3 +20,8 @@ smtp_user = user
smtp_server = smtp.example.com
cookie_secret = \xc3\xb0\xa7\xaa\x8fH'\xf7m\x1c\xa2\x92F\x1d\xdcz\x05\xe6CJN5\x83!
web_server_name = localhost
[zoe_storage]
storage_path = /tmp/zoe
http_listen_address = 192.168.45.25
http_listen_port = 4390
......@@ -23,3 +23,8 @@ smtp_user = user
smtp_server = smtp.example.com
cookie_secret = \xc3\xb0\xa7\xaa\x8fH'\xf7m\x1c\xa2\x92F\x1d\xdcz\x05\xe6CJN5\x83!
web_server_name = localhost
[zoe_storage]
storage_path = /tmp/zoe
http_listen_address = 192.168.45.25
http_listen_port = 4390
fake_data = b"test" * 1024
def test_application_data_upload(application):
ret = init_history_paths()
assert ret is True
application_data_upload(application, fake_data)
data = application_data_download(application)
assert data == fake_data
application_data_delete(application)
......@@ -10,7 +10,7 @@ class ZoeObjectStoreHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
"""
def _read_request(self):
zos = ZoePersistentObjectStore()
zos = ZoePersistentObjectStore(storage_conf().storage_path)
req = self.path.split("/")[1:]
try:
obj_type = req[0]
......@@ -56,7 +56,7 @@ class ZoeObjectStoreHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
self.send_error(400, "Expected form-data POST request")
return
zos = ZoePersistentObjectStore()
zos = ZoePersistentObjectStore(storage_conf().storage_path)
req = self.path.split("/")[1:]
try:
obj_type = req[0]
......@@ -79,7 +79,7 @@ class ZoeObjectStoreHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
self.end_headers()
def do_DELETE(self):
zos = ZoePersistentObjectStore()
zos = ZoePersistentObjectStore(storage_conf().storage_path)
req = self.path.split("/")[1:]
try:
obj_type = req[0]
......
......@@ -3,14 +3,12 @@ import logging
import os
from shutil import copyfileobj
from zoe_storage_server.configuration import storage_conf
log = logging.getLogger(__name__)
class ZoePersistentObjectStore:
def __init__(self):
self.base_path = storage_conf().storage_path
def __init__(self, storage_path):
self.base_path = storage_path
self.objects_path = os.path.join(self.base_path, 'objects')
self._init_paths()
......
import json
import pytest
from zoe_storage_server.configuration import conf_init
def pytest_addoption(parser):
parser.addoption("--test-environment", default="local", help="Test environment: 'local' or 'travis'")
@pytest.fixture(scope='session')
def configuration(request):
env = request.config.getoption("--test-environment")
if env == 'local':
return conf_init('tests/resources/zoe-local.conf')
else:
return conf_init('tests/resources/zoe-travis.conf')
@pytest.fixture(scope='session')
def notebook_test():
jsondata = open("tests/resources/spark-notebook-test.json", "r")
dictdata = json.load(jsondata)
return dictdata
from io import BytesIO
import os
from zoe_storage_server.object_storage import ZoePersistentObjectStore
def test_application_data_upload(configuration):
fake_data = BytesIO(b"test" * 1024)
zos = ZoePersistentObjectStore(configuration.storage_path)
filepath = os.path.join(zos.objects_path, "app-1.zip")
zos.application_data_upload(1, fake_data)
assert os.path.exists(filepath)
data = zos.application_data_download(1)
assert data == fake_data.getvalue()
zos.application_data_delete(1)
assert not os.path.exists(filepath)
def test_logs_upload(configuration):
fake_data = BytesIO(b"test" * 1024)
zos = ZoePersistentObjectStore(configuration.storage_path)
filepath = os.path.join(zos.objects_path, "log-1.zip")
zos.logs_archive_upload(1, fake_data)
assert os.path.exists(filepath)
data = zos.logs_archive_download(1)
assert data == fake_data.getvalue()
zos.logs_archive_delete(1)
assert not os.path.exists(filepath)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment