kevinveenbirkenbach/pkgmgr
1
0
Fork 0
Code Issues Pull Requests Actions 17 Packages Projects Releases Wiki Activity

feat(tests): add unit tests for mirror context, io, commands, and remote helpers
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Details
Mark stable commit / test-integration (push) Has been cancelled
Details
Mark stable commit / test-env-virtual (push) Has been cancelled
Details
Mark stable commit / test-env-nix (push) Has been cancelled
Details
Mark stable commit / test-e2e (push) Has been cancelled
Details
Mark stable commit / test-virgin-user (push) Has been cancelled
Details
Mark stable commit / test-virgin-root (push) Has been cancelled
Details
Mark stable commit / codesniffer-shellcheck (push) Has been cancelled
Details
Mark stable commit / codesniffer-ruff (push) Has been cancelled
Details
Mark stable commit / mark-stable (push) Has been cancelled
Details

Browse Source 
https://chatgpt.com/share/693ed188-eb80-800f-8541-356e3fbd98c5
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-14 16:02:11 +01:00
parent 27c0c7c01f
commit e61b30d9af
10 changed files with 643 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/unit/pkgmgr/actions/mirror/__init__.py
View File

@@ -0,0 +1 @@
# Unit test package for pkgmgr.actions.mirror

51
tests/unit/pkgmgr/actions/mirror/test_context.py Normal file
View File

@@ -0,0 +1,51 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from unittest.mock import patch
from pkgmgr.actions.mirror.context import build_context
class TestMirrorContext(unittest.TestCase):
"""
Unit tests for building RepoMirrorContext from repo + filesystem.
"""
@patch("pkgmgr.actions.mirror.context.read_mirrors_file")
@patch("pkgmgr.actions.mirror.context.load_config_mirrors")
@patch("pkgmgr.actions.mirror.context.get_repo_dir")
@patch("pkgmgr.actions.mirror.context.get_repo_identifier")
def test_build_context_bundles_config_and_file_mirrors(
self,
mock_identifier,
mock_repo_dir,
mock_load_config,
mock_read_file,
) -> None:
mock_identifier.return_value = "id"
mock_repo_dir.return_value = "/tmp/repo"
mock_load_config.return_value = {"origin": "git@github.com:alice/repo.git"}
mock_read_file.return_value = {"backup": "ssh://git@backup/alice/repo.git"}
repo = {"provider": "github.com", "account": "alice", "repository": "repo"}
ctx = build_context(repo, repositories_base_dir="/base", all_repos=[repo])
self.assertEqual(ctx.identifier, "id")
self.assertEqual(ctx.repo_dir, "/tmp/repo")
self.assertEqual(ctx.config_mirrors, {"origin": "git@github.com:alice/repo.git"})
self.assertEqual(ctx.file_mirrors, {"backup": "ssh://git@backup/alice/repo.git"})
self.assertEqual(
ctx.resolved_mirrors,
{
"origin": "git@github.com:alice/repo.git",
"backup": "ssh://git@backup/alice/repo.git",
},
)
if __name__ == "__main__":
unittest.main()

77
tests/unit/pkgmgr/actions/mirror/test_diff_cmd.py Normal file
View File

