Commit 70d9a10f authored by Xavier Thompson's avatar Xavier Thompson

software/jupyter: fix slap parameters

parent c2a7ed05
......@@ -19,7 +19,7 @@ md5sum = de37ec3d4adb0be4c67bcc7397f27c91
[instance-jupyter]
filename = instance-jupyter.cfg.in
md5sum = 9340498841caa5771f40f8c9e561eacd
md5sum = 38bbb350a07ad2bb9f01b46122231708
[jupyter-notebook-config]
filename = jupyter_notebook_config.py.jinja
......
......@@ -51,6 +51,10 @@ frontend-additional-software-url = http://git.erp5.org/gitweb/slapos.git/blob_pl
frontend-additional-instance-guid =
frontend-additional-instance-name = Jupyter Frontend Additional
{% for k, v in slapparameter_dict.items() -%}
{{ k }} = {{ v }}
{% endfor -%}
[dynamic-jinja2-template-base]
recipe = slapos.recipe.template:jinja2
mode = 0644
......
......@@ -30,6 +30,8 @@ import http.client
import json
import os
import requests
import sqlite3
from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass
......@@ -85,36 +87,12 @@ class TestJupyter(InstanceTestCase):
)
class TestJupyterPassword(InstanceTestCase):
def test(self):
parameter_dict = self.computer_partition.getConnectionParameterDict()
self.assertTrue('_' in parameter_dict)
try:
connection_dict = json.loads(parameter_dict['_'])
except Exception as e:
self.fail("Can't parse json in %s, error %s" % (parameter_dict['_'], e))
url = connection_dict['url']
with requests.Session() as s:
resp = s.get(url, verify=False)
result = s.post(
resp.url,
verify = False,
data={"_xsrf": s.cookies["_xsrf"], "password": connection_dict['password']}
)
self.assertEqual(
[http.client.OK, url],
[result.status_code, result.url]
)
class TestJupyterAdditional(InstanceTestCase):
@classmethod
def getInstanceParameterDict(cls):
return {
'frontend-additional-instance-guid': 'SOMETHING'
}
'frontend-additional-instance-guid': 'SOMETHING' }
def test(self):
parameter_dict = self.computer_partition.getConnectionParameterDict()
......@@ -169,3 +147,144 @@ class TestJupyterAdditional(InstanceTestCase):
[http.client.FOUND, True, '/login?next=%2Flab'],
[result.status_code, result.is_redirect, result.headers['Location']]
)
class TestJupyterPassword(InstanceTestCase):
def test(self):
parameter_dict = self.computer_partition.getConnectionParameterDict()
self.assertTrue('_' in parameter_dict)
try:
connection_dict = json.loads(parameter_dict['_'])
except Exception as e:
self.fail("Can't parse json in %s, error %s" % (parameter_dict['_'], e))
url = connection_dict['url']
with requests.Session() as s:
resp = s.get(url, verify=False)
result = s.post(
resp.url,
verify = False,
data={"_xsrf": s.cookies["_xsrf"], "password": connection_dict['password']}
)
self.assertEqual(
[http.client.OK, url],
[result.status_code, result.url]
)
class SelectMixin(object):
def sqlite3_connect(self):
sqlitedb_file = os.path.join(
os.path.abspath(
os.path.join(
self.slap.instance_directory, os.pardir
)
), 'var', 'proxy.db'
)
return sqlite3.connect(sqlitedb_file)
def select(self, fields, table, where={}):
connection = self.sqlite3_connect()
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
connection.row_factory = dict_factory
cursor = connection.cursor()
condition = " AND ".join("%s='%s'" % (k, v) for k, v in where.items())
cursor.execute(
"SELECT %s FROM %s%s"
% (
", ".join(fields),
table,
" WHERE %s" % condition if where else "",
)
)
return cursor.fetchall()
class TestJupyterCustomFrontend(SelectMixin, InstanceTestCase):
instance_parameter_dict = {}
frontend_software_url = 'hello://frontend.url'
frontend_software_type = 'hello-type'
frontend_instance_name = 'Hello Frontend'
@classmethod
def getInstanceParameterDict(cls):
return cls.instance_parameter_dict
def test(self):
# create a fake master instance for the frontend slave request
r = self.slap.request(
software_release=self.frontend_software_url,
software_type=self.frontend_software_type,
partition_reference= "Fake master instance",
)
# update the request parameters of the test instance
self.instance_parameter_dict.update({
'frontend-software-url': self.frontend_software_url,
'frontend-software-type': self.frontend_software_type,
'frontend-instance-name': self.frontend_instance_name,
'frontend-instance-guid': r._partition_id,
})
self.requestDefaultInstance()
# wait for the instance to converge to the new state
try:
self.slap.waitForInstance()
except Exception:
pass
selection = self.select(fields=["*"], table = "slave14", where = {"hosted_by": r._partition_id})
self.assertEqual(len(selection), 1)
# clean up the fake master
r.destroyed()
class TestJupyterCustomAdditional(SelectMixin, InstanceTestCase):
instance_parameter_dict = {}
frontend_additional_software_url = 'hello://frontend.url'
frontend_additional_software_type = 'hello-type'
frontend_additional_instance_name = 'Hello Frontend'
@classmethod
def getInstanceParameterDict(cls):
return cls.instance_parameter_dict
def test(self):
# create a fake master instance for the frontend slave request
r = self.slap.request(
software_release=self.frontend_additional_software_url,
software_type=self.frontend_additional_software_type,
partition_reference= "Fake master instance",
)
# update the request parameters of the test instance
self.instance_parameter_dict.update({
'frontend-additional-software-url': self.frontend_additional_software_url,
'frontend-additional-software-type': self.frontend_additional_software_type,
'frontend-additional-instance-name': self.frontend_additional_instance_name,
'frontend-additional-instance-guid': r._partition_id,
})
self.requestDefaultInstance()
# wait for the instance to converge to the new state
try:
self.slap.waitForInstance()
except Exception:
pass
selection = self.select(fields=["*"], table = "slave14", where = {"hosted_by": r._partition_id})
self.assertEqual(len(selection), 1)
# clean up the fake master
r.destroyed()
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