Merge pull request #55730 from Ch3LL/mp_50089

Master Port #50089

salt/transport/tcp.py

@@ -557,9 +557,16 @@ def connect(self):
             if not self.auth.authenticated:
                 yield self.auth.authenticate()
             if self.auth.authenticated:
+                # if this is changed from the default, we assume it was intentional
+                if int(self.opts.get('publish_port', 4506)) != 4506:
+                    self.publish_port = self.opts.get('publish_port')
+                # else take the relayed publish_port master reports
+                else:
+                    self.publish_port = self.auth.creds['publish_port']
+
                 self.message_client = SaltMessageClientPool(
                     self.opts,
-                    args=(self.opts, self.opts['master_ip'], int(self.auth.creds['publish_port']),),
+                    args=(self.opts, self.opts['master_ip'], int(self.publish_port),),
                     kwargs={'io_loop': self.io_loop,
                             'connect_callback': self.connect_callback,
                             'disconnect_callback': self.disconnect_callback,

salt/transport/zeromq.py

@@ -509,7 +509,14 @@ def __del__(self):
     def connect(self):
         if not self.auth.authenticated:
             yield self.auth.authenticate()
-        self.publish_port = self.auth.creds['publish_port']
+
+        # if this is changed from the default, we assume it was intentional
+        if int(self.opts.get('publish_port', 4506)) != 4506:
+            self.publish_port = self.opts.get('publish_port')
+        # else take the relayed publish_port master reports
+        else:
+            self.publish_port = self.auth.creds['publish_port']
+
         log.debug('Connecting the Minion to the Master publish port, using the URI: %s', self.master_pub)
         self._socket.connect(self.master_pub)
 

tests/integration/__init__.py

@@ -810,6 +810,14 @@ def transplant_configs(cls, transport='zeromq'):
         syndic_opts['cachedir'] = 'cache'
         syndic_opts['root_dir'] = os.path.join(TMP_ROOT_DIR)
 
+        # This is the syndic for master
+        # Let's start with a copy of the syndic master configuration
+        syndic_opts = copy.deepcopy(syndic_master_opts)
+        # Let's update with the syndic configuration
+        syndic_opts.update(salt.config._read_conf_file(os.path.join(RUNTIME_VARS.CONF_DIR, 'syndic')))
+        syndic_opts['cachedir'] = os.path.join(TMP, 'rootdir', 'cache')
+        syndic_opts['config_dir'] = RUNTIME_VARS.TMP_SYNDIC_MINION_CONF_DIR
+
         # This proxy connects to master
         proxy_opts = salt.config._read_conf_file(os.path.join(CONF_DIR, 'proxy'))
         proxy_opts['cachedir'] = 'cache'

tests/unit/transport/test_tcp.py

@@ -231,6 +231,26 @@ def tearDown(self):
         del self._start_handlers
 
 
+class AsyncTCPPubChannelTest(AsyncTestCase, AdaptedConfigurationTestCaseMixin):
+    def test_connect_publish_port(self):
+        '''
+        test when publish_port is not 4506
+        '''
+        opts = self.get_temp_config('master')
+        opts['master_uri'] = ''
+        opts['master_ip'] = '127.0.0.1'
+        opts['publish_port'] = 1234
+        channel = salt.transport.tcp.AsyncTCPPubChannel(opts)
+        patch_auth = MagicMock(return_value=True)
+        patch_client = MagicMock(spec=SaltMessageClientPool)
+        with patch('salt.crypt.AsyncAuth.gen_token', patch_auth), \
+                patch('salt.crypt.AsyncAuth.authenticated', patch_auth), \
+                patch('salt.transport.tcp.SaltMessageClientPool',
+                      patch_client):
+            channel.connect()
+        assert patch_client.call_args[0][0]['publish_port'] == opts['publish_port']
+
+
 @skipIf(True, 'Skip until we can devote time to fix this test')
 class AsyncPubChannelTest(BaseTCPPubCase, PubChannelMixin):
     '''

tests/unit/transport/test_zeromq.py

@@ -495,6 +495,27 @@ def test_publish_to_pubserv_ipc(self):
         server_channel.pub_close()
         assert len(results) == send_num, (len(results), set(expect).difference(results))
 
+    def test_zeromq_publish_port(self):
+        '''
+        test when connecting that we
+        use the publish_port set in opts
+        when its not 4506
+        '''
+        opts = dict(self.master_config, ipc_mode='ipc',
+                    pub_hwm=0, recon_randomize=False,
+                    publish_port=455505,
+                    recon_default=1, recon_max=2, master_ip='127.0.0.1',
+                    acceptance_wait_time=5, acceptance_wait_time_max=5)
+        opts['master_uri'] = 'tcp://{interface}:{publish_port}'.format(**opts)
+
+        channel = salt.transport.zeromq.AsyncZeroMQPubChannel(opts)
+        patch_socket = MagicMock(return_value=True)
+        patch_auth = MagicMock(return_value=True)
+        with patch.object(channel, '_socket', patch_socket), \
+            patch.object(channel, 'auth', patch_auth):
+            channel.connect()
+        assert str(opts['publish_port']) in patch_socket.mock_calls[0][1][0]
+
     def test_zeromq_zeromq_filtering_decode_message_no_match(self):
         '''
         test AsyncZeroMQPubChannel _decode_messages when