282 lines
9.5 KiB
Python
282 lines
9.5 KiB
Python
"""Focused tests for public page safety and mock-route input handling."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
SITE_DIR = Path(__file__).resolve().parents[1] / "site"
|
|
sys.path.insert(0, str(SITE_DIR))
|
|
|
|
import app as polygon_app
|
|
import config.loader as config
|
|
import mock_state
|
|
from routes import mock_routes
|
|
from routes import openapi as openapi_routes
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_polygon_state():
|
|
for stand_id in config.list_stands():
|
|
mock_state.get_state(stand_id).reset()
|
|
yield
|
|
for stand_id in config.list_stands():
|
|
mock_state.get_state(stand_id).reset()
|
|
|
|
|
|
def test_swagger_does_not_embed_mock_auth_token():
|
|
response = polygon_app.app.test_client().get("/swagger")
|
|
|
|
assert response.status_code == 200
|
|
assert b"test-token-123" not in response.data
|
|
assert b"auth_token" not in response.data
|
|
assert b"MOCK_AUTH_TOKEN" in response.data
|
|
assert b"X-Mock-Auth" in response.data
|
|
|
|
|
|
def test_openapi_schema_and_auth_match_runtime_contract():
|
|
spec = openapi_routes._build_spec()
|
|
schemas = spec["components"]["schemas"]
|
|
paths = spec["paths"]
|
|
|
|
assert len(spec["servers"]) == len(config.STANDS)
|
|
assert schemas["CfsParam"]["properties"]["valueList"]["items"]["type"] == "string"
|
|
assert schemas["CfsParam"]["properties"]["dataDescriptor"]["type"] == "object"
|
|
assert schemas["Operation"]["properties"]["kind"]["description"] == "instance | subresource"
|
|
assert schemas["Instance"]["properties"]["status"]["description"] == "creating | running | suspended"
|
|
page_size = paths["/api/v1/svc/instances"]["get"]["parameters"][0]["schema"]
|
|
assert page_size["minimum"] == 1
|
|
assert page_size["maximum"] == 200
|
|
for path, path_item in paths.items():
|
|
if "/_mock/" in path:
|
|
operation = path_item.get("get") or path_item.get("post")
|
|
assert operation["security"] == [{"mockAuth": []}, {}]
|
|
set_param_responses = paths["/api/v1/svc/instanceOperationCfsParams"]["post"]["responses"]
|
|
assert set_param_responses["200"]["content"]["application/json"]["schema"]["type"] == "object"
|
|
assert set_param_responses["409"]["description"] == "Операция уже завершена"
|
|
|
|
|
|
def test_homepage_documents_real_operation_routes():
|
|
response = polygon_app.app.test_client().get("/")
|
|
|
|
assert response.status_code == 200
|
|
assert b"/api/v1/svc/instanceOperations/default/<id>" in response.data
|
|
assert b"/api/v1/svc/instanceOperationCfsParams" in response.data
|
|
assert b"/api/v1/svc/openapi.json" in response.data
|
|
assert b"/api/v1/svc/instanceOperations/<uid>/status" not in response.data
|
|
assert "вставьте токен".encode() in response.data
|
|
|
|
|
|
@pytest.mark.parametrize("seconds", ["nan", "inf", "-inf"])
|
|
def test_mock_delay_rejects_non_finite_numbers(monkeypatch, seconds):
|
|
monkeypatch.setattr(mock_routes, "_MOCK_TOKEN", "")
|
|
previous_delay = config.DELAY
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
f"/api/v1/svc/_mock/delay/{seconds}"
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert config.DELAY == previous_delay
|
|
|
|
|
|
def test_operation_routes_reject_non_object_json():
|
|
client = polygon_app.app.test_client()
|
|
|
|
create_response = client.post(
|
|
"/api/v1/svc/instanceOperations", json=[1]
|
|
)
|
|
param_response = client.post(
|
|
"/api/v1/svc/instanceOperationCfsParams", json=[1]
|
|
)
|
|
|
|
assert create_response.status_code == 400
|
|
assert param_response.status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize("body", [[], "", 0, False, None])
|
|
def test_create_instance_rejects_non_object_json(body):
|
|
response = polygon_app.app.test_client().post(
|
|
"/api/v1/svc/instances", json=body
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.get_json()["error"] == "request body must be a JSON object"
|
|
|
|
|
|
def test_create_instance_rejects_non_string_display_name():
|
|
service_id = next(iter(config.get_services(config.DEFAULT_STAND)))
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
"/api/v1/svc/instances",
|
|
json={"serviceId": service_id, "displayName": 123},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_create_operation_rejects_foreign_service_operation():
|
|
stand_id = config.DEFAULT_STAND
|
|
services = config.get_services(stand_id)
|
|
service_id, service = next(iter(services.items()))
|
|
foreign_operation = next(
|
|
operation
|
|
for other_id, other_service in services.items()
|
|
if other_id != service_id
|
|
for operation in other_service.get("operations", [])
|
|
if operation["svcOperationId"] not in {
|
|
own["svcOperationId"] for own in service.get("operations", [])
|
|
}
|
|
)
|
|
instance_uid = mock_state.get_state(stand_id).create_instance(
|
|
service_id, "route-test", service
|
|
)
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
"/api/v1/svc/instanceOperations",
|
|
json={
|
|
"instanceUid": instance_uid,
|
|
"operation": foreign_operation["operation"],
|
|
"svcOperationId": foreign_operation["svcOperationId"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_create_operation_rejects_mismatched_operation_name():
|
|
stand_id = config.DEFAULT_STAND
|
|
services = config.get_services(stand_id)
|
|
service_id, service = next(
|
|
(service_id, definition)
|
|
for service_id, definition in services.items()
|
|
if len(definition.get("operations", [])) > 1
|
|
)
|
|
target_op, other_op = service["operations"][:2]
|
|
instance_uid = mock_state.get_state(stand_id).create_instance(
|
|
service_id, "route-test", service
|
|
)
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
"/api/v1/svc/instanceOperations",
|
|
json={
|
|
"instanceUid": instance_uid,
|
|
"operation": other_op["operation"],
|
|
"svcOperationId": target_op["svcOperationId"],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert mock_state.get_state(stand_id).operations == {}
|
|
|
|
|
|
@pytest.mark.parametrize("stand_id", config.list_stands())
|
|
def test_homepage_swagger_link_keeps_current_stand(stand_id):
|
|
if len(config.list_stands()) <= 1:
|
|
pytest.skip("stand-specific Swagger link is needed only for multiple stands")
|
|
|
|
response = polygon_app.app.test_client().get(f"/{stand_id}/")
|
|
|
|
assert response.status_code == 200
|
|
assert f'href="/{stand_id}/swagger"'.encode() in response.data
|
|
|
|
|
|
@pytest.mark.parametrize("page_size", [0, -5])
|
|
def test_instances_page_size_is_clamped_to_one(page_size):
|
|
response = polygon_app.app.test_client().get(
|
|
f"/api/v1/svc/instances?pageSize={page_size}"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.get_json()["pageSize"] == 1
|
|
|
|
|
|
def test_run_rejects_operation_for_deleted_instance():
|
|
stand_id = config.DEFAULT_STAND
|
|
state = mock_state.get_state(stand_id)
|
|
op_uid = state.create_operation(
|
|
"missing-instance", 18, "create", "instance", "create"
|
|
)
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
f"/api/v1/svc/instanceOperations/{op_uid}/run"
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_set_param_rejects_completed_operation():
|
|
stand_id = config.DEFAULT_STAND
|
|
services = config.get_services(stand_id)
|
|
service_id, service = next(
|
|
(service_id, definition)
|
|
for service_id, definition in services.items()
|
|
if definition.get("operations")
|
|
)
|
|
operation = service["operations"][0]
|
|
state = mock_state.get_state(stand_id)
|
|
instance_uid = state.create_instance(service_id, "route-test", service)
|
|
op_uid = state.create_operation(
|
|
instance_uid,
|
|
operation["svcOperationId"],
|
|
operation["operation"],
|
|
operation.get("kind", "instance"),
|
|
operation.get("action", operation["operation"]),
|
|
)
|
|
state.get_operation(op_uid)["dtFinish"] = "completed"
|
|
|
|
response = polygon_app.app.test_client().post(
|
|
"/api/v1/svc/instanceOperationCfsParams",
|
|
json={
|
|
"instanceOperationUid": op_uid,
|
|
"svcOperationCfsParamId": 1,
|
|
"paramValue": "value",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_set_param_validates_value_type_and_operation_membership():
|
|
stand_id = config.DEFAULT_STAND
|
|
services = config.get_services(stand_id)
|
|
service_id, service, operation, param_id = next(
|
|
(service_id, definition, operation, param_id)
|
|
for service_id, definition in services.items()
|
|
for operation in definition.get("operations", [])
|
|
for param_id in definition.get("cfsParamsByOp", {}).get(
|
|
operation["svcOperationId"], []
|
|
)
|
|
)
|
|
state = mock_state.get_state(stand_id)
|
|
instance_uid = state.create_instance(service_id, "route-test", service)
|
|
op_uid = state.create_operation(
|
|
instance_uid,
|
|
operation["svcOperationId"],
|
|
operation["operation"],
|
|
operation.get("kind", "instance"),
|
|
operation.get("action", operation["operation"]),
|
|
)
|
|
client = polygon_app.app.test_client()
|
|
|
|
wrong_type = client.post(
|
|
"/api/v1/svc/instanceOperationCfsParams",
|
|
json={
|
|
"instanceOperationUid": op_uid,
|
|
"svcOperationCfsParamId": param_id,
|
|
"paramValue": 7,
|
|
},
|
|
)
|
|
wrong_membership = client.post(
|
|
"/api/v1/svc/instanceOperationCfsParams",
|
|
json={
|
|
"instanceOperationUid": op_uid,
|
|
"svcOperationCfsParamId": -1,
|
|
"paramValue": "value",
|
|
},
|
|
)
|
|
|
|
assert wrong_type.status_code == 400
|
|
assert wrong_membership.status_code == 400
|
|
assert state.get_params(op_uid) == {}
|