Files
wehub-resource-sync e0e362d700
SDK Tests / changes (push) Successful in 2m29s
Real E2E Tests / changes (push) Successful in 2m29s
Deploy Docs Pages / build (push) Has been cancelled
Deploy Docs Pages / deploy (push) Has been cancelled
Real E2E Tests / JavaScript E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Python E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Java E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / C# E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Go E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Real E2E CI (push) Has been cancelled
SDK Tests / SDK CI (push) Has been cancelled
SDK Tests / CLI Tests (push) Has been cancelled
SDK Tests / Python SDK Quality (code-interpreter) (push) Has been cancelled
SDK Tests / Python SDK Quality (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (sandbox) (push) Has been cancelled
SDK Tests / CLI Quality (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Go SDK Quality And Tests (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:39:33 +08:00

90 lines
3.0 KiB
Python

# Copyright 2026 Alibaba Group Holding Ltd.
#
# 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.
"""Kubernetes V1SecurityContext ↔ plain dict helpers for CRD pod specs."""
from typing import Any, Dict, Optional
def build_security_context_from_dict(
security_context_dict: Dict[str, Any],
) -> Optional[Any]:
"""
Convert a security context dict to ``V1SecurityContext``.
Empty dict returns None.
"""
if not security_context_dict:
return None
from kubernetes.client import V1SecurityContext, V1Capabilities
capabilities = None
if "capabilities" in security_context_dict:
caps_dict = security_context_dict["capabilities"]
add_caps = caps_dict.get("add", [])
drop_caps = caps_dict.get("drop", [])
capabilities = V1Capabilities(
add=add_caps if add_caps else None,
drop=drop_caps if drop_caps else None,
)
privileged = security_context_dict.get("privileged")
if capabilities is None and privileged is None:
return None
return V1SecurityContext(
capabilities=capabilities,
privileged=privileged,
)
def serialize_security_context_to_dict(
security_context: Optional[Any],
) -> Optional[Dict[str, Any]]:
"""Serialize ``V1SecurityContext`` to a CRD-friendly dict."""
if not security_context:
return None
result: Dict[str, Any] = {}
if security_context.capabilities:
caps: Dict[str, Any] = {}
if security_context.capabilities.add:
caps["add"] = security_context.capabilities.add
if security_context.capabilities.drop:
caps["drop"] = security_context.capabilities.drop
if caps:
result["capabilities"] = caps
if security_context.privileged is not None:
result["privileged"] = security_context.privileged
if getattr(security_context, "seccomp_profile", None) is not None:
sp = security_context.seccomp_profile
profile_dict: Dict[str, Any] = {"type": sp.type}
if getattr(sp, "localhost_profile", None) is not None:
profile_dict["localhostProfile"] = sp.localhost_profile
result["seccompProfile"] = profile_dict
if getattr(security_context, "app_armor_profile", None) is not None:
ap = security_context.app_armor_profile
profile_dict = {"type": ap.type}
if getattr(ap, "localhost_profile", None) is not None:
profile_dict["localhostProfile"] = ap.localhost_profile
result["appArmorProfile"] = profile_dict
return result if result else None