diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 918e33d..b55fbf8 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -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 diff --git a/src/harbor/cli.py b/src/harbor/cli.py index f8bafff..e7e8b3e 100644 --- a/src/harbor/cli.py +++ b/src/harbor/cli.py @@ -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") diff --git a/src/harbor/commands.py b/src/harbor/commands.py index a3d1b97..5e4f378 100644 --- a/src/harbor/commands.py +++ b/src/harbor/commands.py @@ -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 diff --git a/src/harbor/output.py b/src/harbor/output.py index e9694cd..46406a9 100644 --- a/src/harbor/output.py +++ b/src/harbor/output.py @@ -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)) diff --git a/src/harbor/tests/__init__.py b/src/harbor/tests/__init__.py new file mode 100644 index 0000000..80c94af --- /dev/null +++ b/src/harbor/tests/__init__.py @@ -0,0 +1 @@ +"""harbor-ctl test suite.""" diff --git a/src/harbor/tests/fixtures/__init__.py b/src/harbor/tests/fixtures/__init__.py new file mode 100644 index 0000000..68283a1 --- /dev/null +++ b/src/harbor/tests/fixtures/__init__.py @@ -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) diff --git a/src/harbor/tests/fixtures/projects.json b/src/harbor/tests/fixtures/projects.json new file mode 100644 index 0000000..f60e190 --- /dev/null +++ b/src/harbor/tests/fixtures/projects.json @@ -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" + } +] diff --git a/src/harbor/tests/test_client.py b/src/harbor/tests/test_client.py new file mode 100644 index 0000000..977c24f --- /dev/null +++ b/src/harbor/tests/test_client.py @@ -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() diff --git a/src/harbor/tests/test_commands.py b/src/harbor/tests/test_commands.py new file mode 100644 index 0000000..f4d5011 --- /dev/null +++ b/src/harbor/tests/test_commands.py @@ -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() diff --git a/src/harbor/tests/test_output.py b/src/harbor/tests/test_output.py new file mode 100644 index 0000000..dcb31bf --- /dev/null +++ b/src/harbor/tests/test_output.py @@ -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()