baha/skyline-apiserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: cleanup some invalid tests

Browse Source 
Some tests codes are not used, so just to cleanup them.

Change-Id: I9596a54705150271ae6900a17448d9250af4f2db
This commit is contained in:
zhu.boxiang 2022-07-12 09:52:49 +08:00
parent 22a2895ba9
commit 1b17da1b2c
14 changed files with 0 additions and 906 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
skyline_apiserver/tests/api/__init__.py
View File

0
skyline_apiserver/tests/api/v1/__init__.py
View File

49
skyline_apiserver/tests/api/v1/test_contrib.py
View File

@ -1,49 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import os
import pytest
from httpx import AsyncClient
from skyline_apiserver import main
from skyline_apiserver.config import CONF
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_regions(client: AsyncClient) -> None:
r = await client.get(url=f"{main.API_PREFIX}/contrib/regions")
result = r.json()
assert r.status_code == 200
assert len(result) > 0
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_domains(client: AsyncClient) -> None:
r = await client.get(url=f"{main.API_PREFIX}/contrib/domains")
result = r.json()
assert r.status_code == 200
assert len(result) > 0
assert result[0] not in CONF.openstack.base_domains
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_keystone_endpoints(client: AsyncClient) -> None:
r = await client.get(url=f"{main.API_PREFIX}/contrib/keystone_endpoints")
result = r.json()
assert r.status_code == 200
assert len(result) > 0

278
skyline_apiserver/tests/api/v1/test_extension.py
View File

