Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

SDK - Python support for arbitrary secret, similar to ".use_gcp_secret('user-gcp-sa')" #2639

Merged
merged 16 commits into from
Dec 3, 2019
Prev Previous commit
Next Next commit
the new test after renaming
  • Loading branch information
NikeNano committed Nov 24, 2019
commit 7a11fd487d154f0fed1c890aa8e6552876710909
58 changes: 58 additions & 0 deletions sdk/python/kfp/test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
from kfp.dsl import ContainerOp
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
from utils import use_secret
import unittest
import inspect


class TestAddSecrets(unittest.TestCase):

def test_use_default_use_secret(self):
op1 = ContainerOp(name='op1', image='image')
secret_name = "my-secret"
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
secret_path = "/here/are/my/secret"
op1 = op1.apply(use_secret(secret_name=secret_name,
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
secret_volume_mount_path=secret_path))
self.assertEqual(type(op1.container.env), type(None))
container_dict = op1.container.to_dict()
volume_mounts = container_dict["volume_mounts"][0]
self.assertEqual(type(volume_mounts),dict)
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
self.assertEqual(volume_mounts["name"],secret_name + '_volume')
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
self.assertEqual(volume_mounts["mount_path"], secret_path)

def test_use_set_volume_use_secret(self):
op1 = ContainerOp(name='op1', image='image')
secret_name = "my-secret"
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
secret_path = "/here/are/my/secret"
volume_name = "my_volume"
op1 = op1.apply(use_secret(secret_name=secret_name,
secret_volume_mount_path=secret_path,
volume_name = volume_name))
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
self.assertEqual( type(op1.container.env), type(None))
NikeNano marked this conversation as resolved.
Show resolved Hide resolved
container_dict = op1.container.to_dict()
volume_mounts = container_dict["volume_mounts"][0]
self.assertEqual( type(volume_mounts), dict)
self.assertEqual( volume_mounts["name"], volume_name)
self.assertEqual( volume_mounts["mount_path"], secret_path)

def test_use_set_env_ues_secret(self):
op1 = ContainerOp(name='op1', image='image')
secret_name = "my-secret"
secret_path = "/here/are/my/secret/"
env_variable = "MY_SECRET"
secret_file_path_in_volume = "secret.json"
op1 = op1.apply(use_secret(secret_name=secret_name,
secret_volume_mount_path=secret_path,
env_variable=env_variable,
secret_file_path_in_volume=secret_file_path_in_volume))
self.assertEqual( len(op1.container.env), 1)
container_dict = op1.container.to_dict()
volume_mounts = container_dict["volume_mounts"][0]
self.assertEqual(type(volume_mounts), dict)
self.assertEqual(volume_mounts["name"], secret_name + '_volume')
self.assertEqual(volume_mounts["mount_path"], secret_path)
env_dict = op1.container.env[0].to_dict()
self.assertEqual(env_dict["name"], env_variable)
self.assertEqual(env_dict["value"], secret_path + secret_file_path_in_volume)

if __name__ == '__main__':
unittest.main()