Fix missing required parameter when k8s_obj init

While deploying CNF(deployment) using pod-affinity rules, a
parameter error happended because the topology_key is not
assigned when initializing k8s object.

This patch fixes this bug by adding the topology_key's default
value into `must_param` in translate_outputs._create_k8s_object().
At the same time, in order to deal with the recurrence of this
kind of bug, we checked the required parameters of all resources
in kubernetes again and added them to must_param.
    * Kubernetes version: from v1.18.20 to v1.21.1
    * Kubernetes-python-client version: v18.20.0

The major changes from the "must_param" used in the previous
Wallaby release of Tacker are probably as follows:

RuntimeRawExtension
  The `data` parameter of `V1ControllerRevision`[1] in v11.0.0
  was `RuntimeRawExtension`[2], but the `RuntimeRawExtension` was
  not support for v12.0.0 or later. Therefore `V1ControllerRevision`
  in v18.20.0[3] it has been changed to` object` type.

V1ServiceReference
  The `service` parameter of `V1APIServiceSpec`[4] in v11.0.0 was
  `V1ServiceReference`[5], but the `V1ServiceReference` was not
  support for v12.0.0 or later.
  Therefore `V1APIServiceSpec` in v18.20.0[6] it has been changed
  to `ApiregistrationV1ServiceReference`[7].

V1LimitRangeItem
  The `type` parameter of `V1LimitRangeItem`[8] in v11.0.0 was
  optional type, but the parameter has changed to must param
  in v18.20.0[9].
  Therefore, the parameter should be added into `must_param`.

[1] https://github.com/kubernetes-client/python/blob/v11.0.0/kubernetes/docs/V1ControllerRevision.md
[2] https://github.com/kubernetes-client/python/blob/v11.0.0/kubernetes/docs/RuntimeRawExtension.md
[3] https://github.com/kubernetes-client/python/blob/v18.20.0/kubernetes/docs/V1ControllerRevision.md
[4] https://github.com/kubernetes-client/python/blob/v11.0.0/kubernetes/docs/V1APIServiceSpec.md
[5] https://github.com/kubernetes-client/python/blob/v11.0.0/kubernetes/docs/V1ServiceReference.md
[6] https://github.com/kubernetes-client/python/blob/v18.20.0/kubernetes/docs/V1APIServiceSpec.md
[7] https://github.com/kubernetes-client/python/blob/v18.20.0/kubernetes/docs/ApiregistrationV1ServiceReference.md
[8] https://github.com/kubernetes-client/python/blob/v11.0.0/kubernetes/docs/V1LimitRangeItem.md
[9] https://github.com/kubernetes-client/python/blob/v18.20.0/kubernetes/docs/V1LimitRangeItem.md

Closes-Bug: #1928153
Closes-Bug: #1940602
Change-Id: If77be98d0c4cec6f4c860ae84978f59772f9a6ee
changes/23/791123/14
LiangLu 1 year ago committed by Yi Feng
parent ef982945d9
commit 26a2b5fce2
  1. 4
      lower-constraints.txt
  2. 4
      requirements.txt
  3. 4
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/api-service.yaml
  4. 8
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/daemon-set.yaml
  5. 7
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/deployment.yaml
  6. 3
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/horizontal-pod-autoscaler.yaml
  7. 5
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/job.yaml
  8. 3
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/limit-range.yaml
  9. 6
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/namespace.yaml
  10. 9
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/network-policy.yaml
  11. 35
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/node.yaml
  12. 9
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/persistent-volume-claim.yaml
  13. 42
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/persistent-volume.yaml
  14. 76
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/pod-template.yaml
  15. 88
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/pod.yaml
  16. 7
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/replica-set.yaml
  17. 6
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/resource-quota.yaml
  18. 10
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/self-subject-rule-review.yaml
  19. 5
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/stateful-set.yaml
  20. 5
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/storage-class.yaml
  21. 4
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/subject-access-review.yaml
  22. 4
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/kubernetes_api_resource/volume-attachment.yaml
  23. 739
      tacker/tests/unit/vnfm/infra_drivers/kubernetes/test_translate_outputs.py
  24. 52
      tacker/vnfm/infra_drivers/kubernetes/k8s/translate_outputs.py
  25. 2
      upper-constraints.txt

@ -46,7 +46,7 @@ jsonschema==3.2.0
keystoneauth1==4.3.1
keystonemiddleware==4.17.0
kombu==4.3.0
kubernetes==11.0.0
kubernetes==18.20.0
linecache2==1.0.0
Mako==1.0.7
MarkupSafe==1.1
@ -116,7 +116,7 @@ python-swiftclient==3.5.0
python-tackerclient==0.8.0
python-cinderclient==8.0.0
pytz==2018.3
PyYAML==5.1
PyYAML==5.4.1
repoze.lru==0.7
requests-oauthlib==0.8.0
requests==2.25.1