@@ -0,0 +1,77 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import io
import unittest
from contextlib import redirect_stdout
from unittest.mock import MagicMock, PropertyMock, patch
from pkgmgr.actions.mirror.diff_cmd import diff_mirrors
class TestDiffCmd(unittest.TestCase):
"""
Unit tests for mirror diff output.
"""
@patch("pkgmgr.actions.mirror.diff_cmd.build_context")
def test_diff_mirrors_reports_only_in_config_and_only_in_file(self, mock_build_context) -> None:
ctx = MagicMock()
ctx.identifier = "id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "a", "cfgonly": "b"}
ctx.file_mirrors = {"origin": "a", "fileonly": "c"}
type(ctx).resolved_mirrors = PropertyMock(
return_value={"origin": "a", "cfgonly": "b", "fileonly": "c"}
)
mock_build_context.return_value = ctx
buf = io.StringIO()
with redirect_stdout(buf):
diff_mirrors(selected_repos=[{}], repositories_base_dir="/base", all_repos=[])
out = buf.getvalue()
self.assertIn("[ONLY IN CONFIG] cfgonly: b", out)
self.assertIn("[ONLY IN FILE] fileonly: c", out)
@patch("pkgmgr.actions.mirror.diff_cmd.build_context")
def test_diff_mirrors_reports_url_mismatch(self, mock_build_context) -> None:
ctx = MagicMock()
ctx.identifier = "id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "a"}
ctx.file_mirrors = {"origin": "different"}
type(ctx).resolved_mirrors = PropertyMock(return_value={"origin": "different"})
mock_build_context.return_value = ctx
buf = io.StringIO()
with redirect_stdout(buf):
diff_mirrors(selected_repos=[{}], repositories_base_dir="/base", all_repos=[])
out = buf.getvalue()
self.assertIn("[URL MISMATCH]", out)
self.assertIn("config: a", out)
self.assertIn("file: different", out)
@patch("pkgmgr.actions.mirror.diff_cmd.build_context")
def test_diff_mirrors_reports_in_sync(self, mock_build_context) -> None:
ctx = MagicMock()
ctx.identifier = "id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "a"}
ctx.file_mirrors = {"origin": "a"}
type(ctx).resolved_mirrors = PropertyMock(return_value={"origin": "a"})
mock_build_context.return_value = ctx
buf = io.StringIO()
with redirect_stdout(buf):
diff_mirrors(selected_repos=[{}], repositories_base_dir="/base", all_repos=[])
out = buf.getvalue()
self.assertIn("[OK] Mirrors in config and MIRRORS file are in sync.", out)
if __name__ == "__main__":
unittest.main()

65
tests/unit/pkgmgr/actions/mirror/test_git_remote.py
View File

