ROS version update fixes, #295

Author: akpw

mktxp/collector/address_list_collector.py

@@ -45,7 +45,7 @@ def collect(router_entry):
     @staticmethod
     def _collect_and_yield_metrics(router_entry, address_list_names, ip_version, metric_labels, reduced_metric_labels, translation_table):
         ipv6_suffix = '_ipv6' if ip_version == 'ipv6' else ''
-        
+
         # Collect and yield address list entries
         records = AddressListMetricsDataSource.metric_records(
             router_entry,

mktxp/collector/bandwidth_collector.py

@@ -27,33 +27,33 @@ def get_result(bandwidth_dict):
 
 class BandwidthCollector(BaseCollector):
     ''' MKTXP collector
-    '''    
+    '''
     def __init__(self):
         self.pool = None
-        self.last_call_timestamp = 0        
-    
+        self.last_call_timestamp = 0
+
     def collect(self):
         if not config_handler.system_entry.bandwidth:
             return
 
         if self.pool is None:
             self.pool = get_context("spawn").Pool()
 
-        if result_list:      
+        if result_list:
             result_dict = result_list[0]
-            bandwidth_records = [{'direction': key, 'bandwidth': str(result_dict[key])} for key in ('download', 'upload')]     
-            bandwidth_metrics = BaseCollector.gauge_collector('internet_bandwidth', 'Internet bandwidth in bits per second', 
+            bandwidth_records = [{'direction': key, 'bandwidth': str(result_dict[key])} for key in ('download', 'upload')]
+            bandwidth_metrics = BaseCollector.gauge_collector('internet_bandwidth', 'Internet bandwidth in bits per second',
                                                                             bandwidth_records, 'bandwidth', ['direction'], add_id_labels = False)
             yield bandwidth_metrics
 
             latency_records = [{'latency': str(result_dict['ping'])}]
-            latency_metrics = BaseCollector.gauge_collector('internet_latency', 'Internet latency in milliseconds', 
+            latency_metrics = BaseCollector.gauge_collector('internet_latency', 'Internet latency in milliseconds',
                                                                             latency_records, 'latency', [], add_id_labels = False)
             yield latency_metrics
 
-        ts =  datetime.now().timestamp()       
+        ts =  datetime.now().timestamp()
         if (ts - self.last_call_timestamp) > config_handler.system_entry.bandwidth_test_interval:
-            self.pool.apply_async(BandwidthCollector.bandwidth_worker, callback=get_result)            
+            self.pool.apply_async(BandwidthCollector.bandwidth_worker, callback=get_result)
             self.last_call_timestamp = ts
 
     def __del__(self):

mktxp/collector/base_collector.py

@@ -38,7 +38,7 @@ def _de_duplicate_records(router_records, metric_labels, verbose_reporting = Non
         for record in router_records:
             custom_labels_dict = record.get(MKTXPConfigKeys.CUSTOM_LABELS_METADATA_ID, {})
             label_values_tuple = tuple(record.get(label, custom_labels_dict.get(label, '')) for label in metric_labels)
-            
+
             if label_values_tuple in unique_records:
                 if verbose_reporting:
                     print(f"Warning: Duplicate metric record found for labels {dict(zip(metric_labels, label_values_tuple))}. Keeping last.")
@@ -56,7 +56,7 @@ def info_collector(name: str, documentation: str, router_records, metric_labels=
             metric_labels = BaseCollector._add_id_labels(metric_labels)
         if add_custom_labels:
             metric_labels = BaseCollector._add_custom_labels(metric_labels, records_for_labels)
-        
+
         collector = InfoMetricFamily(f'mktxp_{name}', documentation=documentation, labels=metric_labels)
 
         deduplicated_records = BaseCollector._de_duplicate_records(records_for_processing, metric_labels, verbose_reporting)

mktxp/collector/bgp_collector.py

@@ -30,7 +30,7 @@ def collect(router_entry):
                 'uptime': lambda value: BaseOutputProcessor.parse_timedelta_milliseconds(value) if value else '0'
                 }
         bgp_records = BGPMetricsDataSource.metric_records(router_entry, metric_labels=bgp_labels, translation_table = translation_table)
-        
+
         if bgp_records:
             session_info_labes = ['name', 'remote_address', 'remote_as', 'local_as', 'remote_afi', 'local_afi']
             bgp_sessions_metrics = BaseCollector.info_collector('bgp_sessions', 'BGP sessions info', bgp_records, session_info_labes)

mktxp/collector/capsman_collector.py

@@ -21,7 +21,7 @@
 
 class CapsmanCollector(BaseCollector):
     ''' CAPsMAN Metrics collector
-    '''    
+    '''
     @staticmethod
     def collect(router_entry):
         if not router_entry.config_entry.capsman:
@@ -54,7 +54,7 @@ def collect(router_entry):
                 # translate / trim / augment registration records
                 for registration_record in registration_records:
                     BaseOutputProcessor.augment_record(router_entry, registration_record)
-                    
+
                 tx_byte_metrics = BaseCollector.counter_collector('capsman_clients_tx_bytes', 'Number of sent packet bytes', registration_records, 'tx_bytes', ['dhcp_name', 'mac_address'])
                 yield tx_byte_metrics
 
@@ -64,7 +64,7 @@ def collect(router_entry):
                 signal_strength_metrics = BaseCollector.gauge_collector('capsman_clients_signal_strength', 'Client devices signal strength', registration_records, 'rx_signal', ['dhcp_name', 'mac_address'])
                 yield signal_strength_metrics
 
-                registration_metrics = BaseCollector.info_collector('capsman_clients_devices', 'Registered client devices info', 
+                registration_metrics = BaseCollector.info_collector('capsman_clients_devices', 'Registered client devices info',
                                         registration_records, ['dhcp_name', 'dhcp_address', 'rx_signal', 'ssid', 'tx_rate', 'rx_rate', 'interface', 'mac_address', 'uptime'])
                 yield registration_metrics
 

mktxp/utils/utils.py

@@ -56,7 +56,7 @@ def run_cmd(cmd, shell = False, quiet = False):
 
 def parse_mkt_uptime(time):
     time_dict = re.match(r'((?P<weeks>\d+)w)?((?P<days>\d+)d)?((?P<hours>\d+)h)?((?P<minutes>\d+)m)?((?P<seconds>\d+)s)?', time).groupdict()
-    delta = timedelta(**{key: int(value) for key, value in time_dict.items() if value}).total_seconds() 
+    delta = timedelta(**{key: int(value) for key, value in time_dict.items() if value}).total_seconds()
     return int(delta) if delta else 0
 
 def str2bool(str_value, default = None):
@@ -71,7 +71,7 @@ def str2bool(str_value, default = None):
         return default
     else:
         raise ValueError(f'Invalid truth value: {str_value}')
-        
+
 class FSHelper:
     ''' File System ops helper
     '''
@@ -229,15 +229,15 @@ def __init__(self, interval, func, args=[], kwargs={}, process_name = None, repe
         self.process_name = process_name
         self.interval = interval
         self.restartable = restartable
-        
+
         self.func = func
         self.args = args
         self.kwargs = kwargs
-        
+
         self.finished = Event()
         self.run_once = Event()
         if not repeatable:
-            self.run_once.set()            
+            self.run_once.set()
         self.process = Process(name = self.process_name, target=self._execute)
 
     def start(self):
@@ -256,7 +256,7 @@ def _execute(self):
             self.func(*self.args, **self.kwargs)
             if self.finished.is_set() or self.run_once.is_set():
                 break
-            self.finished.wait(self.interval)     
+            self.finished.wait(self.interval)
 
 class Benchmark:
     def __enter__(self):
@@ -281,29 +281,43 @@ def get_ttl_hash(seconds=3600):
     return round(time.time() / seconds)
 
 
+# Timeout for fetching update RSS feeds (in seconds)
+UPDATE_CHECK_TIMEOUT = 10
+
 @lru_cache(maxsize=5)
 def get_available_updates(channel, ttl_hash=get_ttl_hash()):
     """Check the RSS feed for available updates for a given update channel.
     This method fetches the RSS feed and returns all version from the parsed XML.
-    Version numbers are parsed into version.Version instances (part of setuptools)."""
+    Version numbers are parsed into version.Version instances (part of setuptools).
+
+    Errors are handled internally so that lru_cache caches the (empty) result,
+    preventing repeated blocking retries on transient failures."""
     del ttl_hash
-    rss_feed = CHANNEL_RSS_FEED_MAPPING[channel]
+    rss_feed = CHANNEL_RSS_FEED_MAPPING.get(channel)
+    if not rss_feed:
+        print(f'Unknown update channel: {channel}')
+        return []
 
     print(f'Fetching available ROS releases from {rss_feed}')
     versions = []
-    with urllib.request.urlopen(rss_feed) as response:
-        result = response.read()
-        root = ET.fromstring(result)
-        channel = root[0]
-
-        for child in channel:
-            # iterate over all updates
-            if child.tag == 'item':
-                title = child[0]
-                # extract and parse the version number from title
-                version_text = re.findall(r'[\d+\.]+', title.text)[0]
-                version_number = parse(version_text)
-                versions.append(version_number)
+    try:
+        with urllib.request.urlopen(rss_feed, timeout=UPDATE_CHECK_TIMEOUT) as response:
+            result = response.read()
+            root = ET.fromstring(result)
+            rss_channel = root[0]
+
+            for child in rss_channel:
+                # iterate over all updates
+                if child.tag == 'item':
+                    title = child[0]
+                    # extract and parse the version number from title
+                    version_text = re.findall(r'[\d+\.]+', title.text)[0]
+                    version_number = parse(version_text)
+                    versions.append(version_number)
+    except urllib.error.HTTPError as err:
+        print(f'Update feed returned: {err}')
+    except Exception as err:
+        print(f'Could not check for updates, because: {err}')
     return versions
 
 
@@ -315,7 +329,7 @@ def parse_ros_version(string):
     1.2.3, stable
 
     >>> parse_ros_version('7.14.3 (long-term)')
-    7.14.3, long-term    
+    7.14.3, long-term
     """
 
     match = re.findall(r'([\d\.]+).*?\(([\w-]+)\)', string)
@@ -355,22 +369,13 @@ def check_for_updates(cur_version):
     """Try to check if there is a newer version available.
     If anything goes wrong, it returns the same version.
     Returns a tuple: (<current version>, <newest version>)"""
-    error = False
     try:
         cur_version, channel = parse_ros_version(cur_version)
         available_versions = get_available_updates(channel)
-        newest_version = sorted(available_versions)[-1]
-    except KeyError:
-        print(f'unknown update channel {channel}')
-        error = True
-    except urllib.error.HTTPError as err:
-        print(f'update feed returned: {err}')
-        error = True
+        if available_versions:
+            newest_version = sorted(available_versions)[-1]
+            return cur_version, newest_version
     except Exception as err:
-        print(f'could not check for updates, because: {err}')
-        error = True
-
-    if error:
-        return cur_version, cur_version
+        print(f'Could not check for updates, because: {err}')
 
-    return cur_version, newest_version
+    return cur_version, cur_version

tests/utils/test_utils.py

@@ -12,6 +12,8 @@
 ## GNU General Public License for more details.
 
 from datetime import timedelta
+from unittest.mock import patch, MagicMock
+import urllib.error
 import pytest
 from mktxp.utils import utils
 from packaging.version import parse
@@ -94,3 +96,79 @@ def test_str2bool_raise_value_error(str_value):
 
     assert utils.str2bool(str_value, False) == False
     assert utils.str2bool(str_value, True) == True
+
+
+class TestGetAvailableUpdates:
+    """Tests for get_available_updates and check_for_updates."""
+
+    def setup_method(self):
+        # Clear lru_cache between tests
+        utils.get_available_updates.cache_clear()
+
+    def test_urlopen_called_with_timeout(self):
+        """urlopen must be called with an explicit timeout."""
+        mock_response = MagicMock()
+        mock_response.read.return_value = b'''<?xml version="1.0"?>
+            <rss><channel>
+                <item><title>7.16 changelog</title></item>
+            </channel></rss>'''
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        with patch('urllib.request.urlopen', return_value=mock_response) as mock_urlopen:
+            utils.get_available_updates('stable', ttl_hash=0)
+            mock_urlopen.assert_called_once_with(
+                'https://mikrotik.com/current.rss',
+                timeout=utils.UPDATE_CHECK_TIMEOUT,
+            )
+
+    def test_timeout_returns_empty_list(self):
+        """A timeout should return an empty list (not raise), so lru_cache caches it."""
+        with patch('urllib.request.urlopen', side_effect=TimeoutError('timed out')):
+            result = utils.get_available_updates('stable', ttl_hash=1)
+            assert result == []
+
+    def test_http_error_returns_empty_list(self):
+        """An HTTP error should return an empty list."""
+        with patch('urllib.request.urlopen', side_effect=urllib.error.HTTPError(
+            url=None, code=503, msg='Service Unavailable', hdrs=None, fp=None
+        )):
+            result = utils.get_available_updates('stable', ttl_hash=2)
+            assert result == []
+
+    def test_unknown_channel_returns_empty_list(self):
+        """An unknown channel should return an empty list without attempting a fetch."""
+        with patch('urllib.request.urlopen') as mock_urlopen:
+            result = utils.get_available_updates('nonexistent', ttl_hash=3)
+            assert result == []
+            mock_urlopen.assert_not_called()
+
+    def test_failure_result_is_cached(self):
+        """After a failure, the empty list result should be cached by lru_cache."""
+        with patch('urllib.request.urlopen', side_effect=TimeoutError('timed out')) as mock_urlopen:
+            utils.get_available_updates('stable', ttl_hash=4)
+            utils.get_available_updates('stable', ttl_hash=4)
+            # urlopen should only be called once; second call uses cache
+            mock_urlopen.assert_called_once()
+
+    def test_check_for_updates_on_fetch_failure(self):
+        """check_for_updates should return cur == newest when the feed fetch fails."""
+        with patch('urllib.request.urlopen', side_effect=TimeoutError('timed out')):
+            cur, newest = utils.check_for_updates('7.15 (stable)')
+            assert cur == newest == parse('7.15')
+
+    def test_check_for_updates_success(self):
+        """check_for_updates should return the newest version on success."""
+        mock_response = MagicMock()
+        mock_response.read.return_value = b'''<?xml version="1.0"?>
+            <rss><channel>
+                <item><title>7.16 changelog</title></item>
+                <item><title>7.15.3 changelog</title></item>
+            </channel></rss>'''
+        mock_response.__enter__ = lambda s: s
+        mock_response.__exit__ = MagicMock(return_value=False)
+
+        with patch('urllib.request.urlopen', return_value=mock_response):
+            cur, newest = utils.check_for_updates('7.15 (stable)')
+            assert cur == parse('7.15')
+            assert newest == parse('7.16')