test(ci): cover assurance gate entry points
prd-number-check / require-numbered-prds (pull_request) Successful in 7s
tracker-policy-pr / check-pr (pull_request) Successful in 8s
lint / lint (push) Successful in 58s
test / unit (pull_request) Successful in 48s
test / integration-docker (pull_request) Failing after 19m28s
test / coverage (pull_request) Has been skipped
prd-number-check / require-numbered-prds (pull_request) Successful in 7s
tracker-policy-pr / check-pr (pull_request) Successful in 8s
lint / lint (push) Successful in 58s
test / unit (pull_request) Successful in 48s
test / integration-docker (pull_request) Failing after 19m28s
test / coverage (pull_request) Has been skipped
This commit is contained in:
@@ -4,6 +4,8 @@ from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
from scripts.critical_modules import (
|
||||
@@ -11,6 +13,7 @@ from scripts.critical_modules import (
|
||||
REPO_ROOT,
|
||||
CriticalModulesError,
|
||||
load_critical_modules,
|
||||
main,
|
||||
)
|
||||
|
||||
|
||||
@@ -49,6 +52,50 @@ class TestCriticalModules(unittest.TestCase):
|
||||
with self.assertRaisesRegex(CriticalModulesError, "escapes"):
|
||||
load_critical_modules(manifest, root=root)
|
||||
|
||||
def test_invalid_entry_forms_are_reported_together(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
manifest = root / "critical-modules.txt"
|
||||
manifest.write_text(
|
||||
f"{root / 'absolute.py'}\nREADME.md\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
with self.assertRaises(CriticalModulesError) as raised:
|
||||
load_critical_modules(manifest, root=root)
|
||||
self.assertIn("must be relative", str(raised.exception))
|
||||
self.assertIn("is not a Python module", str(raised.exception))
|
||||
|
||||
def test_empty_manifest_fails(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
manifest = root / "critical-modules.txt"
|
||||
manifest.write_text("# comments do not define modules\n", encoding="utf-8")
|
||||
with self.assertRaisesRegex(CriticalModulesError, "contains no"):
|
||||
load_critical_modules(manifest, root=root)
|
||||
|
||||
def test_unreadable_manifest_fails(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
missing = Path(tmp) / "missing.txt"
|
||||
with self.assertRaisesRegex(CriticalModulesError, "cannot read"):
|
||||
load_critical_modules(missing, root=Path(tmp))
|
||||
|
||||
def test_main_prints_include_list_or_checks_silently(self) -> None:
|
||||
output = StringIO()
|
||||
with redirect_stdout(output):
|
||||
self.assertEqual(0, main([]))
|
||||
self.assertIn("bot_bottle/manifest/egress.py", output.getvalue())
|
||||
|
||||
output = StringIO()
|
||||
with redirect_stdout(output):
|
||||
self.assertEqual(0, main(["--check"]))
|
||||
self.assertEqual("", output.getvalue())
|
||||
|
||||
def test_main_reports_manifest_error(self) -> None:
|
||||
error = StringIO()
|
||||
with redirect_stderr(error):
|
||||
self.assertEqual(1, main(["--manifest", "/definitely/missing"]))
|
||||
self.assertIn("critical-modules:", error.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -3,8 +3,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from io import StringIO
|
||||
from unittest.mock import patch
|
||||
|
||||
from scripts.docker_host_address import default_ipv4_gateway
|
||||
from scripts.docker_host_address import default_ipv4_gateway, main
|
||||
|
||||
|
||||
class TestDefaultIpv4Gateway(unittest.TestCase):
|
||||
@@ -23,6 +26,34 @@ class TestDefaultIpv4Gateway(unittest.TestCase):
|
||||
with self.assertRaisesRegex(ValueError, "no active"):
|
||||
default_ipv4_gateway(routes)
|
||||
|
||||
def test_ignores_short_malformed_and_inactive_routes(self) -> None:
|
||||
routes = (
|
||||
"Iface Destination Gateway Flags\n"
|
||||
"short row\n"
|
||||
"eth0 00000000 nope 0003\n"
|
||||
"eth0 00000000 010011AC 0001\n"
|
||||
)
|
||||
with self.assertRaisesRegex(ValueError, "no active"):
|
||||
default_ipv4_gateway(routes)
|
||||
|
||||
def test_main_prints_gateway(self) -> None:
|
||||
routes = (
|
||||
"Iface Destination Gateway Flags\n"
|
||||
"eth0 00000000 010011AC 0003\n"
|
||||
)
|
||||
output = StringIO()
|
||||
with patch("pathlib.Path.read_text", return_value=routes), \
|
||||
redirect_stdout(output):
|
||||
self.assertEqual(0, main())
|
||||
self.assertEqual("172.17.0.1\n", output.getvalue())
|
||||
|
||||
def test_main_reports_route_error(self) -> None:
|
||||
error = StringIO()
|
||||
with patch("pathlib.Path.read_text", side_effect=OSError("denied")), \
|
||||
redirect_stderr(error):
|
||||
self.assertEqual(1, main())
|
||||
self.assertIn("docker-host-address: denied", error.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -9,6 +9,7 @@ a freshly minted token."""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from bot_bottle.backend.docker.infra import (
|
||||
@@ -75,6 +76,13 @@ class TestDockerInfraService(unittest.TestCase):
|
||||
self.assertEqual("registry-volume", svc.orchestrator()._root_mount_source)
|
||||
self.assertEqual("ca-volume", svc.gateway()._ca_mount_source)
|
||||
|
||||
def test_host_path_and_named_root_mount_are_mutually_exclusive(self) -> None:
|
||||
with self.assertRaisesRegex(ValueError, "host_root or root_mount_source"):
|
||||
DockerInfraService(
|
||||
host_root=Path("/host/path"),
|
||||
root_mount_source="registry-volume",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
import urllib.error
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
from bot_bottle.backend.docker.orchestrator import (
|
||||
@@ -61,6 +62,13 @@ class TestDockerOrchestrator(unittest.TestCase):
|
||||
self.assertEqual("0.0.0.0:8099:8099", argv[argv.index("--publish") + 1])
|
||||
self.assertIn("state-volume:/bot-bottle-root", argv)
|
||||
|
||||
def test_host_path_and_named_root_mount_are_mutually_exclusive(self) -> None:
|
||||
with self.assertRaisesRegex(ValueError, "host_root or root_mount_source"):
|
||||
DockerOrchestrator(
|
||||
host_root=Path("/host/path"),
|
||||
root_mount_source="state-volume",
|
||||
)
|
||||
|
||||
def test_gateway_url_is_the_container_dns_name(self) -> None:
|
||||
# The gateway reaches the orchestrator by name on the control network.
|
||||
self.assertEqual(f"http://{ORCHESTRATOR_NAME}:8099", self.orch.gateway_url())
|
||||
|
||||
@@ -3,8 +3,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from contextlib import redirect_stderr
|
||||
from io import StringIO
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from scripts.unittest_gate import assurance_errors
|
||||
from scripts.unittest_gate import assurance_errors, main
|
||||
|
||||
|
||||
class TestAssuranceErrors(unittest.TestCase):
|
||||
@@ -29,6 +32,60 @@ class TestAssuranceErrors(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(["1 test(s) skipped in a no-skip suite"], errors)
|
||||
|
||||
def test_main_accepts_successful_assured_suite(self) -> None:
|
||||
result = Mock(
|
||||
testsRun=22,
|
||||
skipped=[],
|
||||
wasSuccessful=Mock(return_value=True),
|
||||
)
|
||||
runner = Mock()
|
||||
runner.run.return_value = result
|
||||
with patch(
|
||||
"scripts.unittest_gate.unittest.defaultTestLoader.discover",
|
||||
return_value=Mock(),
|
||||
) as discover, patch(
|
||||
"scripts.unittest_gate.unittest.TextTestRunner",
|
||||
return_value=runner,
|
||||
) as runner_type:
|
||||
self.assertEqual(
|
||||
0,
|
||||
main([
|
||||
"-s", "tests/integration",
|
||||
"-t", ".",
|
||||
"-p", "test_*.py",
|
||||
"--minimum-executed", "22",
|
||||
"--fail-on-skip",
|
||||
"-v",
|
||||
]),
|
||||
)
|
||||
discover.assert_called_once_with(
|
||||
"tests/integration", pattern="test_*.py", top_level_dir="."
|
||||
)
|
||||
runner_type.assert_called_once_with(verbosity=2)
|
||||
|
||||
def test_main_rejects_unsuccessful_underfilled_suite(self) -> None:
|
||||
result = Mock(
|
||||
testsRun=1,
|
||||
skipped=[(Mock(), "not available")],
|
||||
wasSuccessful=Mock(return_value=False),
|
||||
)
|
||||
runner = Mock()
|
||||
runner.run.return_value = result
|
||||
error = StringIO()
|
||||
with patch(
|
||||
"scripts.unittest_gate.unittest.defaultTestLoader.discover",
|
||||
return_value=Mock(),
|
||||
), patch(
|
||||
"scripts.unittest_gate.unittest.TextTestRunner",
|
||||
return_value=runner,
|
||||
), redirect_stderr(error):
|
||||
self.assertEqual(
|
||||
1,
|
||||
main(["--minimum-executed", "2", "--fail-on-skip"]),
|
||||
)
|
||||
self.assertIn("below required minimum", error.getvalue())
|
||||
self.assertIn("skipped in a no-skip suite", error.getvalue())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user