|
1 | | -import os |
2 | 1 | import unittest |
3 | | -from agentstack.update import should_update |
| 2 | +import os |
| 3 | +import json |
| 4 | +from unittest.mock import patch, mock_open |
| 5 | +from parameterized import parameterized |
| 6 | +from pathlib import Path |
| 7 | +from packaging.version import Version |
| 8 | +import requests |
| 9 | +from agentstack.update import ( |
| 10 | + _is_ci_environment, |
| 11 | + CI_ENV_VARS, |
| 12 | + should_update, |
| 13 | + _get_base_dir, |
| 14 | + get_latest_version, |
| 15 | + AGENTSTACK_PACKAGE, |
| 16 | + load_update_data, |
| 17 | + record_update_check, |
| 18 | + INSTALL_PATH, |
| 19 | + CHECK_EVERY, |
| 20 | +) |
| 21 | + |
| 22 | +BASE_DIR = Path(__file__).parent |
| 23 | + |
4 | 24 |
|
| 25 | +class TestUpdate(unittest.TestCase): |
| 26 | + @patch.dict('os.environ', {}, clear=True) |
| 27 | + def test_is_ci_environment_false(self): |
| 28 | + """ |
| 29 | + Test that _is_ci_environment() returns False when no CI environment variables are set. |
| 30 | + """ |
| 31 | + self.assertFalse(_is_ci_environment()) |
| 32 | + |
| 33 | + @parameterized.expand([(var,) for var in CI_ENV_VARS]) |
| 34 | + @patch.dict('os.environ', clear=True) |
| 35 | + def test_is_ci_environment_true(self, env_var): |
| 36 | + """ |
| 37 | + Test that _is_ci_environment() returns True when any CI environment variable is set. |
| 38 | + """ |
| 39 | + with patch.dict('os.environ', {env_var: 'true'}): |
| 40 | + self.assertTrue(_is_ci_environment()) |
5 | 41 |
|
6 | | -class UpdateTest(unittest.TestCase): |
7 | 42 | def test_updates_disabled_by_env_var_in_test(self): |
8 | | - assert 'AGENTSTACK_UPDATE_DISABLE' in os.environ |
9 | | - assert not should_update() |
| 43 | + """ |
| 44 | + Test that should_update() returns False when AGENTSTACK_UPDATE_DISABLE environment variable is set to 'true'. |
| 45 | + """ |
| 46 | + with patch.dict('os.environ', {'AGENTSTACK_UPDATE_DISABLE': 'true'}): |
| 47 | + self.assertFalse(should_update()) |
| 48 | + |
| 49 | + @patch('agentstack.update.user_data_dir') |
| 50 | + def test_get_base_dir_writable(self, mock_user_data_dir): |
| 51 | + """ |
| 52 | + Test that _get_base_dir() returns a writable Path when user_data_dir is accessible. |
| 53 | + """ |
| 54 | + mock_path = '/mock/user/data/dir' |
| 55 | + mock_user_data_dir.return_value = mock_path |
| 56 | + |
| 57 | + result = _get_base_dir() |
| 58 | + |
| 59 | + self.assertIsInstance(result, Path) |
| 60 | + self.assertTrue(result.is_absolute()) |
| 61 | + |
| 62 | + @patch('agentstack.update.user_data_dir') |
| 63 | + def test_get_base_dir_not_writable(self, mock_user_data_dir): |
| 64 | + """ |
| 65 | + Test that _get_base_dir() falls back to a temporary directory when user_data_dir is not writable. |
| 66 | + """ |
| 67 | + mock_user_data_dir.side_effect = PermissionError |
| 68 | + |
| 69 | + result = _get_base_dir() |
| 70 | + |
| 71 | + self.assertIsInstance(result, Path) |
| 72 | + self.assertTrue(result.is_absolute()) |
| 73 | + self.assertIn(str(result), ['/tmp', os.environ.get('TEMP', '/tmp')]) |
| 74 | + |
| 75 | + def test_get_latest_version(self): |
| 76 | + """ |
| 77 | + Test that get_latest_version returns a valid Version object from the actual PyPI. |
| 78 | + """ |
| 79 | + latest_version = get_latest_version(AGENTSTACK_PACKAGE) |
| 80 | + self.assertIsInstance(latest_version, Version) |
| 81 | + |
| 82 | + @patch('requests.get') |
| 83 | + def test_get_latest_version_404(self, mock_get): |
| 84 | + """ |
| 85 | + Test that get_latest_version raises an exception when the request returns a 404. |
| 86 | + """ |
| 87 | + mock_response = mock_get.return_value |
| 88 | + mock_response.status_code = 404 |
| 89 | + mock_response.raise_for_status.side_effect = requests.HTTPError("404 Client Error") |
| 90 | + |
| 91 | + with self.assertRaises(Exception) as context: |
| 92 | + get_latest_version(AGENTSTACK_PACKAGE) |
| 93 | + |
| 94 | + @patch('requests.get') |
| 95 | + def test_get_latest_version_timeout(self, mock_get): |
| 96 | + """ |
| 97 | + Test that get_latest_version handles request timeouts. |
| 98 | + """ |
| 99 | + mock_get.side_effect = requests.Timeout("Request timed out") |
| 100 | + |
| 101 | + with self.assertRaises(Exception) as context: |
| 102 | + get_latest_version(AGENTSTACK_PACKAGE) |
| 103 | + |
| 104 | + @patch( |
| 105 | + 'agentstack.update.LAST_CHECK_FILE_PATH', |
| 106 | + new_callable=lambda: BASE_DIR / 'tests/tmp/test_update/last_check.json', |
| 107 | + ) |
| 108 | + @patch('agentstack.update.time.time') |
| 109 | + @patch('agentstack.update._is_ci_environment') |
| 110 | + def test_record_update_check(self, mock_is_ci, mock_time, mock_file_path): |
| 111 | + """ |
| 112 | + Test that record_update_check correctly saves the current timestamp. |
| 113 | + """ |
| 114 | + mock_is_ci.return_value = False |
| 115 | + mock_time.return_value = 1234567890.0 |
| 116 | + |
| 117 | + record_update_check() |
| 118 | + |
| 119 | + with open(mock_file_path, 'r') as f: |
| 120 | + saved_data = json.load(f) |
| 121 | + self.assertEqual(saved_data, {str(INSTALL_PATH): 1234567890.0}) |
| 122 | + |
| 123 | + os.remove(mock_file_path) |
| 124 | + mock_file_path.parent.rmdir() |
| 125 | + |
| 126 | + @patch('agentstack.update.Path.exists') |
| 127 | + def test_load_update_data_empty(self, mock_exists): |
| 128 | + """ |
| 129 | + Test that load_update_data returns an empty dict when the file doesn't exist. |
| 130 | + """ |
| 131 | + mock_exists.return_value = False |
| 132 | + data = load_update_data() |
| 133 | + self.assertEqual(data, {}) |
| 134 | + |
| 135 | + @patch('builtins.open', new_callable=mock_open) |
| 136 | + @patch('agentstack.update.Path.exists') |
| 137 | + def test_load_update_data_valid(self, mock_exists, mock_file): |
| 138 | + """ |
| 139 | + Test that load_update_data correctly loads data from a valid file. |
| 140 | + """ |
| 141 | + mock_exists.return_value = True |
| 142 | + test_data = {"test_path": 1234567890.0} |
| 143 | + mock_file.return_value.read.return_value = json.dumps(test_data) |
| 144 | + |
| 145 | + data = load_update_data() |
| 146 | + self.assertEqual(data, test_data) |
| 147 | + |
| 148 | + @patch.dict('os.environ', {}, clear=True) # clear env to remove AGENTSTACK_UPDATE_DISABLE |
| 149 | + @patch('agentstack.update.time.time') |
| 150 | + @patch('agentstack.update._is_ci_environment') |
| 151 | + @patch('agentstack.update.load_update_data') |
| 152 | + def test_should_update(self, mock_load_data, mock_is_ci, mock_time): |
| 153 | + # Test CI environment |
| 154 | + mock_is_ci.return_value = True |
| 155 | + self.assertTrue(should_update()) |
| 156 | + |
| 157 | + # Test first run (no data) |
| 158 | + mock_is_ci.return_value = False |
| 159 | + mock_load_data.return_value = {} |
| 160 | + self.assertTrue(should_update()) |
| 161 | + |
| 162 | + # Test recent check |
| 163 | + mock_time.return_value = 1000000 |
| 164 | + mock_load_data.return_value = {str(INSTALL_PATH): 999999} # 1 second ago |
| 165 | + self.assertFalse(should_update()) |
| 166 | + |
| 167 | + # Test old check |
| 168 | + mock_load_data.return_value = { |
| 169 | + str(INSTALL_PATH): 1000000 - CHECK_EVERY - 1 |
| 170 | + } # CHECK_EVERY + 1 second ago |
| 171 | + self.assertTrue(should_update()) |
0 commit comments