98 lines
3.5 KiB
Python
Executable File
98 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
#
|
|
# This file is open source software, licensed to you under the terms
|
|
# of the Apache License, Version 2.0 (the "License"). See the NOTICE file
|
|
# distributed with this work for additional information regarding copyright
|
|
# ownership. 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.
|
|
#
|
|
#
|
|
# Copyright (C) 2024 Scylladb, Ltd.
|
|
#
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import urllib.request
|
|
import urllib.parse
|
|
|
|
|
|
class TestJson2Code(unittest.TestCase):
|
|
rest_api_httpd = None
|
|
server = None
|
|
port = 15789
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
args = [cls.rest_api_httpd, '--port', f'{cls.port}', '--smp=2']
|
|
cls.server = subprocess.Popen(args,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
bufsize=0, text=True)
|
|
# wait until the server is ready for serve
|
|
cls.server.stdout.readline()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.terminate()
|
|
|
|
def test_path_params(self):
|
|
var1 = 'bon'
|
|
var2 = 'jour'
|
|
query_enum = 'VAL2'
|
|
params = urllib.parse.urlencode({'query_enum': query_enum})
|
|
url = f'http://localhost:{self.port}/hello/world/{var1}/{var2}?{params}'
|
|
with urllib.request.urlopen(url) as f:
|
|
response = json.loads(f.read().decode('utf-8'))
|
|
self.assertEqual(response['var1'], f'/{var1}')
|
|
self.assertEqual(response['var2'], f'/{var2}')
|
|
self.assertEqual(response['enum_var'], query_enum)
|
|
|
|
def test_bad_enum(self):
|
|
var1 = 'bon'
|
|
var2 = 'jour'
|
|
query_enum = 'unknown value'
|
|
params = urllib.parse.urlencode({'query_enum': query_enum})
|
|
url = f'http://localhost:{self.port}/hello/world/{var1}/{var2}?{params}'
|
|
with urllib.request.urlopen(url) as f:
|
|
response = json.loads(f.read().decode('utf-8'))
|
|
self.assertEqual(response['var1'], f'/{var1}')
|
|
self.assertEqual(response['var2'], f'/{var2}')
|
|
self.assertEqual(response['enum_var'], 'Unknown')
|
|
|
|
def test_missing_path_param(self):
|
|
query_enum = 'VAL2'
|
|
params = urllib.parse.urlencode({'query_enum': query_enum})
|
|
url = f'http://localhost:{self.port}/hello/world/?{params}'
|
|
with self.assertRaises(urllib.error.HTTPError) as e:
|
|
with urllib.request.urlopen(url):
|
|
pass
|
|
ex = e.exception
|
|
self.assertEqual(ex.code, 404)
|
|
response = json.loads(ex.read().decode('utf-8'))
|
|
self.assertEqual(response['message'], 'Not found')
|
|
self.assertEqual(response['code'], 404)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--rest-api-httpd',
|
|
required=True,
|
|
help='Path of the rest_api_httpd executable')
|
|
opts, remaining = parser.parse_known_args()
|
|
remaining.insert(0, sys.argv[0])
|
|
TestJson2Code.rest_api_httpd = opts.rest_api_httpd
|
|
unittest.main(argv=remaining)
|