Commit 57184ce6 authored by Daniele Venzano's avatar Daniele Venzano

Fix most of the pylint errors

parent 8c398980
#!/usr/bin/env bash
ZOE_TEST_IMAGE=${ZOE_TEST_IMAGE:-zoe:manual}
DOCKER_REGISTRY=${DOCKER_REGISTRY:-docker-registry:5000}
docker build --pull -t ${DOCKER_REGISTRY}/ci/${ZOE_TEST_IMAGE} .
docker push ${DOCKER_REGISTRY}/ci/${ZOE_TEST_IMAGE}
# Copyright (c) 2017, Daniele Venzano
#
# 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.
"""Create the DB tables needed by Zoe. This script is used in the CI pipeline to prevent race conditions with zoe-api automatically creating the tables while zoe-master is starting at the same time."""
import zoe_lib.config as config
import zoe_api.db_init
......
......@@ -2,5 +2,5 @@
set -e
pylint *.py zoe_* ci/*.py tests/*.py
pylint *.py zoe_* tests/*.py
doc8 docs/
import requests
import json
"""Test script for unsuccessful basic authentication."""
import time
import sys
import unittest
import requests
class ZoeRestTestSuccess(unittest.TestCase):
"""Test case class."""
uri = 'http://localhost:5001/api/0.7/'
auth = ('test', '1234')
auth = ('admin', 'admin')
wrong_auth = ('test', '123')
id = '-1'
def tearDown(self):
"""Test end."""
time.sleep(3)
def test_1_userinfo(self):
"""Test userinfo api endpoint."""
print('Test userinfo api endpoint')
r = requests.get(self.__class__.uri + 'userinfo', auth=self.__class__.wrong_auth)
self.assertEqual(r.status_code, 401)
req = requests.get(self.__class__.uri + 'userinfo', auth=self.__class__.wrong_auth)
self.assertEqual(req.status_code, 401)
def test_3_execution_details(self):
"""Test execution details api endpoint."""
print('Test execution details api endpoint')
r = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 404)
req = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 404)
def test_4_terminate_execution(self):
"""Test terminate no execution api endpoint."""
print('Test terminate no execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 404)
req = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 404)
def test_6_delete_execution(self):
"""Test delete execution api endpoint."""
print('Test delete execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 404)
req = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 404)
def test_2_start_failed_execution(self):
"""Test fail submit execution api endpoint."""
print('Test fail submit execution api endpoint')
r = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": "data", "name": "requests"})
self.assertEqual(r.status_code, 400)
req = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": "data", "name": "requests"})
self.assertEqual(req.status_code, 400)
if __name__ == '__main__':
unittest.main()
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
import requests
"""Test script for successful basic authentication."""
import json
import time
import sys
import unittest
import requests
class ZoeRestTestSuccess(unittest.TestCase):
"""Test case class."""
uri = ''
auth = ('admin', 'admin')
id = ''
def tearDown(self):
"""Test end."""
time.sleep(3)
def test_0_info(self):
"""Test info api endpoint."""
print('Test info api endpoint')
r = requests.get(self.__class__.uri + 'info')
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'info')
self.assertEqual(req.status_code, 200)
def test_1_userinfo(self):
"""Test userinfo api endpoint."""
print('Test userinfo api endpoint')
r = requests.get(self.__class__.uri + 'userinfo', auth=self.__class__.auth)
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'userinfo', auth=self.__class__.auth)
self.assertEqual(req.status_code, 200)
def test_3_execution_details(self):
"""Test execution details api endpoint."""
print('Test execution details api endpoint')
r = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 200)
def test_4_service_details(self):
"""Test service details api endpoint."""
print('Test service details api endpoint')
r = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
sid = r.json()["services"][0]
r = requests.get(self.__class__.uri + 'service/' + str(sid), auth=self.__class__.auth)
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
sid = req.json()["services"][0]
req = requests.get(self.__class__.uri + 'service/' + str(sid), auth=self.__class__.auth)
self.assertEqual(req.status_code, 200)
def test_5_terminate_execution(self):
"""Test terminate execution api endpoint."""
print('Test terminate execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 204)
req = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 204)
def test_6_list_all_executions(self):
"""Test list all executions api endpoint."""
print('Test list all executions api endpoint')
r = requests.get(self.__class__.uri + 'execution', auth=self.__class__.auth)
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'execution', auth=self.__class__.auth)
self.assertEqual(req.status_code, 200)
def test_7_delete_execution(self):
"""Test delete execution api endpoint."""
print('Test delete execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 204)
req = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 204)
def test_2_start_execution(self):
"""Test start execution api endpoint."""
print('Test start execution api endpoint')
data = []
with open('tf.json', encoding='utf-8') as data_file:
data = json.loads(data_file.read())
r = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": data, "name": "requests"})
self.assertEqual(r.status_code, 201)
self.__class__.id = str(r.json()['execution_id'])
req = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": data, "name": "requests"})
self.assertEqual(req.status_code, 201)
self.__class__.id = str(req.json()['execution_id'])
if __name__ == '__main__':
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
import requests
import json
"""Test script for unsuccessful cookie authentication."""
import sys
import time
import unittest
import requests
class ZoeRestTestSuccess(unittest.TestCase):
"""Test case class."""
uri = 'http://localhost:5001/api/0.7/'
id = '-1'
s = None
session = None
def tearDown(self):
"""Test end."""
time.sleep(3)
def test_0_login_fail(self):
"""Test failed login api endpoint."""
print('Test failed login api endpoint')
s = requests.Session()
r = s.get(self.__class__.uri + 'login', auth=('test','123'))
session = requests.Session()
req = session.get(self.__class__.uri + 'login', auth=('test', '123'))
self.assertEqual(r.status_code, 401)
self.assertEqual(req.status_code, 401)
def test_1_login(self):
"""Test login api endpoint."""
print('Test login api endpoint')
s = requests.Session()
r = s.get(self.__class__.uri + 'login', auth=('test','1234'))
session = requests.Session()
req = session.get(self.__class__.uri + 'login', auth=('test', '1234'))
self.assertEqual(r.status_code, 200)
self.assertEqual(req.status_code, 200)
self.assertGreater(len(s.cookies.items()), 0)
self.assertGreater(len(session.cookies.items()), 0)
self.__class__.s = s
self.__class__.session = session
def test_4_execution_details(self):
"""Test execution details api endpoint."""
print('Test execution details api endpoint')
s = self.__class__.s
r = s.get(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(r.status_code, 404)
session = self.__class__.session
req = session.get(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(req.status_code, 404)
def test_5_terminate_execution(self):
"""Test terminate execution api endpoint."""
print('Test terminate execution api endpoint')
s = self.__class__.s
r = s.delete(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(r.status_code, 404)
session = self.__class__.session
req = session.delete(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(req.status_code, 404)
def test_7_delete_execution(self):
"""Test delete execution api endpoint."""
print('Test delete execution api endpoint')
s = self.__class__.s
r = s.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id)
self.assertEqual(r.status_code, 404)
session = self.__class__.session
req = session.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id)
self.assertEqual(req.status_code, 404)
def test_3_start_execution(self):
"""Test start execution api endpoint."""
print('Test start execution api endpoint')
s = self.__class__.s
r = s.post(self.__class__.uri + 'execution', json={"application": "data", "name": "requests"})
self.assertEqual(r.status_code, 400)
session = self.__class__.session
req = session.post(self.__class__.uri + 'execution', json={"application": "data", "name": "requests"})
self.assertEqual(req.status_code, 400)
if __name__ == '__main__':
unittest.main()
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
import requests
"""Test script for successful cookie authentication."""
import json
import time
import sys
import unittest
import requests
class ZoeRestTestSuccess(unittest.TestCase):
"""Test case class."""
uri = 'http://192.168.12.2:5100/api/0.7/'
id = ''
s = None
session = None
def tearDown(self):
"""Test end."""
time.sleep(3)
def test_0_login(self):
"""Test login api endpoint."""
print('Test login api endpoint')
s = requests.Session()
r = s.get(self.__class__.uri + 'login', auth=('admin','admin'))
session = requests.Session()
req = session.get(self.__class__.uri + 'login', auth=('admin', 'admin'))
self.assertEqual(r.status_code, 200)
self.assertEqual(req.status_code, 200)
self.assertGreater(len(s.cookies.items()), 0)
self.assertGreater(len(session.cookies.items()), 0)
self.__class__.s = s
self.__class__.session = session
def test_1_info(self):
"""Test info api endpoint."""
print('Test info api endpoint')
s = self.__class__.s
session = self.__class__.session
req = session.get(self.__class__.uri + 'info')
r = s.get(self.__class__.uri + 'info')
self.assertEqual(r.status_code, 200)
self.assertEqual(req.status_code, 200)
def test_2_userinfo(self):
"""Test userinfo api endpoint."""
print('Test userinfo api endpoint')
s = self.__class__.s
session = self.__class__.session
r = s.get(self.__class__.uri + 'userinfo')
req = session.get(self.__class__.uri + 'userinfo')
self.assertEqual(r.status_code, 200)
self.assertEqual(req.status_code, 200)
def test_4_execution_details(self):
"""Test execution details api endpoint."""
print('Test execution details api endpoint')
s = self.__class__.s
r = s.get(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(r.status_code, 200)
session = self.__class__.session
req = session.get(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(req.status_code, 200)
def test_4_service_details(self):
print('Test service details api aendpoint')
s = self.__class__.s
r = s.get(self.__class__.uri + 'execution/' + self.__class__.id)
sid = r.json()["services"][0]
r = s.get(self.__class__.uri + 'service/' + str(sid))
self.assertEqual(r.status_code, 200)
"""Test service details api endpoint."""
print('Test service details api endpoint')
session = self.__class__.session
req = session.get(self.__class__.uri + 'execution/' + self.__class__.id)
sid = req.json()["services"][0]
req = session.get(self.__class__.uri + 'service/' + str(sid))
self.assertEqual(req.status_code, 200)
def test_5_terminate_execution(self):
"""Test terminate execution api endpoint."""
print('Test terminate execution api endpoint')
s = self.__class__.s
r = s.delete(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(r.status_code, 204)
session = self.__class__.session
req = session.delete(self.__class__.uri + 'execution/' + self.__class__.id)
self.assertEqual(req.status_code, 204)
def test_6_list_all_executions(self):
"""Test list all executions api endpoint."""
print('Test list all executions api endpoint')
s = self.__class__.s
r = s.get(self.__class__.uri + 'execution')
self.assertEqual(r.status_code, 200)
session = self.__class__.session
req = session.get(self.__class__.uri + 'execution')
self.assertEqual(req.status_code, 200)
def test_7_delete_execution(self):
"""Test delete execution api endpoint."""
print('Test delete execution api endpoint')
s = self.__class__.s
r = s.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id)
self.assertEqual(r.status_code, 204)
session = self.__class__.session
req = session.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id)
self.assertEqual(req.status_code, 204)
def test_3_start_execution(self):
"""Test start execution api endpoint."""
print('Test start execution api endpoint')
data = []
with open('tf.json', encoding='utf-8') as data_file:
data = json.loads(data_file.read())
s = self.__class__.s
r = s.post(self.__class__.uri + 'execution', json={"application": data, "name": "requests"})
self.assertEqual(r.status_code, 201)
self.__class__.id = str(r.json()['execution_id'])
session = self.__class__.session
req = session.post(self.__class__.uri + 'execution', json={"application": data, "name": "requests"})
self.assertEqual(req.status_code, 201)
self.__class__.id = str(req.json()['execution_id'])
if __name__ == '__main__':
unittest.main()
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
import requests
"""Corner test."""
import json
import sys
import time
import unittest
import requests
class ZoeRestTestSuccess(unittest.TestCase):
"""Test case class."""
uri = 'http://localhost:5001/api/0.6/'
auth = ('test', '1234')
id = ''
def tearDown(self):
"""Test end."""
time.sleep(3)
def test_3_execution_details(self):
"""Test execution details api endpoint."""
print('Test execution details api endpoint')
r = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 200)
req = requests.get(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 200)
def test_4_terminate_execution(self):
"""Test terminate no execution api endpoint."""
print('Test terminate no execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 400)
req = requests.delete(self.__class__.uri + 'execution/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 400)
def test_6_delete_execution(self):
"""Test delete execution api endpoint."""
print('Test delete execution api endpoint')
r = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(r.status_code, 204)
req = requests.delete(self.__class__.uri + 'execution/delete/' + self.__class__.id, auth=self.__class__.auth)
self.assertEqual(req.status_code, 204)
def test_2_start_failed_execution(self):
"""Test fail submit execution api endpoint."""
print('Test fail submit execution api endpoint')
data = []
with open('dummy.json', encoding='utf-8') as data_file:
data = json.loads(data_file.read())
r = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": data, "name": "requests"})
self.assertEqual(r.status_code, 201)
self.__class__.id = str(r.json()['execution_id'])
req = requests.post(self.__class__.uri + 'execution', auth=self.__class__.auth, json={"application": data, "name": "requests"})
self.assertEqual(req.status_code, 201)
self.__class__.id = str(req.json()['execution_id'])
if __name__ == '__main__':
unittest.main()
if len(sys.argv) > 1:
API_SERVER = sys.argv.pop()
ZoeRestTestSuccess.uri = 'http://' + API_SERVER + '/api/0.7/'
unittest.main()
......@@ -25,7 +25,7 @@ import tornado.web
from zoe_lib.config import get_conf
from zoe_api.exceptions import ZoeRestAPIException, ZoeNotFoundException, ZoeAuthException, ZoeException
from zoe_api.auth.base import BaseAuthenticator
from zoe_api.auth.base import BaseAuthenticator # pylint-bug #1063 pylint: disable=unused-import
from zoe_api.auth.ldap import LDAPAuthenticator
from zoe_api.auth.file import PlainTextAuthenticator
from zoe_api.auth.ldapsasl import LDAPSASLAuthenticator
......
......@@ -22,7 +22,6 @@ import re
from zoe_lib.exceptions import InvalidApplicationDescription
import zoe_lib.version
from zoe_lib.config import get_conf
log = logging.getLogger(__name__)
......
......@@ -53,6 +53,7 @@ except AttributeError:
log.error('Docker package does not have the DockerClient attribute')
raise ImportError('Wrong Docker library version')
def zookeeper_swarm(zk_server_list: str, path='/docker') -> str:
"""
Given a Zookeeper server list, find the currently active Swarm master.
......
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