@ -1,278 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import os
import uuid
from typing import Any, Dict
import pytest
from httpx import AsyncClient
from six.moves.urllib import parse
from skyline_apiserver import main
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"limit": 10}, 200),
({"limit": 20, "marker": str(uuid.uuid4())}, 400),
({"sort_dirs": "desc", "sort_keys": "uuid"}, 200),
({"sort_dirs": "desc", "sort_keys": "display_name"}, 200),
({"sort_dirs": "desc", "sort_keys": "vm_state"}, 200),
({"sort_dirs": "desc", "sort_keys": "locked"}, 200),
({"sort_dirs": "desc", "sort_keys": "created_at"}, 200),
({"sort_dirs": "desc", "sort_keys": "host"}, 200),
({"sort_dirs": "desc", "sort_keys": "project_id"}, 200),
({"sort_dirs": "desc", "sort_keys": "abc123"}, 422),
({"all_projects": True}, 200),
({"all_projects": True, "project_id": str(uuid.uuid4())}, 200),
({"all_projects": True, "project_name": "test-project"}, 200),
(
{
"all_projects": True,
"project_id": str(uuid.uuid4()),
"project_name": "test-project",
},
200,
),
({"name": "abc123"}, 200),
({"status": "ACTIVE"}, 200),
({"host": "host01"}, 200),
({"flavor_id": "abc123"}, 200),
({"uuid": [str(uuid.uuid4())]}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_servers(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
# list servers
url = f"{main.API_PREFIX}/extension/servers" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "servers" in result
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"limit": 10}, 200),
({"limit": 20, "marker": str(uuid.uuid4())}, 400),
({"sort_dirs": "desc", "sort_keys": "uuid"}, 200),
({"sort_dirs": "desc", "sort_keys": "display_name"}, 200),
({"sort_dirs": "desc", "sort_keys": "updated_at"}, 200),
({"sort_dirs": "desc", "sort_keys": "project_id"}, 200),
({"sort_dirs": "desc", "sort_keys": "abc123"}, 422),
({"all_projects": True}, 200),
({"all_projects": True, "project_id": str(uuid.uuid4())}, 200),
({"all_projects": True, "project_name": "test-project"}, 200),
(
{
"all_projects": True,
"project_id": str(uuid.uuid4()),
"project_name": "test-project",
},
200,
),
({"name": "abc123"}, 200),
({"uuid": [str(uuid.uuid4())]}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_recycle_servers(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
# list recycle servers
url = f"{main.API_PREFIX}/extension/recycle_servers" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "recycle_servers" in result
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"limit": 10}, 200),
({"limit": 20, "marker": str(uuid.uuid4())}, 500),
({"sort_dirs": "desc", "sort_keys": "id"}, 200),
({"sort_dirs": "desc", "sort_keys": "name"}, 200),
({"sort_dirs": "desc", "sort_keys": "size"}, 200),
({"sort_dirs": "desc", "sort_keys": "status"}, 200),
({"sort_dirs": "desc", "sort_keys": "bootable"}, 200),
({"sort_dirs": "desc", "sort_keys": "created_at"}, 200),
({"sort_dirs": "desc", "sort_keys": "abc123"}, 422),
({"all_projects": True}, 200),
({"all_projects": True, "project_id": str(uuid.uuid4())}, 200),
({"name": "abc123"}, 200),
({"multiattach": True}, 200),
({"status": "available"}, 200),
({"uuid": [str(uuid.uuid4())]}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_volumes(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
# list volumes
url = f"{main.API_PREFIX}/extension/volumes" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "volumes" in result
assert "count" in result
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"limit": 10}, 200),
({"limit": 20, "marker": str(uuid.uuid4())}, 500),
({"sort_dirs": "desc", "sort_keys": "id"}, 200),
({"sort_dirs": "desc", "sort_keys": "name"}, 200),
({"sort_dirs": "desc", "sort_keys": "status"}, 200),
({"sort_dirs": "desc", "sort_keys": "created_at"}, 200),
({"sort_dirs": "desc", "sort_keys": "abc123"}, 422),
({"all_projects": True}, 200),
({"all_projects": True, "project_id": str(uuid.uuid4())}, 200),
({"name": "abc123"}, 200),
({"status": "AVAILABLE"}, 200),
({"volume_id": str(uuid.uuid4())}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_volume_snapshots(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
# list volume snapshots
url = f"{main.API_PREFIX}/extension/volume_snapshots" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "volume_snapshots" in result
assert "count" in result
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"limit": 10}, 200),
({"limit": 20, "marker": str(uuid.uuid4())}, 404),
({"sort_dirs": "desc", "sort_keys": "id"}, 200),
({"sort_dirs": "desc", "sort_keys": "name"}, 200),
({"sort_dirs": "desc", "sort_keys": "mac_address"}, 200),
({"sort_dirs": "desc", "sort_keys": "status"}, 200),
({"sort_dirs": "desc", "sort_keys": "project_id"}, 200),
({"sort_dirs": "desc", "sort_keys": "abc123"}, 422),
({"all_projects": True}, 200),
({"all_projects": True, "project_id": str(uuid.uuid4())}, 200),
({"name": "abc123"}, 200),
({"status": "ACTIVE"}, 200),
({"network_name": "net01"}, 200),
({"network_id": str(uuid.uuid4())}, 200),
({"device_id": str(uuid.uuid4())}, 200),
({"device_owner": "compute:nova"}, 200),
({"uuid": [str(uuid.uuid4())]}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_ports(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
# list ports
url = f"{main.API_PREFIX}/extension/ports" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "ports" in result
@pytest.mark.parametrize(
("params", "status_code"),
(
({}, 200),
({"binary": "nova-compute"}, 200),
({"host": "host01"}, 200),
),
)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_compute_services(
client: AsyncClient,
login_jwt: str,
params: Dict[str, Any],
status_code: int,
) -> None:
url = f"{main.API_PREFIX}/extension/compute-services" + "?%s" % parse.urlencode(params)
r = await client.get(
url=url,
headers={"Content-Type": "application/json"},
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == status_code
if status_code == 200:
assert "services" in result

137
skyline_apiserver/tests/api/v1/test_login.py
View File

@ -1,137 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import os
import time
import pytest
from httpx import AsyncClient
from skyline_apiserver import main, version
from skyline_apiserver.config import CONF
from skyline_apiserver.db import api as db_api, setup as db_setup
from skyline_apiserver.tests.utils import utils
from skyline_apiserver.types import constants
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_login(client: AsyncClient) -> None:
login_data = {
"region": "RegionOne",
"username": CONF.openstack.system_user_name,
"domain": CONF.openstack.system_user_domain,
"password": CONF.openstack.system_user_password,
}
r = await client.post(
url=f"{main.API_PREFIX}/login",
json=login_data,
)
result = r.json()
assert r.status_code == 200
assert "keystone_token" in result
assert "keystone" in result["endpoints"]
assert "projects" in result
await utils._logout(client)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_profile_ok(client: AsyncClient, login_jwt: str) -> None:
r = await client.get(
f"{main.API_PREFIX}/profile",
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == 200
assert "keystone_token" in result
assert "keystone" in result["endpoints"]
assert "projects" in result
assert "base_roles" in result
assert "base_domains" in result
assert result["version"] == version.version_string()
assert len(CONF.openstack.base_domains) == len(result["base_domains"])
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_switch_project(client: AsyncClient, login_jwt: str) -> None:
r = await client.get(
f"{main.API_PREFIX}/profile",
cookies={"session": login_jwt},
)
result = r.json()
assert r.status_code == 200
assert "project" in result
project_id = result["project"]["id"]
r = await client.post(
f"{main.API_PREFIX}/switch_project/{project_id}",
)
result = r.json()
assert r.status_code == 200
assert project_id == result["project"]["id"]
await utils._logout(client)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_profile_no_auth(client: AsyncClient) -> None:
r = await client.get(f"{main.API_PREFIX}/profile")
result = r.json()
assert r.status_code == 401
assert result["detail"] == constants.ERR_MSG_TOKEN_NOTFOUND
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_profile_token_expire(client: AsyncClient) -> None:
profile = utils.get_session_profile()
profile.exp = int(time.time()) - 1
token = profile.toJWTPayload()
r = await client.get(
f"{main.API_PREFIX}/profile",
cookies={CONF.default.session_name: token},
)
result = r.json()
assert r.status_code == 401
assert result["detail"].startswith(constants.ERR_MSG_TOKEN_EXPIRED)
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_profile_token_revoke(client: AsyncClient) -> None:
profile = utils.get_session_profile()
await db_setup()
await db_api.revoke_token(profile.uuid, profile.exp)
token = profile.toJWTPayload()
r = await client.get(
f"{main.API_PREFIX}/profile",
cookies={CONF.default.session_name: token},
)
result = r.json()
assert r.status_code == 401
assert result["detail"] == constants.ERR_MSG_TOKEN_REVOKED
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_logout(client: AsyncClient, session_token: str) -> None:
r = await client.post(
f"{main.API_PREFIX}/logout",
cookies={CONF.default.session_name: session_token},
)
assert r.status_code == 200

118
skyline_apiserver/tests/api/v1/test_setting.py
View File

@ -1,118 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import os
import uuid
import pytest
from httpx import AsyncClient
from skyline_apiserver import main
from skyline_apiserver.config import CONF
from skyline_apiserver.types import constants
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_list_settings(client: AsyncClient, login_jwt: str) -> None:
result = await client.get(
f"{main.API_PREFIX}/settings",
cookies={"session": login_jwt},
)
assert result.status_code == 200
assert "settings" in result.json()
settings = result.json()["settings"]
for setting in settings:
key = setting["key"]
assert setting["hidden"] == (key in constants.SETTINGS_HIDDEN_SET)
assert setting["restart_service"] == (key in constants.SETTINGS_RESTART_SET)
# settings not only include something in CONF.setting.base_settings
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_default_setting(client: AsyncClient, login_jwt: str) -> None:
for key in CONF.setting.base_settings:
result = await client.get(
f"{main.API_PREFIX}/setting/{key}",
cookies={"session": login_jwt},
)
assert result.status_code == 200
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_get_setting_not_found(client: AsyncClient, login_jwt: str) -> None:
key = str(uuid.uuid4().hex)
result = await client.get(
f"{main.API_PREFIX}/setting/{key}",
cookies={"session": login_jwt},
)
assert result.status_code == 404
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_update_setting(client: AsyncClient, login_jwt: str) -> None:
for key in CONF.setting.base_settings:
# Create
update_value = ["test1", "test2"]
result = await client.put(
url=f"{main.API_PREFIX}/setting",
json={"key": key, "value": update_value},
cookies={"session": login_jwt},
)
assert result.status_code == 200
assert result.json()["value"] == update_value
# Update
update_value = ["test1", "test3"]
result = await client.put(
url=f"{main.API_PREFIX}/setting",
json={"key": key, "value": update_value},
cookies={"session": login_jwt},
)
assert result.status_code == 200
assert result.json()["value"] == update_value
break
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_update_setting_not_found(client: AsyncClient, login_jwt: str) -> None:
key = str(uuid.uuid4().hex)
update_value = {"value": "{}"}
result = await client.put(
url=f"{main.API_PREFIX}/setting",
json={"key": key, "value": update_value},
cookies={"session": login_jwt},
)
assert result.status_code == 404
@pytest.mark.skipif(os.getenv("TEST_API") != "true", reason="No backend OpenStack for api-test.")
@pytest.mark.asyncio
async def test_reset_setting(client: AsyncClient, login_jwt: str) -> None:
for key in CONF.setting.base_settings:
result = await client.delete(
f"{main.API_PREFIX}/setting/{key}",
cookies={"session": login_jwt},
)
assert result.status_code == 204
result = await client.get(
f"{main.API_PREFIX}/setting/{key}",
cookies={"session": login_jwt},
)
assert result.json()["value"] == getattr(CONF.setting, key)
break

11
skyline_apiserver/tests/conftest.py
View File

@ -22,7 +22,6 @@ from httpx import AsyncClient
from skyline_apiserver.config import CONF
from skyline_apiserver.main import app
from skyline_apiserver.tests.models import TestData
from skyline_apiserver.tests.utils import utils
if TYPE_CHECKING:
from _pytest.python import Metafunc
@ -37,16 +36,6 @@ async def client() -> Iterator[AsyncClient]:
CONF.cleanup()
@pytest.fixture(scope="function")
async def session_token(client: AsyncClient) -> str:
return utils.get_session_token()
@pytest.fixture(scope="function")
async def login_jwt(client: AsyncClient) -> str:
return await utils.get_jwt_from_cookie(client)
def pytest_generate_tests(metafunc: Optional["Metafunc"]) -> None:
for marker in metafunc.definition.iter_markers(name="ddt"):
test_data: TestData

0
skyline_apiserver/tests/core/__init__.py
View File

0
skyline_apiserver/tests/core/config/__init__.py
View File

143
skyline_apiserver/tests/core/config/backup_test_base.py
View File

@ -1,143 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import os
import pytest
from skyline_apiserver.config import base
@pytest.mark.parametrize(
"opt",
[
{"name": "test0", "help": "this is test0"},
{"name": "test1", "help": "this is test1", "deprecated": True},
{"name": "test2", "help": "this is test2", "schema": {"type": "string"}},
{"name": "test3", "help": "this is test3", "default": "test3"},
],
)
@pytest.mark.parametrize("value", ["test_value"])
def test_opt_from_init(opt, value):
opt = base.Opt(**opt)
assert opt._loaded is False
if opt.default is not None:
value = None
result = opt.default
else:
result = value
opt.load(value)
assert opt._loaded is True
assert opt.value == result
with pytest.raises(ValueError):
opt.load(value)
@pytest.mark.parametrize(
"opt,value",
[
({"name": "test0", "help": "this is test0", "schema": {"type": "null"}}, "test_value"),
({"name": "test1", "help": "this is test1", "schema": {"type": "string"}}, None),
({"name": "test2", "help": "this is test2", "schema": {"type": "array"}}, None),
({"name": "test3", "help": "this is test3", "schema": {"type": "object"}}, None),
],
)
def test_opt_from_init_validate(opt, value):
opt = base.Opt(**opt)
assert opt._loaded is False
# with pytest.raises(ValidationError):
# opt.load(value)
@pytest.mark.parametrize("opt", [{"name": "test0"}, {"help": "this is test1"}])
def test_opt_from_init_error(opt):
with pytest.raises(TypeError):
opt = base.Opt(**opt)
@pytest.mark.parametrize(
"opt_schema",
[
{"title": "test0", "description": "this is test0", "type": "string"},
{"title": "test1", "description": "this is test1", "type": "string", "default": "test"},
{"title": "test2", "description": "this is test2", "type": "string", "deprecated": True},
{
"title": "test3",
"description": "this is test3",
"type": "string",
"default": "test",
"deprecated": True,
},
],
)
@pytest.mark.parametrize("value", ["test_value"])
def test_opt_from_schema(opt_schema, value):
opt = base.Opt.from_schema(opt_schema)
assert opt._loaded is False
if opt.default is not None:
value = None
result = opt.default
else:
result = value
opt.load(value)
assert opt._loaded is True
assert opt.value == result
with pytest.raises(ValueError):
opt.load(value)
@pytest.mark.parametrize(
"opt_schema",
[
{"title": "not_description"},
{"description": "not title"},
{
"title": "test_title",
"description": "deprecated is not boolean",
"deprecated": "somestring",
},
],
)
def test_opt_from_schema_error(opt_schema):
pass
# with pytest.raises(ValidationError):
# base.Opt.from_schema(opt_schema)
# TODO: add test Group & Configuration
@pytest.mark.parametrize("config_dir", [None, "/etc/tests"])
@pytest.mark.parametrize("config_file", [None, "/etc/tests/test.yaml"])
def test_get_config_file(config_dir, config_file):
env = {}
if config_dir is not None:
env["OS_CONFIG_DIR"] = config_dir
if config_file is not None:
env["OS_CONFIG_FILE"] = config_file
result = []
def_config_dir = os.path.join("/etc", "skyline")
result.append(def_config_dir if config_dir is None else config_dir)
def_config_file = os.path.join(result[0], "skyline_apiserver.yaml")
result.append(def_config_file if config_file is None else config_file)
assert base.Configuration._get_config_file(env) == tuple(result)
def test_get_config_file_from_env(monkeypatch):
monkeypatch.setenv("OS_CONFIG_DIR", "/etc/tests_env")
monkeypatch.setenv("OS_CONFIG_FILE", "/etc/tests_env/tests_env.yaml")
assert base.Configuration._get_config_file() == (
"/etc/tests_env",
"/etc/tests_env/tests_env.yaml",
)

35
skyline_apiserver/tests/core/config/backup_test_default.py
View File

@ -1,35 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import pytest
from skyline_apiserver.config.default import ALL_OPTS, GROUP_NAME
def test_group_name():
assert GROUP_NAME == "default"
@pytest.mark.parametrize("log_dir_value", [".", "./", "/", "/tmp", "/qwer"])
@pytest.mark.parametrize("debug_value", [True, False])
def test_all_opts(debug_value, log_dir_value):
for index, opt in enumerate(ALL_OPTS):
object.__setattr__(opt, "_loaded", False)
if index == 0:
opt.load(debug_value)
assert opt.value == debug_value
elif index == 1:
opt.load(log_dir_value)
assert opt.value == log_dir_value
object.__setattr__(opt, "_loaded", False)

44
skyline_apiserver/tests/core/config/backup_test_openstack.py
View File

@ -1,44 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import pytest
from skyline_apiserver.config.openstack import ALL_OPTS, GROUP_NAME
def test_group_name():
assert GROUP_NAME == "openstack"
@pytest.mark.parametrize("host_value", ["localhost", "127.0.0.1", "host-1", "192.168.1.1"])
@pytest.mark.parametrize(
"keystone_url_value",
[
"http://localhost:5000/v3/",
"http://127.0.0.1:5000/v3/",
"https://keystone:5000/v3/",
"https://keystone:5000/",
"https://keystone/",
],
)
def test_all_opts(host_value, keystone_url_value):
for index, opt in enumerate(ALL_OPTS):
object.__setattr__(opt, "_loaded", False)
if index == 0:
opt.load(host_value)
assert opt.value == host_value
elif index == 1:
opt.load(keystone_url_value)
assert opt.value == keystone_url_value
object.__setattr__(opt, "_loaded", False)

0
skyline_apiserver/tests/utils/__init__.py
View File

91
skyline_apiserver/tests/utils/utils.py
View File

@ -1,91 +0,0 @@
# Copyright 2021 99cloud
#
# 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.
import time
import uuid
from httpx import AsyncClient
from skyline_apiserver import config, main, schemas, version
def get_session_profile() -> schemas.Profile:
profile = schemas.Profile(
username="testUser",
keystone_token="testKeystoneToken",
region="testRegion",
project={
"id": uuid.uuid4().hex,
"name": "testProject",
"domain": {"id": uuid.uuid4().hex, "name": "testDomain"},
},
user={
"id": uuid.uuid4().hex,
"name": "testUser",
"domain": {
"id": uuid.uuid4().hex,
"name": "testDomain",
},
},
roles=[{"id": uuid.uuid4().hex, "name": "testRole"}],
keystone_token_exp="2221-01-13T12:29:37.000000Z",
base_roles=[],
exp=int(time.time()) + config.CONF.default.access_token_expire,
uuid=uuid.uuid4().hex,
endpoints={
"placement": "/api/openstack/placement",
"neutron": "/api/openstack/network",
"swift": "/api/openstack/object-storage",
"nova": "/api/openstack/compute",
"heat": "/api/openstack/heat-api",
"nova_legacy": "/api/openstack/compute",
"cinderv2": "/api/openstack/volume",
"heat-cfn": "/api/openstack/heat-api-cfn",
"keystone": "/api/openstack/identity",
"cinder": "/api/openstack/volume",
"cinderv3": "/api/openstack/volume",
"glance": "/api/openstack/image",
},
projects={
"0e064ea01b614943993a28b2c15bd6c4": {"name": "demo", "domain_id": "default"},
"4c017648d2e34d1a8e732b98e3232af9": {"name": "alt_demo", "domain_id": "default"},
"e88226c062094881b7a1f01517b945b4": {"name": "admin", "domain_id": "default"},
},
version=version.version_string(),
)
return profile
def get_session_token() -> str:
profile = get_session_profile()
return profile.toJWTPayload()
async def get_jwt_from_cookie(client: AsyncClient) -> str:
login_data = {
"region": "RegionOne",
"username": config.CONF.openstack.system_user_name,
"domain": config.CONF.openstack.system_user_domain,
"password": config.CONF.openstack.system_user_password,
}
r = await client.post(
url=f"{main.API_PREFIX}/login",
json=login_data,
)
token = r.cookies.get(config.CONF.default.session_name, "")
return token
async def _logout(client: AsyncClient) -> None:
await client.post(f"{main.API_PREFIX}/logout")