feat: apt's proxy through AptProxyMixin (NetOrk/netork#1397)
CI / test (3.10) (push) Failing after 21s
CI / test (3.11) (push) Failing after 19s
CI / test (3.12) (push) Failing after 18s
CI / test (3.10) (pull_request) Failing after 19s
CI / test (3.11) (pull_request) Failing after 17s
CI / test (3.12) (pull_request) Failing after 16s

get_package_proxy() is what apt-config dump says, not 00proxy alone: a
proxy in 90curtin-aptproxy or /etc/apt/apt.conf wins over it and was
missed. The driver now supplies only the transports: reads on the exec
channel, root commands through _root_outcome, the file through the staged
_install_file_as_root. New: get_package_proxy_report() and
check_package_proxy(). configure_package_proxy() also comments out other
proxy lines, reads the result back and fetches through the proxy.

Requires napalm-device-types 4.22.0; version 0.14.0.
This commit is contained in:
Christian Manivong committed 2026-10-22 19:16:40 +02:00
1 parent 0ba5081824
commit 04f071afd0
4 files changed
+93 -31

No files matched your search

+73 -3
View File
@@ -123,21 +123,72 @@ class TestSnmpAgentSteps:
driver._install_file_as_root.assert_not_called()
_REPORT = (
"APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n"
"http://deb.debian.org/debian/dists/trixie/\nAPROXY_END\n"
)
def _host(driver, proxy="", parts="", fetched=0):
"""What the host answers on the exec channel: apt's report, then a fetch."""
ran: list[str] = []
def run_command(command, *, privileged=False, timeout=60, stdin=None):
ran.append(command)
if "apt-config dump" in command:
dump = f'Acquire::http::Proxy "{proxy}";\n' if proxy else ""
return MagicMock(exit_code=0, stdout=_REPORT.format(dump=dump, parts=parts), stderr="")
out = f"APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n{fetched}\nAPROXY_END\n"
return MagicMock(exit_code=0, stdout=out, stderr="")
driver.run_command = run_command
return ran
class TestPackageProxyAndEngineAccess:
def test_the_proxy_is_the_caller_s(self, driver):
driver._install_file_as_root = MagicMock(
return_value=MagicMock(exit_code=0, stdout="", stderr="")
)
_host(driver, proxy="http://cache.lan:3142")
result = driver.configure_package_proxy("http://cache.lan:3142")
assert result["success"] is True
assert result["success"] is True, result["output"]
path, content = driver._install_file_as_root.call_args.args
assert (path, content) == (
"/etc/apt/apt.conf.d/00proxy",
'Acquire::http::Proxy "http://cache.lan:3142";\n',
)
def test_it_reads_back_and_fetches_through_it(self, driver):
driver._install_file_as_root = MagicMock(
return_value=MagicMock(exit_code=0, stdout="", stderr="")
)
ran = _host(driver, proxy="http://cache.lan:3142")
result = driver.configure_package_proxy("http://cache.lan:3142")
assert [s["step"] for s in result["steps"]] == ["configure", "conflicts", "verify", "fetch"]
assert any("dists/trixie/InRelease" in c for c in ran)
def test_another_file_s_proxy_is_commented_out_as_root(self, driver):
driver._install_file_as_root = MagicMock(
return_value=MagicMock(exit_code=0, stdout="", stderr="")
)
_host(
driver,
proxy="http://cache.lan:3142",
parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
)
ran = []
driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
driver.configure_package_proxy("http://cache.lan:3142")
assert ran and ran[0].startswith("sed -i -e '1{")
assert ran[0].endswith("/etc/apt/apt.conf.d/90curtin-aptproxy")
def test_a_proxy_url_that_would_break_the_file_is_refused(self, driver):
driver._install_file_as_root = MagicMock()
@@ -151,6 +202,15 @@ class TestPackageProxyAndEngineAccess:
assert driver.configure_package_proxy("http://cache.lan:3142")["success"] is False
def test_the_proxy_answers_past_apt_s_proxy(self, driver):
out = "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n3\nAPROXY_END\n"
driver.run_command = MagicMock(return_value=MagicMock(exit_code=0, stdout=out, stderr=""))
result = driver.check_package_proxy("http://c:3142/acng-report.html", expect="Cache")
assert result == {"reachable": True, "expected": True, "error": ""}
assert "Proxy::c=DIRECT" in driver.run_command.call_args.args[0]
def test_engine_access_for_the_user_the_caller_names(self, driver):
ran = []
driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
@@ -220,16 +280,26 @@ class TestFactsNotWarnings:
assert "get_device_warnings" not in vars(LinuxDriver)
def test_the_package_proxy_apt_fetches_through(self, driver):
driver._send = lambda command, read_timeout=100: 'Acquire::http::Proxy "http://c:3142";'
_host(driver, proxy="http://c:3142")
assert driver.get_package_proxy() == "http://c:3142"
def test_no_proxy_set(self, driver):
driver._send = lambda command, read_timeout=100: ""
_host(driver)
assert driver.get_package_proxy() is None
def test_where_apt_s_proxy_is_set(self, driver):
_host(
driver,
proxy="http://old:3128",
parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
)
report = driver.get_package_proxy_report()
assert report["settings"][0]["file"] == "/etc/apt/apt.conf.d/90curtin-aptproxy"
def test_not_an_apt_host(self, driver):
driver._pkg_manager = "dnf"
assert driver.get_package_proxy() is None
assert driver.get_package_proxy_report() is None
def test_the_applications_proxy_url_is_no_argument_any_more(self):
d = LinuxDriver("h", "admin", "p", optional_args={"apt_proxy_url": "http://c:3142"})