Properly format IPv6 bind address strings

This fixes:
    Error: '::9443' is not a valid port number.

Introduce unit test skeleton for amphora agent.

Change-Id: Ic7aebf0674ba7036356bb3231c26fa309cd4c475

octavia/cmd/agent.py

@@ -25,6 +25,7 @@ import six
 from octavia.amphorae.backends.agent.api_server import server
 from octavia.amphorae.backends.health_daemon import health_daemon
 from octavia.common import service
+from octavia.common import utils
 from octavia import version
 
 
@@ -66,11 +67,10 @@ def main():
     # Initiate server class
     server_instance = server.Server()
 
+    bind_ip_port = utils.ip_port_str(CONF.haproxy_amphora.bind_host,
+                                     CONF.haproxy_amphora.bind_port)
     options = {
-        'bind': '{host}:{port}'.format(
-            host=CONF.haproxy_amphora.bind_host,
-            port=CONF.haproxy_amphora.bind_port
-        ),
+        'bind': bind_ip_port,
         'workers': 1,
         'timeout': CONF.amphora_agent.agent_request_read_timeout,
         'certfile': CONF.amphora_agent.agent_server_cert,

octavia/common/utils.py

@@ -65,6 +65,15 @@ def is_ipv6_lla(ip_address):
     return ip.version == 6 and ip.is_link_local()
 
 
+def ip_port_str(ip_address, port):
+    """Return IP port as string representation depending on address family."""
+    ip = netaddr.IPAddress(ip_address)
+    if ip.version == 4:
+        return "{ip}:{port}".format(ip=ip, port=port)
+    elif ip.version == 6:
+        return "[{ip}]:{port}".format(ip=ip, port=port)
+
+
 class exception_logger(object):
     """Wrap a function and log raised exception
 

octavia/tests/unit/cmd/test_agent.py

@@ -0,0 +1,40 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import mock
+
+from octavia.cmd import agent
+from octavia.tests.unit import base
+
+
+class TestAmphoraAgentCMD(base.TestCase):
+
+    def setUp(self):
+        super(TestAmphoraAgentCMD, self).setUp()
+
+    @mock.patch('octavia.cmd.agent.AmphoraAgent')
+    @mock.patch('octavia.amphorae.backends.agent.api_server.server.Server')
+    @mock.patch('multiprocessing.Process')
+    @mock.patch('octavia.common.service.prepare_service')
+    def test_main(self, mock_service, mock_process, mock_server, mock_amp):
+        mock_health_proc = mock.MagicMock()
+        mock_server_instance = mock.MagicMock()
+        mock_amp_instance = mock.MagicMock()
+
+        mock_process.return_value = mock_health_proc
+        mock_server.return_value = mock_server_instance
+        mock_amp.return_value = mock_amp_instance
+
+        agent.main()
+
+        mock_health_proc.start.assert_called_once_with()
+        mock_amp_instance.run.assert_called_once()

octavia/tests/unit/common/test_utils.py

@@ -36,3 +36,9 @@ class TestConfig(base.TestCase):
         self.assertFalse(utils.is_ipv6_lla('::'))
         self.assertFalse(utils.is_ipv6_lla('2001:db8::1'))
         self.assertTrue(utils.is_ipv6_lla('fe80::225:90ff:fefb:53ad'))
+
+    def test_ip_port_str(self):
+        self.assertEqual("127.0.0.1:8080",
+                         utils.ip_port_str('127.0.0.1', 8080))
+        self.assertEqual("[::1]:8080",
+                         utils.ip_port_str('::1', 8080))