From 003217a2b84ac0e7312abfb59575d368af59404a Mon Sep 17 00:00:00 2001 From: Jonathan Branan Date: Tue, 27 Oct 2026 00:11:53 -0500 Subject: [PATCH] test: cover komodo-deploy.sh against a mock Komodo API 15 unittest cases exercising the real script end to end: target selection, auth headers, request shape, success/failure, HTTP 401/400, unreachable host, polling to completion, and timeout. Runs in CI via .gitea/workflows/test.yml on push to main/dev* and on pull requests. No network or live Komodo instance required. --- .gitea/workflows/test.yml | 17 +++ test_komodo_deploy.py | 273 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 290 insertions(+) create mode 100644 .gitea/workflows/test.yml create mode 100644 test_komodo_deploy.py diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml new file mode 100644 index 0000000..6e456d5 --- /dev/null +++ b/.gitea/workflows/test.yml @@ -0,0 +1,17 @@ +name: test + +on: + push: + branches: + - main + - dev* + pull_request: + +jobs: + komodo-deploy-script: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Run komodo-deploy.sh tests + run: python3 -m unittest -v test_komodo_deploy diff --git a/test_komodo_deploy.py b/test_komodo_deploy.py new file mode 100644 index 0000000..0db0892 --- /dev/null +++ b/test_komodo_deploy.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""Tests for .gitea/scripts/komodo-deploy.sh. + +Runs the real script against a local mock of the Komodo Core API, so no +network access or live Komodo instance is required. + +Run with: python3 -m unittest -v test_komodo_deploy +""" + +import json +import os +import socket +import subprocess +import threading +import unittest +from http.server import BaseHTTPRequestHandler, HTTPServer + +REPO_ROOT = os.path.dirname(os.path.abspath(__file__)) +SCRIPT = os.path.join(REPO_ROOT, ".gitea", "scripts", "komodo-deploy.sh") + + +def _free_port(): + """Return a port that is currently closed, for the unreachable-host test.""" + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class MockKomodo: + """Minimal mock of the Komodo Core endpoints used by the deploy script. + + ``responses`` is a list of ``(status, body)`` pairs consumed in order. A + single-element list repeats forever, which is what the timeout test needs. + Every request is recorded in ``requests`` for assertions. + """ + + def __init__(self, responses): + self.responses = list(responses) + self.requests = [] + mock = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + raw = self.rfile.read(length).decode() if length else "" + mock.requests.append( + { + "path": self.path, + "key": self.headers.get("x-api-key"), + "secret": self.headers.get("x-api-secret"), + "body": raw, + } + ) + status, body = ( + mock.responses.pop(0) + if len(mock.responses) > 1 + else mock.responses[0] + ) + payload = json.dumps(body).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + self.server = HTTPServer(("127.0.0.1", 0), Handler) + self.port = self.server.server_address[1] + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *exc): + self.server.shutdown() + self.server.server_close() + + @property + def url(self): + return "http://127.0.0.1:%d" % self.port + + +def run_script(env_overrides, timeout=60): + """Run the deploy script with a clean KOMODO_* environment.""" + env = {k: v for k, v in os.environ.items() if not k.startswith("KOMODO_")} + env.update(env_overrides) + return subprocess.run( + ["sh", SCRIPT], + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def base_env(url, **extra): + env = { + "KOMODO_URL": url, + "KOMODO_API_KEY": "test-key", + "KOMODO_API_SECRET": "test-secret", + "KOMODO_TARGET": "qbit-maid", + "KOMODO_WAIT_INTERVAL": "1", + } + env.update(extra) + return env + + +class TestKomodoDeployScript(unittest.TestCase): + # --- target selection ------------------------------------------------- + + def test_skips_when_target_empty(self): + r = run_script({"KOMODO_TARGET": ""}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("skipping", r.stdout) + + def test_fails_when_target_empty_and_required(self): + r = run_script({"KOMODO_TARGET": "", "KOMODO_REQUIRED": "true"}) + self.assertEqual(r.returncode, 1) + self.assertIn("KOMODO_REQUIRED=true", r.stderr) + + def test_fails_when_url_missing(self): + r = run_script({"KOMODO_TARGET": "qbit-maid"}) + self.assertEqual(r.returncode, 1) + self.assertIn("KOMODO_URL is required", r.stderr) + + def test_fails_when_api_key_missing(self): + r = run_script( + { + "KOMODO_URL": "http://127.0.0.1:1", + "KOMODO_TARGET": "qbit-maid", + "KOMODO_API_SECRET": "s", + } + ) + self.assertEqual(r.returncode, 1) + self.assertIn("KOMODO_API_KEY is required", r.stderr) + + def test_invalid_target_type(self): + r = run_script( + { + "KOMODO_URL": "http://127.0.0.1:1", + "KOMODO_API_KEY": "k", + "KOMODO_API_SECRET": "s", + "KOMODO_TARGET": "qbit-maid", + "KOMODO_TARGET_TYPE": "bogus", + } + ) + self.assertEqual(r.returncode, 1) + self.assertIn("must be 'stack' or 'procedure'", r.stderr) + + # --- request shape ---------------------------------------------------- + + def test_sends_auth_headers_and_stack_body(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + (200, {"status": "Complete", "success": True}), + ] + ) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 0, r.stderr) + first = mock.requests[0] + self.assertEqual(first["path"], "/execute/DeployStack") + self.assertEqual(first["key"], "test-key") + self.assertEqual(first["secret"], "test-secret") + self.assertEqual(json.loads(first["body"]), {"stack": "qbit-maid"}) + + def test_procedure_target_type(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + (200, {"status": "Complete", "success": True}), + ] + ) as mock: + r = run_script(base_env(mock.url, KOMODO_TARGET_TYPE="procedure")) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(mock.requests[0]["path"], "/execute/RunProcedure") + self.assertEqual(json.loads(mock.requests[0]["body"]), {"procedure": "qbit-maid"}) + + # --- success and failure --------------------------------------------- + + def test_success_path(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + (200, {"status": "Complete", "success": True}), + ] + ) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("completed successfully", r.stdout) + + def test_deploy_failure_surfaces_logs(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + ( + 200, + { + "status": "Complete", + "success": False, + "logs": [{"stage": "Deploy", "msg": "image pull failed"}], + }, + ), + ] + ) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 1) + self.assertIn("failed", r.stderr) + self.assertIn("image pull failed", r.stderr) + + def test_unauthorized(self): + with MockKomodo([(401, {"error": "bad credentials"})]) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 1) + self.assertIn("HTTP 401", r.stderr) + + def test_bad_request(self): + with MockKomodo([(400, {"error": "stack not found"})]) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 1) + self.assertIn("HTTP 400", r.stderr) + + def test_unreachable_host(self): + r = run_script(base_env("http://127.0.0.1:%d" % _free_port())) + self.assertEqual(r.returncode, 1) + self.assertIn("Could not reach Komodo", r.stderr) + + # --- polling ---------------------------------------------------------- + + def test_no_wait_returns_after_accept(self): + with MockKomodo([(200, {"_id": {"$oid": "u1"}})]) as mock: + r = run_script(base_env(mock.url, KOMODO_WAIT="false")) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("not waiting", r.stdout) + self.assertEqual(len(mock.requests), 1) + + def test_polls_until_complete(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + (200, {"status": "Queued"}), + (200, {"status": "InProgress"}), + (200, {"status": "Complete", "success": True}), + ] + ) as mock: + r = run_script(base_env(mock.url)) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("Queued", r.stdout) + self.assertIn("InProgress", r.stdout) + self.assertIn("completed successfully", r.stdout) + self.assertEqual(len(mock.requests), 4) + + def test_timeout(self): + with MockKomodo( + [ + (200, {"_id": {"$oid": "u1"}}), + (200, {"status": "InProgress"}), + ] + ) as mock: + r = run_script( + base_env(mock.url, KOMODO_WAIT_TIMEOUT="2", KOMODO_WAIT_INTERVAL="1") + ) + self.assertEqual(r.returncode, 1) + self.assertIn("Timed out", r.stderr) + + +if __name__ == "__main__": + unittest.main(verbosity=2)