@@ -4,10 +4,13 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from pkgmgr.actions.mirror.git_remote import (
build_default_ssh_url,
determine_primary_remote_url,
current_origin_url,
has_origin_remote,
)
from pkgmgr.actions.mirror.types import MirrorMap, Repository
@@ -25,10 +28,7 @@ class TestMirrorGitRemote(unittest.TestCase):
}
url = build_default_ssh_url(repo)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
self.assertEqual(url, "git@github.com:kevinveenbirkenbach/package-manager.git")
def test_build_default_ssh_url_with_port(self) -> None:
repo: Repository = {
@@ -39,24 +39,18 @@ class TestMirrorGitRemote(unittest.TestCase):
}
url = build_default_ssh_url(repo)
self.assertEqual(
url,
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
)
self.assertEqual(url, "ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git")
def test_build_default_ssh_url_missing_fields_returns_none(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
# "repository" fehlt absichtlich
}
url = build_default_ssh_url(repo)
self.assertIsNone(url)
def test_determine_primary_remote_url_prefers_origin_in_resolved_mirrors(
self,
) -> None:
def test_determine_primary_remote_url_prefers_origin_in_resolved_mirrors(self) -> None:
repo: Repository = {
"provider": "github.com",
"account": "kevinveenbirkenbach",
@@ -68,10 +62,7 @@ class TestMirrorGitRemote(unittest.TestCase):
}
url = determine_primary_remote_url(repo, mirrors)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
self.assertEqual(url, "git@github.com:kevinveenbirkenbach/package-manager.git")
def test_determine_primary_remote_url_uses_any_mirror_if_no_origin(self) -> None:
repo: Repository = {
@@ -85,11 +76,7 @@ class TestMirrorGitRemote(unittest.TestCase):
}
url = determine_primary_remote_url(repo, mirrors)
# Alphabetisch sortiert: backup, mirror2 → backup gewinnt
self.assertEqual(
url,
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
)
self.assertEqual(url, "ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git")
def test_determine_primary_remote_url_falls_back_to_default_ssh(self) -> None:
repo: Repository = {
@@ -100,10 +87,38 @@ class TestMirrorGitRemote(unittest.TestCase):
mirrors: MirrorMap = {}
url = determine_primary_remote_url(repo, mirrors)
self.assertEqual(
url,
"git@github.com:kevinveenbirkenbach/package-manager.git",
)
self.assertEqual(url, "git@github.com:kevinveenbirkenbach/package-manager.git")
@patch("pkgmgr.actions.mirror.git_remote.run_git")
def test_current_origin_url_returns_value(self, mock_run_git) -> None:
mock_run_git.return_value = "git@github.com:alice/repo.git\n"
self.assertEqual(current_origin_url("/tmp/repo"), "git@github.com:alice/repo.git")
mock_run_git.assert_called_once_with(["remote", "get-url", "origin"], cwd="/tmp/repo")
@patch("pkgmgr.actions.mirror.git_remote.run_git")
def test_current_origin_url_returns_none_on_git_error(self, mock_run_git) -> None:
from pkgmgr.core.git import GitError
mock_run_git.side_effect = GitError("fail")
self.assertIsNone(current_origin_url("/tmp/repo"))
@patch("pkgmgr.actions.mirror.git_remote.run_git")
def test_has_origin_remote_true(self, mock_run_git) -> None:
mock_run_git.return_value = "origin\nupstream\n"
self.assertTrue(has_origin_remote("/tmp/repo"))
mock_run_git.assert_called_once_with(["remote"], cwd="/tmp/repo")
@patch("pkgmgr.actions.mirror.git_remote.run_git")
def test_has_origin_remote_false_on_missing_remote(self, mock_run_git) -> None:
mock_run_git.return_value = "upstream\n"
self.assertFalse(has_origin_remote("/tmp/repo"))
@patch("pkgmgr.actions.mirror.git_remote.run_git")
def test_has_origin_remote_false_on_git_error(self, mock_run_git) -> None:
from pkgmgr.core.git import GitError
mock_run_git.side_effect = GitError("fail")
self.assertFalse(has_origin_remote("/tmp/repo"))
if __name__ == "__main__":

150
tests/unit/pkgmgr/actions/mirror/test_io.py
View File

@@ -7,10 +7,7 @@ import os
import tempfile
import unittest
from pkgmgr.actions.mirror.io import (
load_config_mirrors,
read_mirrors_file,
)
from pkgmgr.actions.mirror.io import load_config_mirrors, read_mirrors_file, write_mirrors_file
class TestMirrorIO(unittest.TestCase):
@@ -18,117 +15,96 @@ class TestMirrorIO(unittest.TestCase):
Unit tests for pkgmgr.actions.mirror.io helpers.
"""
# ------------------------------------------------------------------
# load_config_mirrors
# ------------------------------------------------------------------
def test_load_config_mirrors_from_dict(self) -> None:
def test_load_config_mirrors_from_dict_filters_empty(self) -> None:
repo = {
"mirrors": {
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
"empty": "",
"none": None,
"backup": "",
"invalid": None,
}
}
mirrors = load_config_mirrors(repo)
self.assertEqual(mirrors, {"origin": "ssh://git@example.com/account/repo.git"})
self.assertEqual(
mirrors,
{
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
},
)
def test_load_config_mirrors_from_list(self) -> None:
def test_load_config_mirrors_from_list_filters_invalid_entries(self) -> None:
repo = {
"mirrors": [
{"name": "origin", "url": "ssh://git@example.com/account/repo.git"},
{"name": "backup", "url": "ssh://git@backup/account/repo.git"},
{"name": "", "url": "ssh://git@invalid/ignored.git"},
{"name": "missing-url"},
"not-a-dict",
{"name": "backup", "url": ""},
{"name": "", "url": "ssh://git@example.com/empty-name.git"},
{"url": "ssh://git@example.com/missing-name.git"},
]
}
mirrors = load_config_mirrors(repo)
self.assertEqual(
mirrors,
{
"origin": "ssh://git@example.com/account/repo.git",
"backup": "ssh://git@backup/account/repo.git",
},
)
self.assertEqual(mirrors, {"origin": "ssh://git@example.com/account/repo.git"})
def test_load_config_mirrors_empty_when_missing(self) -> None:
repo = {}
mirrors = load_config_mirrors(repo)
self.assertEqual(mirrors, {})
self.assertEqual(load_config_mirrors({}), {})
# ------------------------------------------------------------------
# read_mirrors_file
# ------------------------------------------------------------------
def test_read_mirrors_file_with_named_and_url_only_entries(self) -> None:
"""
Ensure that the MIRRORS file format is parsed correctly:
- 'name url' → exact name
- 'url' → auto name derived from netloc (host[:port]),
with numeric suffix if duplicated.
"""
def test_read_mirrors_file_parses_named_entries(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
mirrors_path = os.path.join(tmpdir, "MIRRORS")
content = "\n".join(
[
"# comment",
"",
"origin ssh://git@example.com/account/repo.git",
"https://github.com/kevinveenbirkenbach/package-manager",
"https://github.com/kevinveenbirkenbach/another-repo",
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
]
)
with open(mirrors_path, "w", encoding="utf-8") as fh:
fh.write(content + "\n")
p = os.path.join(tmpdir, "MIRRORS")
with open(p, "w", encoding="utf-8") as fh:
fh.write("origin ssh://git@example.com/account/repo.git\n")
mirrors = read_mirrors_file(tmpdir)
# 'origin' is preserved as given
self.assertIn("origin", mirrors)
self.assertEqual(
mirrors["origin"],
"ssh://git@example.com/account/repo.git",
)
self.assertEqual(mirrors, {"origin": "ssh://git@example.com/account/repo.git"})
# Two GitHub URLs → auto names: github.com, github.com2
github_urls = {
mirrors.get("github.com"),
mirrors.get("github.com2"),
}
self.assertIn(
"https://github.com/kevinveenbirkenbach/package-manager",
github_urls,
)
self.assertIn(
"https://github.com/kevinveenbirkenbach/another-repo",
github_urls,
)
def test_read_mirrors_file_url_only_uses_netloc_basename_and_suffix(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
p = os.path.join(tmpdir, "MIRRORS")
with open(p, "w", encoding="utf-8") as fh:
fh.write(
"\n".join(
[
"https://github.com/alice/repo1",
"https://github.com/alice/repo2",
"ssh://git@git.veen.world:2201/alice/repo3.git",
]
)
+ "\n"
)
mirrors = read_mirrors_file(tmpdir)
self.assertIn("github.com", mirrors)
self.assertIn("github.com2", mirrors)
self.assertEqual(mirrors["github.com"], "https://github.com/alice/repo1")
self.assertEqual(mirrors["github.com2"], "https://github.com/alice/repo2")
# SSH-URL mit User-Teil → netloc ist "git@git.veen.world:2201"
# → host = "git@git.veen.world"
self.assertIn("git@git.veen.world", mirrors)
self.assertEqual(
mirrors["git@git.veen.world"],
"ssh://git@git.veen.world:2201/kevinveenbirkenbach/pkgmgr.git",
)
self.assertEqual(mirrors["git@git.veen.world"], "ssh://git@git.veen.world:2201/alice/repo3.git")
def test_read_mirrors_file_missing_returns_empty(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
mirrors = read_mirrors_file(tmpdir) # no MIRRORS file
self.assertEqual(mirrors, {})
self.assertEqual(read_mirrors_file(tmpdir), {})
def test_write_mirrors_file_writes_sorted_lines(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
mirrors = {
"b": "ssh://b.example/repo.git",
"a": "ssh://a.example/repo.git",
}
write_mirrors_file(tmpdir, mirrors, preview=False)
p = os.path.join(tmpdir, "MIRRORS")
self.assertTrue(os.path.exists(p))
with open(p, "r", encoding="utf-8") as fh:
content = fh.read()
self.assertEqual(content, "a ssh://a.example/repo.git\nb ssh://b.example/repo.git\n")
def test_write_mirrors_file_preview_does_not_create_file(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
mirrors = {"a": "ssh://a.example/repo.git"}
write_mirrors_file(tmpdir, mirrors, preview=True)
p = os.path.join(tmpdir, "MIRRORS")
self.assertFalse(os.path.exists(p))
if __name__ == "__main__":

72
tests/unit/pkgmgr/actions/mirror/test_list_cmd.py Normal file
View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import io
import unittest
from contextlib import redirect_stdout
from unittest.mock import MagicMock, PropertyMock, patch
from pkgmgr.actions.mirror.list_cmd import list_mirrors
class TestListCmd(unittest.TestCase):
"""
Unit tests for mirror list output.
"""
@patch("pkgmgr.actions.mirror.list_cmd.build_context")
def test_list_mirrors_all_sources_prints_sections(self, mock_build_context) -> None:
ctx = MagicMock()
ctx.identifier = "id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "a"}
ctx.file_mirrors = {"backup": "b"}
type(ctx).resolved_mirrors = PropertyMock(return_value={"origin": "a", "backup": "b"})
mock_build_context.return_value = ctx
buf = io.StringIO()
with redirect_stdout(buf):
list_mirrors(
selected_repos=[{}],
repositories_base_dir="/base",
all_repos=[],
source="all",
)
out = buf.getvalue()
self.assertIn("[config mirrors]", out)
self.assertIn("[MIRRORS file]", out)
self.assertIn("[resolved mirrors]", out)
self.assertIn("origin: a", out)
self.assertIn("backup: b", out)
@patch("pkgmgr.actions.mirror.list_cmd.build_context")
def test_list_mirrors_config_only(self, mock_build_context) -> None:
ctx = MagicMock()
ctx.identifier = "id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "a"}
ctx.file_mirrors = {"backup": "b"}
type(ctx).resolved_mirrors = PropertyMock(return_value={"origin": "a", "backup": "b"})
mock_build_context.return_value = ctx
buf = io.StringIO()
with redirect_stdout(buf):
list_mirrors(
selected_repos=[{}],
repositories_base_dir="/base",
all_repos=[],
source="config",
)
out = buf.getvalue()
self.assertIn("[config mirrors]", out)
self.assertIn("origin: a", out)
self.assertNotIn("[MIRRORS file]", out)
self.assertNotIn("[resolved mirrors]", out)
if __name__ == "__main__":
unittest.main()

52
tests/unit/pkgmgr/actions/mirror/test_remote_check.py Normal file
View File

@@ -0,0 +1,52 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from unittest.mock import patch
from pkgmgr.actions.mirror.remote_check import probe_mirror
from pkgmgr.core.git import GitError
class TestRemoteCheck(unittest.TestCase):
"""
Unit tests for non-destructive remote probing (git ls-remote).
"""
@patch("pkgmgr.actions.mirror.remote_check.run_git")
def test_probe_mirror_success_returns_true_and_empty_message(self, mock_run_git) -> None:
mock_run_git.return_value = "dummy-output"
ok, message = probe_mirror(
"ssh://git@code.example.org:2201/alice/repo.git",
"/tmp/some-repo",
)
self.assertTrue(ok)
self.assertEqual(message, "")
mock_run_git.assert_called_once_with(
["ls-remote", "ssh://git@code.example.org:2201/alice/repo.git"],
cwd="/tmp/some-repo",
)
@patch("pkgmgr.actions.mirror.remote_check.run_git")
def test_probe_mirror_failure_returns_false_and_error_message(self, mock_run_git) -> None:
mock_run_git.side_effect = GitError("Git command failed (simulated)")
ok, message = probe_mirror(
"ssh://git@code.example.org:2201/alice/repo.git",
"/tmp/some-repo",
)
self.assertFalse(ok)
self.assertIn("Git command failed", message)
mock_run_git.assert_called_once_with(
["ls-remote", "ssh://git@code.example.org:2201/alice/repo.git"],
cwd="/tmp/some-repo",
)
if __name__ == "__main__":
unittest.main()

114
tests/unit/pkgmgr/actions/mirror/test_remote_provision.py Normal file
View File

@@ -0,0 +1,114 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from unittest.mock import MagicMock, PropertyMock, patch
from pkgmgr.actions.mirror.remote_provision import ensure_remote_repository
class TestRemoteProvision(unittest.TestCase):
"""
Unit tests for remote provisioning wrapper logic (action layer).
"""
@patch("pkgmgr.actions.mirror.remote_provision.ensure_remote_repo")
@patch("pkgmgr.actions.mirror.remote_provision.determine_primary_remote_url")
@patch("pkgmgr.actions.mirror.remote_provision.build_context")
def test_ensure_remote_repository_builds_spec_from_url_and_calls_core(
self,
mock_build_context,
mock_determine_primary,
mock_ensure_remote_repo,
) -> None:
ctx = MagicMock()
type(ctx).resolved_mirrors = PropertyMock(
return_value={"origin": "ssh://git@git.veen.world:2201/alice/repo.git"}
)
ctx.identifier = "repo-id"
mock_build_context.return_value = ctx
mock_determine_primary.return_value = "ssh://git@git.veen.world:2201/alice/repo.git"
result = MagicMock()
result.status = "created"
result.message = "Repository created (user)."
result.url = "https://git.veen.world/alice/repo"
mock_ensure_remote_repo.return_value = result
repo = {
"provider": "gitea",
"account": "SHOULD_NOT_BE_USED_ANYMORE",
"repository": "SHOULD_NOT_BE_USED_ANYMORE",
"private": True,
"description": "desc",
}
ensure_remote_repository(
repo=repo,
repositories_base_dir="/base",
all_repos=[],
preview=False,
)
self.assertTrue(mock_ensure_remote_repo.called)
called_spec = mock_ensure_remote_repo.call_args[0][0]
self.assertEqual(called_spec.host, "git.veen.world")
self.assertEqual(called_spec.owner, "alice")
self.assertEqual(called_spec.name, "repo")
@patch("pkgmgr.actions.mirror.remote_provision.ensure_remote_repo")
@patch("pkgmgr.actions.mirror.remote_provision.determine_primary_remote_url")
@patch("pkgmgr.actions.mirror.remote_provision.build_context")
def test_ensure_remote_repository_skips_when_no_primary_url(
self,
mock_build_context,
mock_determine_primary,
mock_ensure_remote_repo,
) -> None:
ctx = MagicMock()
type(ctx).resolved_mirrors = PropertyMock(return_value={})
ctx.identifier = "repo-id"
mock_build_context.return_value = ctx
mock_determine_primary.return_value = None
ensure_remote_repository(
repo={"provider": "gitea"},
repositories_base_dir="/base",
all_repos=[],
preview=False,
)
mock_ensure_remote_repo.assert_not_called()
@patch("pkgmgr.actions.mirror.remote_provision.ensure_remote_repo")
@patch("pkgmgr.actions.mirror.remote_provision.determine_primary_remote_url")
@patch("pkgmgr.actions.mirror.remote_provision.build_context")
def test_ensure_remote_repository_skips_when_url_not_parseable(
self,
mock_build_context,
mock_determine_primary,
mock_ensure_remote_repo,
) -> None:
ctx = MagicMock()
type(ctx).resolved_mirrors = PropertyMock(
return_value={"origin": "ssh://git@host:2201/not-enough-parts"}
)
ctx.identifier = "repo-id"
mock_build_context.return_value = ctx
mock_determine_primary.return_value = "ssh://git@host:2201/not-enough-parts"
ensure_remote_repository(
repo={"provider": "gitea"},
repositories_base_dir="/base",
all_repos=[],
preview=False,
)
mock_ensure_remote_repo.assert_not_called()
if __name__ == "__main__":
unittest.main()

127
tests/unit/pkgmgr/actions/mirror/test_setup_cmd.py
View File

@@ -4,55 +4,120 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from unittest.mock import MagicMock, PropertyMock, patch
from pkgmgr.actions.mirror.setup_cmd import _probe_mirror
from pkgmgr.core.git import GitError
from pkgmgr.actions.mirror.setup_cmd import setup_mirrors
class TestMirrorSetupCmd(unittest.TestCase):
"""
Unit tests for the non-destructive remote probing logic in setup_cmd.
Unit tests for mirror setup orchestration (local + remote).
"""
@patch("pkgmgr.actions.mirror.setup_cmd.run_git")
def test_probe_mirror_success_returns_true_and_empty_message(
@patch("pkgmgr.actions.mirror.setup_cmd.ensure_origin_remote")
@patch("pkgmgr.actions.mirror.setup_cmd.build_context")
def test_setup_mirrors_local_calls_ensure_origin_remote(
self,
mock_run_git,
mock_build_context,
mock_ensure_origin,
) -> None:
"""
If run_git returns successfully, _probe_mirror must report (True, "").
"""
mock_run_git.return_value = "dummy-output"
ctx = MagicMock()
ctx.identifier = "repo-id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {}
ctx.file_mirrors = {}
type(ctx).resolved_mirrors = PropertyMock(return_value={})
mock_build_context.return_value = ctx
ok, message = _probe_mirror(
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
"/tmp/some-repo",
repo = {"provider": "github.com", "account": "alice", "repository": "repo"}
setup_mirrors(
selected_repos=[repo],
repositories_base_dir="/base",
all_repos=[repo],
preview=True,
local=True,
remote=False,
ensure_remote=False,
)
self.assertTrue(ok)
self.assertEqual(message, "")
mock_run_git.assert_called_once()
mock_ensure_origin.assert_called_once()
args, kwargs = mock_ensure_origin.call_args
self.assertEqual(args[0], repo)
self.assertEqual(kwargs.get("preview"), True)
@patch("pkgmgr.actions.mirror.setup_cmd.run_git")
def test_probe_mirror_failure_returns_false_and_error_message(
@patch("pkgmgr.actions.mirror.setup_cmd.ensure_remote_repository")
@patch("pkgmgr.actions.mirror.setup_cmd.probe_mirror")
@patch("pkgmgr.actions.mirror.setup_cmd.build_context")
def test_setup_mirrors_remote_provisions_when_enabled(
self,
mock_run_git,
mock_build_context,
mock_probe,
mock_ensure_remote_repository,
) -> None:
"""
If run_git raises GitError, _probe_mirror must report (False, <message>),
and not re-raise the exception.
"""
mock_run_git.side_effect = GitError("Git command failed (simulated)")
ctx = MagicMock()
ctx.identifier = "repo-id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {"origin": "git@github.com:alice/repo.git"}
ctx.file_mirrors = {}
type(ctx).resolved_mirrors = PropertyMock(return_value={"origin": "git@github.com:alice/repo.git"})
mock_build_context.return_value = ctx
ok, message = _probe_mirror(
"ssh://git@code.cymais.cloud:2201/kevinveenbirkenbach/pkgmgr.git",
"/tmp/some-repo",
mock_probe.return_value = (True, "")
repo = {"provider": "github.com", "account": "alice", "repository": "repo"}
setup_mirrors(
selected_repos=[repo],
repositories_base_dir="/base",
all_repos=[repo],
preview=False,
local=False,
remote=True,
ensure_remote=True,
)
self.assertFalse(ok)
self.assertIn("Git command failed", message)
mock_run_git.assert_called_once()
mock_ensure_remote_repository.assert_called_once()
mock_probe.assert_called_once()
@patch("pkgmgr.actions.mirror.setup_cmd.ensure_remote_repository")
@patch("pkgmgr.actions.mirror.setup_cmd.probe_mirror")
@patch("pkgmgr.actions.mirror.setup_cmd.build_context")
def test_setup_mirrors_remote_probes_all_resolved_mirrors(
self,
mock_build_context,
mock_probe,
mock_ensure_remote_repository,
) -> None:
ctx = MagicMock()
ctx.identifier = "repo-id"
ctx.repo_dir = "/tmp/repo"
ctx.config_mirrors = {}
ctx.file_mirrors = {}
type(ctx).resolved_mirrors = PropertyMock(
return_value={
"mirror": "git@github.com:alice/repo.git",
"backup": "ssh://git@git.veen.world:2201/alice/repo.git",
}
)
mock_build_context.return_value = ctx
mock_probe.return_value = (True, "")
repo = {"provider": "github.com", "account": "alice", "repository": "repo"}
setup_mirrors(
selected_repos=[repo],
repositories_base_dir="/base",
all_repos=[repo],
preview=False,
local=False,
remote=True,
ensure_remote=False,
)
mock_ensure_remote_repository.assert_not_called()
self.assertEqual(mock_probe.call_count, 2)
if __name__ == "__main__":

77
tests/unit/pkgmgr/actions/mirror/test_url_utils.py Normal file
View File

@@ -0,0 +1,77 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations
import unittest
from pkgmgr.actions.mirror.url_utils import hostport_from_git_url, normalize_provider_host, parse_repo_from_git_url
class TestUrlUtils(unittest.TestCase):
"""
Unit tests for URL parsing helpers used in mirror setup/provisioning.
"""
def test_hostport_from_git_url_ssh_url_with_port(self) -> None:
host, port = hostport_from_git_url("ssh://git@code.example.org:2201/alice/repo.git")
self.assertEqual(host, "code.example.org")
self.assertEqual(port, "2201")
def test_hostport_from_git_url_https_url_no_port(self) -> None:
host, port = hostport_from_git_url("https://github.com/alice/repo.git")
self.assertEqual(host, "github.com")
self.assertIsNone(port)
def test_hostport_from_git_url_scp_like(self) -> None:
host, port = hostport_from_git_url("git@github.com:alice/repo.git")
self.assertEqual(host, "github.com")
self.assertIsNone(port)
def test_hostport_from_git_url_empty(self) -> None:
host, port = hostport_from_git_url("")
self.assertEqual(host, "")
self.assertIsNone(port)
def test_normalize_provider_host_strips_port_and_lowercases(self) -> None:
self.assertEqual(normalize_provider_host("GIT.VEEN.WORLD:2201"), "git.veen.world")
def test_normalize_provider_host_ipv6_brackets(self) -> None:
self.assertEqual(normalize_provider_host("[::1]"), "::1")
def test_normalize_provider_host_empty(self) -> None:
self.assertEqual(normalize_provider_host(""), "")
def test_parse_repo_from_git_url_ssh_url(self) -> None:
host, owner, name = parse_repo_from_git_url("ssh://git@code.example.org:2201/alice/repo.git")
self.assertEqual(host, "code.example.org")
self.assertEqual(owner, "alice")
self.assertEqual(name, "repo")
def test_parse_repo_from_git_url_https_url(self) -> None:
host, owner, name = parse_repo_from_git_url("https://github.com/alice/repo.git")
self.assertEqual(host, "github.com")
self.assertEqual(owner, "alice")
self.assertEqual(name, "repo")
def test_parse_repo_from_git_url_scp_like(self) -> None:
host, owner, name = parse_repo_from_git_url("git@github.com:alice/repo.git")
self.assertEqual(host, "github.com")
self.assertEqual(owner, "alice")
self.assertEqual(name, "repo")
def test_parse_repo_from_git_url_best_effort_host_owner_repo(self) -> None:
host, owner, name = parse_repo_from_git_url("git.veen.world/alice/repo.git")
self.assertEqual(host, "git.veen.world")
self.assertEqual(owner, "alice")
self.assertEqual(name, "repo")
def test_parse_repo_from_git_url_missing_owner_repo_returns_none(self) -> None:
host, owner, name = parse_repo_from_git_url("https://github.com/")
self.assertEqual(host, "github.com")
self.assertIsNone(owner)
self.assertIsNone(name)
if __name__ == "__main__":
unittest.main()