33 lines
1.5 KiB
Python
33 lines
1.5 KiB
Python
"""Ensure port coordination never treats another service's port as ours."""
|
|
import json
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
import deploy
|
|
|
|
|
|
class PortGuardTests(unittest.TestCase):
|
|
def test_foreign_listener_blocks_deployment(self):
|
|
probe = MagicMock()
|
|
probe.__enter__.return_value.bind.side_effect = OSError("Address already in use")
|
|
with patch.object(deploy, "run", return_value=""), patch.object(deploy.socket, "socket", return_value=probe):
|
|
with self.assertRaisesRegex(RuntimeError, "already occupied"):
|
|
deploy.check_port(18135)
|
|
|
|
def test_current_service_can_keep_its_port(self):
|
|
details = [{"NetworkSettings": {"Ports": {"3000/tcp": [{"HostPort": "18135"}]}}}]
|
|
with patch.object(deploy, "run", side_effect=["our-container", json.dumps(details)]), patch.object(deploy.socket, "socket") as probe:
|
|
deploy.check_port(18135)
|
|
probe.assert_not_called()
|
|
|
|
def test_current_service_on_another_port_does_not_bypass_conflict(self):
|
|
details = [{"NetworkSettings": {"Ports": {"3000/tcp": [{"HostPort": "18134"}]}}}]
|
|
probe = MagicMock()
|
|
probe.__enter__.return_value.bind.side_effect = OSError("Address already in use")
|
|
with patch.object(deploy, "run", side_effect=["our-container", json.dumps(details)]), patch.object(deploy.socket, "socket", return_value=probe):
|
|
with self.assertRaisesRegex(RuntimeError, "already occupied"):
|
|
deploy.check_port(18135)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|