fix: harden Polygon operation validation

2026-10-25 14:07:13 +03:00
parent 3df89ec9e7
commit e3a031ce50
18 changed files with 803 additions and 346 deletions
@@ -0,0 +1,228 @@
+"""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
+
+
+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": []}, {}]
+
+
+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/&lt;id&gt;" 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/&lt;uid&gt;/status" not 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
+
+
+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
+
+
+@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) == {}