1 # coding: utf-8 2 """Tests for the IPython tab-completion machinery. 3 """ 4 #----------------------------------------------------------------------------- 5 # Module imports 6 #----------------------------------------------------------------------------- 7 8 # stdlib 9 import io 10 import sys 11 import tempfile 12 from datetime import datetime 13 from pathlib import Path 14 15 from tempfile import TemporaryDirectory 16 # our own packages 17 from traitlets.config.loader import Config 18 19 from IPython.core.history import HistoryAccessor, HistoryManager, extract_hist_ranges 20 21 22 def test_proper_default_encoding(): 23 assert sys.getdefaultencoding() == "utf-8" 24 25 def test_history(): 26 ip = get_ipython() 27 with TemporaryDirectory() as tmpdir: 28 tmp_path = Path(tmpdir) 29 hist_manager_ori = ip.history_manager 30 hist_file = tmp_path / "history.sqlite" 31 try: 32 ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file) 33 hist = ["a=1", "def f():\n test = 1\n return test", "b='€Æ¾÷ß'"] 34 for i, h in enumerate(hist, start=1): 35 ip.history_manager.store_inputs(i, h) 36 37 ip.history_manager.db_log_output = True 38 # Doesn't match the input, but we'll just check it's stored. 39 ip.history_manager.output_hist_reprs[3] = "spam" 40 ip.history_manager.store_output(3) 41 42 assert ip.history_manager.input_hist_raw == [""] + hist 43 44 # Detailed tests for _get_range_session 45 grs = ip.history_manager._get_range_session 46 assert list(grs(start=2, stop=-1)) == list(zip([0], [2], hist[1:-1])) 47 assert list(grs(start=-2)) == list(zip([0, 0], [2, 3], hist[-2:])) 48 assert list(grs(output=True)) == list( 49 zip([0, 0, 0], [1, 2, 3], zip(hist, [None, None, "spam"])) 50 ) 51 52 # Check whether specifying a range beyond the end of the current 53 # session results in an error (gh-804) 54 ip.run_line_magic("hist", "2-500") 55 56 # Check that we can write non-ascii characters to a file 57 ip.run_line_magic("hist", "-f %s" % (tmp_path / "test1")) 58 ip.run_line_magic("hist", "-pf %s" % (tmp_path / "test2")) 59 ip.run_line_magic("hist", "-nf %s" % (tmp_path / "test3")) 60 ip.run_line_magic("save", "%s 1-10" % (tmp_path / "test4")) 61 62 # New session 63 ip.history_manager.reset() 64 newcmds = ["z=5", "class X(object):\n pass", "k='p'", "z=5"] 65 for i, cmd in enumerate(newcmds, start=1): 66 ip.history_manager.store_inputs(i, cmd) 67 gothist = ip.history_manager.get_range(start=1, stop=4) 68 assert list(gothist) == list(zip([0, 0, 0], [1, 2, 3], newcmds)) 69 # Previous session: 70 gothist = ip.history_manager.get_range(-1, 1, 4) 71 assert list(gothist) == list(zip([1, 1, 1], [1, 2, 3], hist)) 72 73 newhist = [(2, i, c) for (i, c) in enumerate(newcmds, 1)] 74 75 # Check get_hist_tail 76 gothist = ip.history_manager.get_tail(5, output=True, 77 include_latest=True) 78 expected = [(1, 3, (hist[-1], "spam"))] \ 79 + [(s, n, (c, None)) for (s, n, c) in newhist] 80 assert list(gothist) == expected 81 82 gothist = ip.history_manager.get_tail(2) 83 expected = newhist[-3:-1] 84 assert list(gothist) == expected 85 86 # Check get_hist_search 87 88 gothist = ip.history_manager.search("*test*") 89 assert list(gothist) == [(1, 2, hist[1])] 90 91 gothist = ip.history_manager.search("*=*") 92 assert list(gothist) == [ 93 (1, 1, hist[0]), 94 (1, 2, hist[1]), 95 (1, 3, hist[2]), 96 newhist[0], 97 newhist[2], 98 newhist[3], 99 ] 100 101 gothist = ip.history_manager.search("*=*", n=4) 102 assert list(gothist) == [ 103 (1, 3, hist[2]), 104 newhist[0], 105 newhist[2], 106 newhist[3], 107 ] 108 109 gothist = ip.history_manager.search("*=*", unique=True) 110 assert list(gothist) == [ 111 (1, 1, hist[0]), 112 (1, 2, hist[1]), 113 (1, 3, hist[2]), 114 newhist[2], 115 newhist[3], 116 ] 117 118 gothist = ip.history_manager.search("*=*", unique=True, n=3) 119 assert list(gothist) == [(1, 3, hist[2]), newhist[2], newhist[3]] 120 121 gothist = ip.history_manager.search("b*", output=True) 122 assert list(gothist) == [(1, 3, (hist[2], "spam"))] 123 124 # Cross testing: check that magic %save can get previous session. 125 testfilename = (tmp_path / "test.py").resolve() 126 ip.run_line_magic("save", str(testfilename) + " ~1/1-3") 127 with io.open(testfilename, encoding="utf-8") as testfile: 128 assert testfile.read() == "# coding: utf-8\n" + "\n".join(hist) + "\n" 129 130 # Duplicate line numbers - check that it doesn't crash, and 131 # gets a new session 132 ip.history_manager.store_inputs(1, "rogue") 133 ip.history_manager.writeout_cache() 134 assert ip.history_manager.session_number == 3 135 136 # Check that session and line values are not just max values 137 sessid, lineno, entry = newhist[-1] 138 assert lineno > 1 139 ip.history_manager.reset() 140 lineno = 1 141 ip.history_manager.store_inputs(lineno, entry) 142 gothist = ip.history_manager.search("*=*", unique=True) 143 hist = list(gothist)[-1] 144 assert sessid < hist[0] 145 assert hist[1:] == (lineno, entry) 146 finally: 147 # Ensure saving thread is shut down before we try to clean up the files 148 ip.history_manager.save_thread.stop() 149 # Forcibly close database rather than relying on garbage collection 150 ip.history_manager.db.close() 151 # Restore history manager 152 ip.history_manager = hist_manager_ori 153 154 155 def test_extract_hist_ranges(): 156 instr = "1 2/3 ~4/5-6 ~4/7-~4/9 ~9/2-~7/5 ~10/" 157 expected = [(0, 1, 2), # 0 == current session 158 (2, 3, 4), 159 (-4, 5, 7), 160 (-4, 7, 10), 161 (-9, 2, None), # None == to end 162 (-8, 1, None), 163 (-7, 1, 6), 164 (-10, 1, None)] 165 actual = list(extract_hist_ranges(instr)) 166 assert actual == expected 167 168 169 def test_extract_hist_ranges_empty_str(): 170 instr = "" 171 expected = [(0, 1, None)] # 0 == current session, None == to end 172 actual = list(extract_hist_ranges(instr)) 173 assert actual == expected 174 175 176 def test_magic_rerun(): 177 """Simple test for %rerun (no args -> rerun last line)""" 178 ip = get_ipython() 179 ip.run_cell("a = 10", store_history=True) 180 ip.run_cell("a += 1", store_history=True) 181 assert ip.user_ns["a"] == 11 182 ip.run_cell("%rerun", store_history=True) 183 assert ip.user_ns["a"] == 12 184 185 def test_timestamp_type(): 186 ip = get_ipython() 187 info = ip.history_manager.get_session_info() 188 assert isinstance(info[1], datetime) 189 190 def test_hist_file_config(): 191 cfg = Config() 192 tfile = tempfile.NamedTemporaryFile(delete=False) 193 cfg.HistoryManager.hist_file = Path(tfile.name) 194 try: 195 hm = HistoryManager(shell=get_ipython(), config=cfg) 196 assert hm.hist_file == cfg.HistoryManager.hist_file 197 finally: 198 try: 199 Path(tfile.name).unlink() 200 except OSError: 201 # same catch as in testing.tools.TempFileMixin 202 # On Windows, even though we close the file, we still can't 203 # delete it. I have no clue why 204 pass 205 206 def test_histmanager_disabled(): 207 """Ensure that disabling the history manager doesn't create a database.""" 208 cfg = Config() 209 cfg.HistoryAccessor.enabled = False 210 211 ip = get_ipython() 212 with TemporaryDirectory() as tmpdir: 213 hist_manager_ori = ip.history_manager 214 hist_file = Path(tmpdir) / "history.sqlite" 215 cfg.HistoryManager.hist_file = hist_file 216 try: 217 ip.history_manager = HistoryManager(shell=ip, config=cfg) 218 hist = ["a=1", "def f():\n test = 1\n return test", "b='€Æ¾÷ß'"] 219 for i, h in enumerate(hist, start=1): 220 ip.history_manager.store_inputs(i, h) 221 assert ip.history_manager.input_hist_raw == [""] + hist 222 ip.history_manager.reset() 223 ip.history_manager.end_session() 224 finally: 225 ip.history_manager = hist_manager_ori 226 227 # hist_file should not be created 228 assert hist_file.exists() is False 229 230 231 def test_get_tail_session_awareness(): 232 """Test .get_tail() is: 233 - session specific in HistoryManager 234 - session agnostic in HistoryAccessor 235 same for .get_last_session_id() 236 """ 237 ip = get_ipython() 238 with TemporaryDirectory() as tmpdir: 239 tmp_path = Path(tmpdir) 240 hist_file = tmp_path / "history.sqlite" 241 get_source = lambda x: x[2] 242 hm1 = None 243 hm2 = None 244 ha = None 245 try: 246 # hm1 creates a new session and adds history entries, 247 # ha catches up 248 hm1 = HistoryManager(shell=ip, hist_file=hist_file) 249 hm1_last_sid = hm1.get_last_session_id 250 ha = HistoryAccessor(hist_file=hist_file) 251 ha_last_sid = ha.get_last_session_id 252 253 hist1 = ["a=1", "b=1", "c=1"] 254 for i, h in enumerate(hist1 + [""], start=1): 255 hm1.store_inputs(i, h) 256 assert list(map(get_source, hm1.get_tail())) == hist1 257 assert list(map(get_source, ha.get_tail())) == hist1 258 sid1 = hm1_last_sid() 259 assert sid1 is not None 260 assert ha_last_sid() == sid1 261 262 # hm2 creates a new session and adds entries, 263 # ha catches up 264 hm2 = HistoryManager(shell=ip, hist_file=hist_file) 265 hm2_last_sid = hm2.get_last_session_id 266 267 hist2 = ["a=2", "b=2", "c=2"] 268 for i, h in enumerate(hist2 + [""], start=1): 269 hm2.store_inputs(i, h) 270 tail = hm2.get_tail(n=3) 271 assert list(map(get_source, tail)) == hist2 272 tail = ha.get_tail(n=3) 273 assert list(map(get_source, tail)) == hist2 274 sid2 = hm2_last_sid() 275 assert sid2 is not None 276 assert ha_last_sid() == sid2 277 assert sid2 != sid1 278 279 # but hm1 still maintains its point of reference 280 # and adding more entries to it doesn't change others 281 # immediate perspective 282 assert hm1_last_sid() == sid1 283 tail = hm1.get_tail(n=3) 284 assert list(map(get_source, tail)) == hist1 285 286 hist3 = ["a=3", "b=3", "c=3"] 287 for i, h in enumerate(hist3 + [""], start=5): 288 hm1.store_inputs(i, h) 289 tail = hm1.get_tail(n=7) 290 assert list(map(get_source, tail)) == hist1 + [""] + hist3 291 tail = hm2.get_tail(n=3) 292 assert list(map(get_source, tail)) == hist2 293 tail = ha.get_tail(n=3) 294 assert list(map(get_source, tail)) == hist2 295 assert hm1_last_sid() == sid1 296 assert hm2_last_sid() == sid2 297 assert ha_last_sid() == sid2 298 finally: 299 if hm1: 300 hm1.save_thread.stop() 301 hm1.db.close() 302 if hm2: 303 hm2.save_thread.stop() 304 hm2.db.close() 305 if ha: 306 ha.db.close() 307