test added, error boundaries set

This commit is contained in:
Akshit Agrawal
2023-12-19 13:27:35 +05:30
parent 0f331248ca
commit e71939800d
5 changed files with 115 additions and 55 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
FROM alpine:3.18
FROM --platform=${BUILDPLATFORM:-linux/amd64} alpine:3.18 as base
WORKDIR /app
Binary file not shown.
+5
View File
@@ -0,0 +1,5 @@
apiVersion: v2
name: mywebapp
description: a simple web app
version: 5.0.0
appVersion: 5.0.0
+62 -54
View File
@@ -1,64 +1,72 @@
# Plugin Name: Push OCI Chart to Registry
# Description: Pushes an Helm Chart to a Docker Registry
# your_script_file.py
import os
import subprocess
# Environment Variables
# Helm Chart
CHART_NAME = os.getenv("PLUGIN_CHART_NAME") # ! required
CHART_VERSION = os.getenv("PLUGIN_CHART_VERSION", "1.0.0")
def main_function():
# Environment Variables
# Docker Repository Details
DOCKER_REGISTRY = os.getenv(
"PLUGIN_DOCKER_REGISTRY", 'registry.hub.docker.com') # ? optional
# Helm Chart
CHART_NAME = os.getenv("PLUGIN_CHART_NAME") # ! required
CHART_VERSION = os.getenv("PLUGIN_CHART_VERSION", "1.0.0")
# Docker Hub Credentials
DOCKER_USERNAME = os.getenv(
"PLUGIN_DOCKER_USERNAME") # ! required
DOCKER_PASSWORD = os.getenv(
"PLUGIN_DOCKER_PASSWORD") # ! required
# Docker Repository Details
DOCKER_REGISTRY = os.getenv(
"PLUGIN_DOCKER_REGISTRY", 'registry.hub.docker.com') # ? optional
# Path to Chart
CHART_PATH = os.getenv("PLUGIN_CHART_PATH") # ? optional
# Docker Hub Credentials
DOCKER_USERNAME = os.getenv(
"PLUGIN_DOCKER_USERNAME") # ! required
DOCKER_PASSWORD = os.getenv(
"PLUGIN_DOCKER_PASSWORD") # ! required
# Path to Chart
CHART_PATH = os.getenv("PLUGIN_CHART_PATH") # ? optional
# ? validate environment variables
if (CHART_NAME is None):
print("Please provide a chart name")
exit(1)
if (DOCKER_USERNAME is None or DOCKER_PASSWORD is None):
print("Please provide a username and a password")
exit(1)
# cd into the chart directory
if (CHART_PATH is not None):
os.chdir(CHART_PATH)
# Package the helm chart
try:
subprocess.run(["helm", "package", "--dependency-update", "."])
except subprocess.CalledProcessError:
print("Failed to package chart!")
exit(1)
# Construct the chart name
chart_filename = f"{CHART_NAME}-{CHART_VERSION}.tgz"
# Login to Docker Registry
try:
login_command = ['helm', 'registry', 'login', DOCKER_REGISTRY,
'-u', DOCKER_USERNAME, '-p', DOCKER_PASSWORD]
subprocess.run(login_command)
except subprocess.CalledProcessError:
print("Failed to login!")
exit(1)
# Push the chart to Docker Hub
try:
docker_push_command = ["helm", "push", chart_filename,
f"oci://{DOCKER_REGISTRY}/{DOCKER_USERNAME}"]
subprocess.run(docker_push_command)
print("Chart pushed successfully.")
except subprocess.CalledProcessError:
print("Failed to push chart!")
exit(1)
# ? validate environment variables
if (CHART_NAME is None):
print("Please provide a chart name")
exit(1)
if (DOCKER_USERNAME is None or DOCKER_PASSWORD is None):
print("Please provide a username and a password")
exit(1)
# cd into the chart directory
if (CHART_PATH is not None):
os.chdir(CHART_PATH)
# Package the helm chart
subprocess.run(["helm", "package", "--dependency-update", "."])
# Construct the chart name
chart_filename = f"{CHART_NAME}-{CHART_VERSION}.tgz"
# Login to Docker Registry
try:
login_command = ['helm', 'registry', 'login', DOCKER_REGISTRY,
'-u', DOCKER_USERNAME, '-p', DOCKER_PASSWORD]
subprocess.run(login_command)
except:
print("Failed to login!")
exit(1)
# Push the chart to Docker Hub
try:
docker_push_command = ["helm", "push", chart_filename,
f"oci://{DOCKER_REGISTRY}/{DOCKER_USERNAME}"]
subprocess.run(docker_push_command)
except:
print("Failed to push chart!")
exit(1)
if __name__ == '__main__':
main_function()
+47
View File
@@ -0,0 +1,47 @@
import unittest
from unittest.mock import patch
import os
from io import StringIO
from main import main_function
class TestPushOCIChartToRegistry(unittest.TestCase):
@patch("subprocess.run")
def test_successful_chart_push(self, mock_subprocess_run):
os.environ["PLUGIN_CHART_NAME"] = "mywebapp"
os.environ["PLUGIN_DOCKER_USERNAME"] = "testuser"
os.environ["PLUGIN_DOCKER_PASSWORD"] = "testpassword"
os.environ["PLUGIN_CHART_PATH"] = "chart"
mock_subprocess_run.return_value.returncode = 0
with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
main_function()
mock_subprocess_run.assert_called_with(
["helm", "push", "mywebapp-1.0.0.tgz", "oci://registry.hub.docker.com/testuser"])
expected_output = 'Chart pushed successfully.'
self.assertEqual(mock_stdout.getvalue().strip(), expected_output)
# @patch("subprocess.run")
# def test_failed_chart_push(self, mock_subprocess_run):
# mock_subprocess_run.return_value.returncode = 1
# os.environ["PLUGIN_CHART_NAME"] = "mywebapp"
# os.environ["PLUGIN_DOCKER_USERNAME"] = "testuser"
# os.environ["PLUGIN_DOCKER_PASSWORD"] = "testpassword"
# with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
# main_function()
# mock_subprocess_run.assert_called_with(
# ["helm", "push", "mywebapp-1.0.0.tgz", "oci://registry.hub.docker.com/testuser"])
# expected_output = "Failed to push chart!" # Update the expected output
# self.assertEqual(mock_stdout.getvalue().strip(), expected_output)
if __name__ == '__main__':
unittest.main()