<s> '''<STR_LIT>''' <EOL> from __future__ import absolute_import <EOL> import json <EOL> import hashlib <EOL> import base64 <EOL> import time <EOL> from subprocess import Popen , PIPE <EOL> from salttesting import TestCase <EOL> from salttesting . helpers import ensure_in_syspath , skip_if_binaries_missing <EOL> ensure_in_syspath ( '<STR_LIT>' ) <EOL> import salt . modules . k8s as k8s <EOL> TestCase . maxDiff = None <EOL> @ skip_if_binaries_missing ( [ '<STR_LIT>' ] ) <EOL> class TestK8SNamespace ( TestCase ) : <EOL> def test_get_namespaces ( self ) : <EOL> res = k8s . get_namespaces ( apiserver_url = "<STR_LIT>" ) <EOL> a = len ( res . get ( "<STR_LIT>" ) ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = len ( kubectl_out . get ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( a , b ) <EOL> def test_get_one_namespace ( self ) : <EOL> res = k8s . get_namespaces ( "<STR_LIT:default>" , apiserver_url = "<STR_LIT>" ) <EOL> a = res . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:a>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:default>" , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:b>" ) <EOL> self . assertEqual ( a , b ) <EOL> def test_create_namespace ( self ) : <EOL> hash = hashlib . sha1 ( ) <EOL> hash . update ( str ( time . time ( ) ) ) <EOL> nsname = hash . hexdigest ( ) [ : <NUM_LIT:16> ] <EOL> res = k8s . create_namespace ( nsname , apiserver_url = "<STR_LIT>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , nsname , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( isinstance ( kubectl_out , dict ) ) <EOL> @ skip_if_binaries_missing ( [ '<STR_LIT>' ] ) <EOL> class TestK8SSecrets ( TestCase ) : <EOL> def setUp ( self ) : <EOL> hash = hashlib . sha1 ( ) <EOL> hash . update ( str ( time . time ( ) ) ) <EOL> self . name = hash . hexdigest ( ) [ : <NUM_LIT:16> ] <EOL> data = { "<STR_LIT>" : base64 . encodestring ( "<STR_LIT>" ) } <EOL> self . request = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : self . name , <EOL> "<STR_LIT>" : "<STR_LIT:default>" , <EOL> } , <EOL> "<STR_LIT:data>" : data , <EOL> } <EOL> def test_get_secrets ( self ) : <EOL> res = k8s . get_secrets ( "<STR_LIT:default>" , apiserver_url = "<STR_LIT>" ) <EOL> a = len ( res . get ( "<STR_LIT>" , [ ] ) ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = len ( kubectl_out . get ( "<STR_LIT>" , [ ] ) ) <EOL> self . assertEqual ( a , b ) <EOL> def test_get_one_secret ( self ) : <EOL> name = self . name <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> json . dump ( self . request , f ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> res = k8s . get_secrets ( "<STR_LIT:default>" , name , apiserver_url = "<STR_LIT>" ) <EOL> a = res . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:a>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:b>" ) <EOL> self . assertEqual ( a , b ) <EOL> def test_get_decoded_secret ( self ) : <EOL> name = self . name <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> json . dump ( self . request , f ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> res = k8s . get_secrets ( "<STR_LIT:default>" , name , apiserver_url = "<STR_LIT>" , decode = True ) <EOL> a = res . get ( "<STR_LIT:data>" , { } ) . get ( "<STR_LIT>" , ) <EOL> self . assertEqual ( a , "<STR_LIT>" ) <EOL> def test_create_secret ( self ) : <EOL> name = self . name <EOL> names = [ ] <EOL> expected_data = { } <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> names . append ( "<STR_LIT>" . format ( name , i ) ) <EOL> with open ( "<STR_LIT>" . format ( name , i ) , '<STR_LIT:w>' ) as f : <EOL> expected_data [ "<STR_LIT>" . format ( name , i ) ] = base64 . b64encode ( "<STR_LIT>" . format ( name , i ) ) <EOL> f . write ( "<STR_LIT>" . format ( name , i ) ) <EOL> res = k8s . create_secret ( "<STR_LIT:default>" , name , names , apiserver_url = "<STR_LIT>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT:data>" , { } ) <EOL> self . assertTrue ( isinstance ( kubectl_out , dict ) ) <EOL> self . assertEqual ( expected_data , b ) <EOL> def test_update_secret ( self ) : <EOL> name = self . name <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> json . dump ( self . request , f ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> expected_data = { } <EOL> names = [ ] <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> names . append ( "<STR_LIT>" . format ( name , i ) ) <EOL> with open ( "<STR_LIT>" . format ( name , i ) , '<STR_LIT:w>' ) as f : <EOL> expected_data [ "<STR_LIT>" . format ( name , i ) ] = base64 . b64encode ( "<STR_LIT>" . format ( name , i ) ) <EOL> f . write ( "<STR_LIT>" . format ( name , i ) ) <EOL> res = k8s . update_secret ( "<STR_LIT:default>" , name , names , apiserver_url = "<STR_LIT>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT:data>" , { } ) <EOL> self . assertTrue ( isinstance ( kubectl_out , dict ) ) <EOL> self . assertEqual ( expected_data , b ) <EOL> def test_delete_secret ( self ) : <EOL> name = self . name <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> json . dump ( self . request , f ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> res = k8s . delete_secret ( "<STR_LIT:default>" , name , apiserver_url = "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE , stderr = PIPE ) <EOL> kubectl_out , err = proc . communicate ( ) <EOL> self . assertEqual ( '<STR_LIT>' , kubectl_out ) <EOL> self . assertEqual ( '<STR_LIT>' . format ( name ) , err ) <EOL> @ skip_if_binaries_missing ( [ '<STR_LIT>' ] ) <EOL> class TestK8SResourceQuotas ( TestCase ) : <EOL> def setUp ( self ) : <EOL> hash = hashlib . sha1 ( ) <EOL> hash . update ( str ( time . time ( ) ) ) <EOL> self . name = hash . hexdigest ( ) [ : <NUM_LIT:16> ] <EOL> def test_get_resource_quotas ( self ) : <EOL> name = self . name <EOL> namespace = self . name <EOL> create_namespace = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , namespace ] , stdout = PIPE ) <EOL> create_namespace = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , namespace ] , stdout = PIPE ) <EOL> request = """<STR_LIT>""" . format ( name ) <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> f . write ( request ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> res = k8s . get_resource_quotas ( namespace , apiserver_url = "<STR_LIT>" ) <EOL> a = len ( res . get ( "<STR_LIT>" , [ ] ) ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = len ( kubectl_out . get ( "<STR_LIT>" , [ ] ) ) <EOL> self . assertEqual ( a , b ) <EOL> def test_get_one_resource_quota ( self ) : <EOL> name = self . name <EOL> namespace = self . name <EOL> create_namespace = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , namespace ] , stdout = PIPE ) <EOL> request = """<STR_LIT>""" . format ( name ) <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> f . write ( request ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> res = k8s . get_resource_quotas ( namespace , name , apiserver_url = "<STR_LIT>" ) <EOL> a = res . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:a>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:b>" ) <EOL> self . assertEqual ( a , b ) <EOL> def test_create_resource_quota ( self ) : <EOL> name = self . name <EOL> namespace = self . name <EOL> create_namespace = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , namespace ] , stdout = PIPE ) <EOL> quota = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> res = k8s . create_resource_quota ( namespace , quota , name = name , apiserver_url = "<STR_LIT>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( isinstance ( kubectl_out , dict ) ) <EOL> def test_update_resource_quota ( self ) : <EOL> name = self . name <EOL> namespace = self . name <EOL> create_namespace = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , namespace ] , stdout = PIPE ) <EOL> request = """<STR_LIT>""" . format ( name ) <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> f . write ( request ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> quota = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> res = k8s . create_resource_quota ( namespace , quota , name = name , apiserver_url = "<STR_LIT>" , update = True ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" . format ( namespace ) , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> limit = kubectl_out . get ( "<STR_LIT>" ) . get ( "<STR_LIT>" ) . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT>" , limit ) <EOL> @ skip_if_binaries_missing ( [ '<STR_LIT>' ] ) <EOL> class TestK8SLimitRange ( TestCase ) : <EOL> def setUp ( self ) : <EOL> hash = hashlib . sha1 ( ) <EOL> hash . update ( str ( time . time ( ) ) ) <EOL> self . name = hash . hexdigest ( ) [ : <NUM_LIT:16> ] <EOL> def test_create_limit_range ( self ) : <EOL> name = self . name <EOL> limits = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> } <EOL> res = k8s . create_limit_range ( "<STR_LIT:default>" , limits , name = name , apiserver_url = "<STR_LIT>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( isinstance ( kubectl_out , dict ) ) <EOL> def test_update_limit_range ( self ) : <EOL> name = self . name <EOL> request = """<STR_LIT>""" . format ( name ) <EOL> limits = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } <EOL> } <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> f . write ( request ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> res = k8s . create_limit_range ( "<STR_LIT:default>" , limits , name = name , apiserver_url = "<STR_LIT>" , update = True ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> limit = kubectl_out . get ( "<STR_LIT>" ) . get ( "<STR_LIT>" ) [ <NUM_LIT:0> ] . get ( "<STR_LIT>" ) . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( "<STR_LIT>" , limit ) <EOL> def test_get_limit_ranges ( self ) : <EOL> res = k8s . get_limit_ranges ( "<STR_LIT:default>" , apiserver_url = "<STR_LIT>" ) <EOL> a = len ( res . get ( "<STR_LIT>" , [ ] ) ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = len ( kubectl_out . get ( "<STR_LIT>" , [ ] ) ) <EOL> self . assertEqual ( a , b ) <EOL> def test_get_one_limit_range ( self ) : <EOL> name = self . name <EOL> request = """<STR_LIT>""" . format ( name ) <EOL> filename = "<STR_LIT>" . format ( name ) <EOL> with open ( filename , '<STR_LIT:w>' ) as f : <EOL> f . write ( request ) <EOL> create = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , filename ] , stdout = PIPE ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> res = k8s . get_limit_ranges ( "<STR_LIT:default>" , name , apiserver_url = "<STR_LIT>" ) <EOL> a = res . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:a>" ) <EOL> proc = Popen ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , name , "<STR_LIT>" , "<STR_LIT>" ] , stdout = PIPE ) <EOL> kubectl_out = json . loads ( proc . communicate ( ) [ <NUM_LIT:0> ] ) <EOL> b = kubectl_out . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT:name>" , "<STR_LIT:b>" ) <EOL> self . assertEqual ( a , b ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from integration import run_tests <EOL> run_tests ( TestK8SNamespace , <EOL> TestK8SSecrets , <EOL> TestK8SResourceQuotas , <EOL> TestK8SLimitRange , <EOL> needs_daemon = False ) </s>