@ -45,10 +45,10 @@ pyroute2>=0.4.21;sys_platform!='win32' # Apache-2.0 (+ dual licensed GPL2)
python-mistralclient>=4.2.0 # Apache-2.0
python-barbicanclient>=4.5.2 # Apache-2.0
castellan>=0.16.0 # Apache-2.0
kubernetes>=11.0.0 # Apache-2.0
kubernetes>=18.20.0 # Apache-2.0
setuptools!=24.0.0,!=34.0.0,!=34.0.1,!=34.0.2,!=34.0.3,!=34.1.0,!=34.1.1,!=34.2.0,!=34.3.0,!=34.3.1,!=34.3.2,!=36.2.0,>=21.0.0 # PSF/ZPL
tooz>=1.58.0 # Apache-2.0
PyYAML>=5.1 # MIT
PyYAML>=5.4.1 # MIT
PyMySQL>=0.10.1 # MIT
# Glance Store

@ -6,4 +6,6 @@ spec:
group: currytest.k8s.io
groupPriorityMinimum: 17000
version: v1beta1
versionPriority: 5
versionPriority: 5
service:
name: test

@ -14,3 +14,11 @@ spec:
containers:
- image: nginx
name: nginx
status:
currentNumberScheduled: 1
desiredNumberScheduled: 1
numberMisscheduled: 1
numberReady: 1
conditions:
- status: True
type: DaemonSet

@ -7,6 +7,9 @@ spec:
selector:
matchLabels:
selector: curry-probe-test001
matchExpressions:
- key: test
operator: test
template:
metadata:
labels:
@ -38,3 +41,7 @@ spec:
path: /
failureThreshold: 2
periodSeconds: 2
status:
conditions:
- status: True
type: Deployment

@ -11,3 +11,6 @@ spec:
kind: Deployment
name: curry-svc-vdu001
targetCPUUtilizationPercentage: 40
status:
currentReplicas: 1
desiredReplicas: 1

@ -24,4 +24,7 @@ spec:
limits: {}
requests: {}
restartPolicy: OnFailure
status: {}
status:
conditions:
- status: True
type: Job

@ -7,4 +7,5 @@ spec:
limits:
- default:
cpu: 500m
memory: 512M
memory: 512M
type: test

@ -1,4 +1,8 @@
apiVersion: v1
kind: Namespace
metadata:
name: curry-ns
name: curry-ns
status:
conditions:
- status: True
type: Namespace

@ -5,4 +5,11 @@ metadata:
spec:
podSelector: {}
policyTypes:
- Egress
- Egress
egress:
- to:
- ipBlock:
cidr: 10.0.0.0/24
ports:
- protocol: TCP
port: 5978

@ -4,3 +4,38 @@ metadata:
name: curry-node-test
labels:
name: curry-node-test
spec:
configSource:
configMap:
name: CONFIG_MAP_NAME
namespace: kube-system
kubeletConfigKey: kubelet
taints:
- effect: 'test'
key: 'test'
status:
addresses:
- address: '1.1.1.1'
type: 'test'
conditions:
- status: True
type: Node
daemonEndpoints:
kubeletEndpoint:
port: 8080
images:
- names: 'test'
nodeInfo:
architecture: 'test'
bootId: 'test'
containerRuntimeVersion: 'test'
kernelVersion: 'test'
kubeProxyVersion: 'test'
kubeletVersion: 'test'
machineId: 'test'
operatingSystem: 'test'
osImage: 'test'
systemUuid: 'test'
volumesAttached:
- device_path: 'test'
name: 'test'

@ -9,4 +9,11 @@ spec:
resources:
requests:
storage: 2Gi
storageClassName: curry-sc-local
storageClassName: curry-sc-local
dataSource:
name: existing-src-pvc-name
kind: PersistentVolumeClaim
status:
conditions:
- status: True
type: PersistentVolumeClaim

@ -11,4 +11,44 @@ spec:
path: /data/curry-sc-test
type: DirectoryOrCreate
persistentVolumeReclaimPolicy: Delete
storageClassName: curry-sc-local
storageClassName: curry-sc-local
azureFile:
secretName: azure-secret
shareName: aksshare
readOnly: false
cephfs:
monitors:
- 10.16.154.78:6789
cinder:
volumeID: "90d6900d-808f-4ddb-a30e-5ef821f58b4e"
fsType: ext4
csi:
driver: csi-nfsplugin
volume_handle: data-id
flexVolume:
driver: "kubernetes.io/lvm"
fsType: "ext4"
glusterfs:
endpoints: glusterfs-cluster
path: kube_vol
readOnly: true
iscsi:
targetPortal: 10.0.2.15:3260
iqn: iqn.2001-04.com.example:storage.kube.sys1.xyz
lun: 0
local:
path: /mnt/disks/ssd1
rbd:
monitors:
- '10.16.154.78:6789'
image: foo
scaleIO:
gateway: https://localhost:443/api
system: scaleio
secretRef:
name: sio-secret
awsElasticBlockStore:
volumeID: "123"
azureDisk:
diskName: test.vhd
diskURI: https://someaccount.blob.microsoft.net/vhds/test.vhd

