94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
from pathlib import Path
|
|
import sys
|
|
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
|
|
str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
|
|
|
|
from informationscollector import InformationsCollector
|
|
|
|
|
|
class WarningCollectorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = object.__new__(InformationsCollector)
|
|
self.app.warning_list = []
|
|
self.app.warning_count_sensor = Mock()
|
|
self.app.warning_count_sensor.get_state.return_value = '2'
|
|
|
|
def test_warning_is_counted_and_exposed_in_attributes(self):
|
|
self.app.on_log_warning(
|
|
'example', None, 'WARNING', None, 'Something happened', {}
|
|
)
|
|
|
|
self.app.warning_count_sensor.set_state.assert_called_once_with(
|
|
state=3,
|
|
attributes={
|
|
'last_warning': 'Something happened',
|
|
'warning_list': ['Something happened'],
|
|
},
|
|
)
|
|
|
|
def test_warning_reset_clears_count_and_attributes(self):
|
|
self.app.warning_list = ['Old warning']
|
|
|
|
self.app.on_reset_warning_count(None, None, None, None, {})
|
|
|
|
self.assertEqual(self.app.warning_list, [])
|
|
self.app.warning_count_sensor.set_state.assert_called_once_with(
|
|
state=0,
|
|
attributes={'last_warning': '', 'warning_list': []},
|
|
)
|
|
|
|
|
|
class AppDaemonDockerVersionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.app = object.__new__(InformationsCollector)
|
|
self.app.log_warning = Mock()
|
|
self.app.dataset = {'appdaemon_latest_official_version': '4.5.14'}
|
|
self.app.appdaemon_version_sensor = Mock()
|
|
|
|
@staticmethod
|
|
def response(tags, next_url=None):
|
|
return io.StringIO(json.dumps({'results': tags, 'next': next_url}))
|
|
|
|
def test_selects_highest_stable_published_tag_across_pages(self):
|
|
pages = [
|
|
self.response([{'name': name, 'images': [{}]} for name in
|
|
['latest', 'dev', '4.5.14rc1', '4.5.9']], 'https://hub.docker.com/page2'),
|
|
self.response([{'name': '4.5.13', 'images': [{}]},
|
|
{'name': '4.5.14', 'images': []},
|
|
{'name': '4.5.12', 'images': [{}]}]),
|
|
]
|
|
with patch('informationscollector.urlopen', side_effect=pages) as fetch:
|
|
self.assertEqual(self.app._get_latest_official_appdaemon_version(), '4.5.13')
|
|
self.assertEqual(fetch.call_count, 2)
|
|
fetch.assert_any_call(self.app.APPDAEMON_DOCKER_TAGS_URL, timeout=10)
|
|
self.app.log_warning.assert_not_called()
|
|
|
|
def test_empty_stable_results_are_reported(self):
|
|
with patch('informationscollector.urlopen', return_value=self.response([])):
|
|
self.assertIsNone(self.app._get_latest_official_appdaemon_version())
|
|
self.app.log_warning.assert_called_once()
|
|
|
|
def test_sensor_uses_docker_cache_without_reusing_pypi_cache(self):
|
|
with patch.object(InformationsCollector, '_get_installed_appdaemon_version', return_value='4.5.10'):
|
|
with patch('informationscollector.urlopen', side_effect=OSError('offline')):
|
|
self.app.update_appdaemon_version_sensor()
|
|
self.app.appdaemon_version_sensor.set_state.assert_called_with(
|
|
state='4.5.10', attributes={'latest_official_version': None, 'update_available': None})
|
|
with patch('informationscollector.urlopen', return_value=self.response([
|
|
{'name': '4.5.13', 'images': [{}]}])):
|
|
self.app.update_appdaemon_version_sensor()
|
|
self.assertEqual(self.app.dataset['appdaemon_latest_docker_version'], '4.5.13')
|
|
with patch('informationscollector.urlopen', side_effect=OSError('offline')):
|
|
self.app.update_appdaemon_version_sensor()
|
|
self.app.appdaemon_version_sensor.set_state.assert_called_with(
|
|
state='4.5.10', attributes={'latest_official_version': '4.5.13', 'update_available': True})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|