add test infrastructure and --quiet mode
Some checks failed
CI / Lint & Check (push) Failing after 11s

- stdlib unittest suite: 38 tests for client, commands, output
- fixtures/ with load_fixture() helper and sample data
- -q/--quiet flag: tab-separated output, no headers
- ci: package syntax check, lint, test job
This commit is contained in:
Username
2026-01-18 21:08:20 +01:00
parent 969f0a5207
commit 9b1df32ffe
10 changed files with 598 additions and 8 deletions

View File

@@ -27,8 +27,22 @@ jobs:
- name: Install tools
run: pip install -q ruff
- name: Syntax check
- name: Syntax check (entrypoint)
run: python -m py_compile harbor-ctl.py
- name: Lint
- name: Syntax check (package)
run: |
for f in src/harbor/*.py; do
python -m py_compile "$f"
done
- name: Lint (entrypoint)
run: ruff check harbor-ctl.py
- name: Lint (package)
run: ruff check src/harbor/
- name: Run tests
run: python -m unittest discover -s src/harbor/tests -v
env:
PYTHONPATH: src

View File

@@ -66,6 +66,8 @@ Admin commands (require admin privileges):
parser.add_argument("--verify-ssl", action="store_true", help="Verify SSL certificates (default: skip)")
parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT,
help=f"API request timeout in seconds (default: {DEFAULT_TIMEOUT})")
parser.add_argument("-q", "--quiet", action="store_true",
help="Quiet mode: tab-separated output, no headers")
subparsers = parser.add_subparsers(dest="command", help="Commands")

View File

@@ -36,7 +36,14 @@ from .constants import (
HTTP_OK,
SCAN_POLL_INTERVAL,
)
from .output import confirm_action, format_size, format_timestamp, print_error
from .output import (
confirm_action,
format_size,
format_timestamp,
print_error,
print_table_header,
print_table_row,
)
def cmd_projects(args: Namespace, user: str, password: str, url: str) -> int:
@@ -46,12 +53,27 @@ def cmd_projects(args: Namespace, user: str, password: str, url: str) -> int:
print_error(data)
return 1
print(f"{'Project':<20} {'Public':<8} {'Repos':<6} {'Auto-Scan':<10} {'Auto-SBOM':<10}")
print("-" * 60)
quiet = getattr(args, "quiet", False)
columns = [
("Project", 20),
("Public", 8),
("Repos", 6),
("Auto-Scan", 10),
("Auto-SBOM", 10),
]
widths = [w for _, w in columns]
print_table_header(columns, quiet=quiet)
for p in data:
meta = p.get("metadata", {})
print(f"{p['name']:<20} {meta.get('public', 'false'):<8} {p.get('repo_count', 0):<6} "
f"{meta.get('auto_scan', 'false'):<10} {meta.get('auto_sbom_generation', 'false'):<10}")
values = [
p["name"],
meta.get("public", "false"),
p.get("repo_count", 0),
meta.get("auto_scan", "false"),
meta.get("auto_sbom_generation", "false"),
]
print_table_row(values, widths, quiet=quiet)
return 0

View File

@@ -1,7 +1,7 @@
"""Output formatting helpers."""
import sys
from typing import Any
from typing import Any, Sequence
def print_error(data: dict[str, Any] | str) -> None:
@@ -75,3 +75,42 @@ def format_timestamp(ts: str | None) -> str:
return "N/A"
# Strip timezone suffix and fractional seconds for display
return ts.replace("T", " ").split(".")[0].rstrip("Z")
def print_table_header(
columns: Sequence[tuple[str, int]],
quiet: bool = False,
) -> None:
"""Print table header with column names.
Args:
columns: Sequence of (name, width) tuples
quiet: If True, skip header entirely
"""
if quiet:
return
header = " ".join(f"{name:<{width}}" for name, width in columns)
print(header)
print("-" * len(header))
def print_table_row(
values: Sequence[str],
widths: Sequence[int],
quiet: bool = False,
) -> None:
"""Print a single table row.
Args:
values: Column values
widths: Column widths (ignored in quiet mode)
quiet: If True, use tab-separated raw values
"""
if quiet:
print("\t".join(str(v) for v in values))
else:
parts = []
for val, width in zip(values, widths):
parts.append(f"{str(val):<{width}}")
print(" ".join(parts))

View File

@@ -0,0 +1 @@
"""harbor-ctl test suite."""

20
src/harbor/tests/fixtures/__init__.py vendored Normal file
View File

@@ -0,0 +1,20 @@
"""Test fixtures loader."""
import json
from pathlib import Path
FIXTURES_DIR = Path(__file__).parent
def load_fixture(name: str) -> dict | list:
"""Load JSON fixture by name.
Args:
name: Fixture filename (without .json extension)
Returns:
Parsed JSON data
"""
path = FIXTURES_DIR / f"{name}.json"
with open(path) as f:
return json.load(f)

40
src/harbor/tests/fixtures/projects.json vendored Normal file
View File

@@ -0,0 +1,40 @@
[
{
"name": "library",
"project_id": 1,
"owner_id": 1,
"owner_name": "admin",
"repo_count": 5,
"metadata": {
"public": "true",
"auto_scan": "true",
"auto_sbom_generation": "false"
},
"creation_time": "2024-01-15T10:30:00.000Z",
"update_time": "2024-06-20T14:22:00.000Z"
},
{
"name": "private",
"project_id": 2,
"owner_id": 1,
"owner_name": "admin",
"repo_count": 3,
"metadata": {
"public": "false",
"auto_scan": "false",
"auto_sbom_generation": "true"
},
"creation_time": "2024-02-01T08:00:00.000Z",
"update_time": "2024-05-15T11:45:00.000Z"
},
{
"name": "staging",
"project_id": 3,
"owner_id": 2,
"owner_name": "developer",
"repo_count": 0,
"metadata": {},
"creation_time": "2024-03-10T16:00:00.000Z",
"update_time": "2024-03-10T16:00:00.000Z"
}
]

View File

@@ -0,0 +1,175 @@
"""Tests for client module."""
import unittest
from unittest.mock import patch
from harbor.client import build_url, resolve_digest
from harbor.constants import SHA256_PREFIX
class TestBuildUrl(unittest.TestCase):
"""Tests for build_url() function."""
def test_simple_path(self):
"""Single path component appends to /api/v2.0/."""
result = build_url("https://harbor.example.com", "projects")
self.assertEqual(result, "https://harbor.example.com/api/v2.0/projects")
def test_multiple_parts(self):
"""Multiple path components join with /."""
result = build_url(
"https://harbor.example.com",
"projects", "library", "repositories"
)
self.assertEqual(
result,
"https://harbor.example.com/api/v2.0/projects/library/repositories"
)
def test_with_params(self):
"""Query parameters append as ?key=value."""
result = build_url(
"https://harbor.example.com",
"projects",
page="1",
page_size="25"
)
self.assertIn("?", result)
self.assertIn("page=1", result)
self.assertIn("page_size=25", result)
def test_empty_params(self):
"""Empty params dict does not add ?."""
result = build_url("https://harbor.example.com", "projects")
self.assertNotIn("?", result)
def test_trailing_slash_base(self):
"""Base URL with trailing slash handled correctly."""
result = build_url("https://harbor.example.com/", "projects")
self.assertEqual(result, "https://harbor.example.com//api/v2.0/projects")
class TestResolveDigest(unittest.TestCase):
"""Tests for resolve_digest() function."""
def setUp(self):
"""Set up test fixtures."""
self.user = "admin"
self.password = "secret"
self.url = "https://harbor.example.com"
self.project = "library"
self.repo = "nginx"
# SHA256 digest: 7 (prefix) + 64 (hex) = 71 chars
self.full_digest = "sha256:abc123def456789012345678901234567890123456789012345678901234abcd"
@patch("harbor.client.api_request")
def test_none_returns_latest(self, mock_api):
"""None input returns latest artifact digest."""
mock_api.return_value = [{"digest": self.full_digest}]
result = resolve_digest(
self.project, self.repo, None,
self.user, self.password, self.url
)
self.assertEqual(result, self.full_digest)
mock_api.assert_called_once()
self.assertIn("page_size=1", mock_api.call_args[0][0])
@patch("harbor.client.api_request")
def test_full_digest_passthrough(self, mock_api):
"""Full 71-char digest passes through without API call."""
result = resolve_digest(
self.project, self.repo, self.full_digest,
self.user, self.password, self.url
)
self.assertEqual(result, self.full_digest)
mock_api.assert_not_called()
@patch("harbor.client.api_request")
def test_tag_lookup(self, mock_api):
"""Tag name resolves via artifact endpoint."""
mock_api.return_value = {"digest": self.full_digest}
result = resolve_digest(
self.project, self.repo, "latest",
self.user, self.password, self.url
)
self.assertEqual(result, self.full_digest)
self.assertIn("/artifacts/latest", mock_api.call_args[0][0])
@patch("harbor.client.api_request")
def test_partial_digest_match(self, mock_api):
"""Partial digest matches in artifact list."""
mock_api.side_effect = [
{"error": 404, "message": "not found"}, # tag lookup fails
[{"digest": self.full_digest}], # list returns match
]
result = resolve_digest(
self.project, self.repo, "sha256:abc123",
self.user, self.password, self.url
)
self.assertEqual(result, self.full_digest)
self.assertEqual(mock_api.call_count, 2)
@patch("harbor.client.api_request")
def test_partial_without_prefix(self, mock_api):
"""Partial digest without sha256: prefix still matches."""
mock_api.side_effect = [
{"error": 404, "message": "not found"},
[{"digest": self.full_digest}],
]
result = resolve_digest(
self.project, self.repo, "abc123",
self.user, self.password, self.url
)
self.assertEqual(result, self.full_digest)
@patch("harbor.client.api_request")
def test_not_found_returns_original(self, mock_api):
"""Unresolved digest/tag returns original value."""
mock_api.side_effect = [
{"error": 404, "message": "not found"},
[], # empty list
]
result = resolve_digest(
self.project, self.repo, "nonexistent",
self.user, self.password, self.url
)
self.assertEqual(result, "nonexistent")
@patch("harbor.client.api_request")
def test_api_error_on_latest(self, mock_api):
"""API error when fetching latest returns None."""
mock_api.return_value = {"error": 500, "message": "server error"}
result = resolve_digest(
self.project, self.repo, None,
self.user, self.password, self.url
)
self.assertIsNone(result)
@patch("harbor.client.api_request")
def test_empty_latest_returns_none(self, mock_api):
"""Empty artifact list for latest returns None."""
mock_api.return_value = []
result = resolve_digest(
self.project, self.repo, None,
self.user, self.password, self.url
)
self.assertIsNone(result)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,180 @@
"""Tests for commands module."""
import io
import sys
import unittest
from argparse import Namespace
from unittest.mock import patch
from harbor.commands import cmd_projects
from harbor.tests.fixtures import load_fixture
class TestCmdProjects(unittest.TestCase):
"""Tests for cmd_projects() function."""
def setUp(self):
"""Set up test fixtures."""
self.user = "admin"
self.password = "secret"
self.url = "https://harbor.example.com"
self.args = Namespace()
self.projects = load_fixture("projects")
def _capture_output(self, func, *args, **kwargs):
"""Capture stdout from function execution."""
captured = io.StringIO()
old_stdout = sys.stdout
sys.stdout = captured
try:
result = func(*args, **kwargs)
finally:
sys.stdout = old_stdout
return result, captured.getvalue()
@patch("harbor.commands.api_request")
def test_success_returns_zero(self, mock_api):
"""Successful listing returns exit code 0."""
mock_api.return_value = self.projects
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 0)
@patch("harbor.commands.api_request")
def test_prints_header(self, mock_api):
"""Output includes column headers."""
mock_api.return_value = self.projects
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertIn("Project", output)
self.assertIn("Public", output)
self.assertIn("Repos", output)
@patch("harbor.commands.api_request")
def test_prints_project_names(self, mock_api):
"""Output includes project names from response."""
mock_api.return_value = self.projects
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertIn("library", output)
self.assertIn("private", output)
self.assertIn("staging", output)
@patch("harbor.commands.api_request")
def test_prints_metadata_values(self, mock_api):
"""Output includes metadata values (public, auto_scan, etc.)."""
mock_api.return_value = self.projects
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
# library has public=true, auto_scan=true
self.assertIn("true", output)
# private has public=false
self.assertIn("false", output)
@patch("harbor.commands.api_request")
def test_api_error_returns_one(self, mock_api):
"""API error returns exit code 1."""
mock_api.return_value = {"error": 401, "message": "unauthorized"}
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 1)
self.assertIn("Error", output)
@patch("harbor.commands.api_request")
def test_empty_metadata_handled(self, mock_api):
"""Projects with empty metadata dict don't crash."""
mock_api.return_value = [
{
"name": "minimal",
"repo_count": 1,
"metadata": {}
}
]
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 0)
self.assertIn("minimal", output)
@patch("harbor.commands.api_request")
def test_missing_metadata_handled(self, mock_api):
"""Projects without metadata key don't crash."""
mock_api.return_value = [
{
"name": "no-meta",
"repo_count": 0
}
]
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 0)
self.assertIn("no-meta", output)
@patch("harbor.commands.api_request")
def test_empty_list_returns_zero(self, mock_api):
"""Empty project list returns exit code 0."""
mock_api.return_value = []
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 0)
# Header still printed
self.assertIn("Project", output)
@patch("harbor.commands.api_request")
def test_quiet_mode_no_header(self, mock_api):
"""Quiet mode suppresses header."""
mock_api.return_value = self.projects
self.args.quiet = True
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
self.assertEqual(result, 0)
# Header should not be present
self.assertNotIn("Project", output)
self.assertNotIn("---", output)
# Data should be present
self.assertIn("library", output)
@patch("harbor.commands.api_request")
def test_quiet_mode_tab_separated(self, mock_api):
"""Quiet mode outputs tab-separated values."""
mock_api.return_value = self.projects
self.args.quiet = True
result, output = self._capture_output(
cmd_projects, self.args, self.user, self.password, self.url
)
lines = output.strip().split("\n")
# Each line should have tabs
for line in lines:
self.assertIn("\t", line)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,97 @@
"""Tests for output module."""
import unittest
from harbor.output import format_size, format_timestamp
class TestFormatSize(unittest.TestCase):
"""Tests for format_size() function."""
def test_bytes(self):
"""Small values display as bytes."""
self.assertEqual(format_size(0), "0.0 B")
self.assertEqual(format_size(500), "500.0 B")
self.assertEqual(format_size(1023), "1023.0 B")
def test_kilobytes(self):
"""Values 1024+ display as KB."""
self.assertEqual(format_size(1024), "1.0 KB")
self.assertEqual(format_size(1536), "1.5 KB")
self.assertEqual(format_size(10240), "10.0 KB")
def test_megabytes(self):
"""Values 1MB+ display as MB."""
self.assertEqual(format_size(1024 * 1024), "1.0 MB")
self.assertEqual(format_size(1024 * 1024 * 50), "50.0 MB")
self.assertEqual(format_size(1024 * 1024 * 512), "512.0 MB")
def test_gigabytes(self):
"""Values 1GB+ display as GB."""
self.assertEqual(format_size(1024 ** 3), "1.0 GB")
self.assertEqual(format_size(1024 ** 3 * 2), "2.0 GB")
self.assertEqual(format_size(1024 ** 3 * 10), "10.0 GB")
def test_terabytes(self):
"""Values 1TB+ display as TB."""
self.assertEqual(format_size(1024 ** 4), "1.0 TB")
self.assertEqual(format_size(1024 ** 4 * 5), "5.0 TB")
def test_petabytes(self):
"""Values 1PB+ display as PB."""
self.assertEqual(format_size(1024 ** 5), "1.0 PB")
def test_fractional_display(self):
"""Fractional values display with one decimal."""
self.assertIn(".", format_size(1500))
self.assertIn(".", format_size(1024 * 1024 * 1.5))
class TestFormatTimestamp(unittest.TestCase):
"""Tests for format_timestamp() function."""
def test_none_returns_na(self):
"""None input returns N/A."""
self.assertEqual(format_timestamp(None), "N/A")
def test_empty_string_returns_na(self):
"""Empty string returns N/A."""
self.assertEqual(format_timestamp(""), "N/A")
def test_strips_t_separator(self):
"""T between date and time replaced with space."""
result = format_timestamp("2024-06-15T10:30:00Z")
self.assertNotIn("T", result)
self.assertIn(" ", result)
def test_truncates_fractional_seconds(self):
"""Fractional seconds removed."""
result = format_timestamp("2024-06-15T10:30:00.123456Z")
self.assertNotIn(".", result)
self.assertNotIn("123456", result)
def test_strips_z_suffix(self):
"""Trailing Z removed."""
result = format_timestamp("2024-06-15T10:30:00Z")
self.assertNotIn("Z", result)
def test_basic_format(self):
"""Basic ISO timestamp converts correctly."""
result = format_timestamp("2024-06-15T10:30:00Z")
self.assertEqual(result, "2024-06-15 10:30:00")
def test_with_timezone_offset(self):
"""Timestamp with timezone offset handled."""
result = format_timestamp("2024-06-15T10:30:00.000+00:00")
# Should at least have date and time
self.assertIn("2024-06-15", result)
self.assertIn("10:30:00", result)
def test_full_harbor_format(self):
"""Full Harbor API timestamp format."""
result = format_timestamp("2024-01-15T10:30:00.000Z")
self.assertEqual(result, "2024-01-15 10:30:00")
if __name__ == "__main__":
unittest.main()