@ -40,4 +40,80 @@ template:
- name: curry-claim-volume
persistentVolumeClaim:
claimName: curry-pv-claim
azureFile:
secretName: azure-secret
shareName: aksshare
readOnly: false
cephfs:
monitors:
- 10.16.154.78:6789
cinder:
volumeID: "90d6900d-808f-4ddb-a30e-5ef821f58b4e"
fsType: ext4
configMap:
name: log-config
items:
- key: log_level
path: log_level
csi:
driver: csi-nfsplugin
volume_handle: data-id
downwardAPI:
items:
- path: "labels"
fieldRef:
fieldPath: metadata.labels
resourceFieldRef:
resource: limits.cpu
flexVolume:
driver: "kubernetes.io/lvm"
fsType: "ext4"
glusterfs:
endpoints: glusterfs-cluster
path: kube_vol
readOnly: true
gcePersistentDisk:
pdName: my-data-disk
fsType: ext4
gitRepo:
repository: "git@somewhere:me/my-git-repository.git"
revision: "22f1d8406d464b0c0874075539c1f2e96c253775"
hostPath:
path: /var/local/aaa
type: DirectoryOrCreate
iscsi:
targetPortal: 10.0.2.15:3260
iqn: iqn.2001-04.com.example:storage.kube.sys1.xyz
lun: 0
nfs:
server: nfs-server.default.svc.cluster.local
path: "/"
photonPersistentDisk:
pdId: 'test'
portworxVolume:
volumeID: "pxvol"
projected:
sources:
- secret:
name: mysecret
items:
- key: username
path: my-group/my-username
serviceAccountToken:
path: 'test'
quobyte:
registry: 'test'
volume: 'test'
rbd:
monitors:
- '10.16.154.78:6789'
image: foo
scaleIO:
gateway: https://localhost:443/api
system: scaleio
secretRef:
name: sio-secret
vsphereVolume:
volumePath: "[DatastoreName] volumes/myDisk"
fsType: ext4
terminationGracePeriodSeconds: 0

@ -2,6 +2,13 @@ apiVersion: v1
kind: Pod
metadata:
name: curry-endpoint-test001
ownerReferences:
- apiVersion: apps/v1
controller: true
blockOwnerDeletion: true
kind: ReplicaSet
name: my-repset
uid: d9607e19-f88f-11e6-a518-42010a800195
spec:
containers:
- image: celebdor/kuryr-demo
@ -9,3 +16,84 @@ spec:
name: web-server
ports:
- containerPort: 8080
livenessProbe:
httpGet:
path: /healthz
port: 8080
httpHeaders:
- name: Custom-Header
value: Awesome
tcpSocket:
port: 8080
initialDelaySeconds: 3
periodSeconds: 3
volumeDevices:
- name: data
devicePath: /dev/xvda
volumeMounts:
- name: redis-storage
mountPath: /data/redis
env:
- value_from:
config_map_key_ref:
key: test
secret_key_ref:
key: test
name: test
readinessGates:
- conditionType: "www.example.com/feature-1"
securityContext:
sysctls:
- name: kernel.shm_rmid_forced
value: "0"
topologySpreadConstraints:
- maxSkew: 1
topologyKey: zone
whenUnsatisfiable: DoNotSchedule
labelSelector:
matchLabels:
foo: bar
ephemeralContainers:
- name: debugger
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/e2e-az-name
operator: In
values:
- e2e-az1
- e2e-az2
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 1
preference:
matchExpressions:
- key: another-node-label-key
operator: In
values:
- another-node-label-value
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: security
operator: In
values:
- S2
topologyKey: topology.kubernetes.io/zone
status:
containerStatuses:
- lastState:
terminated:
exitCode: 1
image: test
image_id: 123
name: test
ready: True
restart_count: 1
conditions:
- status: True
type: Pod

@ -14,4 +14,9 @@ spec:
spec:
containers:
- image: nginx
name: nginx
name: nginx
status:
replicas: 1
conditions:
- status: True
type: ReplicaSet

@ -9,3 +9,9 @@ spec:
memory: 2Gi
scopes:
- NotBestEffort
scopeSelector:
matchExpressions:
- scopeName: PriorityClass
operator: In
values:
- middle

@ -1,4 +1,12 @@
apiVersion: authorization.k8s.io/v1
kind: SelfSubjectRulesReview
spec:
namespace: curry-ns
namespace: curry-ns
status:
resourceRules:
- verbs:
- 'test'
incomplete: True
nonResourceRules:
- verbs:
- 'test'

@ -31,3 +31,8 @@ spec:
resources:
requests:
storage: 1Gi
status:
conditions:
- status: True
type: StatefulSet
replicas: 1

