github-such-software/adaptive-multi-protocol-gateway
1
0
Fork 0
mirror of https://github.com/Such-Software/adaptive-multi-protocol-gateway.git synced 2026-10-27 19:39:30 +01:00
Code Issues Packages Projects Releases Wiki Activity

Model transport provider sources

Browse Source
This commit is contained in:
jwinterm
2026-07-08 10:12:59 -04:00
parent e272ec1afb
commit f0d26933e6
10 changed files with 148 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_install_plan.py
+2 -1
View File
@@ -3,6 +3,7 @@ import io
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from ampg.approvals import ApprovalInput, approve_artifacts
from ampg.cli import main
@@ -170,7 +171,7 @@ daemon_policy = "auto"
)
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
with contextlib.redirect_stdout(stdout), patch("ampg.status.shutil.which", return_value=None):
status = main(
[
"--config",
tests/test_protocol_selection.py
+3 -2
View File
@@ -3,6 +3,7 @@ import io
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from ampg.cli import main
from ampg.config import load_config
@@ -149,7 +150,7 @@ class ProtocolSelectionTest(unittest.TestCase):
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
with contextlib.redirect_stdout(stdout), patch("ampg.status.shutil.which", return_value=None):
status = main(
[
"--config",
@@ -179,7 +180,7 @@ class ProtocolSelectionTest(unittest.TestCase):
(output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
with contextlib.redirect_stdout(stdout), patch("ampg.status.shutil.which", return_value=None):
status = main(
[
"--config",
tests/test_status.py
+3
View File
@@ -74,6 +74,7 @@ daemon_policy = "auto"
self.assertEqual(1, len(statuses))
self.assertEqual("ok", statuses[0].status)
self.assertEqual("adopt-existing", statuses[0].action)
self.assertEqual("system-adopted", statuses[0].provider_source)
self.assertTrue(statuses[0].adoptable)
def test_auto_policy_manages_missing_daemon_on_manageable_platform(self):
@@ -98,6 +99,7 @@ daemon_policy = "auto"
self.assertEqual("ok", statuses[0].status)
self.assertEqual("manage-owned", statuses[0].action)
self.assertEqual("platform-package", statuses[0].provider_source)
self.assertTrue(statuses[0].manageable)
def test_adopt_policy_errors_when_daemon_is_missing(self):
@@ -122,6 +124,7 @@ daemon_policy = "adopt"
self.assertEqual("error", statuses[0].status)
self.assertEqual("unavailable", statuses[0].action)
self.assertEqual("unavailable", statuses[0].provider_source)
self.assertIn("nginx is not installed", statuses[0].message)
def test_unknown_adapter_errors_unless_external(self):