@ -4,5 +4,10 @@ metadata:
name: curry-sc-local
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer
allowedTopologies:
- matchLabelExpressions:
- key: failure-domain.beta.kubernetes.io/zone
values:
- us-central1-a
#volumeBindingMode: Immediate
#reclaimPolicy: Retain

@ -6,4 +6,6 @@ spec:
group: apps
resource: deployments
verb: create
namespace: curry-ns
namespace: curry-ns
status:
allowed: True

@ -7,4 +7,6 @@ spec:
attacher: nginx
node_name: nginx
source:
persistent_volume_name: curry-sc-pvc
persistent_volume_name: curry-sc-pvc
status:
attached: True

@ -63,369 +63,794 @@ class TestTransformer(base.TestCase):
def test_deployment(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['deployment.yaml'], self.yaml_path)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'Deployment')
self.assertEqual(k8s_objs[0].get('object').api_version, 'apps/v1')
self.assertEqual(k8s_obj.kind, 'Deployment')
self.assertEqual(k8s_obj.api_version, 'apps/v1')
# V1DeploymentCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'Deployment')
# V1DeploymentSpec
self.assertIsNotNone(k8s_obj.spec.selector)
self.assertIsNotNone(k8s_obj.spec.template)
# V1LabelSelectorRequirement
self.assertEqual(k8s_obj.spec.selector.
match_expressions[0].key, 'test')
self.assertEqual(k8s_obj.spec.selector.
match_expressions[0].operator, 'test')
def test_api_service(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['api-service.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'APIService')
self.assertEqual(k8s_objs[0].get('object').api_version,
'apiregistration.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'APIService')
self.assertEqual(k8s_obj.api_version, 'apiregistration.k8s.io/v1')
# V1APIServiceSpec
self.assertEqual(k8s_obj.spec.group_priority_minimum, 17000)
self.assertIsNotNone(k8s_obj.spec.service)
self.assertEqual(k8s_obj.spec.version_priority, 5)
def test_cluster_role(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['cluster-role.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'ClusterRole')
self.assertEqual(k8s_objs[0].get('object').api_version,
'rbac.authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'ClusterRole')
self.assertEqual(k8s_obj.api_version, 'rbac.authorization.k8s.io/v1')
# V1PolicyRule
self.assertIsNotNone(k8s_obj.rules[0].verbs)
def test_cluster_role_binding(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['cluster-role-binding.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'ClusterRoleBinding')
self.assertEqual(k8s_objs[0].get('object').api_version,
'rbac.authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'ClusterRoleBinding')
self.assertEqual(k8s_obj.api_version, 'rbac.authorization.k8s.io/v1')
# V1ClusterRoleBinding
self.assertIsNotNone(k8s_obj.role_ref)
# V1RoleRef
self.assertEqual(k8s_obj.role_ref.api_group,
'rbac.authorization.k8s.io')
self.assertEqual(k8s_obj.role_ref.kind, 'ClusterRole')
self.assertEqual(k8s_obj.role_ref.name, 'curry-cluster-role')
# V1Subject
self.assertEqual(k8s_obj.subjects[0].kind, 'ServiceAccount')
self.assertEqual(k8s_obj.subjects[0].name, 'curry-cluster-sa')
def test_config_map(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['config-map.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'curryns')
self.assertEqual(k8s_objs[0].get('object').kind,
'ConfigMap')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
self.assertEqual(k8s_obj.kind, 'ConfigMap')
self.assertEqual(k8s_obj.api_version, 'v1')
def test_daemon_set(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['daemon-set.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'DaemonSet')
self.assertEqual(k8s_objs[0].get('object').api_version, 'apps/v1')
self.assertEqual(k8s_obj.kind, 'DaemonSet')
self.assertEqual(k8s_obj.api_version, 'apps/v1')
# V1DaemonSetStatus
self.assertEqual(k8s_obj.status.current_number_scheduled, 1)
self.assertEqual(k8s_obj.status.desired_number_scheduled, 1)
self.assertEqual(k8s_obj.status.number_misscheduled, 1)
self.assertEqual(k8s_obj.status.number_ready, 1)
# V1DaemonSetCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'DaemonSet')
# V1DaemonSetSpec
self.assertIsNotNone(k8s_obj.spec.selector)
self.assertIsNotNone(k8s_obj.spec.template)
def test_horizontal_pod_autoscaler(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['horizontal-pod-autoscaler.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'default')
self.assertEqual(k8s_objs[0].get('object').kind,
'HorizontalPodAutoscaler')
self.assertEqual(k8s_objs[0].get('object').api_version,
'autoscaling/v1')
self.assertEqual(k8s_obj.kind, 'HorizontalPodAutoscaler')
self.assertEqual(k8s_obj.api_version, 'autoscaling/v1')
# V1HorizontalPodAutoscalerSpec
self.assertEqual(k8s_obj.spec.max_replicas, 3)
self.assertIsNotNone(k8s_obj.spec.scale_target_ref)
# V1CrossVersionObjectReference
self.assertEqual(k8s_obj.spec.scale_target_ref.kind, 'Deployment')
self.assertEqual(k8s_obj.spec.scale_target_ref.name,
'curry-svc-vdu001')
# V1HorizontalPodAutoscalerStatus
self.assertEqual(k8s_obj.status.current_replicas, 1)
self.assertEqual(k8s_obj.status.desired_replicas, 1)
def test_job(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['job.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'Job')
self.assertEqual(k8s_objs[0].get('object').api_version,
'batch/v1')
self.assertEqual(k8s_obj.kind, 'Job')
self.assertEqual(k8s_obj.api_version, 'batch/v1')
# V1JobCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'Job')
# V1JobSpec
self.assertIsNotNone(k8s_obj.spec.template)
def test_lease(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['lease.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'default')
self.assertEqual(k8s_objs[0].get('object').kind, 'Lease')
self.assertEqual(k8s_objs[0].get('object').api_version,
'coordination.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'Lease')
self.assertEqual(k8s_obj.api_version, 'coordination.k8s.io/v1')
def test_local_subject_access_review(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['local-subject-access-review.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'curry-ns')
self.assertEqual(k8s_objs[0].get('object').kind,
'LocalSubjectAccessReview')
self.assertEqual(k8s_objs[0].get('object').api_version,
'authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'LocalSubjectAccessReview')
self.assertEqual(k8s_obj.api_version, 'authorization.k8s.io/v1')
self.assertIsNotNone(k8s_obj.spec)
def test_namespace(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['namespace.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'Namespace')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
self.assertEqual(k8s_obj.kind, 'Namespace')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1NamespaceCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'Namespace')
def test_network_policy(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['network-policy.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'NetworkPolicy')
self.assertEqual(k8s_objs[0].get('object').api_version,
'networking.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'NetworkPolicy')
self.assertEqual(k8s_obj.api_version, 'networking.k8s.io/v1')
# V1IPBlock
self.assertEqual(k8s_obj.spec.egress[0].to[0].ip_block.cidr,
'10.0.0.0/24')
# V1NetworkPolicySpec
self.assertIsNotNone(k8s_obj.spec.pod_selector)
def test_node(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['node.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'Node')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
self.assertEqual(k8s_obj.kind, 'Node')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1ConfigMapNodeConfigSource
self.assertEqual(k8s_obj.spec.config_source.
config_map.kubelet_config_key, 'kubelet')
self.assertEqual(k8s_obj.spec.config_source.
config_map.namespace, 'kube-system')
self.assertEqual(k8s_obj.spec.config_source.
config_map.name, 'CONFIG_MAP_NAME')
# V1Taint
self.assertEqual(k8s_obj.spec.taints[0].key, 'test')
self.assertEqual(k8s_obj.spec.taints[0].effect, 'test')
# V1NodeAddress
self.assertEqual(k8s_obj.status.addresses[0].address, '1.1.1.1')
self.assertEqual(k8s_obj.status.addresses[0].type, 'test')
# V1NodeCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'Node')
# V1DaemonEndpoint
self.assertEqual(k8s_obj.status.daemon_endpoints.
kubelet_endpoint.port, 8080)
# V1ContainerImage
self.assertEqual(k8s_obj.status.images[0].names, 'test')
# V1NodeSystemInfo
self.assertEqual(k8s_obj.status.node_info.architecture, 'test')
self.assertEqual(k8s_obj.status.node_info.boot_id, 'test')
self.assertEqual(k8s_obj.status.node_info.
container_runtime_version, 'test')
self.assertEqual(k8s_obj.status.node_info.kube_proxy_version, 'test')
self.assertEqual(k8s_obj.status.node_info.kubelet_version, 'test')
self.assertEqual(k8s_obj.status.node_info.machine_id, 'test')
self.assertEqual(k8s_obj.status.node_info.operating_system, 'test')
self.assertEqual(k8s_obj.status.node_info.os_image, 'test')
self.assertEqual(k8s_obj.status.node_info.system_uuid, 'test')
# V1AttachedVolume
self.assertEqual(k8s_obj.status.volumes_attached[0].
device_path, 'test')
self.assertEqual(k8s_obj.status.volumes_attached[0].name, 'test')
def test_persistent_volume(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['persistent-volume.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind, 'PersistentVolume')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
self.assertEqual(k8s_obj.kind, 'PersistentVolume')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1AzureFilePersistentVolumeSource
self.assertEqual(k8s_obj.spec.azure_file.secret_name, 'azure-secret')
self.assertEqual(k8s_obj.spec.azure_file.share_name, 'aksshare')
# V1CephFSPersistentVolumeSource
self.assertEqual(k8s_obj.spec.cephfs.monitors[0], '10.16.154.78:6789')
# V1CinderPersistentVolumeSource
self.assertEqual(k8s_obj.spec.cinder.volume_id,
'90d6900d-808f-4ddb-a30e-5ef821f58b4e')
# V1CSIPersistentVolumeSource
self.assertEqual(k8s_obj.spec.csi.driver, 'csi-nfsplugin')
self.assertEqual(k8s_objs[0].get('object').spec.csi.volume_handle,
'data-id')
# V1FlexPersistentVolumeSource
self.assertEqual(k8s_objs[0].get('object').spec.flex_volume.driver,
'kubernetes.io/lvm')
# V1GlusterfsPersistentVolumeSource
self.assertEqual(k8s_objs[0].get('object').spec.glusterfs.endpoints,
'glusterfs-cluster')
self.assertEqual(k8s_obj.spec.glusterfs.path, 'kube_vol')
# V1ISCSIPersistentVolumeSource
self.assertEqual(k8s_obj.spec.iscsi.target_portal, '10.0.2.15:3260')
self.assertEqual(k8s_obj.spec.iscsi.iqn,
'iqn.2001-04.com.example:storage.kube.sys1.xyz')
self.assertEqual(k8s_obj.spec.iscsi.lun, 0)
# V1LocalVolumeSource
self.assertEqual(k8s_obj.spec.local.path, '/mnt/disks/ssd1')
# V1RBDPersistentVolumeSource
self.assertEqual(k8s_obj.spec.rbd.monitors[0], '10.16.154.78:6789')
self.assertEqual(k8s_obj.spec.rbd.image, 'foo')
# V1ScaleIOPersistentVolumeSource
self.assertEqual(k8s_obj.spec.scale_io.gateway,
'https://localhost:443/api')
self.assertIsNotNone(k8s_obj.spec.scale_io.secret_ref)
self.assertEqual(k8s_obj.spec.scale_io.system, 'scaleio')
# V1AWSElasticBlockStoreVolumeSource
self.assertEqual(k8s_obj.spec.aws_elastic_block_store.volume_id,
'123')
# V1AzureDiskVolumeSource
self.assertEqual(k8s_obj.spec.azure_disk.disk_name, 'test.vhd')
self.assertEqual(
k8s_obj.spec.azure_disk.disk_uri,
'https://someaccount.blob.microsoft.net/vhds/test.vhd')
def test_persistent_volume_claim(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['persistent-volume-claim.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
self.assertEqual(k8s_obj.kind, 'PersistentVolumeClaim')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1PersistentVolumeClaimCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type,
'PersistentVolumeClaim')
# V1TypedLocalObjectReference
self.assertEqual(k8s_obj.spec.data_source.name,
'existing-src-pvc-name')
self.assertEqual(k8s_obj.spec.data_source.kind,
'PersistentVolumeClaim')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
def test_pod(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['pod.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'Pod')
self.assertEqual(k8s_objs[0].get('object').api_version, 'v1')
self.assertEqual(k8s_obj.kind, 'Pod')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1NodeSelector
self.assertIsNotNone(
k8s_obj.spec.affinity.node_affinity.
required_during_scheduling_ignored_during_execution.
node_selector_terms)
# V1NodeSelectorRequirement
self.assertEqual(
k8s_obj.spec.affinity.node_affinity.
required_during_scheduling_ignored_during_execution.
node_selector_terms[0].match_expressions[0].key,
'kubernetes.io/e2e-az-name')
self.assertEqual(
k8s_obj.spec.affinity.node_affinity.
required_during_scheduling_ignored_during_execution.
node_selector_terms[0].match_expressions[0].operator,
'In')
# V1PreferredSchedulingTerm
self.assertEqual(
k8s_obj.spec.affinity.node_affinity.
preferred_during_scheduling_ignored_during_execution[0].
weight, 1)
self.assertIsNotNone(
k8s_obj.spec.affinity.node_affinity.
preferred_during_scheduling_ignored_during_execution[0].
preference)
# V1PodAffinityTerm
self.assertEqual(
k8s_obj.spec.affinity.pod_anti_affinity.
preferred_during_scheduling_ignored_during_execution[0].
pod_affinity_term.topology_key, 'topology.kubernetes.io/zone')
# V1WeightedPodAffinityTerm
self.assertEqual(
k8s_obj.spec.affinity.pod_anti_affinity.
preferred_during_scheduling_ignored_during_execution[0].
weight, 100)
self.assertIsNotNone(
k8s_obj.spec.affinity.pod_anti_affinity.
preferred_during_scheduling_ignored_during_execution[0].
pod_affinity_term)
# V1OwnerReference
self.assertEqual(k8s_obj.metadata.owner_references[0].api_version,
'apps/v1')
self.assertEqual(k8s_obj.metadata.owner_references[0].kind,
'ReplicaSet')
self.assertEqual(k8s_obj.metadata.owner_references[0].name,
'my-repset')
self.assertEqual(k8s_obj.metadata.owner_references[0].uid,
'd9607e19-f88f-11e6-a518-42010a800195')
# V1HTTPHeader
self.assertEqual(k8s_obj.spec.containers[0].liveness_probe.http_get.
http_headers[0].name, 'Custom-Header')
self.assertEqual(k8s_obj.spec.containers[0].liveness_probe.http_get.
http_headers[0].value, 'Awesome')
# V1TCPSocketAction
self.assertEqual(k8s_obj.spec.containers[0].liveness_probe.
tcp_socket.port, 8080)
# V1VolumeDevice
self.assertEqual(k8s_obj.spec.containers[0].volume_devices[0].
device_path, '/dev/xvda')
self.assertEqual(k8s_obj.spec.containers[0].volume_devices[0].name,
'data')
# V1PodReadinessGate
self.assertEqual(k8s_obj.spec.readiness_gates[0].condition_type,
'www.example.com/feature-1')
# V1Sysctl
self.assertEqual(k8s_obj.spec.security_context.sysctls[0].name,
'kernel.shm_rmid_forced')
self.assertEqual(k8s_obj.spec.security_context.sysctls[0].value, '0')
# V1ContainerStateTerminated
self.assertEqual(k8s_obj.status.container_statuses[0].last_state.
terminated.exit_code, 1)
# V1EphemeralContainer
self.assertEqual(k8s_obj.spec.topology_spread_constraints[0].
topology_key, 'zone')
# V1TopologySpreadConstraint
self.assertEqual(k8s_obj.spec.ephemeral_containers[0].name,
'debugger')
# V1HTTPGetAction
self.assertEqual(k8s_obj.spec.containers[0].liveness_probe.
http_get.port, 8080)
# V1ConfigMapKeySelector
self.assertEqual(k8s_obj.spec.containers[0].env[0].value_from.
config_map_key_ref.key, 'test')
# V1EnvVar
self.assertEqual(k8s_obj.spec.containers[0].env[0].name, 'test')
# V1SecretKeySelector
self.assertEqual(k8s_obj.spec.containers[0].env[0].value_from.
secret_key_ref.key, 'test')
# V1ContainerPort
self.assertEqual(k8s_obj.spec.containers[0].ports[0].
container_port, 8080)
# V1VolumeMount
self.assertEqual(k8s_obj.spec.containers[0].volume_mounts[0].
mount_path, '/data/redis')
self.assertEqual(k8s_obj.spec.containers[0].volume_mounts[0].
name, 'redis-storage')
# V1PodCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'Pod')
# V1ContainerStatus
self.assertEqual(k8s_obj.status.container_statuses[0].image, 'test')
self.assertEqual(k8s_obj.status.container_statuses[0].image_id, 123)
self.assertEqual(k8s_obj.status.container_statuses[0].name, 'test')
self.assertEqual(k8s_obj.status.container_statuses[0].ready, True)
self.assertEqual(k8s_obj.status.container_statuses[0].restart_count, 1)
def test_priority_class(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['priority-class.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'PriorityClass')
self.assertEqual(k8s_objs[0].get('object').api_version,
'scheduling.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'PriorityClass')
self.assertEqual(k8s_obj.api_version, 'scheduling.k8s.io/v1')
# V1PriorityClass
self.assertEqual(k8s_obj.value, 1000000)
def test_replica_set(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['replica-set.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'ReplicaSet')
self.assertEqual(k8s_objs[0].get('object').api_version,
'apps/v1')
self.assertEqual(k8s_obj.kind, 'ReplicaSet')
self.assertEqual(k8s_obj.api_version, 'apps/v1')
# V1ReplicaSetStatus
self.assertEqual(k8s_obj.status.replicas, 1)
# V1ReplicaSetCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'ReplicaSet')
# V1ReplicaSetSpec
self.assertIsNotNone(k8s_obj.spec.selector)
self.assertIsNotNone(k8s_obj.spec.template)
def test_resource_quota(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['resource-quota.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'curryns')
self.assertEqual(k8s_objs[0].get('object').kind,
'ResourceQuota')
self.assertEqual(k8s_objs[0].get('object').api_version,
'v1')
self.assertEqual(k8s_obj.kind, 'ResourceQuota')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1ScopedResourceSelectorRequirement
self.assertEqual(k8s_obj.spec.scope_selector.
match_expressions[0].operator, 'In')
self.assertEqual(k8s_obj.spec.scope_selector.
match_expressions[0].scope_name, 'PriorityClass')
def test_role(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['role.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'curry-ns')
self.assertEqual(k8s_objs[0].get('object').kind,
'Role')
self.assertEqual(k8s_objs[0].get('object').api_version,
'rbac.authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'Role')
self.assertEqual(k8s_obj.api_version, 'rbac.authorization.k8s.io/v1')
def test_role_binding(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['role-bindings.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'curry-ns')
self.assertEqual(k8s_objs[0].get('object').kind,
'RoleBinding')
self.assertEqual(k8s_objs[0].get('object').api_version,
'rbac.authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'RoleBinding')
self.assertEqual(k8s_obj.api_version, 'rbac.authorization.k8s.io/v1')
# V1RoleBinding
self.assertIsNotNone(k8s_obj.role_ref)
def test_secret(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['secret.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'default')
self.assertEqual(k8s_objs[0].get('object').kind,
'Secret')
self.assertEqual(k8s_objs[0].get('object').api_version,
'v1')
self.assertEqual(k8s_obj.kind, 'Secret')
self.assertEqual(k8s_obj.api_version, 'v1')
def test_self_subject_access_review(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['self-subject-access-review.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'SelfSubjectAccessReview')
self.assertEqual(k8s_objs[0].get('object').api_version,
'authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'SelfSubjectAccessReview')
self.assertEqual(k8s_obj.api_version, 'authorization.k8s.io/v1')
# V1SelfSubjectAccessReview
self.assertIsNotNone(k8s_obj.spec)
def test_self_subject_rules_review(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['self-subject-rule-review.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'SelfSubjectRulesReview')
self.assertEqual(k8s_objs[0].get('object').api_version,
'authorization.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'SelfSubjectRulesReview')
self.assertEqual(k8s_obj.api_version, 'authorization.k8s.io/v1')
# V1ResourceRule
self.assertEqual(k8s_obj.status.resource_rules[0].verbs[0], 'test')
# V1SelfSubjectRulesReview
self.assertIsNotNone(k8s_obj.spec)
# V1SubjectRulesReviewStatus
self.assertIsNotNone(k8s_obj.status.resource_rules)
self.assertIsNotNone(k8s_obj.status.non_resource_rules)
self.assertEqual(k8s_obj.status.incomplete, True)
# V1NonResourceRule
self.assertEqual(k8s_obj.status.non_resource_rules[0].verbs[0], 'test')
def test_service(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['service.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'default')
self.assertEqual(k8s_objs[0].get('object').kind,
'Service')
self.assertEqual(k8s_objs[0].get('object').api_version,
'v1')
self.assertEqual(k8s_obj.kind, 'Service')
self.assertEqual(k8s_obj.api_version, 'v1')
# V1ServicePort
self.assertEqual(k8s_obj.spec.ports[0].port, 80)
def test_service_account(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['service-account.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), 'default')
self.assertEqual(k8s_objs[0].get('object').kind,
'ServiceAccount')
self.assertEqual(k8s_objs[0].get('object').api_version,
'v1')
self.assertEqual(k8s_obj.kind, 'ServiceAccount')
self.assertEqual(k8s_obj.api_version, 'v1')
def test_stateful_set(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['stateful-set.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'StatefulSet')
self.assertEqual(k8s_objs[0].get('object').api_version,
'apps/v1')
self.assertEqual(k8s_obj.kind, 'StatefulSet')
self.assertEqual(k8s_obj.api_version, 'apps/v1')
# V1StatefulSetSpec
self.assertIsNotNone(k8s_obj.spec.selector)
self.assertIsNotNone(k8s_obj.spec.template)
self.assertEqual(k8s_obj.spec.service_name, 'nginx')
# V1StatefulSetCondition
self.assertEqual(k8s_obj.status.conditions[0].status, True)
self.assertEqual(k8s_obj.status.conditions[0].type, 'StatefulSet')
# V1StatefulSetStatus
self.assertEqual(k8s_obj.status.replicas, 1)
def test_storage_class(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['storage-class.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)
self.assertEqual(k8s_objs[0].get('namespace'), '')
self.assertEqual(k8s_objs[0].get('object').kind,
'StorageClass')
self.assertEqual(k8s_objs[0].get('object').api_version,
'storage.k8s.io/v1')
self.assertEqual(k8s_obj.kind, 'StorageClass')
self.assertEqual(k8s_obj.api_version, 'storage.k8s.io/v1')
# V1StorageClass
self.assertEqual(k8s_obj.provisioner, 'kubernetes.io/no-provisioner')
# V1TopologySelectorLabelRequirement
self.assertEqual(k8s_obj.allowed_topologies[0].
match_label_expressions[0].key,
'failure-domain.beta.kubernetes.io/zone')
self.assertEqual(k8s_obj.allowed_topologies[0].
match_label_expressions[0].values[0],
'us-central1-a')
def test_subject_access_review(self):
k8s_objs = self.transfromer.get_k8s_objs_from_yaml(
['subject-access-review.yaml'], self.yaml_path
)
self.assertIsNotNone(k8s_objs[0].get('object'))
k8s_obj = k8s_objs[0].get('object')
self.assertIsNotNone(k8s_obj)