##// END OF EJS Templates
Merge pull request #13049 from MrMino/empty_histrange_means_all...
Matthias Bussonnier -
r26671:be4887fe merge
parent child Browse files
Show More

The requested changes are too big and content was truncated. Show full diff

@@ -0,0 +1,19 b''
1 Empty History Ranges
2 ====================
3
4 A number of magics that take history ranges can now be used with an empty
5 range. These magics are:
6
7 * ``%save``
8 * ``%load``
9 * ``%pastebin``
10 * ``%pycat``
11
12 Using them this way will make them take the history of the current session up
13 to the point of the magic call (such that the magic itself will not be
14 included).
15
16 Therefore it is now possible to save the whole history to a file using simple
17 ``%save <filename>``, load and edit it using ``%load`` (makes for a nice usage
18 when followed with :kbd:`F2`), send it to dpaste.org using ``%pastebin``, or
19 view the whole thing syntax-highlighted with a single ``%pycat``.
@@ -1,897 +1,907 b''
1 1 """ History related magics and functionality """
2 2
3 3 # Copyright (c) IPython Development Team.
4 4 # Distributed under the terms of the Modified BSD License.
5 5
6 6
7 7 import atexit
8 8 import datetime
9 9 from pathlib import Path
10 10 import re
11 11 import sqlite3
12 12 import threading
13 13
14 14 from traitlets.config.configurable import LoggingConfigurable
15 15 from decorator import decorator
16 16 from IPython.utils.decorators import undoc
17 17 from IPython.paths import locate_profile
18 18 from traitlets import (
19 19 Any,
20 20 Bool,
21 21 Dict,
22 22 Instance,
23 23 Integer,
24 24 List,
25 25 Unicode,
26 26 Union,
27 27 TraitError,
28 28 default,
29 29 observe,
30 30 )
31 31
32 32 #-----------------------------------------------------------------------------
33 33 # Classes and functions
34 34 #-----------------------------------------------------------------------------
35 35
36 36 @undoc
37 37 class DummyDB(object):
38 38 """Dummy DB that will act as a black hole for history.
39 39
40 40 Only used in the absence of sqlite"""
41 41 def execute(*args, **kwargs):
42 42 return []
43 43
44 44 def commit(self, *args, **kwargs):
45 45 pass
46 46
47 47 def __enter__(self, *args, **kwargs):
48 48 pass
49 49
50 50 def __exit__(self, *args, **kwargs):
51 51 pass
52 52
53 53
54 54 @decorator
55 55 def only_when_enabled(f, self, *a, **kw):
56 56 """Decorator: return an empty list in the absence of sqlite."""
57 57 if not self.enabled:
58 58 return []
59 59 else:
60 60 return f(self, *a, **kw)
61 61
62 62
63 63 # use 16kB as threshold for whether a corrupt history db should be saved
64 64 # that should be at least 100 entries or so
65 65 _SAVE_DB_SIZE = 16384
66 66
67 67 @decorator
68 68 def catch_corrupt_db(f, self, *a, **kw):
69 69 """A decorator which wraps HistoryAccessor method calls to catch errors from
70 70 a corrupt SQLite database, move the old database out of the way, and create
71 71 a new one.
72 72
73 73 We avoid clobbering larger databases because this may be triggered due to filesystem issues,
74 74 not just a corrupt file.
75 75 """
76 76 try:
77 77 return f(self, *a, **kw)
78 78 except (sqlite3.DatabaseError, sqlite3.OperationalError) as e:
79 79 self._corrupt_db_counter += 1
80 80 self.log.error("Failed to open SQLite history %s (%s).", self.hist_file, e)
81 81 if self.hist_file != ':memory:':
82 82 if self._corrupt_db_counter > self._corrupt_db_limit:
83 83 self.hist_file = ':memory:'
84 84 self.log.error("Failed to load history too many times, history will not be saved.")
85 85 elif self.hist_file.is_file():
86 86 # move the file out of the way
87 87 base = str(self.hist_file.parent / self.hist_file.stem)
88 88 ext = self.hist_file.suffix
89 89 size = self.hist_file.stat().st_size
90 90 if size >= _SAVE_DB_SIZE:
91 91 # if there's significant content, avoid clobbering
92 92 now = datetime.datetime.now().isoformat().replace(':', '.')
93 93 newpath = base + '-corrupt-' + now + ext
94 94 # don't clobber previous corrupt backups
95 95 for i in range(100):
96 96 if not Path(newpath).exists():
97 97 break
98 98 else:
99 99 newpath = base + '-corrupt-' + now + (u'-%i' % i) + ext
100 100 else:
101 101 # not much content, possibly empty; don't worry about clobbering
102 102 # maybe we should just delete it?
103 103 newpath = base + '-corrupt' + ext
104 104 self.hist_file.rename(newpath)
105 105 self.log.error("History file was moved to %s and a new file created.", newpath)
106 106 self.init_db()
107 107 return []
108 108 else:
109 109 # Failed with :memory:, something serious is wrong
110 110 raise
111 111
112 112
113 113 class HistoryAccessorBase(LoggingConfigurable):
114 114 """An abstract class for History Accessors """
115 115
116 116 def get_tail(self, n=10, raw=True, output=False, include_latest=False):
117 117 raise NotImplementedError
118 118
119 119 def search(self, pattern="*", raw=True, search_raw=True,
120 120 output=False, n=None, unique=False):
121 121 raise NotImplementedError
122 122
123 123 def get_range(self, session, start=1, stop=None, raw=True,output=False):
124 124 raise NotImplementedError
125 125
126 126 def get_range_by_str(self, rangestr, raw=True, output=False):
127 127 raise NotImplementedError
128 128
129 129
130 130 class HistoryAccessor(HistoryAccessorBase):
131 131 """Access the history database without adding to it.
132 132
133 133 This is intended for use by standalone history tools. IPython shells use
134 134 HistoryManager, below, which is a subclass of this."""
135 135
136 136 # counter for init_db retries, so we don't keep trying over and over
137 137 _corrupt_db_counter = 0
138 138 # after two failures, fallback on :memory:
139 139 _corrupt_db_limit = 2
140 140
141 141 # String holding the path to the history file
142 142 hist_file = Union(
143 143 [Instance(Path), Unicode()],
144 144 help="""Path to file to use for SQLite history database.
145 145
146 146 By default, IPython will put the history database in the IPython
147 147 profile directory. If you would rather share one history among
148 148 profiles, you can set this value in each, so that they are consistent.
149 149
150 150 Due to an issue with fcntl, SQLite is known to misbehave on some NFS
151 151 mounts. If you see IPython hanging, try setting this to something on a
152 152 local disk, e.g::
153 153
154 154 ipython --HistoryManager.hist_file=/tmp/ipython_hist.sqlite
155 155
156 156 you can also use the specific value `:memory:` (including the colon
157 157 at both end but not the back ticks), to avoid creating an history file.
158 158
159 159 """,
160 160 ).tag(config=True)
161 161
162 162 enabled = Bool(True,
163 163 help="""enable the SQLite history
164 164
165 165 set enabled=False to disable the SQLite history,
166 166 in which case there will be no stored history, no SQLite connection,
167 167 and no background saving thread. This may be necessary in some
168 168 threaded environments where IPython is embedded.
169 169 """
170 170 ).tag(config=True)
171 171
172 172 connection_options = Dict(
173 173 help="""Options for configuring the SQLite connection
174 174
175 175 These options are passed as keyword args to sqlite3.connect
176 176 when establishing database connections.
177 177 """
178 178 ).tag(config=True)
179 179
180 180 # The SQLite database
181 181 db = Any()
182 182 @observe('db')
183 183 def _db_changed(self, change):
184 184 """validate the db, since it can be an Instance of two different types"""
185 185 new = change['new']
186 186 connection_types = (DummyDB, sqlite3.Connection)
187 187 if not isinstance(new, connection_types):
188 188 msg = "%s.db must be sqlite3 Connection or DummyDB, not %r" % \
189 189 (self.__class__.__name__, new)
190 190 raise TraitError(msg)
191 191
192 192 def __init__(self, profile="default", hist_file="", **traits):
193 193 """Create a new history accessor.
194 194
195 195 Parameters
196 196 ----------
197 197 profile : str
198 198 The name of the profile from which to open history.
199 199 hist_file : str
200 200 Path to an SQLite history database stored by IPython. If specified,
201 201 hist_file overrides profile.
202 202 config : :class:`~traitlets.config.loader.Config`
203 203 Config object. hist_file can also be set through this.
204 204 """
205 205 # We need a pointer back to the shell for various tasks.
206 206 super(HistoryAccessor, self).__init__(**traits)
207 207 # defer setting hist_file from kwarg until after init,
208 208 # otherwise the default kwarg value would clobber any value
209 209 # set by config
210 210 if hist_file:
211 211 self.hist_file = hist_file
212 212
213 213 try:
214 214 self.hist_file
215 215 except TraitError:
216 216 # No one has set the hist_file, yet.
217 217 self.hist_file = self._get_hist_file_name(profile)
218 218
219 219 self.init_db()
220 220
221 221 def _get_hist_file_name(self, profile='default'):
222 222 """Find the history file for the given profile name.
223 223
224 224 This is overridden by the HistoryManager subclass, to use the shell's
225 225 active profile.
226 226
227 227 Parameters
228 228 ----------
229 229 profile : str
230 230 The name of a profile which has a history file.
231 231 """
232 232 return Path(locate_profile(profile)) / "history.sqlite"
233 233
234 234 @catch_corrupt_db
235 235 def init_db(self):
236 236 """Connect to the database, and create tables if necessary."""
237 237 if not self.enabled:
238 238 self.db = DummyDB()
239 239 return
240 240
241 241 # use detect_types so that timestamps return datetime objects
242 242 kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
243 243 kwargs.update(self.connection_options)
244 244 self.db = sqlite3.connect(str(self.hist_file), **kwargs)
245 245 self.db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
246 246 primary key autoincrement, start timestamp,
247 247 end timestamp, num_cmds integer, remark text)""")
248 248 self.db.execute("""CREATE TABLE IF NOT EXISTS history
249 249 (session integer, line integer, source text, source_raw text,
250 250 PRIMARY KEY (session, line))""")
251 251 # Output history is optional, but ensure the table's there so it can be
252 252 # enabled later.
253 253 self.db.execute("""CREATE TABLE IF NOT EXISTS output_history
254 254 (session integer, line integer, output text,
255 255 PRIMARY KEY (session, line))""")
256 256 self.db.commit()
257 257 # success! reset corrupt db count
258 258 self._corrupt_db_counter = 0
259 259
260 260 def writeout_cache(self):
261 261 """Overridden by HistoryManager to dump the cache before certain
262 262 database lookups."""
263 263 pass
264 264
265 265 ## -------------------------------
266 266 ## Methods for retrieving history:
267 267 ## -------------------------------
268 268 def _run_sql(self, sql, params, raw=True, output=False):
269 269 """Prepares and runs an SQL query for the history database.
270 270
271 271 Parameters
272 272 ----------
273 273 sql : str
274 274 Any filtering expressions to go after SELECT ... FROM ...
275 275 params : tuple
276 276 Parameters passed to the SQL query (to replace "?")
277 277 raw, output : bool
278 278 See :meth:`get_range`
279 279
280 280 Returns
281 281 -------
282 282 Tuples as :meth:`get_range`
283 283 """
284 284 toget = 'source_raw' if raw else 'source'
285 285 sqlfrom = "history"
286 286 if output:
287 287 sqlfrom = "history LEFT JOIN output_history USING (session, line)"
288 288 toget = "history.%s, output_history.output" % toget
289 289 cur = self.db.execute("SELECT session, line, %s FROM %s " %\
290 290 (toget, sqlfrom) + sql, params)
291 291 if output: # Regroup into 3-tuples, and parse JSON
292 292 return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
293 293 return cur
294 294
295 295 @only_when_enabled
296 296 @catch_corrupt_db
297 297 def get_session_info(self, session):
298 298 """Get info about a session.
299 299
300 300 Parameters
301 301 ----------
302 302
303 303 session : int
304 304 Session number to retrieve.
305 305
306 306 Returns
307 307 -------
308 308
309 309 session_id : int
310 310 Session ID number
311 311 start : datetime
312 312 Timestamp for the start of the session.
313 313 end : datetime
314 314 Timestamp for the end of the session, or None if IPython crashed.
315 315 num_cmds : int
316 316 Number of commands run, or None if IPython crashed.
317 317 remark : unicode
318 318 A manually set description.
319 319 """
320 320 query = "SELECT * from sessions where session == ?"
321 321 return self.db.execute(query, (session,)).fetchone()
322 322
323 323 @catch_corrupt_db
324 324 def get_last_session_id(self):
325 325 """Get the last session ID currently in the database.
326 326
327 327 Within IPython, this should be the same as the value stored in
328 328 :attr:`HistoryManager.session_number`.
329 329 """
330 330 for record in self.get_tail(n=1, include_latest=True):
331 331 return record[0]
332 332
333 333 @catch_corrupt_db
334 334 def get_tail(self, n=10, raw=True, output=False, include_latest=False):
335 335 """Get the last n lines from the history database.
336 336
337 337 Parameters
338 338 ----------
339 339 n : int
340 340 The number of lines to get
341 341 raw, output : bool
342 342 See :meth:`get_range`
343 343 include_latest : bool
344 344 If False (default), n+1 lines are fetched, and the latest one
345 345 is discarded. This is intended to be used where the function
346 346 is called by a user command, which it should not return.
347 347
348 348 Returns
349 349 -------
350 350 Tuples as :meth:`get_range`
351 351 """
352 352 self.writeout_cache()
353 353 if not include_latest:
354 354 n += 1
355 355 cur = self._run_sql("ORDER BY session DESC, line DESC LIMIT ?",
356 356 (n,), raw=raw, output=output)
357 357 if not include_latest:
358 358 return reversed(list(cur)[1:])
359 359 return reversed(list(cur))
360 360
361 361 @catch_corrupt_db
362 362 def search(self, pattern="*", raw=True, search_raw=True,
363 363 output=False, n=None, unique=False):
364 364 """Search the database using unix glob-style matching (wildcards
365 365 * and ?).
366 366
367 367 Parameters
368 368 ----------
369 369 pattern : str
370 370 The wildcarded pattern to match when searching
371 371 search_raw : bool
372 372 If True, search the raw input, otherwise, the parsed input
373 373 raw, output : bool
374 374 See :meth:`get_range`
375 375 n : None or int
376 376 If an integer is given, it defines the limit of
377 377 returned entries.
378 378 unique : bool
379 379 When it is true, return only unique entries.
380 380
381 381 Returns
382 382 -------
383 383 Tuples as :meth:`get_range`
384 384 """
385 385 tosearch = "source_raw" if search_raw else "source"
386 386 if output:
387 387 tosearch = "history." + tosearch
388 388 self.writeout_cache()
389 389 sqlform = "WHERE %s GLOB ?" % tosearch
390 390 params = (pattern,)
391 391 if unique:
392 392 sqlform += ' GROUP BY {0}'.format(tosearch)
393 393 if n is not None:
394 394 sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
395 395 params += (n,)
396 396 elif unique:
397 397 sqlform += " ORDER BY session, line"
398 398 cur = self._run_sql(sqlform, params, raw=raw, output=output)
399 399 if n is not None:
400 400 return reversed(list(cur))
401 401 return cur
402 402
403 403 @catch_corrupt_db
404 404 def get_range(self, session, start=1, stop=None, raw=True,output=False):
405 405 """Retrieve input by session.
406 406
407 407 Parameters
408 408 ----------
409 409 session : int
410 410 Session number to retrieve.
411 411 start : int
412 412 First line to retrieve.
413 413 stop : int
414 414 End of line range (excluded from output itself). If None, retrieve
415 415 to the end of the session.
416 416 raw : bool
417 417 If True, return untranslated input
418 418 output : bool
419 419 If True, attempt to include output. This will be 'real' Python
420 420 objects for the current session, or text reprs from previous
421 421 sessions if db_log_output was enabled at the time. Where no output
422 422 is found, None is used.
423 423
424 424 Returns
425 425 -------
426 426 entries
427 427 An iterator over the desired lines. Each line is a 3-tuple, either
428 428 (session, line, input) if output is False, or
429 429 (session, line, (input, output)) if output is True.
430 430 """
431 431 if stop:
432 432 lineclause = "line >= ? AND line < ?"
433 433 params = (session, start, stop)
434 434 else:
435 435 lineclause = "line>=?"
436 436 params = (session, start)
437 437
438 438 return self._run_sql("WHERE session==? AND %s" % lineclause,
439 439 params, raw=raw, output=output)
440 440
441 441 def get_range_by_str(self, rangestr, raw=True, output=False):
442 442 """Get lines of history from a string of ranges, as used by magic
443 443 commands %hist, %save, %macro, etc.
444 444
445 445 Parameters
446 446 ----------
447 447 rangestr : str
448 A string specifying ranges, e.g. "5 ~2/1-4". See
449 :func:`magic_history` for full details.
448 A string specifying ranges, e.g. "5 ~2/1-4". If empty string is used,
449 this will return everything from current session's history.
450
451 See the documentation of :func:`%history` for the full details.
452
450 453 raw, output : bool
451 454 As :meth:`get_range`
452 455
453 456 Returns
454 457 -------
455 458 Tuples as :meth:`get_range`
456 459 """
457 460 for sess, s, e in extract_hist_ranges(rangestr):
458 461 for line in self.get_range(sess, s, e, raw=raw, output=output):
459 462 yield line
460 463
461 464
462 465 class HistoryManager(HistoryAccessor):
463 466 """A class to organize all history-related functionality in one place.
464 467 """
465 468 # Public interface
466 469
467 470 # An instance of the IPython shell we are attached to
468 471 shell = Instance('IPython.core.interactiveshell.InteractiveShellABC',
469 472 allow_none=True)
470 473 # Lists to hold processed and raw history. These start with a blank entry
471 474 # so that we can index them starting from 1
472 475 input_hist_parsed = List([""])
473 476 input_hist_raw = List([""])
474 477 # A list of directories visited during session
475 478 dir_hist = List()
476 479 @default('dir_hist')
477 480 def _dir_hist_default(self):
478 481 try:
479 482 return [Path.cwd()]
480 483 except OSError:
481 484 return []
482 485
483 486 # A dict of output history, keyed with ints from the shell's
484 487 # execution count.
485 488 output_hist = Dict()
486 489 # The text/plain repr of outputs.
487 490 output_hist_reprs = Dict()
488 491
489 492 # The number of the current session in the history database
490 493 session_number = Integer()
491 494
492 495 db_log_output = Bool(False,
493 496 help="Should the history database include output? (default: no)"
494 497 ).tag(config=True)
495 498 db_cache_size = Integer(0,
496 499 help="Write to database every x commands (higher values save disk access & power).\n"
497 500 "Values of 1 or less effectively disable caching."
498 501 ).tag(config=True)
499 502 # The input and output caches
500 503 db_input_cache = List()
501 504 db_output_cache = List()
502 505
503 506 # History saving in separate thread
504 507 save_thread = Instance('IPython.core.history.HistorySavingThread',
505 508 allow_none=True)
506 509 save_flag = Instance(threading.Event, allow_none=True)
507 510
508 511 # Private interface
509 512 # Variables used to store the three last inputs from the user. On each new
510 513 # history update, we populate the user's namespace with these, shifted as
511 514 # necessary.
512 515 _i00 = Unicode(u'')
513 516 _i = Unicode(u'')
514 517 _ii = Unicode(u'')
515 518 _iii = Unicode(u'')
516 519
517 520 # A regex matching all forms of the exit command, so that we don't store
518 521 # them in the history (it's annoying to rewind the first entry and land on
519 522 # an exit call).
520 523 _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
521 524
522 525 def __init__(self, shell=None, config=None, **traits):
523 526 """Create a new history manager associated with a shell instance.
524 527 """
525 528 # We need a pointer back to the shell for various tasks.
526 529 super(HistoryManager, self).__init__(shell=shell, config=config,
527 530 **traits)
528 531 self.save_flag = threading.Event()
529 532 self.db_input_cache_lock = threading.Lock()
530 533 self.db_output_cache_lock = threading.Lock()
531 534
532 535 try:
533 536 self.new_session()
534 537 except sqlite3.OperationalError:
535 538 self.log.error("Failed to create history session in %s. History will not be saved.",
536 539 self.hist_file, exc_info=True)
537 540 self.hist_file = ':memory:'
538 541
539 542 if self.enabled and self.hist_file != ':memory:':
540 543 self.save_thread = HistorySavingThread(self)
541 544 self.save_thread.start()
542 545
543 546 def _get_hist_file_name(self, profile=None):
544 547 """Get default history file name based on the Shell's profile.
545 548
546 549 The profile parameter is ignored, but must exist for compatibility with
547 550 the parent class."""
548 551 profile_dir = self.shell.profile_dir.location
549 552 return Path(profile_dir) / "history.sqlite"
550 553
551 554 @only_when_enabled
552 555 def new_session(self, conn=None):
553 556 """Get a new session number."""
554 557 if conn is None:
555 558 conn = self.db
556 559
557 560 with conn:
558 561 cur = conn.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
559 562 NULL, "") """, (datetime.datetime.now(),))
560 563 self.session_number = cur.lastrowid
561 564
562 565 def end_session(self):
563 566 """Close the database session, filling in the end time and line count."""
564 567 self.writeout_cache()
565 568 with self.db:
566 569 self.db.execute("""UPDATE sessions SET end=?, num_cmds=? WHERE
567 570 session==?""", (datetime.datetime.now(),
568 571 len(self.input_hist_parsed)-1, self.session_number))
569 572 self.session_number = 0
570 573
571 574 def name_session(self, name):
572 575 """Give the current session a name in the history database."""
573 576 with self.db:
574 577 self.db.execute("UPDATE sessions SET remark=? WHERE session==?",
575 578 (name, self.session_number))
576 579
577 580 def reset(self, new_session=True):
578 581 """Clear the session history, releasing all object references, and
579 582 optionally open a new session."""
580 583 self.output_hist.clear()
581 584 # The directory history can't be completely empty
582 585 self.dir_hist[:] = [Path.cwd()]
583 586
584 587 if new_session:
585 588 if self.session_number:
586 589 self.end_session()
587 590 self.input_hist_parsed[:] = [""]
588 591 self.input_hist_raw[:] = [""]
589 592 self.new_session()
590 593
591 594 # ------------------------------
592 595 # Methods for retrieving history
593 596 # ------------------------------
594 597 def get_session_info(self, session=0):
595 598 """Get info about a session.
596 599
597 600 Parameters
598 601 ----------
599 602
600 603 session : int
601 604 Session number to retrieve. The current session is 0, and negative
602 605 numbers count back from current session, so -1 is the previous session.
603 606
604 607 Returns
605 608 -------
606 609
607 610 session_id : int
608 611 Session ID number
609 612 start : datetime
610 613 Timestamp for the start of the session.
611 614 end : datetime
612 615 Timestamp for the end of the session, or None if IPython crashed.
613 616 num_cmds : int
614 617 Number of commands run, or None if IPython crashed.
615 618 remark : unicode
616 619 A manually set description.
617 620 """
618 621 if session <= 0:
619 622 session += self.session_number
620 623
621 624 return super(HistoryManager, self).get_session_info(session=session)
622 625
623 626 def _get_range_session(self, start=1, stop=None, raw=True, output=False):
624 627 """Get input and output history from the current session. Called by
625 628 get_range, and takes similar parameters."""
626 629 input_hist = self.input_hist_raw if raw else self.input_hist_parsed
627 630
628 631 n = len(input_hist)
629 632 if start < 0:
630 633 start += n
631 634 if not stop or (stop > n):
632 635 stop = n
633 636 elif stop < 0:
634 637 stop += n
635 638
636 639 for i in range(start, stop):
637 640 if output:
638 641 line = (input_hist[i], self.output_hist_reprs.get(i))
639 642 else:
640 643 line = input_hist[i]
641 644 yield (0, i, line)
642 645
643 646 def get_range(self, session=0, start=1, stop=None, raw=True,output=False):
644 647 """Retrieve input by session.
645 648
646 649 Parameters
647 650 ----------
648 651 session : int
649 652 Session number to retrieve. The current session is 0, and negative
650 653 numbers count back from current session, so -1 is previous session.
651 654 start : int
652 655 First line to retrieve.
653 656 stop : int
654 657 End of line range (excluded from output itself). If None, retrieve
655 658 to the end of the session.
656 659 raw : bool
657 660 If True, return untranslated input
658 661 output : bool
659 662 If True, attempt to include output. This will be 'real' Python
660 663 objects for the current session, or text reprs from previous
661 664 sessions if db_log_output was enabled at the time. Where no output
662 665 is found, None is used.
663 666
664 667 Returns
665 668 -------
666 669 entries
667 670 An iterator over the desired lines. Each line is a 3-tuple, either
668 671 (session, line, input) if output is False, or
669 672 (session, line, (input, output)) if output is True.
670 673 """
671 674 if session <= 0:
672 675 session += self.session_number
673 676 if session==self.session_number: # Current session
674 677 return self._get_range_session(start, stop, raw, output)
675 678 return super(HistoryManager, self).get_range(session, start, stop, raw,
676 679 output)
677 680
678 681 ## ----------------------------
679 682 ## Methods for storing history:
680 683 ## ----------------------------
681 684 def store_inputs(self, line_num, source, source_raw=None):
682 685 """Store source and raw input in history and create input cache
683 686 variables ``_i*``.
684 687
685 688 Parameters
686 689 ----------
687 690 line_num : int
688 691 The prompt number of this input.
689 692
690 693 source : str
691 694 Python input.
692 695
693 696 source_raw : str, optional
694 697 If given, this is the raw input without any IPython transformations
695 698 applied to it. If not given, ``source`` is used.
696 699 """
697 700 if source_raw is None:
698 701 source_raw = source
699 702 source = source.rstrip('\n')
700 703 source_raw = source_raw.rstrip('\n')
701 704
702 705 # do not store exit/quit commands
703 706 if self._exit_re.match(source_raw.strip()):
704 707 return
705 708
706 709 self.input_hist_parsed.append(source)
707 710 self.input_hist_raw.append(source_raw)
708 711
709 712 with self.db_input_cache_lock:
710 713 self.db_input_cache.append((line_num, source, source_raw))
711 714 # Trigger to flush cache and write to DB.
712 715 if len(self.db_input_cache) >= self.db_cache_size:
713 716 self.save_flag.set()
714 717
715 718 # update the auto _i variables
716 719 self._iii = self._ii
717 720 self._ii = self._i
718 721 self._i = self._i00
719 722 self._i00 = source_raw
720 723
721 724 # hackish access to user namespace to create _i1,_i2... dynamically
722 725 new_i = '_i%s' % line_num
723 726 to_main = {'_i': self._i,
724 727 '_ii': self._ii,
725 728 '_iii': self._iii,
726 729 new_i : self._i00 }
727 730
728 731 if self.shell is not None:
729 732 self.shell.push(to_main, interactive=False)
730 733
731 734 def store_output(self, line_num):
732 735 """If database output logging is enabled, this saves all the
733 736 outputs from the indicated prompt number to the database. It's
734 737 called by run_cell after code has been executed.
735 738
736 739 Parameters
737 740 ----------
738 741 line_num : int
739 742 The line number from which to save outputs
740 743 """
741 744 if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
742 745 return
743 746 output = self.output_hist_reprs[line_num]
744 747
745 748 with self.db_output_cache_lock:
746 749 self.db_output_cache.append((line_num, output))
747 750 if self.db_cache_size <= 1:
748 751 self.save_flag.set()
749 752
750 753 def _writeout_input_cache(self, conn):
751 754 with conn:
752 755 for line in self.db_input_cache:
753 756 conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
754 757 (self.session_number,)+line)
755 758
756 759 def _writeout_output_cache(self, conn):
757 760 with conn:
758 761 for line in self.db_output_cache:
759 762 conn.execute("INSERT INTO output_history VALUES (?, ?, ?)",
760 763 (self.session_number,)+line)
761 764
762 765 @only_when_enabled
763 766 def writeout_cache(self, conn=None):
764 767 """Write any entries in the cache to the database."""
765 768 if conn is None:
766 769 conn = self.db
767 770
768 771 with self.db_input_cache_lock:
769 772 try:
770 773 self._writeout_input_cache(conn)
771 774 except sqlite3.IntegrityError:
772 775 self.new_session(conn)
773 776 print("ERROR! Session/line number was not unique in",
774 777 "database. History logging moved to new session",
775 778 self.session_number)
776 779 try:
777 780 # Try writing to the new session. If this fails, don't
778 781 # recurse
779 782 self._writeout_input_cache(conn)
780 783 except sqlite3.IntegrityError:
781 784 pass
782 785 finally:
783 786 self.db_input_cache = []
784 787
785 788 with self.db_output_cache_lock:
786 789 try:
787 790 self._writeout_output_cache(conn)
788 791 except sqlite3.IntegrityError:
789 792 print("!! Session/line number for output was not unique",
790 793 "in database. Output will not be stored.")
791 794 finally:
792 795 self.db_output_cache = []
793 796
794 797
795 798 class HistorySavingThread(threading.Thread):
796 799 """This thread takes care of writing history to the database, so that
797 800 the UI isn't held up while that happens.
798 801
799 802 It waits for the HistoryManager's save_flag to be set, then writes out
800 803 the history cache. The main thread is responsible for setting the flag when
801 804 the cache size reaches a defined threshold."""
802 805 daemon = True
803 806 stop_now = False
804 807 enabled = True
805 808 def __init__(self, history_manager):
806 809 super(HistorySavingThread, self).__init__(name="IPythonHistorySavingThread")
807 810 self.history_manager = history_manager
808 811 self.enabled = history_manager.enabled
809 812 atexit.register(self.stop)
810 813
811 814 @only_when_enabled
812 815 def run(self):
813 816 # We need a separate db connection per thread:
814 817 try:
815 818 self.db = sqlite3.connect(
816 819 str(self.history_manager.hist_file),
817 820 **self.history_manager.connection_options
818 821 )
819 822 while True:
820 823 self.history_manager.save_flag.wait()
821 824 if self.stop_now:
822 825 self.db.close()
823 826 return
824 827 self.history_manager.save_flag.clear()
825 828 self.history_manager.writeout_cache(self.db)
826 829 except Exception as e:
827 830 print(("The history saving thread hit an unexpected error (%s)."
828 831 "History will not be written to the database.") % repr(e))
829 832
830 833 def stop(self):
831 834 """This can be called from the main thread to safely stop this thread.
832 835
833 836 Note that it does not attempt to write out remaining history before
834 837 exiting. That should be done by calling the HistoryManager's
835 838 end_session method."""
836 839 self.stop_now = True
837 840 self.history_manager.save_flag.set()
838 841 self.join()
839 842
840 843
841 844 # To match, e.g. ~5/8-~2/3
842 845 range_re = re.compile(r"""
843 846 ((?P<startsess>~?\d+)/)?
844 847 (?P<start>\d+)?
845 848 ((?P<sep>[\-:])
846 849 ((?P<endsess>~?\d+)/)?
847 850 (?P<end>\d+))?
848 851 $""", re.VERBOSE)
849 852
850 853
851 854 def extract_hist_ranges(ranges_str):
852 855 """Turn a string of history ranges into 3-tuples of (session, start, stop).
853 856
857 Empty string results in a `[(0, 1, None)]`, i.e. "everything from current
858 session".
859
854 860 Examples
855 861 --------
856 862 >>> list(extract_hist_ranges("~8/5-~7/4 2"))
857 863 [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
858 864 """
865 if ranges_str == "":
866 yield (0, 1, None) # Everything from current session
867 return
868
859 869 for range_str in ranges_str.split():
860 870 rmatch = range_re.match(range_str)
861 871 if not rmatch:
862 872 continue
863 873 start = rmatch.group("start")
864 874 if start:
865 875 start = int(start)
866 876 end = rmatch.group("end")
867 877 # If no end specified, get (a, a + 1)
868 878 end = int(end) if end else start + 1
869 879 else: # start not specified
870 880 if not rmatch.group('startsess'): # no startsess
871 881 continue
872 882 start = 1
873 883 end = None # provide the entire session hist
874 884
875 885 if rmatch.group("sep") == "-": # 1-3 == 1:4 --> [1, 2, 3]
876 886 end += 1
877 887 startsess = rmatch.group("startsess") or "0"
878 888 endsess = rmatch.group("endsess") or startsess
879 889 startsess = int(startsess.replace("~","-"))
880 890 endsess = int(endsess.replace("~","-"))
881 891 assert endsess >= startsess, "start session must be earlier than end session"
882 892
883 893 if endsess == startsess:
884 894 yield (startsess, start, end)
885 895 continue
886 896 # Multiple sessions in one range:
887 897 yield (startsess, start, None)
888 898 for sess in range(startsess+1, endsess):
889 899 yield (sess, 1, None)
890 900 yield (endsess, 1, end)
891 901
892 902
893 903 def _format_lineno(session, line):
894 904 """Helper function to format line numbers properly."""
895 905 if session == 0:
896 906 return str(line)
897 907 return "%s#%s" % (session, line)
1 NO CONTENT: modified file
The requested commit or file is too big and content was truncated. Show full diff
@@ -1,750 +1,754 b''
1 1 """Implementation of code management magic functions.
2 2 """
3 3 #-----------------------------------------------------------------------------
4 4 # Copyright (c) 2012 The IPython Development Team.
5 5 #
6 6 # Distributed under the terms of the Modified BSD License.
7 7 #
8 8 # The full license is in the file COPYING.txt, distributed with this software.
9 9 #-----------------------------------------------------------------------------
10 10
11 11 #-----------------------------------------------------------------------------
12 12 # Imports
13 13 #-----------------------------------------------------------------------------
14 14
15 15 # Stdlib
16 16 import inspect
17 17 import io
18 18 import os
19 19 import re
20 20 import sys
21 21 import ast
22 22 from itertools import chain
23 23 from urllib.request import Request, urlopen
24 24 from urllib.parse import urlencode
25 25 from pathlib import Path
26 26
27 27 # Our own packages
28 28 from IPython.core.error import TryNext, StdinNotImplementedError, UsageError
29 29 from IPython.core.macro import Macro
30 30 from IPython.core.magic import Magics, magics_class, line_magic
31 31 from IPython.core.oinspect import find_file, find_source_lines
32 32 from IPython.core.release import version
33 33 from IPython.testing.skipdoctest import skip_doctest
34 34 from IPython.utils.contexts import preserve_keys
35 35 from IPython.utils.path import get_py_filename
36 36 from warnings import warn
37 37 from logging import error
38 38 from IPython.utils.text import get_text_list
39 39
40 40 #-----------------------------------------------------------------------------
41 41 # Magic implementation classes
42 42 #-----------------------------------------------------------------------------
43 43
44 44 # Used for exception handling in magic_edit
45 45 class MacroToEdit(ValueError): pass
46 46
47 47 ipython_input_pat = re.compile(r"<ipython\-input\-(\d+)-[a-z\d]+>$")
48 48
49 49 # To match, e.g. 8-10 1:5 :10 3-
50 50 range_re = re.compile(r"""
51 51 (?P<start>\d+)?
52 52 ((?P<sep>[\-:])
53 53 (?P<end>\d+)?)?
54 54 $""", re.VERBOSE)
55 55
56 56
57 57 def extract_code_ranges(ranges_str):
58 58 """Turn a string of range for %%load into 2-tuples of (start, stop)
59 59 ready to use as a slice of the content split by lines.
60 60
61 61 Examples
62 62 --------
63 63 list(extract_input_ranges("5-10 2"))
64 64 [(4, 10), (1, 2)]
65 65 """
66 66 for range_str in ranges_str.split():
67 67 rmatch = range_re.match(range_str)
68 68 if not rmatch:
69 69 continue
70 70 sep = rmatch.group("sep")
71 71 start = rmatch.group("start")
72 72 end = rmatch.group("end")
73 73
74 74 if sep == '-':
75 75 start = int(start) - 1 if start else None
76 76 end = int(end) if end else None
77 77 elif sep == ':':
78 78 start = int(start) - 1 if start else None
79 79 end = int(end) - 1 if end else None
80 80 else:
81 81 end = int(start)
82 82 start = int(start) - 1
83 83 yield (start, end)
84 84
85 85
86 86 def extract_symbols(code, symbols):
87 87 """
88 88 Return a tuple (blocks, not_found)
89 89 where ``blocks`` is a list of code fragments
90 90 for each symbol parsed from code, and ``not_found`` are
91 91 symbols not found in the code.
92 92
93 93 For example::
94 94
95 95 In [1]: code = '''a = 10
96 96 ...: def b(): return 42
97 97 ...: class A: pass'''
98 98
99 99 In [2]: extract_symbols(code, 'A,b,z')
100 100 Out[2]: (['class A: pass\\n', 'def b(): return 42\\n'], ['z'])
101 101 """
102 102 symbols = symbols.split(',')
103 103
104 104 # this will raise SyntaxError if code isn't valid Python
105 105 py_code = ast.parse(code)
106 106
107 107 marks = [(getattr(s, 'name', None), s.lineno) for s in py_code.body]
108 108 code = code.split('\n')
109 109
110 110 symbols_lines = {}
111 111
112 112 # we already know the start_lineno of each symbol (marks).
113 113 # To find each end_lineno, we traverse in reverse order until each
114 114 # non-blank line
115 115 end = len(code)
116 116 for name, start in reversed(marks):
117 117 while not code[end - 1].strip():
118 118 end -= 1
119 119 if name:
120 120 symbols_lines[name] = (start - 1, end)
121 121 end = start - 1
122 122
123 123 # Now symbols_lines is a map
124 124 # {'symbol_name': (start_lineno, end_lineno), ...}
125 125
126 126 # fill a list with chunks of codes for each requested symbol
127 127 blocks = []
128 128 not_found = []
129 129 for symbol in symbols:
130 130 if symbol in symbols_lines:
131 131 start, end = symbols_lines[symbol]
132 132 blocks.append('\n'.join(code[start:end]) + '\n')
133 133 else:
134 134 not_found.append(symbol)
135 135
136 136 return blocks, not_found
137 137
138 138 def strip_initial_indent(lines):
139 139 """For %load, strip indent from lines until finding an unindented line.
140 140
141 141 https://github.com/ipython/ipython/issues/9775
142 142 """
143 143 indent_re = re.compile(r'\s+')
144 144
145 145 it = iter(lines)
146 146 first_line = next(it)
147 147 indent_match = indent_re.match(first_line)
148 148
149 149 if indent_match:
150 150 # First line was indented
151 151 indent = indent_match.group()
152 152 yield first_line[len(indent):]
153 153
154 154 for line in it:
155 155 if line.startswith(indent):
156 156 yield line[len(indent):]
157 157 else:
158 158 # Less indented than the first line - stop dedenting
159 159 yield line
160 160 break
161 161 else:
162 162 yield first_line
163 163
164 164 # Pass the remaining lines through without dedenting
165 165 for line in it:
166 166 yield line
167 167
168 168
169 169 class InteractivelyDefined(Exception):
170 170 """Exception for interactively defined variable in magic_edit"""
171 171 def __init__(self, index):
172 172 self.index = index
173 173
174 174
175 175 @magics_class
176 176 class CodeMagics(Magics):
177 177 """Magics related to code management (loading, saving, editing, ...)."""
178 178
179 179 def __init__(self, *args, **kwargs):
180 180 self._knowntemps = set()
181 181 super(CodeMagics, self).__init__(*args, **kwargs)
182 182
183 183 @line_magic
184 184 def save(self, parameter_s=''):
185 185 """Save a set of lines or a macro to a given filename.
186 186
187 187 Usage:\\
188 188 %save [options] filename n1-n2 n3-n4 ... n5 .. n6 ...
189 189
190 190 Options:
191 191
192 192 -r: use 'raw' input. By default, the 'processed' history is used,
193 193 so that magics are loaded in their transformed version to valid
194 194 Python. If this option is given, the raw input as typed as the
195 195 command line is used instead.
196 196
197 197 -f: force overwrite. If file exists, %save will prompt for overwrite
198 198 unless -f is given.
199 199
200 200 -a: append to the file instead of overwriting it.
201 201
202 202 This function uses the same syntax as %history for input ranges,
203 203 then saves the lines to the filename you specify.
204 204
205 If no ranges are specified, saves history of the current session up to
206 this point.
207
205 208 It adds a '.py' extension to the file if you don't do so yourself, and
206 209 it asks for confirmation before overwriting existing files.
207 210
208 211 If `-r` option is used, the default extension is `.ipy`.
209 212 """
210 213
211 214 opts,args = self.parse_options(parameter_s,'fra',mode='list')
212 215 if not args:
213 216 raise UsageError('Missing filename.')
214 217 raw = 'r' in opts
215 218 force = 'f' in opts
216 219 append = 'a' in opts
217 220 mode = 'a' if append else 'w'
218 221 ext = '.ipy' if raw else '.py'
219 222 fname, codefrom = args[0], " ".join(args[1:])
220 223 if not fname.endswith(('.py','.ipy')):
221 224 fname += ext
222 225 file_exists = os.path.isfile(fname)
223 226 if file_exists and not force and not append:
224 227 try:
225 228 overwrite = self.shell.ask_yes_no('File `%s` exists. Overwrite (y/[N])? ' % fname, default='n')
226 229 except StdinNotImplementedError:
227 230 print("File `%s` exists. Use `%%save -f %s` to force overwrite" % (fname, parameter_s))
228 231 return
229 232 if not overwrite :
230 233 print('Operation cancelled.')
231 234 return
232 235 try:
233 236 cmds = self.shell.find_user_code(codefrom,raw)
234 237 except (TypeError, ValueError) as e:
235 238 print(e.args[0])
236 239 return
237 240 with io.open(fname, mode, encoding="utf-8") as f:
238 241 if not file_exists or not append:
239 242 f.write("# coding: utf-8\n")
240 243 f.write(cmds)
241 244 # make sure we end on a newline
242 245 if not cmds.endswith('\n'):
243 246 f.write('\n')
244 247 print('The following commands were written to file `%s`:' % fname)
245 248 print(cmds)
246 249
247 250 @line_magic
248 251 def pastebin(self, parameter_s=''):
249 252 """Upload code to dpaste.com, returning the URL.
250 253
251 254 Usage:\\
252 255 %pastebin [-d "Custom description"][-e 24] 1-7
253 256
254 257 The argument can be an input history range, a filename, or the name of a
255 258 string or macro.
256 259
260 If no arguments are given, uploads the history of this session up to
261 this point.
262
257 263 Options:
258 264
259 265 -d: Pass a custom description. The default will say
260 266 "Pasted from IPython".
261 267 -e: Pass number of days for the link to be expired.
262 268 The default will be 7 days.
263 269 """
264 270 opts, args = self.parse_options(parameter_s, "d:e:")
265 271
266 272 try:
267 273 code = self.shell.find_user_code(args)
268 274 except (ValueError, TypeError) as e:
269 275 print(e.args[0])
270 276 return
271 277
272 278 expiry_days = 7
273 279 try:
274 280 expiry_days = int(opts.get("e", 7))
275 281 except ValueError as e:
276 282 print(e.args[0].capitalize())
277 283 return
278 284 if expiry_days < 1 or expiry_days > 365:
279 285 print("Expiry days should be in range of 1 to 365")
280 286 return
281 287
282 288 post_data = urlencode(
283 289 {
284 290 "title": opts.get("d", "Pasted from IPython"),
285 291 "syntax": "python",
286 292 "content": code,
287 293 "expiry_days": expiry_days,
288 294 }
289 295 ).encode("utf-8")
290 296
291 297 request = Request(
292 298 "https://dpaste.com/api/v2/",
293 299 headers={"User-Agent": "IPython v{}".format(version)},
294 300 )
295 301 response = urlopen(request, post_data)
296 302 return response.headers.get('Location')
297 303
298 304 @line_magic
299 305 def loadpy(self, arg_s):
300 306 """Alias of `%load`
301 307
302 308 `%loadpy` has gained some flexibility and dropped the requirement of a `.py`
303 309 extension. So it has been renamed simply into %load. You can look at
304 310 `%load`'s docstring for more info.
305 311 """
306 312 self.load(arg_s)
307 313
308 314 @line_magic
309 315 def load(self, arg_s):
310 316 """Load code into the current frontend.
311 317
312 318 Usage:\\
313 319 %load [options] source
314 320
315 321 where source can be a filename, URL, input history range, macro, or
316 322 element in the user namespace
317 323
324 If no arguments are given, loads the history of this session up to this
325 point.
326
318 327 Options:
319 328
320 329 -r <lines>: Specify lines or ranges of lines to load from the source.
321 330 Ranges could be specified as x-y (x..y) or in python-style x:y
322 331 (x..(y-1)). Both limits x and y can be left blank (meaning the
323 332 beginning and end of the file, respectively).
324 333
325 334 -s <symbols>: Specify function or classes to load from python source.
326 335
327 336 -y : Don't ask confirmation for loading source above 200 000 characters.
328 337
329 338 -n : Include the user's namespace when searching for source code.
330 339
331 340 This magic command can either take a local filename, a URL, an history
332 341 range (see %history) or a macro as argument, it will prompt for
333 342 confirmation before loading source with more than 200 000 characters, unless
334 343 -y flag is passed or if the frontend does not support raw_input::
335 344
345 %load
336 346 %load myscript.py
337 347 %load 7-27
338 348 %load myMacro
339 349 %load http://www.example.com/myscript.py
340 350 %load -r 5-10 myscript.py
341 351 %load -r 10-20,30,40: foo.py
342 352 %load -s MyClass,wonder_function myscript.py
343 353 %load -n MyClass
344 354 %load -n my_module.wonder_function
345 355 """
346 356 opts,args = self.parse_options(arg_s,'yns:r:')
347
348 if not args:
349 raise UsageError('Missing filename, URL, input history range, '
350 'macro, or element in the user namespace.')
351
352 357 search_ns = 'n' in opts
353
354 358 contents = self.shell.find_user_code(args, search_ns=search_ns)
355 359
356 360 if 's' in opts:
357 361 try:
358 362 blocks, not_found = extract_symbols(contents, opts['s'])
359 363 except SyntaxError:
360 364 # non python code
361 365 error("Unable to parse the input as valid Python code")
362 366 return
363 367
364 368 if len(not_found) == 1:
365 369 warn('The symbol `%s` was not found' % not_found[0])
366 370 elif len(not_found) > 1:
367 371 warn('The symbols %s were not found' % get_text_list(not_found,
368 372 wrap_item_with='`')
369 373 )
370 374
371 375 contents = '\n'.join(blocks)
372 376
373 377 if 'r' in opts:
374 378 ranges = opts['r'].replace(',', ' ')
375 379 lines = contents.split('\n')
376 380 slices = extract_code_ranges(ranges)
377 381 contents = [lines[slice(*slc)] for slc in slices]
378 382 contents = '\n'.join(strip_initial_indent(chain.from_iterable(contents)))
379 383
380 384 l = len(contents)
381 385
382 386 # 200 000 is ~ 2500 full 80 character lines
383 387 # so in average, more than 5000 lines
384 388 if l > 200000 and 'y' not in opts:
385 389 try:
386 390 ans = self.shell.ask_yes_no(("The text you're trying to load seems pretty big"\
387 391 " (%d characters). Continue (y/[N]) ?" % l), default='n' )
388 392 except StdinNotImplementedError:
389 393 #assume yes if raw input not implemented
390 394 ans = True
391 395
392 396 if ans is False :
393 397 print('Operation cancelled.')
394 398 return
395 399
396 400 contents = "# %load {}\n".format(arg_s) + contents
397 401
398 402 self.shell.set_next_input(contents, replace=True)
399 403
400 404 @staticmethod
401 405 def _find_edit_target(shell, args, opts, last_call):
402 406 """Utility method used by magic_edit to find what to edit."""
403 407
404 408 def make_filename(arg):
405 409 "Make a filename from the given args"
406 410 try:
407 411 filename = get_py_filename(arg)
408 412 except IOError:
409 413 # If it ends with .py but doesn't already exist, assume we want
410 414 # a new file.
411 415 if arg.endswith('.py'):
412 416 filename = arg
413 417 else:
414 418 filename = None
415 419 return filename
416 420
417 421 # Set a few locals from the options for convenience:
418 422 opts_prev = 'p' in opts
419 423 opts_raw = 'r' in opts
420 424
421 425 # custom exceptions
422 426 class DataIsObject(Exception): pass
423 427
424 428 # Default line number value
425 429 lineno = opts.get('n',None)
426 430
427 431 if opts_prev:
428 432 args = '_%s' % last_call[0]
429 433 if args not in shell.user_ns:
430 434 args = last_call[1]
431 435
432 436 # by default this is done with temp files, except when the given
433 437 # arg is a filename
434 438 use_temp = True
435 439
436 440 data = ''
437 441
438 442 # First, see if the arguments should be a filename.
439 443 filename = make_filename(args)
440 444 if filename:
441 445 use_temp = False
442 446 elif args:
443 447 # Mode where user specifies ranges of lines, like in %macro.
444 448 data = shell.extract_input_lines(args, opts_raw)
445 449 if not data:
446 450 try:
447 451 # Load the parameter given as a variable. If not a string,
448 452 # process it as an object instead (below)
449 453
450 454 #print '*** args',args,'type',type(args) # dbg
451 455 data = eval(args, shell.user_ns)
452 456 if not isinstance(data, str):
453 457 raise DataIsObject
454 458
455 459 except (NameError,SyntaxError):
456 460 # given argument is not a variable, try as a filename
457 461 filename = make_filename(args)
458 462 if filename is None:
459 463 warn("Argument given (%s) can't be found as a variable "
460 464 "or as a filename." % args)
461 465 return (None, None, None)
462 466 use_temp = False
463 467
464 468 except DataIsObject as e:
465 469 # macros have a special edit function
466 470 if isinstance(data, Macro):
467 471 raise MacroToEdit(data) from e
468 472
469 473 # For objects, try to edit the file where they are defined
470 474 filename = find_file(data)
471 475 if filename:
472 476 if 'fakemodule' in filename.lower() and \
473 477 inspect.isclass(data):
474 478 # class created by %edit? Try to find source
475 479 # by looking for method definitions instead, the
476 480 # __module__ in those classes is FakeModule.
477 481 attrs = [getattr(data, aname) for aname in dir(data)]
478 482 for attr in attrs:
479 483 if not inspect.ismethod(attr):
480 484 continue
481 485 filename = find_file(attr)
482 486 if filename and \
483 487 'fakemodule' not in filename.lower():
484 488 # change the attribute to be the edit
485 489 # target instead
486 490 data = attr
487 491 break
488 492
489 493 m = ipython_input_pat.match(os.path.basename(filename))
490 494 if m:
491 495 raise InteractivelyDefined(int(m.groups()[0])) from e
492 496
493 497 datafile = 1
494 498 if filename is None:
495 499 filename = make_filename(args)
496 500 datafile = 1
497 501 if filename is not None:
498 502 # only warn about this if we get a real name
499 503 warn('Could not find file where `%s` is defined.\n'
500 504 'Opening a file named `%s`' % (args, filename))
501 505 # Now, make sure we can actually read the source (if it was
502 506 # in a temp file it's gone by now).
503 507 if datafile:
504 508 if lineno is None:
505 509 lineno = find_source_lines(data)
506 510 if lineno is None:
507 511 filename = make_filename(args)
508 512 if filename is None:
509 513 warn('The file where `%s` was defined '
510 514 'cannot be read or found.' % data)
511 515 return (None, None, None)
512 516 use_temp = False
513 517
514 518 if use_temp:
515 519 filename = shell.mktempfile(data)
516 520 print('IPython will make a temporary file named:',filename)
517 521
518 522 # use last_call to remember the state of the previous call, but don't
519 523 # let it be clobbered by successive '-p' calls.
520 524 try:
521 525 last_call[0] = shell.displayhook.prompt_count
522 526 if not opts_prev:
523 527 last_call[1] = args
524 528 except:
525 529 pass
526 530
527 531
528 532 return filename, lineno, use_temp
529 533
530 534 def _edit_macro(self,mname,macro):
531 535 """open an editor with the macro data in a file"""
532 536 filename = self.shell.mktempfile(macro.value)
533 537 self.shell.hooks.editor(filename)
534 538
535 539 # and make a new macro object, to replace the old one
536 540 mvalue = Path(filename).read_text()
537 541 self.shell.user_ns[mname] = Macro(mvalue)
538 542
539 543 @skip_doctest
540 544 @line_magic
541 545 def edit(self, parameter_s='',last_call=['','']):
542 546 """Bring up an editor and execute the resulting code.
543 547
544 548 Usage:
545 549 %edit [options] [args]
546 550
547 551 %edit runs IPython's editor hook. The default version of this hook is
548 552 set to call the editor specified by your $EDITOR environment variable.
549 553 If this isn't found, it will default to vi under Linux/Unix and to
550 554 notepad under Windows. See the end of this docstring for how to change
551 555 the editor hook.
552 556
553 557 You can also set the value of this editor via the
554 558 ``TerminalInteractiveShell.editor`` option in your configuration file.
555 559 This is useful if you wish to use a different editor from your typical
556 560 default with IPython (and for Windows users who typically don't set
557 561 environment variables).
558 562
559 563 This command allows you to conveniently edit multi-line code right in
560 564 your IPython session.
561 565
562 566 If called without arguments, %edit opens up an empty editor with a
563 567 temporary file and will execute the contents of this file when you
564 568 close it (don't forget to save it!).
565 569
566 570
567 571 Options:
568 572
569 573 -n <number>: open the editor at a specified line number. By default,
570 574 the IPython editor hook uses the unix syntax 'editor +N filename', but
571 575 you can configure this by providing your own modified hook if your
572 576 favorite editor supports line-number specifications with a different
573 577 syntax.
574 578
575 579 -p: this will call the editor with the same data as the previous time
576 580 it was used, regardless of how long ago (in your current session) it
577 581 was.
578 582
579 583 -r: use 'raw' input. This option only applies to input taken from the
580 584 user's history. By default, the 'processed' history is used, so that
581 585 magics are loaded in their transformed version to valid Python. If
582 586 this option is given, the raw input as typed as the command line is
583 587 used instead. When you exit the editor, it will be executed by
584 588 IPython's own processor.
585 589
586 590 -x: do not execute the edited code immediately upon exit. This is
587 591 mainly useful if you are editing programs which need to be called with
588 592 command line arguments, which you can then do using %run.
589 593
590 594
591 595 Arguments:
592 596
593 597 If arguments are given, the following possibilities exist:
594 598
595 599 - If the argument is a filename, IPython will load that into the
596 600 editor. It will execute its contents with execfile() when you exit,
597 601 loading any code in the file into your interactive namespace.
598 602
599 603 - The arguments are ranges of input history, e.g. "7 ~1/4-6".
600 604 The syntax is the same as in the %history magic.
601 605
602 606 - If the argument is a string variable, its contents are loaded
603 607 into the editor. You can thus edit any string which contains
604 608 python code (including the result of previous edits).
605 609
606 610 - If the argument is the name of an object (other than a string),
607 611 IPython will try to locate the file where it was defined and open the
608 612 editor at the point where it is defined. You can use `%edit function`
609 613 to load an editor exactly at the point where 'function' is defined,
610 614 edit it and have the file be executed automatically.
611 615
612 616 - If the object is a macro (see %macro for details), this opens up your
613 617 specified editor with a temporary file containing the macro's data.
614 618 Upon exit, the macro is reloaded with the contents of the file.
615 619
616 620 Note: opening at an exact line is only supported under Unix, and some
617 621 editors (like kedit and gedit up to Gnome 2.8) do not understand the
618 622 '+NUMBER' parameter necessary for this feature. Good editors like
619 623 (X)Emacs, vi, jed, pico and joe all do.
620 624
621 625 After executing your code, %edit will return as output the code you
622 626 typed in the editor (except when it was an existing file). This way
623 627 you can reload the code in further invocations of %edit as a variable,
624 628 via _<NUMBER> or Out[<NUMBER>], where <NUMBER> is the prompt number of
625 629 the output.
626 630
627 631 Note that %edit is also available through the alias %ed.
628 632
629 633 This is an example of creating a simple function inside the editor and
630 634 then modifying it. First, start up the editor::
631 635
632 636 In [1]: edit
633 637 Editing... done. Executing edited code...
634 638 Out[1]: 'def foo():\\n print "foo() was defined in an editing
635 639 session"\\n'
636 640
637 641 We can then call the function foo()::
638 642
639 643 In [2]: foo()
640 644 foo() was defined in an editing session
641 645
642 646 Now we edit foo. IPython automatically loads the editor with the
643 647 (temporary) file where foo() was previously defined::
644 648
645 649 In [3]: edit foo
646 650 Editing... done. Executing edited code...
647 651
648 652 And if we call foo() again we get the modified version::
649 653
650 654 In [4]: foo()
651 655 foo() has now been changed!
652 656
653 657 Here is an example of how to edit a code snippet successive
654 658 times. First we call the editor::
655 659
656 660 In [5]: edit
657 661 Editing... done. Executing edited code...
658 662 hello
659 663 Out[5]: "print 'hello'\\n"
660 664
661 665 Now we call it again with the previous output (stored in _)::
662 666
663 667 In [6]: edit _
664 668 Editing... done. Executing edited code...
665 669 hello world
666 670 Out[6]: "print 'hello world'\\n"
667 671
668 672 Now we call it with the output #8 (stored in _8, also as Out[8])::
669 673
670 674 In [7]: edit _8
671 675 Editing... done. Executing edited code...
672 676 hello again
673 677 Out[7]: "print 'hello again'\\n"
674 678
675 679
676 680 Changing the default editor hook:
677 681
678 682 If you wish to write your own editor hook, you can put it in a
679 683 configuration file which you load at startup time. The default hook
680 684 is defined in the IPython.core.hooks module, and you can use that as a
681 685 starting example for further modifications. That file also has
682 686 general instructions on how to set a new hook for use once you've
683 687 defined it."""
684 688 opts,args = self.parse_options(parameter_s,'prxn:')
685 689
686 690 try:
687 691 filename, lineno, is_temp = self._find_edit_target(self.shell,
688 692 args, opts, last_call)
689 693 except MacroToEdit as e:
690 694 self._edit_macro(args, e.args[0])
691 695 return
692 696 except InteractivelyDefined as e:
693 697 print("Editing In[%i]" % e.index)
694 698 args = str(e.index)
695 699 filename, lineno, is_temp = self._find_edit_target(self.shell,
696 700 args, opts, last_call)
697 701 if filename is None:
698 702 # nothing was found, warnings have already been issued,
699 703 # just give up.
700 704 return
701 705
702 706 if is_temp:
703 707 self._knowntemps.add(filename)
704 708 elif (filename in self._knowntemps):
705 709 is_temp = True
706 710
707 711
708 712 # do actual editing here
709 713 print('Editing...', end=' ')
710 714 sys.stdout.flush()
711 715 filepath = Path(filename)
712 716 try:
713 717 # Quote filenames that may have spaces in them when opening
714 718 # the editor
715 719 quoted = filename = str(filepath.absolute())
716 720 if " " in quoted:
717 721 quoted = "'%s'" % quoted
718 722 self.shell.hooks.editor(quoted, lineno)
719 723 except TryNext:
720 724 warn('Could not open editor')
721 725 return
722 726
723 727 # XXX TODO: should this be generalized for all string vars?
724 728 # For now, this is special-cased to blocks created by cpaste
725 729 if args.strip() == "pasted_block":
726 730 self.shell.user_ns["pasted_block"] = filepath.read_text()
727 731
728 732 if 'x' in opts: # -x prevents actual execution
729 733 print()
730 734 else:
731 735 print('done. Executing edited code...')
732 736 with preserve_keys(self.shell.user_ns, '__file__'):
733 737 if not is_temp:
734 738 self.shell.user_ns['__file__'] = filename
735 739 if 'r' in opts: # Untranslated IPython code
736 740 source = filepath.read_text()
737 741 self.shell.run_cell(source, store_history=False)
738 742 else:
739 743 self.shell.safe_execfile(filename, self.shell.user_ns,
740 744 self.shell.user_ns)
741 745
742 746 if is_temp:
743 747 try:
744 748 return filepath.read_text()
745 749 except IOError as msg:
746 750 if Path(msg.filename) == filepath:
747 751 warn('File not found. Did you forget to save?')
748 752 return
749 753 else:
750 754 self.shell.showtraceback()
@@ -1,335 +1,333 b''
1 1 """Implementation of magic functions related to History.
2 2 """
3 3 #-----------------------------------------------------------------------------
4 4 # Copyright (c) 2012, IPython Development Team.
5 5 #
6 6 # Distributed under the terms of the Modified BSD License.
7 7 #
8 8 # The full license is in the file COPYING.txt, distributed with this software.
9 9 #-----------------------------------------------------------------------------
10 10
11 11 #-----------------------------------------------------------------------------
12 12 # Imports
13 13 #-----------------------------------------------------------------------------
14 14
15 15 # Stdlib
16 16 import os
17 17 import sys
18 18 from io import open as io_open
19 19 import fnmatch
20 20
21 21 # Our own packages
22 22 from IPython.core.error import StdinNotImplementedError
23 23 from IPython.core.magic import Magics, magics_class, line_magic
24 24 from IPython.core.magic_arguments import (argument, magic_arguments,
25 25 parse_argstring)
26 26 from IPython.testing.skipdoctest import skip_doctest
27 27 from IPython.utils import io
28 28
29 29 #-----------------------------------------------------------------------------
30 30 # Magics class implementation
31 31 #-----------------------------------------------------------------------------
32 32
33 33
34 34 _unspecified = object()
35 35
36 36
37 37 @magics_class
38 38 class HistoryMagics(Magics):
39 39
40 40 @magic_arguments()
41 41 @argument(
42 42 '-n', dest='print_nums', action='store_true', default=False,
43 43 help="""
44 44 print line numbers for each input.
45 45 This feature is only available if numbered prompts are in use.
46 46 """)
47 47 @argument(
48 48 '-o', dest='get_output', action='store_true', default=False,
49 49 help="also print outputs for each input.")
50 50 @argument(
51 51 '-p', dest='pyprompts', action='store_true', default=False,
52 52 help="""
53 53 print classic '>>>' python prompts before each input.
54 54 This is useful for making documentation, and in conjunction
55 55 with -o, for producing doctest-ready output.
56 56 """)
57 57 @argument(
58 58 '-t', dest='raw', action='store_false', default=True,
59 59 help="""
60 60 print the 'translated' history, as IPython understands it.
61 61 IPython filters your input and converts it all into valid Python
62 62 source before executing it (things like magics or aliases are turned
63 63 into function calls, for example). With this option, you'll see the
64 64 native history instead of the user-entered version: '%%cd /' will be
65 65 seen as 'get_ipython().run_line_magic("cd", "/")' instead of '%%cd /'.
66 66 """)
67 67 @argument(
68 68 '-f', dest='filename',
69 69 help="""
70 70 FILENAME: instead of printing the output to the screen, redirect
71 71 it to the given file. The file is always overwritten, though *when
72 72 it can*, IPython asks for confirmation first. In particular, running
73 73 the command 'history -f FILENAME' from the IPython Notebook
74 74 interface will replace FILENAME even if it already exists *without*
75 75 confirmation.
76 76 """)
77 77 @argument(
78 78 '-g', dest='pattern', nargs='*', default=None,
79 79 help="""
80 80 treat the arg as a glob pattern to search for in (full) history.
81 81 This includes the saved history (almost all commands ever written).
82 82 The pattern may contain '?' to match one unknown character and '*'
83 83 to match any number of unknown characters. Use '%%hist -g' to show
84 84 full saved history (may be very long).
85 85 """)
86 86 @argument(
87 87 '-l', dest='limit', type=int, nargs='?', default=_unspecified,
88 88 help="""
89 89 get the last n lines from all sessions. Specify n as a single
90 90 arg, or the default is the last 10 lines.
91 91 """)
92 92 @argument(
93 93 '-u', dest='unique', action='store_true',
94 94 help="""
95 95 when searching history using `-g`, show only unique history.
96 96 """)
97 97 @argument('range', nargs='*')
98 98 @skip_doctest
99 99 @line_magic
100 100 def history(self, parameter_s = ''):
101 101 """Print input history (_i<n> variables), with most recent last.
102 102
103 103 By default, input history is printed without line numbers so it can be
104 104 directly pasted into an editor. Use -n to show them.
105 105
106 106 By default, all input history from the current session is displayed.
107 107 Ranges of history can be indicated using the syntax:
108 108
109 109 ``4``
110 110 Line 4, current session
111 111 ``4-6``
112 112 Lines 4-6, current session
113 113 ``243/1-5``
114 114 Lines 1-5, session 243
115 115 ``~2/7``
116 116 Line 7, session 2 before current
117 117 ``~8/1-~6/5``
118 118 From the first line of 8 sessions ago, to the fifth line of 6
119 119 sessions ago.
120 120
121 121 Multiple ranges can be entered, separated by spaces
122 122
123 123 The same syntax is used by %macro, %save, %edit, %rerun
124 124
125 125 Examples
126 126 --------
127 127 ::
128 128
129 129 In [6]: %history -n 4-6
130 130 4:a = 12
131 131 5:print a**2
132 132 6:%history -n 4-6
133 133
134 134 """
135 135
136 136 args = parse_argstring(self.history, parameter_s)
137 137
138 138 # For brevity
139 139 history_manager = self.shell.history_manager
140 140
141 141 def _format_lineno(session, line):
142 142 """Helper function to format line numbers properly."""
143 143 if session in (0, history_manager.session_number):
144 144 return str(line)
145 145 return "%s/%s" % (session, line)
146 146
147 147 # Check if output to specific file was requested.
148 148 outfname = args.filename
149 149 if not outfname:
150 150 outfile = sys.stdout # default
151 151 # We don't want to close stdout at the end!
152 152 close_at_end = False
153 153 else:
154 154 if os.path.exists(outfname):
155 155 try:
156 156 ans = io.ask_yes_no("File %r exists. Overwrite?" % outfname)
157 157 except StdinNotImplementedError:
158 158 ans = True
159 159 if not ans:
160 160 print('Aborting.')
161 161 return
162 162 print("Overwriting file.")
163 163 outfile = io_open(outfname, 'w', encoding='utf-8')
164 164 close_at_end = True
165 165
166 166 print_nums = args.print_nums
167 167 get_output = args.get_output
168 168 pyprompts = args.pyprompts
169 169 raw = args.raw
170 170
171 171 pattern = None
172 172 limit = None if args.limit is _unspecified else args.limit
173 173
174 174 range_pattern = False
175 175 if args.pattern is not None and not args.range:
176 176 if args.pattern:
177 177 pattern = "*" + " ".join(args.pattern) + "*"
178 178 else:
179 179 pattern = "*"
180 180 hist = history_manager.search(pattern, raw=raw, output=get_output,
181 181 n=limit, unique=args.unique)
182 182 print_nums = True
183 183 elif args.limit is not _unspecified:
184 184 n = 10 if limit is None else limit
185 185 hist = history_manager.get_tail(n, raw=raw, output=get_output)
186 186 else:
187 if args.range: # Get history by ranges
188 if args.pattern:
189 range_pattern = "*" + " ".join(args.pattern) + "*"
190 print_nums = True
191 hist = history_manager.get_range_by_str(" ".join(args.range),
192 raw, get_output)
193 else: # Just get history for the current session
194 hist = history_manager.get_range(raw=raw, output=get_output)
187 if args.pattern:
188 range_pattern = "*" + " ".join(args.pattern) + "*"
189 print_nums = True
190 hist = history_manager.get_range_by_str(
191 " ".join(args.range), raw, get_output
192 )
195 193
196 194 # We could be displaying the entire history, so let's not try to pull
197 195 # it into a list in memory. Anything that needs more space will just
198 196 # misalign.
199 197 width = 4
200 198
201 199 for session, lineno, inline in hist:
202 200 # Print user history with tabs expanded to 4 spaces. The GUI
203 201 # clients use hard tabs for easier usability in auto-indented code,
204 202 # but we want to produce PEP-8 compliant history for safe pasting
205 203 # into an editor.
206 204 if get_output:
207 205 inline, output = inline
208 206 if range_pattern:
209 207 if not fnmatch.fnmatch(inline, range_pattern):
210 208 continue
211 209 inline = inline.expandtabs(4).rstrip()
212 210
213 211 multiline = "\n" in inline
214 212 line_sep = '\n' if multiline else ' '
215 213 if print_nums:
216 214 print(u'%s:%s' % (_format_lineno(session, lineno).rjust(width),
217 215 line_sep), file=outfile, end=u'')
218 216 if pyprompts:
219 217 print(u">>> ", end=u"", file=outfile)
220 218 if multiline:
221 219 inline = "\n... ".join(inline.splitlines()) + "\n..."
222 220 print(inline, file=outfile)
223 221 if get_output and output:
224 222 print(output, file=outfile)
225 223
226 224 if close_at_end:
227 225 outfile.close()
228 226
229 227 @line_magic
230 228 def recall(self, arg):
231 229 r"""Repeat a command, or get command to input line for editing.
232 230
233 231 %recall and %rep are equivalent.
234 232
235 233 - %recall (no arguments):
236 234
237 235 Place a string version of last computation result (stored in the
238 236 special '_' variable) to the next input prompt. Allows you to create
239 237 elaborate command lines without using copy-paste::
240 238
241 239 In[1]: l = ["hei", "vaan"]
242 240 In[2]: "".join(l)
243 241 Out[2]: heivaan
244 242 In[3]: %recall
245 243 In[4]: heivaan_ <== cursor blinking
246 244
247 245 %recall 45
248 246
249 247 Place history line 45 on the next input prompt. Use %hist to find
250 248 out the number.
251 249
252 250 %recall 1-4
253 251
254 252 Combine the specified lines into one cell, and place it on the next
255 253 input prompt. See %history for the slice syntax.
256 254
257 255 %recall foo+bar
258 256
259 257 If foo+bar can be evaluated in the user namespace, the result is
260 258 placed at the next input prompt. Otherwise, the history is searched
261 259 for lines which contain that substring, and the most recent one is
262 260 placed at the next input prompt.
263 261 """
264 262 if not arg: # Last output
265 263 self.shell.set_next_input(str(self.shell.user_ns["_"]))
266 264 return
267 265 # Get history range
268 266 histlines = self.shell.history_manager.get_range_by_str(arg)
269 267 cmd = "\n".join(x[2] for x in histlines)
270 268 if cmd:
271 269 self.shell.set_next_input(cmd.rstrip())
272 270 return
273 271
274 272 try: # Variable in user namespace
275 273 cmd = str(eval(arg, self.shell.user_ns))
276 274 except Exception: # Search for term in history
277 275 histlines = self.shell.history_manager.search("*"+arg+"*")
278 276 for h in reversed([x[2] for x in histlines]):
279 277 if 'recall' in h or 'rep' in h:
280 278 continue
281 279 self.shell.set_next_input(h.rstrip())
282 280 return
283 281 else:
284 282 self.shell.set_next_input(cmd.rstrip())
285 283 return
286 284 print("Couldn't evaluate or find in history:", arg)
287 285
288 286 @line_magic
289 287 def rerun(self, parameter_s=''):
290 288 """Re-run previous input
291 289
292 290 By default, you can specify ranges of input history to be repeated
293 291 (as with %history). With no arguments, it will repeat the last line.
294 292
295 293 Options:
296 294
297 295 -l <n> : Repeat the last n lines of input, not including the
298 296 current command.
299 297
300 298 -g foo : Repeat the most recent line which contains foo
301 299 """
302 300 opts, args = self.parse_options(parameter_s, 'l:g:', mode='string')
303 301 if "l" in opts: # Last n lines
304 302 n = int(opts['l'])
305 303
306 304 if n == 0:
307 305 print("Requested 0 last lines - nothing to run")
308 306 return
309 307 elif n < 0:
310 308 print("Number of lines to rerun cannot be negative")
311 309 return
312 310
313 311 hist = self.shell.history_manager.get_tail(n)
314 312 elif "g" in opts: # Search
315 313 p = "*"+opts['g']+"*"
316 314 hist = list(self.shell.history_manager.search(p))
317 315 for l in reversed(hist):
318 316 if "rerun" not in l[2]:
319 317 hist = [l] # The last match which isn't a %rerun
320 318 break
321 319 else:
322 320 hist = [] # No matches except %rerun
323 321 elif args: # Specify history ranges
324 322 hist = self.shell.history_manager.get_range_by_str(args)
325 323 else: # Last line
326 324 hist = self.shell.history_manager.get_tail(1)
327 325 hist = [x[2] for x in hist]
328 326 if not hist:
329 327 print("No lines in history match specification")
330 328 return
331 329 histlines = "\n".join(hist)
332 330 print("=== Executing: ===")
333 331 print(histlines)
334 332 print("=== Output: ===")
335 333 self.shell.run_cell("\n".join(hist), store_history=False)
@@ -1,860 +1,859 b''
1 1 """Implementation of magic functions for interaction with the OS.
2 2
3 3 Note: this module is named 'osm' instead of 'os' to avoid a collision with the
4 4 builtin.
5 5 """
6 6 # Copyright (c) IPython Development Team.
7 7 # Distributed under the terms of the Modified BSD License.
8 8
9 9 import io
10 10 import os
11 11 import re
12 12 import sys
13 13 from pprint import pformat
14 14
15 15 from IPython.core import magic_arguments
16 16 from IPython.core import oinspect
17 17 from IPython.core import page
18 18 from IPython.core.alias import AliasError, Alias
19 19 from IPython.core.error import UsageError
20 20 from IPython.core.magic import (
21 21 Magics, compress_dhist, magics_class, line_magic, cell_magic, line_cell_magic
22 22 )
23 23 from IPython.testing.skipdoctest import skip_doctest
24 24 from IPython.utils.openpy import source_to_unicode
25 25 from IPython.utils.process import abbrev_cwd
26 26 from IPython.utils.terminal import set_term_title
27 27 from traitlets import Bool
28 28 from warnings import warn
29 29
30 30
31 31 @magics_class
32 32 class OSMagics(Magics):
33 33 """Magics to interact with the underlying OS (shell-type functionality).
34 34 """
35 35
36 36 cd_force_quiet = Bool(False,
37 37 help="Force %cd magic to be quiet even if -q is not passed."
38 38 ).tag(config=True)
39 39
40 40 def __init__(self, shell=None, **kwargs):
41 41
42 42 # Now define isexec in a cross platform manner.
43 43 self.is_posix = False
44 44 self.execre = None
45 45 if os.name == 'posix':
46 46 self.is_posix = True
47 47 else:
48 48 try:
49 49 winext = os.environ['pathext'].replace(';','|').replace('.','')
50 50 except KeyError:
51 51 winext = 'exe|com|bat|py'
52 52 try:
53 53 self.execre = re.compile(r'(.*)\.(%s)$' % winext,re.IGNORECASE)
54 54 except re.error:
55 55 warn("Seems like your pathext environmental "
56 56 "variable is malformed. Please check it to "
57 57 "enable a proper handle of file extensions "
58 58 "managed for your system")
59 59 winext = 'exe|com|bat|py'
60 60 self.execre = re.compile(r'(.*)\.(%s)$' % winext,re.IGNORECASE)
61 61
62 62 # call up the chain
63 63 super().__init__(shell=shell, **kwargs)
64 64
65 65
66 66 @skip_doctest
67 67 def _isexec_POSIX(self, file):
68 68 """
69 69 Test for executable on a POSIX system
70 70 """
71 71 if os.access(file.path, os.X_OK):
72 72 # will fail on maxOS if access is not X_OK
73 73 return file.is_file()
74 74 return False
75 75
76 76
77 77
78 78 @skip_doctest
79 79 def _isexec_WIN(self, file):
80 80 """
81 81 Test for executable file on non POSIX system
82 82 """
83 83 return file.is_file() and self.execre.match(file.name) is not None
84 84
85 85 @skip_doctest
86 86 def isexec(self, file):
87 87 """
88 88 Test for executable file on non POSIX system
89 89 """
90 90 if self.is_posix:
91 91 return self._isexec_POSIX(file)
92 92 else:
93 93 return self._isexec_WIN(file)
94 94
95 95
96 96 @skip_doctest
97 97 @line_magic
98 98 def alias(self, parameter_s=''):
99 99 """Define an alias for a system command.
100 100
101 101 '%alias alias_name cmd' defines 'alias_name' as an alias for 'cmd'
102 102
103 103 Then, typing 'alias_name params' will execute the system command 'cmd
104 104 params' (from your underlying operating system).
105 105
106 106 Aliases have lower precedence than magic functions and Python normal
107 107 variables, so if 'foo' is both a Python variable and an alias, the
108 108 alias can not be executed until 'del foo' removes the Python variable.
109 109
110 110 You can use the %l specifier in an alias definition to represent the
111 111 whole line when the alias is called. For example::
112 112
113 113 In [2]: alias bracket echo "Input in brackets: <%l>"
114 114 In [3]: bracket hello world
115 115 Input in brackets: <hello world>
116 116
117 117 You can also define aliases with parameters using %s specifiers (one
118 118 per parameter)::
119 119
120 120 In [1]: alias parts echo first %s second %s
121 121 In [2]: %parts A B
122 122 first A second B
123 123 In [3]: %parts A
124 124 Incorrect number of arguments: 2 expected.
125 125 parts is an alias to: 'echo first %s second %s'
126 126
127 127 Note that %l and %s are mutually exclusive. You can only use one or
128 128 the other in your aliases.
129 129
130 130 Aliases expand Python variables just like system calls using ! or !!
131 131 do: all expressions prefixed with '$' get expanded. For details of
132 132 the semantic rules, see PEP-215:
133 133 http://www.python.org/peps/pep-0215.html. This is the library used by
134 134 IPython for variable expansion. If you want to access a true shell
135 135 variable, an extra $ is necessary to prevent its expansion by
136 136 IPython::
137 137
138 138 In [6]: alias show echo
139 139 In [7]: PATH='A Python string'
140 140 In [8]: show $PATH
141 141 A Python string
142 142 In [9]: show $$PATH
143 143 /usr/local/lf9560/bin:/usr/local/intel/compiler70/ia32/bin:...
144 144
145 145 You can use the alias facility to access all of $PATH. See the %rehashx
146 146 function, which automatically creates aliases for the contents of your
147 147 $PATH.
148 148
149 149 If called with no parameters, %alias prints the current alias table
150 150 for your system. For posix systems, the default aliases are 'cat',
151 151 'cp', 'mv', 'rm', 'rmdir', and 'mkdir', and other platform-specific
152 152 aliases are added. For windows-based systems, the default aliases are
153 153 'copy', 'ddir', 'echo', 'ls', 'ldir', 'mkdir', 'ren', and 'rmdir'.
154 154
155 155 You can see the definition of alias by adding a question mark in the
156 156 end::
157 157
158 158 In [1]: cat?
159 159 Repr: <alias cat for 'cat'>"""
160 160
161 161 par = parameter_s.strip()
162 162 if not par:
163 163 aliases = sorted(self.shell.alias_manager.aliases)
164 164 # stored = self.shell.db.get('stored_aliases', {} )
165 165 # for k, v in stored:
166 166 # atab.append(k, v[0])
167 167
168 168 print("Total number of aliases:", len(aliases))
169 169 sys.stdout.flush()
170 170 return aliases
171 171
172 172 # Now try to define a new one
173 173 try:
174 174 alias,cmd = par.split(None, 1)
175 175 except TypeError:
176 176 print(oinspect.getdoc(self.alias))
177 177 return
178 178
179 179 try:
180 180 self.shell.alias_manager.define_alias(alias, cmd)
181 181 except AliasError as e:
182 182 print(e)
183 183 # end magic_alias
184 184
185 185 @line_magic
186 186 def unalias(self, parameter_s=''):
187 187 """Remove an alias"""
188 188
189 189 aname = parameter_s.strip()
190 190 try:
191 191 self.shell.alias_manager.undefine_alias(aname)
192 192 except ValueError as e:
193 193 print(e)
194 194 return
195 195
196 196 stored = self.shell.db.get('stored_aliases', {} )
197 197 if aname in stored:
198 198 print("Removing %stored alias",aname)
199 199 del stored[aname]
200 200 self.shell.db['stored_aliases'] = stored
201 201
202 202 @line_magic
203 203 def rehashx(self, parameter_s=''):
204 204 """Update the alias table with all executable files in $PATH.
205 205
206 206 rehashx explicitly checks that every entry in $PATH is a file
207 207 with execute access (os.X_OK).
208 208
209 209 Under Windows, it checks executability as a match against a
210 210 '|'-separated string of extensions, stored in the IPython config
211 211 variable win_exec_ext. This defaults to 'exe|com|bat'.
212 212
213 213 This function also resets the root module cache of module completer,
214 214 used on slow filesystems.
215 215 """
216 216 from IPython.core.alias import InvalidAliasError
217 217
218 218 # for the benefit of module completer in ipy_completers.py
219 219 del self.shell.db['rootmodules_cache']
220 220
221 221 path = [os.path.abspath(os.path.expanduser(p)) for p in
222 222 os.environ.get('PATH','').split(os.pathsep)]
223 223
224 224 syscmdlist = []
225 225 savedir = os.getcwd()
226 226
227 227 # Now walk the paths looking for executables to alias.
228 228 try:
229 229 # write the whole loop for posix/Windows so we don't have an if in
230 230 # the innermost part
231 231 if self.is_posix:
232 232 for pdir in path:
233 233 try:
234 234 os.chdir(pdir)
235 235 except OSError:
236 236 continue
237 237
238 238 # for python 3.6+ rewrite to: with os.scandir(pdir) as dirlist:
239 239 dirlist = os.scandir(path=pdir)
240 240 for ff in dirlist:
241 241 if self.isexec(ff):
242 242 fname = ff.name
243 243 try:
244 244 # Removes dots from the name since ipython
245 245 # will assume names with dots to be python.
246 246 if not self.shell.alias_manager.is_alias(fname):
247 247 self.shell.alias_manager.define_alias(
248 248 fname.replace('.',''), fname)
249 249 except InvalidAliasError:
250 250 pass
251 251 else:
252 252 syscmdlist.append(fname)
253 253 else:
254 254 no_alias = Alias.blacklist
255 255 for pdir in path:
256 256 try:
257 257 os.chdir(pdir)
258 258 except OSError:
259 259 continue
260 260
261 261 # for python 3.6+ rewrite to: with os.scandir(pdir) as dirlist:
262 262 dirlist = os.scandir(pdir)
263 263 for ff in dirlist:
264 264 fname = ff.name
265 265 base, ext = os.path.splitext(fname)
266 266 if self.isexec(ff) and base.lower() not in no_alias:
267 267 if ext.lower() == '.exe':
268 268 fname = base
269 269 try:
270 270 # Removes dots from the name since ipython
271 271 # will assume names with dots to be python.
272 272 self.shell.alias_manager.define_alias(
273 273 base.lower().replace('.',''), fname)
274 274 except InvalidAliasError:
275 275 pass
276 276 syscmdlist.append(fname)
277 277
278 278 self.shell.db['syscmdlist'] = syscmdlist
279 279 finally:
280 280 os.chdir(savedir)
281 281
282 282 @skip_doctest
283 283 @line_magic
284 284 def pwd(self, parameter_s=''):
285 285 """Return the current working directory path.
286 286
287 287 Examples
288 288 --------
289 289 ::
290 290
291 291 In [9]: pwd
292 292 Out[9]: '/home/tsuser/sprint/ipython'
293 293 """
294 294 try:
295 295 return os.getcwd()
296 296 except FileNotFoundError as e:
297 297 raise UsageError("CWD no longer exists - please use %cd to change directory.") from e
298 298
299 299 @skip_doctest
300 300 @line_magic
301 301 def cd(self, parameter_s=''):
302 302 """Change the current working directory.
303 303
304 304 This command automatically maintains an internal list of directories
305 305 you visit during your IPython session, in the variable ``_dh``. The
306 306 command :magic:`%dhist` shows this history nicely formatted. You can
307 307 also do ``cd -<tab>`` to see directory history conveniently.
308 308 Usage:
309 309
310 310 - ``cd 'dir'``: changes to directory 'dir'.
311 311 - ``cd -``: changes to the last visited directory.
312 312 - ``cd -<n>``: changes to the n-th directory in the directory history.
313 313 - ``cd --foo``: change to directory that matches 'foo' in history
314 314 - ``cd -b <bookmark_name>``: jump to a bookmark set by %bookmark
315 315 - Hitting a tab key after ``cd -b`` allows you to tab-complete
316 316 bookmark names.
317 317
318 318 .. note::
319 319 ``cd <bookmark_name>`` is enough if there is no directory
320 320 ``<bookmark_name>``, but a bookmark with the name exists.
321 321
322 322
323 323 Options:
324 324
325 325 -q Be quiet. Do not print the working directory after the
326 326 cd command is executed. By default IPython's cd
327 327 command does print this directory, since the default
328 328 prompts do not display path information.
329 329
330 330 .. note::
331 331 Note that ``!cd`` doesn't work for this purpose because the shell
332 332 where ``!command`` runs is immediately discarded after executing
333 333 'command'.
334 334
335 335
336 336 Examples
337 337 --------
338 338 ::
339 339
340 340 In [10]: cd parent/child
341 341 /home/tsuser/parent/child
342 342 """
343 343
344 344 try:
345 345 oldcwd = os.getcwd()
346 346 except FileNotFoundError:
347 347 # Happens if the CWD has been deleted.
348 348 oldcwd = None
349 349
350 350 numcd = re.match(r'(-)(\d+)$',parameter_s)
351 351 # jump in directory history by number
352 352 if numcd:
353 353 nn = int(numcd.group(2))
354 354 try:
355 355 ps = self.shell.user_ns['_dh'][nn]
356 356 except IndexError:
357 357 print('The requested directory does not exist in history.')
358 358 return
359 359 else:
360 360 opts = {}
361 361 elif parameter_s.startswith('--'):
362 362 ps = None
363 363 fallback = None
364 364 pat = parameter_s[2:]
365 365 dh = self.shell.user_ns['_dh']
366 366 # first search only by basename (last component)
367 367 for ent in reversed(dh):
368 368 if pat in os.path.basename(ent) and os.path.isdir(ent):
369 369 ps = ent
370 370 break
371 371
372 372 if fallback is None and pat in ent and os.path.isdir(ent):
373 373 fallback = ent
374 374
375 375 # if we have no last part match, pick the first full path match
376 376 if ps is None:
377 377 ps = fallback
378 378
379 379 if ps is None:
380 380 print("No matching entry in directory history")
381 381 return
382 382 else:
383 383 opts = {}
384 384
385 385
386 386 else:
387 387 opts, ps = self.parse_options(parameter_s, 'qb', mode='string')
388 388 # jump to previous
389 389 if ps == '-':
390 390 try:
391 391 ps = self.shell.user_ns['_dh'][-2]
392 392 except IndexError as e:
393 393 raise UsageError('%cd -: No previous directory to change to.') from e
394 394 # jump to bookmark if needed
395 395 else:
396 396 if not os.path.isdir(ps) or 'b' in opts:
397 397 bkms = self.shell.db.get('bookmarks', {})
398 398
399 399 if ps in bkms:
400 400 target = bkms[ps]
401 401 print('(bookmark:%s) -> %s' % (ps, target))
402 402 ps = target
403 403 else:
404 404 if 'b' in opts:
405 405 raise UsageError("Bookmark '%s' not found. "
406 406 "Use '%%bookmark -l' to see your bookmarks." % ps)
407 407
408 408 # at this point ps should point to the target dir
409 409 if ps:
410 410 try:
411 411 os.chdir(os.path.expanduser(ps))
412 412 if hasattr(self.shell, 'term_title') and self.shell.term_title:
413 413 set_term_title(self.shell.term_title_format.format(cwd=abbrev_cwd()))
414 414 except OSError:
415 415 print(sys.exc_info()[1])
416 416 else:
417 417 cwd = os.getcwd()
418 418 dhist = self.shell.user_ns['_dh']
419 419 if oldcwd != cwd:
420 420 dhist.append(cwd)
421 421 self.shell.db['dhist'] = compress_dhist(dhist)[-100:]
422 422
423 423 else:
424 424 os.chdir(self.shell.home_dir)
425 425 if hasattr(self.shell, 'term_title') and self.shell.term_title:
426 426 set_term_title(self.shell.term_title_format.format(cwd="~"))
427 427 cwd = os.getcwd()
428 428 dhist = self.shell.user_ns['_dh']
429 429
430 430 if oldcwd != cwd:
431 431 dhist.append(cwd)
432 432 self.shell.db['dhist'] = compress_dhist(dhist)[-100:]
433 433 if not 'q' in opts and not self.cd_force_quiet and self.shell.user_ns['_dh']:
434 434 print(self.shell.user_ns['_dh'][-1])
435 435
436 436 @line_magic
437 437 def env(self, parameter_s=''):
438 438 """Get, set, or list environment variables.
439 439
440 440 Usage:\\
441 441
442 442 :``%env``: lists all environment variables/values
443 443 :``%env var``: get value for var
444 444 :``%env var val``: set value for var
445 445 :``%env var=val``: set value for var
446 446 :``%env var=$val``: set value for var, using python expansion if possible
447 447 """
448 448 if parameter_s.strip():
449 449 split = '=' if '=' in parameter_s else ' '
450 450 bits = parameter_s.split(split)
451 451 if len(bits) == 1:
452 452 key = parameter_s.strip()
453 453 if key in os.environ:
454 454 return os.environ[key]
455 455 else:
456 456 err = "Environment does not have key: {0}".format(key)
457 457 raise UsageError(err)
458 458 if len(bits) > 1:
459 459 return self.set_env(parameter_s)
460 460 env = dict(os.environ)
461 461 # hide likely secrets when printing the whole environment
462 462 for key in list(env):
463 463 if any(s in key.lower() for s in ('key', 'token', 'secret')):
464 464 env[key] = '<hidden>'
465 465
466 466 return env
467 467
468 468 @line_magic
469 469 def set_env(self, parameter_s):
470 470 """Set environment variables. Assumptions are that either "val" is a
471 471 name in the user namespace, or val is something that evaluates to a
472 472 string.
473 473
474 474 Usage:\\
475 475 %set_env var val: set value for var
476 476 %set_env var=val: set value for var
477 477 %set_env var=$val: set value for var, using python expansion if possible
478 478 """
479 479 split = '=' if '=' in parameter_s else ' '
480 480 bits = parameter_s.split(split, 1)
481 481 if not parameter_s.strip() or len(bits)<2:
482 482 raise UsageError("usage is 'set_env var=val'")
483 483 var = bits[0].strip()
484 484 val = bits[1].strip()
485 485 if re.match(r'.*\s.*', var):
486 486 # an environment variable with whitespace is almost certainly
487 487 # not what the user intended. what's more likely is the wrong
488 488 # split was chosen, ie for "set_env cmd_args A=B", we chose
489 489 # '=' for the split and should have chosen ' '. to get around
490 490 # this, users should just assign directly to os.environ or use
491 491 # standard magic {var} expansion.
492 492 err = "refusing to set env var with whitespace: '{0}'"
493 493 err = err.format(val)
494 494 raise UsageError(err)
495 495 os.environ[var] = val
496 496 print('env: {0}={1}'.format(var,val))
497 497
498 498 @line_magic
499 499 def pushd(self, parameter_s=''):
500 500 """Place the current dir on stack and change directory.
501 501
502 502 Usage:\\
503 503 %pushd ['dirname']
504 504 """
505 505
506 506 dir_s = self.shell.dir_stack
507 507 tgt = os.path.expanduser(parameter_s)
508 508 cwd = os.getcwd().replace(self.shell.home_dir,'~')
509 509 if tgt:
510 510 self.cd(parameter_s)
511 511 dir_s.insert(0,cwd)
512 512 return self.shell.run_line_magic('dirs', '')
513 513
514 514 @line_magic
515 515 def popd(self, parameter_s=''):
516 516 """Change to directory popped off the top of the stack.
517 517 """
518 518 if not self.shell.dir_stack:
519 519 raise UsageError("%popd on empty stack")
520 520 top = self.shell.dir_stack.pop(0)
521 521 self.cd(top)
522 522 print("popd ->",top)
523 523
524 524 @line_magic
525 525 def dirs(self, parameter_s=''):
526 526 """Return the current directory stack."""
527 527
528 528 return self.shell.dir_stack
529 529
530 530 @line_magic
531 531 def dhist(self, parameter_s=''):
532 532 """Print your history of visited directories.
533 533
534 534 %dhist -> print full history\\
535 535 %dhist n -> print last n entries only\\
536 536 %dhist n1 n2 -> print entries between n1 and n2 (n2 not included)\\
537 537
538 538 This history is automatically maintained by the %cd command, and
539 539 always available as the global list variable _dh. You can use %cd -<n>
540 540 to go to directory number <n>.
541 541
542 542 Note that most of time, you should view directory history by entering
543 543 cd -<TAB>.
544 544
545 545 """
546 546
547 547 dh = self.shell.user_ns['_dh']
548 548 if parameter_s:
549 549 try:
550 550 args = map(int,parameter_s.split())
551 551 except:
552 552 self.arg_err(self.dhist)
553 553 return
554 554 if len(args) == 1:
555 555 ini,fin = max(len(dh)-(args[0]),0),len(dh)
556 556 elif len(args) == 2:
557 557 ini,fin = args
558 558 fin = min(fin, len(dh))
559 559 else:
560 560 self.arg_err(self.dhist)
561 561 return
562 562 else:
563 563 ini,fin = 0,len(dh)
564 564 print('Directory history (kept in _dh)')
565 565 for i in range(ini, fin):
566 566 print("%d: %s" % (i, dh[i]))
567 567
568 568 @skip_doctest
569 569 @line_magic
570 570 def sc(self, parameter_s=''):
571 571 """Shell capture - run shell command and capture output (DEPRECATED use !).
572 572
573 573 DEPRECATED. Suboptimal, retained for backwards compatibility.
574 574
575 575 You should use the form 'var = !command' instead. Example:
576 576
577 577 "%sc -l myfiles = ls ~" should now be written as
578 578
579 579 "myfiles = !ls ~"
580 580
581 581 myfiles.s, myfiles.l and myfiles.n still apply as documented
582 582 below.
583 583
584 584 --
585 585 %sc [options] varname=command
586 586
587 587 IPython will run the given command using commands.getoutput(), and
588 588 will then update the user's interactive namespace with a variable
589 589 called varname, containing the value of the call. Your command can
590 590 contain shell wildcards, pipes, etc.
591 591
592 592 The '=' sign in the syntax is mandatory, and the variable name you
593 593 supply must follow Python's standard conventions for valid names.
594 594
595 595 (A special format without variable name exists for internal use)
596 596
597 597 Options:
598 598
599 599 -l: list output. Split the output on newlines into a list before
600 600 assigning it to the given variable. By default the output is stored
601 601 as a single string.
602 602
603 603 -v: verbose. Print the contents of the variable.
604 604
605 605 In most cases you should not need to split as a list, because the
606 606 returned value is a special type of string which can automatically
607 607 provide its contents either as a list (split on newlines) or as a
608 608 space-separated string. These are convenient, respectively, either
609 609 for sequential processing or to be passed to a shell command.
610 610
611 611 For example::
612 612
613 613 # Capture into variable a
614 614 In [1]: sc a=ls *py
615 615
616 616 # a is a string with embedded newlines
617 617 In [2]: a
618 618 Out[2]: 'setup.py\\nwin32_manual_post_install.py'
619 619
620 620 # which can be seen as a list:
621 621 In [3]: a.l
622 622 Out[3]: ['setup.py', 'win32_manual_post_install.py']
623 623
624 624 # or as a whitespace-separated string:
625 625 In [4]: a.s
626 626 Out[4]: 'setup.py win32_manual_post_install.py'
627 627
628 628 # a.s is useful to pass as a single command line:
629 629 In [5]: !wc -l $a.s
630 630 146 setup.py
631 631 130 win32_manual_post_install.py
632 632 276 total
633 633
634 634 # while the list form is useful to loop over:
635 635 In [6]: for f in a.l:
636 636 ...: !wc -l $f
637 637 ...:
638 638 146 setup.py
639 639 130 win32_manual_post_install.py
640 640
641 641 Similarly, the lists returned by the -l option are also special, in
642 642 the sense that you can equally invoke the .s attribute on them to
643 643 automatically get a whitespace-separated string from their contents::
644 644
645 645 In [7]: sc -l b=ls *py
646 646
647 647 In [8]: b
648 648 Out[8]: ['setup.py', 'win32_manual_post_install.py']
649 649
650 650 In [9]: b.s
651 651 Out[9]: 'setup.py win32_manual_post_install.py'
652 652
653 653 In summary, both the lists and strings used for output capture have
654 654 the following special attributes::
655 655
656 656 .l (or .list) : value as list.
657 657 .n (or .nlstr): value as newline-separated string.
658 658 .s (or .spstr): value as space-separated string.
659 659 """
660 660
661 661 opts,args = self.parse_options(parameter_s, 'lv')
662 662 # Try to get a variable name and command to run
663 663 try:
664 664 # the variable name must be obtained from the parse_options
665 665 # output, which uses shlex.split to strip options out.
666 666 var,_ = args.split('=', 1)
667 667 var = var.strip()
668 668 # But the command has to be extracted from the original input
669 669 # parameter_s, not on what parse_options returns, to avoid the
670 670 # quote stripping which shlex.split performs on it.
671 671 _,cmd = parameter_s.split('=', 1)
672 672 except ValueError:
673 673 var,cmd = '',''
674 674 # If all looks ok, proceed
675 675 split = 'l' in opts
676 676 out = self.shell.getoutput(cmd, split=split)
677 677 if 'v' in opts:
678 678 print('%s ==\n%s' % (var, pformat(out)))
679 679 if var:
680 680 self.shell.user_ns.update({var:out})
681 681 else:
682 682 return out
683 683
684 684 @line_cell_magic
685 685 def sx(self, line='', cell=None):
686 686 """Shell execute - run shell command and capture output (!! is short-hand).
687 687
688 688 %sx command
689 689
690 690 IPython will run the given command using commands.getoutput(), and
691 691 return the result formatted as a list (split on '\\n'). Since the
692 692 output is _returned_, it will be stored in ipython's regular output
693 693 cache Out[N] and in the '_N' automatic variables.
694 694
695 695 Notes:
696 696
697 697 1) If an input line begins with '!!', then %sx is automatically
698 698 invoked. That is, while::
699 699
700 700 !ls
701 701
702 702 causes ipython to simply issue system('ls'), typing::
703 703
704 704 !!ls
705 705
706 706 is a shorthand equivalent to::
707 707
708 708 %sx ls
709 709
710 710 2) %sx differs from %sc in that %sx automatically splits into a list,
711 711 like '%sc -l'. The reason for this is to make it as easy as possible
712 712 to process line-oriented shell output via further python commands.
713 713 %sc is meant to provide much finer control, but requires more
714 714 typing.
715 715
716 716 3) Just like %sc -l, this is a list with special attributes:
717 717 ::
718 718
719 719 .l (or .list) : value as list.
720 720 .n (or .nlstr): value as newline-separated string.
721 721 .s (or .spstr): value as whitespace-separated string.
722 722
723 723 This is very useful when trying to use such lists as arguments to
724 724 system commands."""
725 725
726 726 if cell is None:
727 727 # line magic
728 728 return self.shell.getoutput(line)
729 729 else:
730 730 opts,args = self.parse_options(line, '', 'out=')
731 731 output = self.shell.getoutput(cell)
732 732 out_name = opts.get('out', opts.get('o'))
733 733 if out_name:
734 734 self.shell.user_ns[out_name] = output
735 735 else:
736 736 return output
737 737
738 738 system = line_cell_magic('system')(sx)
739 739 bang = cell_magic('!')(sx)
740 740
741 741 @line_magic
742 742 def bookmark(self, parameter_s=''):
743 743 """Manage IPython's bookmark system.
744 744
745 745 %bookmark <name> - set bookmark to current dir
746 746 %bookmark <name> <dir> - set bookmark to <dir>
747 747 %bookmark -l - list all bookmarks
748 748 %bookmark -d <name> - remove bookmark
749 749 %bookmark -r - remove all bookmarks
750 750
751 751 You can later on access a bookmarked folder with::
752 752
753 753 %cd -b <name>
754 754
755 755 or simply '%cd <name>' if there is no directory called <name> AND
756 756 there is such a bookmark defined.
757 757
758 758 Your bookmarks persist through IPython sessions, but they are
759 759 associated with each profile."""
760 760
761 761 opts,args = self.parse_options(parameter_s,'drl',mode='list')
762 762 if len(args) > 2:
763 763 raise UsageError("%bookmark: too many arguments")
764 764
765 765 bkms = self.shell.db.get('bookmarks',{})
766 766
767 767 if 'd' in opts:
768 768 try:
769 769 todel = args[0]
770 770 except IndexError as e:
771 771 raise UsageError(
772 772 "%bookmark -d: must provide a bookmark to delete") from e
773 773 else:
774 774 try:
775 775 del bkms[todel]
776 776 except KeyError as e:
777 777 raise UsageError(
778 778 "%%bookmark -d: Can't delete bookmark '%s'" % todel) from e
779 779
780 780 elif 'r' in opts:
781 781 bkms = {}
782 782 elif 'l' in opts:
783 783 bks = sorted(bkms)
784 784 if bks:
785 785 size = max(map(len, bks))
786 786 else:
787 787 size = 0
788 788 fmt = '%-'+str(size)+'s -> %s'
789 789 print('Current bookmarks:')
790 790 for bk in bks:
791 791 print(fmt % (bk, bkms[bk]))
792 792 else:
793 793 if not args:
794 794 raise UsageError("%bookmark: You must specify the bookmark name")
795 795 elif len(args)==1:
796 796 bkms[args[0]] = os.getcwd()
797 797 elif len(args)==2:
798 798 bkms[args[0]] = args[1]
799 799 self.shell.db['bookmarks'] = bkms
800 800
801 801 @line_magic
802 802 def pycat(self, parameter_s=''):
803 803 """Show a syntax-highlighted file through a pager.
804 804
805 805 This magic is similar to the cat utility, but it will assume the file
806 806 to be Python source and will show it with syntax highlighting.
807 807
808 808 This magic command can either take a local filename, an url,
809 an history range (see %history) or a macro as argument ::
809 an history range (see %history) or a macro as argument.
810
811 If no parameter is given, prints out history of current session up to
812 this point. ::
810 813
811 814 %pycat myscript.py
812 815 %pycat 7-27
813 816 %pycat myMacro
814 817 %pycat http://www.example.com/myscript.py
815 818 """
816 if not parameter_s:
817 raise UsageError('Missing filename, URL, input history range, '
818 'or macro.')
819
820 try :
819 try:
821 820 cont = self.shell.find_user_code(parameter_s, skip_encoding_cookie=False)
822 821 except (ValueError, IOError):
823 822 print("Error: no such file, variable, URL, history range or macro")
824 823 return
825 824
826 825 page.page(self.shell.pycolorize(source_to_unicode(cont)))
827 826
828 827 @magic_arguments.magic_arguments()
829 828 @magic_arguments.argument(
830 829 '-a', '--append', action='store_true', default=False,
831 830 help='Append contents of the cell to an existing file. '
832 831 'The file will be created if it does not exist.'
833 832 )
834 833 @magic_arguments.argument(
835 834 'filename', type=str,
836 835 help='file to write'
837 836 )
838 837 @cell_magic
839 838 def writefile(self, line, cell):
840 839 """Write the contents of the cell to a file.
841 840
842 841 The file will be overwritten unless the -a (--append) flag is specified.
843 842 """
844 843 args = magic_arguments.parse_argstring(self.writefile, line)
845 844 if re.match(r'^(\'.*\')|(".*")$', args.filename):
846 845 filename = os.path.expanduser(args.filename[1:-1])
847 846 else:
848 847 filename = os.path.expanduser(args.filename)
849 848
850 849 if os.path.exists(filename):
851 850 if args.append:
852 851 print("Appending to %s" % filename)
853 852 else:
854 853 print("Overwriting %s" % filename)
855 854 else:
856 855 print("Writing %s" % filename)
857 856
858 857 mode = 'a' if args.append else 'w'
859 858 with io.open(filename, mode, encoding='utf-8') as f:
860 859 f.write(cell)
@@ -1,215 +1,223 b''
1 1 # coding: utf-8
2 2 """Tests for the IPython tab-completion machinery.
3 3 """
4 4 #-----------------------------------------------------------------------------
5 5 # Module imports
6 6 #-----------------------------------------------------------------------------
7 7
8 8 # stdlib
9 9 import io
10 10 from pathlib import Path
11 11 import sys
12 12 import tempfile
13 13 from datetime import datetime
14 14 import sqlite3
15 15
16 16 # third party
17 17 import nose.tools as nt
18 18
19 19 # our own packages
20 20 from traitlets.config.loader import Config
21 21 from IPython.utils.tempdir import TemporaryDirectory
22 22 from IPython.core.history import HistoryManager, extract_hist_ranges
23 23 from IPython.testing.decorators import skipif
24 24
25 25 def test_proper_default_encoding():
26 26 nt.assert_equal(sys.getdefaultencoding(), "utf-8")
27 27
28 28 @skipif(sqlite3.sqlite_version_info > (3,24,0))
29 29 def test_history():
30 30 ip = get_ipython()
31 31 with TemporaryDirectory() as tmpdir:
32 32 tmp_path = Path(tmpdir)
33 33 hist_manager_ori = ip.history_manager
34 34 hist_file = tmp_path / "history.sqlite"
35 35 try:
36 36 ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file)
37 37 hist = [u'a=1', u'def f():\n test = 1\n return test', u"b='β‚¬Γ†ΒΎΓ·ΓŸ'"]
38 38 for i, h in enumerate(hist, start=1):
39 39 ip.history_manager.store_inputs(i, h)
40 40
41 41 ip.history_manager.db_log_output = True
42 42 # Doesn't match the input, but we'll just check it's stored.
43 43 ip.history_manager.output_hist_reprs[3] = "spam"
44 44 ip.history_manager.store_output(3)
45 45
46 46 nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
47 47
48 48 # Detailed tests for _get_range_session
49 49 grs = ip.history_manager._get_range_session
50 50 nt.assert_equal(list(grs(start=2,stop=-1)), list(zip([0], [2], hist[1:-1])))
51 51 nt.assert_equal(list(grs(start=-2)), list(zip([0,0], [2,3], hist[-2:])))
52 52 nt.assert_equal(list(grs(output=True)), list(zip([0,0,0], [1,2,3], zip(hist, [None,None,'spam']))))
53 53
54 54 # Check whether specifying a range beyond the end of the current
55 55 # session results in an error (gh-804)
56 56 ip.magic('%hist 2-500')
57 57
58 58 # Check that we can write non-ascii characters to a file
59 59 ip.magic("%%hist -f %s" % (tmp_path / "test1"))
60 60 ip.magic("%%hist -pf %s" % (tmp_path / "test2"))
61 61 ip.magic("%%hist -nf %s" % (tmp_path / "test3"))
62 62 ip.magic("%%save %s 1-10" % (tmp_path / "test4"))
63 63
64 64 # New session
65 65 ip.history_manager.reset()
66 66 newcmds = [u"z=5",
67 67 u"class X(object):\n pass",
68 68 u"k='p'",
69 69 u"z=5"]
70 70 for i, cmd in enumerate(newcmds, start=1):
71 71 ip.history_manager.store_inputs(i, cmd)
72 72 gothist = ip.history_manager.get_range(start=1, stop=4)
73 73 nt.assert_equal(list(gothist), list(zip([0,0,0],[1,2,3], newcmds)))
74 74 # Previous session:
75 75 gothist = ip.history_manager.get_range(-1, 1, 4)
76 76 nt.assert_equal(list(gothist), list(zip([1,1,1],[1,2,3], hist)))
77 77
78 78 newhist = [(2, i, c) for (i, c) in enumerate(newcmds, 1)]
79 79
80 80 # Check get_hist_tail
81 81 gothist = ip.history_manager.get_tail(5, output=True,
82 82 include_latest=True)
83 83 expected = [(1, 3, (hist[-1], "spam"))] \
84 84 + [(s, n, (c, None)) for (s, n, c) in newhist]
85 85 nt.assert_equal(list(gothist), expected)
86 86
87 87 gothist = ip.history_manager.get_tail(2)
88 88 expected = newhist[-3:-1]
89 89 nt.assert_equal(list(gothist), expected)
90 90
91 91 # Check get_hist_search
92 92
93 93 gothist = ip.history_manager.search("*test*")
94 94 nt.assert_equal(list(gothist), [(1,2,hist[1])] )
95 95
96 96 gothist = ip.history_manager.search("*=*")
97 97 nt.assert_equal(list(gothist),
98 98 [(1, 1, hist[0]),
99 99 (1, 2, hist[1]),
100 100 (1, 3, hist[2]),
101 101 newhist[0],
102 102 newhist[2],
103 103 newhist[3]])
104 104
105 105 gothist = ip.history_manager.search("*=*", n=4)
106 106 nt.assert_equal(list(gothist),
107 107 [(1, 3, hist[2]),
108 108 newhist[0],
109 109 newhist[2],
110 110 newhist[3]])
111 111
112 112 gothist = ip.history_manager.search("*=*", unique=True)
113 113 nt.assert_equal(list(gothist),
114 114 [(1, 1, hist[0]),
115 115 (1, 2, hist[1]),
116 116 (1, 3, hist[2]),
117 117 newhist[2],
118 118 newhist[3]])
119 119
120 120 gothist = ip.history_manager.search("*=*", unique=True, n=3)
121 121 nt.assert_equal(list(gothist),
122 122 [(1, 3, hist[2]),
123 123 newhist[2],
124 124 newhist[3]])
125 125
126 126 gothist = ip.history_manager.search("b*", output=True)
127 127 nt.assert_equal(list(gothist), [(1,3,(hist[2],"spam"))] )
128 128
129 129 # Cross testing: check that magic %save can get previous session.
130 130 testfilename = (tmp_path / "test.py").resolve()
131 131 ip.magic("save " + str(testfilename) + " ~1/1-3")
132 132 with io.open(testfilename, encoding='utf-8') as testfile:
133 133 nt.assert_equal(testfile.read(),
134 134 u"# coding: utf-8\n" + u"\n".join(hist)+u"\n")
135 135
136 136 # Duplicate line numbers - check that it doesn't crash, and
137 137 # gets a new session
138 138 ip.history_manager.store_inputs(1, "rogue")
139 139 ip.history_manager.writeout_cache()
140 140 nt.assert_equal(ip.history_manager.session_number, 3)
141 141 finally:
142 142 # Ensure saving thread is shut down before we try to clean up the files
143 143 ip.history_manager.save_thread.stop()
144 144 # Forcibly close database rather than relying on garbage collection
145 145 ip.history_manager.db.close()
146 146 # Restore history manager
147 147 ip.history_manager = hist_manager_ori
148 148
149 149
150 150 def test_extract_hist_ranges():
151 151 instr = "1 2/3 ~4/5-6 ~4/7-~4/9 ~9/2-~7/5 ~10/"
152 152 expected = [(0, 1, 2), # 0 == current session
153 153 (2, 3, 4),
154 154 (-4, 5, 7),
155 155 (-4, 7, 10),
156 156 (-9, 2, None), # None == to end
157 157 (-8, 1, None),
158 158 (-7, 1, 6),
159 159 (-10, 1, None)]
160 160 actual = list(extract_hist_ranges(instr))
161 161 nt.assert_equal(actual, expected)
162 162
163
164 def test_extract_hist_ranges_empty_str():
165 instr = ""
166 expected = [(0, 1, None)] # 0 == current session, None == to end
167 actual = list(extract_hist_ranges(instr))
168 nt.assert_equal(actual, expected)
169
170
163 171 def test_magic_rerun():
164 172 """Simple test for %rerun (no args -> rerun last line)"""
165 173 ip = get_ipython()
166 174 ip.run_cell("a = 10", store_history=True)
167 175 ip.run_cell("a += 1", store_history=True)
168 176 nt.assert_equal(ip.user_ns["a"], 11)
169 177 ip.run_cell("%rerun", store_history=True)
170 178 nt.assert_equal(ip.user_ns["a"], 12)
171 179
172 180 def test_timestamp_type():
173 181 ip = get_ipython()
174 182 info = ip.history_manager.get_session_info()
175 183 nt.assert_true(isinstance(info[1], datetime))
176 184
177 185 def test_hist_file_config():
178 186 cfg = Config()
179 187 tfile = tempfile.NamedTemporaryFile(delete=False)
180 188 cfg.HistoryManager.hist_file = Path(tfile.name)
181 189 try:
182 190 hm = HistoryManager(shell=get_ipython(), config=cfg)
183 191 nt.assert_equal(hm.hist_file, cfg.HistoryManager.hist_file)
184 192 finally:
185 193 try:
186 194 Path(tfile.name).unlink()
187 195 except OSError:
188 196 # same catch as in testing.tools.TempFileMixin
189 197 # On Windows, even though we close the file, we still can't
190 198 # delete it. I have no clue why
191 199 pass
192 200
193 201 def test_histmanager_disabled():
194 202 """Ensure that disabling the history manager doesn't create a database."""
195 203 cfg = Config()
196 204 cfg.HistoryAccessor.enabled = False
197 205
198 206 ip = get_ipython()
199 207 with TemporaryDirectory() as tmpdir:
200 208 hist_manager_ori = ip.history_manager
201 209 hist_file = Path(tmpdir) / "history.sqlite"
202 210 cfg.HistoryManager.hist_file = hist_file
203 211 try:
204 212 ip.history_manager = HistoryManager(shell=ip, config=cfg)
205 213 hist = [u'a=1', u'def f():\n test = 1\n return test', u"b='β‚¬Γ†ΒΎΓ·ΓŸ'"]
206 214 for i, h in enumerate(hist, start=1):
207 215 ip.history_manager.store_inputs(i, h)
208 216 nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
209 217 ip.history_manager.reset()
210 218 ip.history_manager.end_session()
211 219 finally:
212 220 ip.history_manager = hist_manager_ori
213 221
214 222 # hist_file should not be created
215 223 nt.assert_false(hist_file.exists())
@@ -1,1279 +1,1302 b''
1 1 # -*- coding: utf-8 -*-
2 2 """Tests for various magic functions.
3 3
4 4 Needs to be run by nose (to make ipython session available).
5 5 """
6 6
7 7 import io
8 8 import os
9 9 import re
10 10 import sys
11 11 import warnings
12 12 from textwrap import dedent
13 13 from unittest import TestCase
14 14 from unittest import mock
15 15 from importlib import invalidate_caches
16 16 from io import StringIO
17 17 from pathlib import Path
18 18
19 19 import nose.tools as nt
20 20
21 21 import shlex
22 22
23 23 from IPython import get_ipython
24 24 from IPython.core import magic
25 25 from IPython.core.error import UsageError
26 26 from IPython.core.magic import (Magics, magics_class, line_magic,
27 27 cell_magic,
28 28 register_line_magic, register_cell_magic)
29 29 from IPython.core.magics import execution, script, code, logging, osm
30 30 from IPython.testing import decorators as dec
31 31 from IPython.testing import tools as tt
32 32 from IPython.utils.io import capture_output
33 33 from IPython.utils.tempdir import (TemporaryDirectory,
34 34 TemporaryWorkingDirectory)
35 35 from IPython.utils.process import find_cmd
36 36 from .test_debugger import PdbTestInput
37 37
38 38 import pytest
39 39
40 40
41 41 @magic.magics_class
42 42 class DummyMagics(magic.Magics): pass
43 43
44 44 def test_extract_code_ranges():
45 45 instr = "1 3 5-6 7-9 10:15 17: :10 10- -13 :"
46 46 expected = [(0, 1),
47 47 (2, 3),
48 48 (4, 6),
49 49 (6, 9),
50 50 (9, 14),
51 51 (16, None),
52 52 (None, 9),
53 53 (9, None),
54 54 (None, 13),
55 55 (None, None)]
56 56 actual = list(code.extract_code_ranges(instr))
57 57 nt.assert_equal(actual, expected)
58 58
59 59 def test_extract_symbols():
60 60 source = """import foo\na = 10\ndef b():\n return 42\n\n\nclass A: pass\n\n\n"""
61 61 symbols_args = ["a", "b", "A", "A,b", "A,a", "z"]
62 62 expected = [([], ['a']),
63 63 (["def b():\n return 42\n"], []),
64 64 (["class A: pass\n"], []),
65 65 (["class A: pass\n", "def b():\n return 42\n"], []),
66 66 (["class A: pass\n"], ['a']),
67 67 ([], ['z'])]
68 68 for symbols, exp in zip(symbols_args, expected):
69 69 nt.assert_equal(code.extract_symbols(source, symbols), exp)
70 70
71 71
72 72 def test_extract_symbols_raises_exception_with_non_python_code():
73 73 source = ("=begin A Ruby program :)=end\n"
74 74 "def hello\n"
75 75 "puts 'Hello world'\n"
76 76 "end")
77 77 with nt.assert_raises(SyntaxError):
78 78 code.extract_symbols(source, "hello")
79 79
80 80
81 81 def test_magic_not_found():
82 82 # magic not found raises UsageError
83 83 with nt.assert_raises(UsageError):
84 84 _ip.magic('doesntexist')
85 85
86 86 # ensure result isn't success when a magic isn't found
87 87 result = _ip.run_cell('%doesntexist')
88 88 assert isinstance(result.error_in_exec, UsageError)
89 89
90 90
91 91 def test_cell_magic_not_found():
92 92 # magic not found raises UsageError
93 93 with nt.assert_raises(UsageError):
94 94 _ip.run_cell_magic('doesntexist', 'line', 'cell')
95 95
96 96 # ensure result isn't success when a magic isn't found
97 97 result = _ip.run_cell('%%doesntexist')
98 98 assert isinstance(result.error_in_exec, UsageError)
99 99
100 100
101 101 def test_magic_error_status():
102 102 def fail(shell):
103 103 1/0
104 104 _ip.register_magic_function(fail)
105 105 result = _ip.run_cell('%fail')
106 106 assert isinstance(result.error_in_exec, ZeroDivisionError)
107 107
108 108
109 109 def test_config():
110 110 """ test that config magic does not raise
111 111 can happen if Configurable init is moved too early into
112 112 Magics.__init__ as then a Config object will be registered as a
113 113 magic.
114 114 """
115 115 ## should not raise.
116 116 _ip.magic('config')
117 117
118 118 def test_config_available_configs():
119 119 """ test that config magic prints available configs in unique and
120 120 sorted order. """
121 121 with capture_output() as captured:
122 122 _ip.magic('config')
123 123
124 124 stdout = captured.stdout
125 125 config_classes = stdout.strip().split('\n')[1:]
126 126 nt.assert_list_equal(config_classes, sorted(set(config_classes)))
127 127
128 128 def test_config_print_class():
129 129 """ test that config with a classname prints the class's options. """
130 130 with capture_output() as captured:
131 131 _ip.magic('config TerminalInteractiveShell')
132 132
133 133 stdout = captured.stdout
134 134 if not re.match("TerminalInteractiveShell.* options", stdout.splitlines()[0]):
135 135 print(stdout)
136 136 raise AssertionError("1st line of stdout not like "
137 137 "'TerminalInteractiveShell.* options'")
138 138
139 139 def test_rehashx():
140 140 # clear up everything
141 141 _ip.alias_manager.clear_aliases()
142 142 del _ip.db['syscmdlist']
143 143
144 144 _ip.magic('rehashx')
145 145 # Practically ALL ipython development systems will have more than 10 aliases
146 146
147 147 nt.assert_true(len(_ip.alias_manager.aliases) > 10)
148 148 for name, cmd in _ip.alias_manager.aliases:
149 149 # we must strip dots from alias names
150 150 nt.assert_not_in('.', name)
151 151
152 152 # rehashx must fill up syscmdlist
153 153 scoms = _ip.db['syscmdlist']
154 154 nt.assert_true(len(scoms) > 10)
155 155
156 156
157 157
158 158 def test_magic_parse_options():
159 159 """Test that we don't mangle paths when parsing magic options."""
160 160 ip = get_ipython()
161 161 path = 'c:\\x'
162 162 m = DummyMagics(ip)
163 163 opts = m.parse_options('-f %s' % path,'f:')[0]
164 164 # argv splitting is os-dependent
165 165 if os.name == 'posix':
166 166 expected = 'c:x'
167 167 else:
168 168 expected = path
169 169 nt.assert_equal(opts['f'], expected)
170 170
171 171 def test_magic_parse_long_options():
172 172 """Magic.parse_options can handle --foo=bar long options"""
173 173 ip = get_ipython()
174 174 m = DummyMagics(ip)
175 175 opts, _ = m.parse_options('--foo --bar=bubble', 'a', 'foo', 'bar=')
176 176 nt.assert_in('foo', opts)
177 177 nt.assert_in('bar', opts)
178 178 nt.assert_equal(opts['bar'], "bubble")
179 179
180 180
181 181 def doctest_hist_f():
182 182 """Test %hist -f with temporary filename.
183 183
184 184 In [9]: import tempfile
185 185
186 186 In [10]: tfile = tempfile.mktemp('.py','tmp-ipython-')
187 187
188 188 In [11]: %hist -nl -f $tfile 3
189 189
190 190 In [13]: import os; os.unlink(tfile)
191 191 """
192 192
193 193
194 194 def doctest_hist_op():
195 195 """Test %hist -op
196 196
197 197 In [1]: class b(float):
198 198 ...: pass
199 199 ...:
200 200
201 201 In [2]: class s(object):
202 202 ...: def __str__(self):
203 203 ...: return 's'
204 204 ...:
205 205
206 206 In [3]:
207 207
208 208 In [4]: class r(b):
209 209 ...: def __repr__(self):
210 210 ...: return 'r'
211 211 ...:
212 212
213 213 In [5]: class sr(s,r): pass
214 214 ...:
215 215
216 216 In [6]:
217 217
218 218 In [7]: bb=b()
219 219
220 220 In [8]: ss=s()
221 221
222 222 In [9]: rr=r()
223 223
224 224 In [10]: ssrr=sr()
225 225
226 226 In [11]: 4.5
227 227 Out[11]: 4.5
228 228
229 229 In [12]: str(ss)
230 230 Out[12]: 's'
231 231
232 232 In [13]:
233 233
234 234 In [14]: %hist -op
235 235 >>> class b:
236 236 ... pass
237 237 ...
238 238 >>> class s(b):
239 239 ... def __str__(self):
240 240 ... return 's'
241 241 ...
242 242 >>>
243 243 >>> class r(b):
244 244 ... def __repr__(self):
245 245 ... return 'r'
246 246 ...
247 247 >>> class sr(s,r): pass
248 248 >>>
249 249 >>> bb=b()
250 250 >>> ss=s()
251 251 >>> rr=r()
252 252 >>> ssrr=sr()
253 253 >>> 4.5
254 254 4.5
255 255 >>> str(ss)
256 256 's'
257 257 >>>
258 258 """
259 259
260 260 def test_hist_pof():
261 261 ip = get_ipython()
262 262 ip.run_cell(u"1+2", store_history=True)
263 263 #raise Exception(ip.history_manager.session_number)
264 264 #raise Exception(list(ip.history_manager._get_range_session()))
265 265 with TemporaryDirectory() as td:
266 266 tf = os.path.join(td, 'hist.py')
267 267 ip.run_line_magic('history', '-pof %s' % tf)
268 268 assert os.path.isfile(tf)
269 269
270 270
271 271 def test_macro():
272 272 ip = get_ipython()
273 273 ip.history_manager.reset() # Clear any existing history.
274 274 cmds = ["a=1", "def b():\n return a**2", "print(a,b())"]
275 275 for i, cmd in enumerate(cmds, start=1):
276 276 ip.history_manager.store_inputs(i, cmd)
277 277 ip.magic("macro test 1-3")
278 278 nt.assert_equal(ip.user_ns["test"].value, "\n".join(cmds)+"\n")
279 279
280 280 # List macros
281 281 nt.assert_in("test", ip.magic("macro"))
282 282
283 283
284 284 def test_macro_run():
285 285 """Test that we can run a multi-line macro successfully."""
286 286 ip = get_ipython()
287 287 ip.history_manager.reset()
288 288 cmds = ["a=10", "a+=1", "print(a)", "%macro test 2-3"]
289 289 for cmd in cmds:
290 290 ip.run_cell(cmd, store_history=True)
291 291 nt.assert_equal(ip.user_ns["test"].value, "a+=1\nprint(a)\n")
292 292 with tt.AssertPrints("12"):
293 293 ip.run_cell("test")
294 294 with tt.AssertPrints("13"):
295 295 ip.run_cell("test")
296 296
297 297
298 298 def test_magic_magic():
299 299 """Test %magic"""
300 300 ip = get_ipython()
301 301 with capture_output() as captured:
302 302 ip.magic("magic")
303 303
304 304 stdout = captured.stdout
305 305 nt.assert_in('%magic', stdout)
306 306 nt.assert_in('IPython', stdout)
307 307 nt.assert_in('Available', stdout)
308 308
309 309
310 310 @dec.skipif_not_numpy
311 311 def test_numpy_reset_array_undec():
312 312 "Test '%reset array' functionality"
313 313 _ip.ex('import numpy as np')
314 314 _ip.ex('a = np.empty(2)')
315 315 nt.assert_in('a', _ip.user_ns)
316 316 _ip.magic('reset -f array')
317 317 nt.assert_not_in('a', _ip.user_ns)
318 318
319 319 def test_reset_out():
320 320 "Test '%reset out' magic"
321 321 _ip.run_cell("parrot = 'dead'", store_history=True)
322 322 # test '%reset -f out', make an Out prompt
323 323 _ip.run_cell("parrot", store_history=True)
324 324 nt.assert_true('dead' in [_ip.user_ns[x] for x in ('_','__','___')])
325 325 _ip.magic('reset -f out')
326 326 nt.assert_false('dead' in [_ip.user_ns[x] for x in ('_','__','___')])
327 327 nt.assert_equal(len(_ip.user_ns['Out']), 0)
328 328
329 329 def test_reset_in():
330 330 "Test '%reset in' magic"
331 331 # test '%reset -f in'
332 332 _ip.run_cell("parrot", store_history=True)
333 333 nt.assert_true('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
334 334 _ip.magic('%reset -f in')
335 335 nt.assert_false('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
336 336 nt.assert_equal(len(set(_ip.user_ns['In'])), 1)
337 337
338 338 def test_reset_dhist():
339 339 "Test '%reset dhist' magic"
340 340 _ip.run_cell("tmp = [d for d in _dh]") # copy before clearing
341 341 _ip.magic('cd ' + os.path.dirname(nt.__file__))
342 342 _ip.magic('cd -')
343 343 nt.assert_true(len(_ip.user_ns['_dh']) > 0)
344 344 _ip.magic('reset -f dhist')
345 345 nt.assert_equal(len(_ip.user_ns['_dh']), 0)
346 346 _ip.run_cell("_dh = [d for d in tmp]") #restore
347 347
348 348 def test_reset_in_length():
349 349 "Test that '%reset in' preserves In[] length"
350 350 _ip.run_cell("print 'foo'")
351 351 _ip.run_cell("reset -f in")
352 352 nt.assert_equal(len(_ip.user_ns['In']), _ip.displayhook.prompt_count+1)
353 353
354 354 class TestResetErrors(TestCase):
355 355
356 356 def test_reset_redefine(self):
357 357
358 358 @magics_class
359 359 class KernelMagics(Magics):
360 360 @line_magic
361 361 def less(self, shell): pass
362 362
363 363 _ip.register_magics(KernelMagics)
364 364
365 365 with self.assertLogs() as cm:
366 366 # hack, we want to just capture logs, but assertLogs fails if not
367 367 # logs get produce.
368 368 # so log one things we ignore.
369 369 import logging as log_mod
370 370 log = log_mod.getLogger()
371 371 log.info('Nothing')
372 372 # end hack.
373 373 _ip.run_cell("reset -f")
374 374
375 375 assert len(cm.output) == 1
376 376 for out in cm.output:
377 377 assert "Invalid alias" not in out
378 378
379 379 def test_tb_syntaxerror():
380 380 """test %tb after a SyntaxError"""
381 381 ip = get_ipython()
382 382 ip.run_cell("for")
383 383
384 384 # trap and validate stdout
385 385 save_stdout = sys.stdout
386 386 try:
387 387 sys.stdout = StringIO()
388 388 ip.run_cell("%tb")
389 389 out = sys.stdout.getvalue()
390 390 finally:
391 391 sys.stdout = save_stdout
392 392 # trim output, and only check the last line
393 393 last_line = out.rstrip().splitlines()[-1].strip()
394 394 nt.assert_equal(last_line, "SyntaxError: invalid syntax")
395 395
396 396
397 397 def test_time():
398 398 ip = get_ipython()
399 399
400 400 with tt.AssertPrints("Wall time: "):
401 401 ip.run_cell("%time None")
402 402
403 403 ip.run_cell("def f(kmjy):\n"
404 404 " %time print (2*kmjy)")
405 405
406 406 with tt.AssertPrints("Wall time: "):
407 407 with tt.AssertPrints("hihi", suppress=False):
408 408 ip.run_cell("f('hi')")
409 409
410 410 def test_time_last_not_expression():
411 411 ip.run_cell("%%time\n"
412 412 "var_1 = 1\n"
413 413 "var_2 = 2\n")
414 414 assert ip.user_ns['var_1'] == 1
415 415 del ip.user_ns['var_1']
416 416 assert ip.user_ns['var_2'] == 2
417 417 del ip.user_ns['var_2']
418 418
419 419
420 420 @dec.skip_win32
421 421 def test_time2():
422 422 ip = get_ipython()
423 423
424 424 with tt.AssertPrints("CPU times: user "):
425 425 ip.run_cell("%time None")
426 426
427 427 def test_time3():
428 428 """Erroneous magic function calls, issue gh-3334"""
429 429 ip = get_ipython()
430 430 ip.user_ns.pop('run', None)
431 431
432 432 with tt.AssertNotPrints("not found", channel='stderr'):
433 433 ip.run_cell("%%time\n"
434 434 "run = 0\n"
435 435 "run += 1")
436 436
437 437 def test_multiline_time():
438 438 """Make sure last statement from time return a value."""
439 439 ip = get_ipython()
440 440 ip.user_ns.pop('run', None)
441 441
442 442 ip.run_cell(dedent("""\
443 443 %%time
444 444 a = "ho"
445 445 b = "hey"
446 446 a+b
447 447 """))
448 448 nt.assert_equal(ip.user_ns_hidden['_'], 'hohey')
449 449
450 450 def test_time_local_ns():
451 451 """
452 452 Test that local_ns is actually global_ns when running a cell magic
453 453 """
454 454 ip = get_ipython()
455 455 ip.run_cell("%%time\n"
456 456 "myvar = 1")
457 457 nt.assert_equal(ip.user_ns['myvar'], 1)
458 458 del ip.user_ns['myvar']
459 459
460 460 def test_doctest_mode():
461 461 "Toggle doctest_mode twice, it should be a no-op and run without error"
462 462 _ip.magic('doctest_mode')
463 463 _ip.magic('doctest_mode')
464 464
465 465
466 466 def test_parse_options():
467 467 """Tests for basic options parsing in magics."""
468 468 # These are only the most minimal of tests, more should be added later. At
469 469 # the very least we check that basic text/unicode calls work OK.
470 470 m = DummyMagics(_ip)
471 471 nt.assert_equal(m.parse_options('foo', '')[1], 'foo')
472 472 nt.assert_equal(m.parse_options(u'foo', '')[1], u'foo')
473 473
474 474
475 475 def test_parse_options_preserve_non_option_string():
476 476 """Test to assert preservation of non-option part of magic-block, while parsing magic options."""
477 477 m = DummyMagics(_ip)
478 478 opts, stmt = m.parse_options(
479 479 " -n1 -r 13 _ = 314 + foo", "n:r:", preserve_non_opts=True
480 480 )
481 481 nt.assert_equal(opts, {"n": "1", "r": "13"})
482 482 nt.assert_equal(stmt, "_ = 314 + foo")
483 483
484 484
485 485 def test_run_magic_preserve_code_block():
486 486 """Test to assert preservation of non-option part of magic-block, while running magic."""
487 487 _ip.user_ns["spaces"] = []
488 488 _ip.magic("timeit -n1 -r1 spaces.append([s.count(' ') for s in ['document']])")
489 489 assert _ip.user_ns["spaces"] == [[0]]
490 490
491 491
492 492 def test_dirops():
493 493 """Test various directory handling operations."""
494 494 # curpath = lambda :os.path.splitdrive(os.getcwd())[1].replace('\\','/')
495 495 curpath = os.getcwd
496 496 startdir = os.getcwd()
497 497 ipdir = os.path.realpath(_ip.ipython_dir)
498 498 try:
499 499 _ip.magic('cd "%s"' % ipdir)
500 500 nt.assert_equal(curpath(), ipdir)
501 501 _ip.magic('cd -')
502 502 nt.assert_equal(curpath(), startdir)
503 503 _ip.magic('pushd "%s"' % ipdir)
504 504 nt.assert_equal(curpath(), ipdir)
505 505 _ip.magic('popd')
506 506 nt.assert_equal(curpath(), startdir)
507 507 finally:
508 508 os.chdir(startdir)
509 509
510 510
511 511 def test_cd_force_quiet():
512 512 """Test OSMagics.cd_force_quiet option"""
513 513 _ip.config.OSMagics.cd_force_quiet = True
514 514 osmagics = osm.OSMagics(shell=_ip)
515 515
516 516 startdir = os.getcwd()
517 517 ipdir = os.path.realpath(_ip.ipython_dir)
518 518
519 519 try:
520 520 with tt.AssertNotPrints(ipdir):
521 521 osmagics.cd('"%s"' % ipdir)
522 522 with tt.AssertNotPrints(startdir):
523 523 osmagics.cd('-')
524 524 finally:
525 525 os.chdir(startdir)
526 526
527 527
528 528 def test_xmode():
529 529 # Calling xmode three times should be a no-op
530 530 xmode = _ip.InteractiveTB.mode
531 531 for i in range(4):
532 532 _ip.magic("xmode")
533 533 nt.assert_equal(_ip.InteractiveTB.mode, xmode)
534 534
535 535 def test_reset_hard():
536 536 monitor = []
537 537 class A(object):
538 538 def __del__(self):
539 539 monitor.append(1)
540 540 def __repr__(self):
541 541 return "<A instance>"
542 542
543 543 _ip.user_ns["a"] = A()
544 544 _ip.run_cell("a")
545 545
546 546 nt.assert_equal(monitor, [])
547 547 _ip.magic("reset -f")
548 548 nt.assert_equal(monitor, [1])
549 549
550 550 class TestXdel(tt.TempFileMixin):
551 551 def test_xdel(self):
552 552 """Test that references from %run are cleared by xdel."""
553 553 src = ("class A(object):\n"
554 554 " monitor = []\n"
555 555 " def __del__(self):\n"
556 556 " self.monitor.append(1)\n"
557 557 "a = A()\n")
558 558 self.mktmp(src)
559 559 # %run creates some hidden references...
560 560 _ip.magic("run %s" % self.fname)
561 561 # ... as does the displayhook.
562 562 _ip.run_cell("a")
563 563
564 564 monitor = _ip.user_ns["A"].monitor
565 565 nt.assert_equal(monitor, [])
566 566
567 567 _ip.magic("xdel a")
568 568
569 569 # Check that a's __del__ method has been called.
570 570 nt.assert_equal(monitor, [1])
571 571
572 572 def doctest_who():
573 573 """doctest for %who
574 574
575 575 In [1]: %reset -f
576 576
577 577 In [2]: alpha = 123
578 578
579 579 In [3]: beta = 'beta'
580 580
581 581 In [4]: %who int
582 582 alpha
583 583
584 584 In [5]: %who str
585 585 beta
586 586
587 587 In [6]: %whos
588 588 Variable Type Data/Info
589 589 ----------------------------
590 590 alpha int 123
591 591 beta str beta
592 592
593 593 In [7]: %who_ls
594 594 Out[7]: ['alpha', 'beta']
595 595 """
596 596
597 597 def test_whos():
598 598 """Check that whos is protected against objects where repr() fails."""
599 599 class A(object):
600 600 def __repr__(self):
601 601 raise Exception()
602 602 _ip.user_ns['a'] = A()
603 603 _ip.magic("whos")
604 604
605 605 def doctest_precision():
606 606 """doctest for %precision
607 607
608 608 In [1]: f = get_ipython().display_formatter.formatters['text/plain']
609 609
610 610 In [2]: %precision 5
611 611 Out[2]: '%.5f'
612 612
613 613 In [3]: f.float_format
614 614 Out[3]: '%.5f'
615 615
616 616 In [4]: %precision %e
617 617 Out[4]: '%e'
618 618
619 619 In [5]: f(3.1415927)
620 620 Out[5]: '3.141593e+00'
621 621 """
622 622
623 623 def test_debug_magic():
624 624 """Test debugging a small code with %debug
625 625
626 626 In [1]: with PdbTestInput(['c']):
627 627 ...: %debug print("a b") #doctest: +ELLIPSIS
628 628 ...:
629 629 ...
630 630 ipdb> c
631 631 a b
632 632 In [2]:
633 633 """
634 634
635 635 def test_psearch():
636 636 with tt.AssertPrints("dict.fromkeys"):
637 637 _ip.run_cell("dict.fr*?")
638 638 with tt.AssertPrints("Ο€.is_integer"):
639 639 _ip.run_cell("Ο€ = 3.14;\nΟ€.is_integ*?")
640 640
641 641 def test_timeit_shlex():
642 642 """test shlex issues with timeit (#1109)"""
643 643 _ip.ex("def f(*a,**kw): pass")
644 644 _ip.magic('timeit -n1 "this is a bug".count(" ")')
645 645 _ip.magic('timeit -r1 -n1 f(" ", 1)')
646 646 _ip.magic('timeit -r1 -n1 f(" ", 1, " ", 2, " ")')
647 647 _ip.magic('timeit -r1 -n1 ("a " + "b")')
648 648 _ip.magic('timeit -r1 -n1 f("a " + "b")')
649 649 _ip.magic('timeit -r1 -n1 f("a " + "b ")')
650 650
651 651
652 652 def test_timeit_special_syntax():
653 653 "Test %%timeit with IPython special syntax"
654 654 @register_line_magic
655 655 def lmagic(line):
656 656 ip = get_ipython()
657 657 ip.user_ns['lmagic_out'] = line
658 658
659 659 # line mode test
660 660 _ip.run_line_magic('timeit', '-n1 -r1 %lmagic my line')
661 661 nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line')
662 662 # cell mode test
663 663 _ip.run_cell_magic('timeit', '-n1 -r1', '%lmagic my line2')
664 664 nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line2')
665 665
666 666 def test_timeit_return():
667 667 """
668 668 test whether timeit -o return object
669 669 """
670 670
671 671 res = _ip.run_line_magic('timeit','-n10 -r10 -o 1')
672 672 assert(res is not None)
673 673
674 674 def test_timeit_quiet():
675 675 """
676 676 test quiet option of timeit magic
677 677 """
678 678 with tt.AssertNotPrints("loops"):
679 679 _ip.run_cell("%timeit -n1 -r1 -q 1")
680 680
681 681 def test_timeit_return_quiet():
682 682 with tt.AssertNotPrints("loops"):
683 683 res = _ip.run_line_magic('timeit', '-n1 -r1 -q -o 1')
684 684 assert (res is not None)
685 685
686 686 def test_timeit_invalid_return():
687 687 with nt.assert_raises_regex(SyntaxError, "outside function"):
688 688 _ip.run_line_magic('timeit', 'return')
689 689
690 690 @dec.skipif(execution.profile is None)
691 691 def test_prun_special_syntax():
692 692 "Test %%prun with IPython special syntax"
693 693 @register_line_magic
694 694 def lmagic(line):
695 695 ip = get_ipython()
696 696 ip.user_ns['lmagic_out'] = line
697 697
698 698 # line mode test
699 699 _ip.run_line_magic('prun', '-q %lmagic my line')
700 700 nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line')
701 701 # cell mode test
702 702 _ip.run_cell_magic('prun', '-q', '%lmagic my line2')
703 703 nt.assert_equal(_ip.user_ns['lmagic_out'], 'my line2')
704 704
705 705 @dec.skipif(execution.profile is None)
706 706 def test_prun_quotes():
707 707 "Test that prun does not clobber string escapes (GH #1302)"
708 708 _ip.magic(r"prun -q x = '\t'")
709 709 nt.assert_equal(_ip.user_ns['x'], '\t')
710 710
711 711 def test_extension():
712 712 # Debugging information for failures of this test
713 713 print('sys.path:')
714 714 for p in sys.path:
715 715 print(' ', p)
716 716 print('CWD', os.getcwd())
717 717
718 718 nt.assert_raises(ImportError, _ip.magic, "load_ext daft_extension")
719 719 daft_path = os.path.join(os.path.dirname(__file__), "daft_extension")
720 720 sys.path.insert(0, daft_path)
721 721 try:
722 722 _ip.user_ns.pop('arq', None)
723 723 invalidate_caches() # Clear import caches
724 724 _ip.magic("load_ext daft_extension")
725 725 nt.assert_equal(_ip.user_ns['arq'], 185)
726 726 _ip.magic("unload_ext daft_extension")
727 727 assert 'arq' not in _ip.user_ns
728 728 finally:
729 729 sys.path.remove(daft_path)
730 730
731 731
732 732 def test_notebook_export_json():
733 733 _ip = get_ipython()
734 734 _ip.history_manager.reset() # Clear any existing history.
735 735 cmds = [u"a=1", u"def b():\n return a**2", u"print('noΓ«l, Γ©tΓ©', b())"]
736 736 for i, cmd in enumerate(cmds, start=1):
737 737 _ip.history_manager.store_inputs(i, cmd)
738 738 with TemporaryDirectory() as td:
739 739 outfile = os.path.join(td, "nb.ipynb")
740 740 _ip.magic("notebook -e %s" % outfile)
741 741
742 742
743 743 class TestEnv(TestCase):
744 744
745 745 def test_env(self):
746 746 env = _ip.magic("env")
747 747 self.assertTrue(isinstance(env, dict))
748 748
749 749 def test_env_secret(self):
750 750 env = _ip.magic("env")
751 751 hidden = "<hidden>"
752 752 with mock.patch.dict(
753 753 os.environ,
754 754 {
755 755 "API_KEY": "abc123",
756 756 "SECRET_THING": "ssshhh",
757 757 "JUPYTER_TOKEN": "",
758 758 "VAR": "abc"
759 759 }
760 760 ):
761 761 env = _ip.magic("env")
762 762 assert env["API_KEY"] == hidden
763 763 assert env["SECRET_THING"] == hidden
764 764 assert env["JUPYTER_TOKEN"] == hidden
765 765 assert env["VAR"] == "abc"
766 766
767 767 def test_env_get_set_simple(self):
768 768 env = _ip.magic("env var val1")
769 769 self.assertEqual(env, None)
770 770 self.assertEqual(os.environ['var'], 'val1')
771 771 self.assertEqual(_ip.magic("env var"), 'val1')
772 772 env = _ip.magic("env var=val2")
773 773 self.assertEqual(env, None)
774 774 self.assertEqual(os.environ['var'], 'val2')
775 775
776 776 def test_env_get_set_complex(self):
777 777 env = _ip.magic("env var 'val1 '' 'val2")
778 778 self.assertEqual(env, None)
779 779 self.assertEqual(os.environ['var'], "'val1 '' 'val2")
780 780 self.assertEqual(_ip.magic("env var"), "'val1 '' 'val2")
781 781 env = _ip.magic('env var=val2 val3="val4')
782 782 self.assertEqual(env, None)
783 783 self.assertEqual(os.environ['var'], 'val2 val3="val4')
784 784
785 785 def test_env_set_bad_input(self):
786 786 self.assertRaises(UsageError, lambda: _ip.magic("set_env var"))
787 787
788 788 def test_env_set_whitespace(self):
789 789 self.assertRaises(UsageError, lambda: _ip.magic("env var A=B"))
790 790
791 791
792 792 class CellMagicTestCase(TestCase):
793 793
794 794 def check_ident(self, magic):
795 795 # Manually called, we get the result
796 796 out = _ip.run_cell_magic(magic, 'a', 'b')
797 797 nt.assert_equal(out, ('a','b'))
798 798 # Via run_cell, it goes into the user's namespace via displayhook
799 799 _ip.run_cell('%%' + magic +' c\nd\n')
800 800 nt.assert_equal(_ip.user_ns['_'], ('c','d\n'))
801 801
802 802 def test_cell_magic_func_deco(self):
803 803 "Cell magic using simple decorator"
804 804 @register_cell_magic
805 805 def cellm(line, cell):
806 806 return line, cell
807 807
808 808 self.check_ident('cellm')
809 809
810 810 def test_cell_magic_reg(self):
811 811 "Cell magic manually registered"
812 812 def cellm(line, cell):
813 813 return line, cell
814 814
815 815 _ip.register_magic_function(cellm, 'cell', 'cellm2')
816 816 self.check_ident('cellm2')
817 817
818 818 def test_cell_magic_class(self):
819 819 "Cell magics declared via a class"
820 820 @magics_class
821 821 class MyMagics(Magics):
822 822
823 823 @cell_magic
824 824 def cellm3(self, line, cell):
825 825 return line, cell
826 826
827 827 _ip.register_magics(MyMagics)
828 828 self.check_ident('cellm3')
829 829
830 830 def test_cell_magic_class2(self):
831 831 "Cell magics declared via a class, #2"
832 832 @magics_class
833 833 class MyMagics2(Magics):
834 834
835 835 @cell_magic('cellm4')
836 836 def cellm33(self, line, cell):
837 837 return line, cell
838 838
839 839 _ip.register_magics(MyMagics2)
840 840 self.check_ident('cellm4')
841 841 # Check that nothing is registered as 'cellm33'
842 842 c33 = _ip.find_cell_magic('cellm33')
843 843 nt.assert_equal(c33, None)
844 844
845 845 def test_file():
846 846 """Basic %%writefile"""
847 847 ip = get_ipython()
848 848 with TemporaryDirectory() as td:
849 849 fname = os.path.join(td, 'file1')
850 850 ip.run_cell_magic("writefile", fname, u'\n'.join([
851 851 'line1',
852 852 'line2',
853 853 ]))
854 854 s = Path(fname).read_text()
855 855 nt.assert_in('line1\n', s)
856 856 nt.assert_in('line2', s)
857 857
858 858 @dec.skip_win32
859 859 def test_file_single_quote():
860 860 """Basic %%writefile with embedded single quotes"""
861 861 ip = get_ipython()
862 862 with TemporaryDirectory() as td:
863 863 fname = os.path.join(td, '\'file1\'')
864 864 ip.run_cell_magic("writefile", fname, u'\n'.join([
865 865 'line1',
866 866 'line2',
867 867 ]))
868 868 s = Path(fname).read_text()
869 869 nt.assert_in('line1\n', s)
870 870 nt.assert_in('line2', s)
871 871
872 872 @dec.skip_win32
873 873 def test_file_double_quote():
874 874 """Basic %%writefile with embedded double quotes"""
875 875 ip = get_ipython()
876 876 with TemporaryDirectory() as td:
877 877 fname = os.path.join(td, '"file1"')
878 878 ip.run_cell_magic("writefile", fname, u'\n'.join([
879 879 'line1',
880 880 'line2',
881 881 ]))
882 882 s = Path(fname).read_text()
883 883 nt.assert_in('line1\n', s)
884 884 nt.assert_in('line2', s)
885 885
886 886 def test_file_var_expand():
887 887 """%%writefile $filename"""
888 888 ip = get_ipython()
889 889 with TemporaryDirectory() as td:
890 890 fname = os.path.join(td, 'file1')
891 891 ip.user_ns['filename'] = fname
892 892 ip.run_cell_magic("writefile", '$filename', u'\n'.join([
893 893 'line1',
894 894 'line2',
895 895 ]))
896 896 s = Path(fname).read_text()
897 897 nt.assert_in('line1\n', s)
898 898 nt.assert_in('line2', s)
899 899
900 900 def test_file_unicode():
901 901 """%%writefile with unicode cell"""
902 902 ip = get_ipython()
903 903 with TemporaryDirectory() as td:
904 904 fname = os.path.join(td, 'file1')
905 905 ip.run_cell_magic("writefile", fname, u'\n'.join([
906 906 u'linΓ©1',
907 907 u'linΓ©2',
908 908 ]))
909 909 with io.open(fname, encoding='utf-8') as f:
910 910 s = f.read()
911 911 nt.assert_in(u'linΓ©1\n', s)
912 912 nt.assert_in(u'linΓ©2', s)
913 913
914 914 def test_file_amend():
915 915 """%%writefile -a amends files"""
916 916 ip = get_ipython()
917 917 with TemporaryDirectory() as td:
918 918 fname = os.path.join(td, 'file2')
919 919 ip.run_cell_magic("writefile", fname, u'\n'.join([
920 920 'line1',
921 921 'line2',
922 922 ]))
923 923 ip.run_cell_magic("writefile", "-a %s" % fname, u'\n'.join([
924 924 'line3',
925 925 'line4',
926 926 ]))
927 927 s = Path(fname).read_text()
928 928 nt.assert_in('line1\n', s)
929 929 nt.assert_in('line3\n', s)
930 930
931 931 def test_file_spaces():
932 932 """%%file with spaces in filename"""
933 933 ip = get_ipython()
934 934 with TemporaryWorkingDirectory() as td:
935 935 fname = "file name"
936 936 ip.run_cell_magic("file", '"%s"'%fname, u'\n'.join([
937 937 'line1',
938 938 'line2',
939 939 ]))
940 940 s = Path(fname).read_text()
941 941 nt.assert_in('line1\n', s)
942 942 nt.assert_in('line2', s)
943 943
944 944 def test_script_config():
945 945 ip = get_ipython()
946 946 ip.config.ScriptMagics.script_magics = ['whoda']
947 947 sm = script.ScriptMagics(shell=ip)
948 948 nt.assert_in('whoda', sm.magics['cell'])
949 949
950 950 @dec.skip_win32
951 951 def test_script_out():
952 952 ip = get_ipython()
953 953 ip.run_cell_magic("script", "--out output sh", "echo 'hi'")
954 954 nt.assert_equal(ip.user_ns['output'], 'hi\n')
955 955
956 956 @dec.skip_win32
957 957 def test_script_err():
958 958 ip = get_ipython()
959 959 ip.run_cell_magic("script", "--err error sh", "echo 'hello' >&2")
960 960 nt.assert_equal(ip.user_ns['error'], 'hello\n')
961 961
962 962 @dec.skip_win32
963 963 def test_script_out_err():
964 964 ip = get_ipython()
965 965 ip.run_cell_magic("script", "--out output --err error sh", "echo 'hi'\necho 'hello' >&2")
966 966 nt.assert_equal(ip.user_ns['output'], 'hi\n')
967 967 nt.assert_equal(ip.user_ns['error'], 'hello\n')
968 968
969 969 @dec.skip_win32
970 970 async def test_script_bg_out():
971 971 ip = get_ipython()
972 972 ip.run_cell_magic("script", "--bg --out output sh", "echo 'hi'")
973 973 nt.assert_equal((await ip.user_ns["output"].read()), b"hi\n")
974 974 ip.user_ns['output'].close()
975 975
976 976 @dec.skip_win32
977 977 async def test_script_bg_err():
978 978 ip = get_ipython()
979 979 ip.run_cell_magic("script", "--bg --err error sh", "echo 'hello' >&2")
980 980 nt.assert_equal((await ip.user_ns["error"].read()), b"hello\n")
981 981 ip.user_ns["error"].close()
982 982
983 983
984 984 @dec.skip_win32
985 985 async def test_script_bg_out_err():
986 986 ip = get_ipython()
987 987 ip.run_cell_magic(
988 988 "script", "--bg --out output --err error sh", "echo 'hi'\necho 'hello' >&2"
989 989 )
990 990 nt.assert_equal((await ip.user_ns["output"].read()), b"hi\n")
991 991 nt.assert_equal((await ip.user_ns["error"].read()), b"hello\n")
992 992 ip.user_ns["output"].close()
993 993 ip.user_ns["error"].close()
994 994
995 995
996 996 def test_script_defaults():
997 997 ip = get_ipython()
998 998 for cmd in ['sh', 'bash', 'perl', 'ruby']:
999 999 try:
1000 1000 find_cmd(cmd)
1001 1001 except Exception:
1002 1002 pass
1003 1003 else:
1004 1004 nt.assert_in(cmd, ip.magics_manager.magics['cell'])
1005 1005
1006 1006
1007 1007 @magics_class
1008 1008 class FooFoo(Magics):
1009 1009 """class with both %foo and %%foo magics"""
1010 1010 @line_magic('foo')
1011 1011 def line_foo(self, line):
1012 1012 "I am line foo"
1013 1013 pass
1014 1014
1015 1015 @cell_magic("foo")
1016 1016 def cell_foo(self, line, cell):
1017 1017 "I am cell foo, not line foo"
1018 1018 pass
1019 1019
1020 1020 def test_line_cell_info():
1021 1021 """%%foo and %foo magics are distinguishable to inspect"""
1022 1022 ip = get_ipython()
1023 1023 ip.magics_manager.register(FooFoo)
1024 1024 oinfo = ip.object_inspect('foo')
1025 1025 nt.assert_true(oinfo['found'])
1026 1026 nt.assert_true(oinfo['ismagic'])
1027 1027
1028 1028 oinfo = ip.object_inspect('%%foo')
1029 1029 nt.assert_true(oinfo['found'])
1030 1030 nt.assert_true(oinfo['ismagic'])
1031 1031 nt.assert_equal(oinfo['docstring'], FooFoo.cell_foo.__doc__)
1032 1032
1033 1033 oinfo = ip.object_inspect('%foo')
1034 1034 nt.assert_true(oinfo['found'])
1035 1035 nt.assert_true(oinfo['ismagic'])
1036 1036 nt.assert_equal(oinfo['docstring'], FooFoo.line_foo.__doc__)
1037 1037
1038 1038 def test_multiple_magics():
1039 1039 ip = get_ipython()
1040 1040 foo1 = FooFoo(ip)
1041 1041 foo2 = FooFoo(ip)
1042 1042 mm = ip.magics_manager
1043 1043 mm.register(foo1)
1044 1044 nt.assert_true(mm.magics['line']['foo'].__self__ is foo1)
1045 1045 mm.register(foo2)
1046 1046 nt.assert_true(mm.magics['line']['foo'].__self__ is foo2)
1047 1047
1048 1048 def test_alias_magic():
1049 1049 """Test %alias_magic."""
1050 1050 ip = get_ipython()
1051 1051 mm = ip.magics_manager
1052 1052
1053 1053 # Basic operation: both cell and line magics are created, if possible.
1054 1054 ip.run_line_magic('alias_magic', 'timeit_alias timeit')
1055 1055 nt.assert_in('timeit_alias', mm.magics['line'])
1056 1056 nt.assert_in('timeit_alias', mm.magics['cell'])
1057 1057
1058 1058 # --cell is specified, line magic not created.
1059 1059 ip.run_line_magic('alias_magic', '--cell timeit_cell_alias timeit')
1060 1060 nt.assert_not_in('timeit_cell_alias', mm.magics['line'])
1061 1061 nt.assert_in('timeit_cell_alias', mm.magics['cell'])
1062 1062
1063 1063 # Test that line alias is created successfully.
1064 1064 ip.run_line_magic('alias_magic', '--line env_alias env')
1065 1065 nt.assert_equal(ip.run_line_magic('env', ''),
1066 1066 ip.run_line_magic('env_alias', ''))
1067 1067
1068 1068 # Test that line alias with parameters passed in is created successfully.
1069 1069 ip.run_line_magic('alias_magic', '--line history_alias history --params ' + shlex.quote('3'))
1070 1070 nt.assert_in('history_alias', mm.magics['line'])
1071 1071
1072 1072
1073 1073 def test_save():
1074 1074 """Test %save."""
1075 1075 ip = get_ipython()
1076 1076 ip.history_manager.reset() # Clear any existing history.
1077 1077 cmds = [u"a=1", u"def b():\n return a**2", u"print(a, b())"]
1078 1078 for i, cmd in enumerate(cmds, start=1):
1079 1079 ip.history_manager.store_inputs(i, cmd)
1080 1080 with TemporaryDirectory() as tmpdir:
1081 1081 file = os.path.join(tmpdir, "testsave.py")
1082 1082 ip.run_line_magic("save", "%s 1-10" % file)
1083 1083 content = Path(file).read_text()
1084 1084 nt.assert_equal(content.count(cmds[0]), 1)
1085 1085 nt.assert_in("coding: utf-8", content)
1086 1086 ip.run_line_magic("save", "-a %s 1-10" % file)
1087 1087 content = Path(file).read_text()
1088 1088 nt.assert_equal(content.count(cmds[0]), 2)
1089 1089 nt.assert_in("coding: utf-8", content)
1090 1090
1091 1091
1092 def test_save_with_no_args():
1093 ip = get_ipython()
1094 ip.history_manager.reset() # Clear any existing history.
1095 cmds = [u"a=1", u"def b():\n return a**2", u"print(a, b())", "%save"]
1096 for i, cmd in enumerate(cmds, start=1):
1097 ip.history_manager.store_inputs(i, cmd)
1098
1099 with TemporaryDirectory() as tmpdir:
1100 path = os.path.join(tmpdir, "testsave.py")
1101 ip.run_line_magic("save", path)
1102 content = Path(path).read_text()
1103 expected_content = dedent(
1104 """\
1105 # coding: utf-8
1106 a=1
1107 def b():
1108 return a**2
1109 print(a, b())
1110 """
1111 )
1112 nt.assert_equal(content, expected_content)
1113
1114
1092 1115 def test_store():
1093 1116 """Test %store."""
1094 1117 ip = get_ipython()
1095 1118 ip.run_line_magic('load_ext', 'storemagic')
1096 1119
1097 1120 # make sure the storage is empty
1098 1121 ip.run_line_magic('store', '-z')
1099 1122 ip.user_ns['var'] = 42
1100 1123 ip.run_line_magic('store', 'var')
1101 1124 ip.user_ns['var'] = 39
1102 1125 ip.run_line_magic('store', '-r')
1103 1126 nt.assert_equal(ip.user_ns['var'], 42)
1104 1127
1105 1128 ip.run_line_magic('store', '-d var')
1106 1129 ip.user_ns['var'] = 39
1107 1130 ip.run_line_magic('store' , '-r')
1108 1131 nt.assert_equal(ip.user_ns['var'], 39)
1109 1132
1110 1133
1111 1134 def _run_edit_test(arg_s, exp_filename=None,
1112 1135 exp_lineno=-1,
1113 1136 exp_contents=None,
1114 1137 exp_is_temp=None):
1115 1138 ip = get_ipython()
1116 1139 M = code.CodeMagics(ip)
1117 1140 last_call = ['','']
1118 1141 opts,args = M.parse_options(arg_s,'prxn:')
1119 1142 filename, lineno, is_temp = M._find_edit_target(ip, args, opts, last_call)
1120 1143
1121 1144 if exp_filename is not None:
1122 1145 nt.assert_equal(exp_filename, filename)
1123 1146 if exp_contents is not None:
1124 1147 with io.open(filename, 'r', encoding='utf-8') as f:
1125 1148 contents = f.read()
1126 1149 nt.assert_equal(exp_contents, contents)
1127 1150 if exp_lineno != -1:
1128 1151 nt.assert_equal(exp_lineno, lineno)
1129 1152 if exp_is_temp is not None:
1130 1153 nt.assert_equal(exp_is_temp, is_temp)
1131 1154
1132 1155
1133 1156 def test_edit_interactive():
1134 1157 """%edit on interactively defined objects"""
1135 1158 ip = get_ipython()
1136 1159 n = ip.execution_count
1137 1160 ip.run_cell(u"def foo(): return 1", store_history=True)
1138 1161
1139 1162 try:
1140 1163 _run_edit_test("foo")
1141 1164 except code.InteractivelyDefined as e:
1142 1165 nt.assert_equal(e.index, n)
1143 1166 else:
1144 1167 raise AssertionError("Should have raised InteractivelyDefined")
1145 1168
1146 1169
1147 1170 def test_edit_cell():
1148 1171 """%edit [cell id]"""
1149 1172 ip = get_ipython()
1150 1173
1151 1174 ip.run_cell(u"def foo(): return 1", store_history=True)
1152 1175
1153 1176 # test
1154 1177 _run_edit_test("1", exp_contents=ip.user_ns['In'][1], exp_is_temp=True)
1155 1178
1156 1179 def test_edit_fname():
1157 1180 """%edit file"""
1158 1181 # test
1159 1182 _run_edit_test("test file.py", exp_filename="test file.py")
1160 1183
1161 1184 def test_bookmark():
1162 1185 ip = get_ipython()
1163 1186 ip.run_line_magic('bookmark', 'bmname')
1164 1187 with tt.AssertPrints('bmname'):
1165 1188 ip.run_line_magic('bookmark', '-l')
1166 1189 ip.run_line_magic('bookmark', '-d bmname')
1167 1190
1168 1191 def test_ls_magic():
1169 1192 ip = get_ipython()
1170 1193 json_formatter = ip.display_formatter.formatters['application/json']
1171 1194 json_formatter.enabled = True
1172 1195 lsmagic = ip.magic('lsmagic')
1173 1196 with warnings.catch_warnings(record=True) as w:
1174 1197 j = json_formatter(lsmagic)
1175 1198 nt.assert_equal(sorted(j), ['cell', 'line'])
1176 1199 nt.assert_equal(w, []) # no warnings
1177 1200
1178 1201 def test_strip_initial_indent():
1179 1202 def sii(s):
1180 1203 lines = s.splitlines()
1181 1204 return '\n'.join(code.strip_initial_indent(lines))
1182 1205
1183 1206 nt.assert_equal(sii(" a = 1\nb = 2"), "a = 1\nb = 2")
1184 1207 nt.assert_equal(sii(" a\n b\nc"), "a\n b\nc")
1185 1208 nt.assert_equal(sii("a\n b"), "a\n b")
1186 1209
1187 1210 def test_logging_magic_quiet_from_arg():
1188 1211 _ip.config.LoggingMagics.quiet = False
1189 1212 lm = logging.LoggingMagics(shell=_ip)
1190 1213 with TemporaryDirectory() as td:
1191 1214 try:
1192 1215 with tt.AssertNotPrints(re.compile("Activating.*")):
1193 1216 lm.logstart('-q {}'.format(
1194 1217 os.path.join(td, "quiet_from_arg.log")))
1195 1218 finally:
1196 1219 _ip.logger.logstop()
1197 1220
1198 1221 def test_logging_magic_quiet_from_config():
1199 1222 _ip.config.LoggingMagics.quiet = True
1200 1223 lm = logging.LoggingMagics(shell=_ip)
1201 1224 with TemporaryDirectory() as td:
1202 1225 try:
1203 1226 with tt.AssertNotPrints(re.compile("Activating.*")):
1204 1227 lm.logstart(os.path.join(td, "quiet_from_config.log"))
1205 1228 finally:
1206 1229 _ip.logger.logstop()
1207 1230
1208 1231
1209 1232 def test_logging_magic_not_quiet():
1210 1233 _ip.config.LoggingMagics.quiet = False
1211 1234 lm = logging.LoggingMagics(shell=_ip)
1212 1235 with TemporaryDirectory() as td:
1213 1236 try:
1214 1237 with tt.AssertPrints(re.compile("Activating.*")):
1215 1238 lm.logstart(os.path.join(td, "not_quiet.log"))
1216 1239 finally:
1217 1240 _ip.logger.logstop()
1218 1241
1219 1242
1220 1243 def test_time_no_var_expand():
1221 1244 _ip.user_ns['a'] = 5
1222 1245 _ip.user_ns['b'] = []
1223 1246 _ip.magic('time b.append("{a}")')
1224 1247 assert _ip.user_ns['b'] == ['{a}']
1225 1248
1226 1249
1227 1250 # this is slow, put at the end for local testing.
1228 1251 def test_timeit_arguments():
1229 1252 "Test valid timeit arguments, should not cause SyntaxError (GH #1269)"
1230 1253 if sys.version_info < (3,7):
1231 1254 _ip.magic("timeit -n1 -r1 ('#')")
1232 1255 else:
1233 1256 # 3.7 optimize no-op statement like above out, and complain there is
1234 1257 # nothing in the for loop.
1235 1258 _ip.magic("timeit -n1 -r1 a=('#')")
1236 1259
1237 1260
1238 1261 TEST_MODULE = """
1239 1262 print('Loaded my_tmp')
1240 1263 if __name__ == "__main__":
1241 1264 print('I just ran a script')
1242 1265 """
1243 1266
1244 1267
1245 1268 def test_run_module_from_import_hook():
1246 1269 "Test that a module can be loaded via an import hook"
1247 1270 with TemporaryDirectory() as tmpdir:
1248 1271 fullpath = os.path.join(tmpdir, 'my_tmp.py')
1249 1272 Path(fullpath).write_text(TEST_MODULE)
1250 1273
1251 1274 class MyTempImporter(object):
1252 1275 def __init__(self):
1253 1276 pass
1254 1277
1255 1278 def find_module(self, fullname, path=None):
1256 1279 if 'my_tmp' in fullname:
1257 1280 return self
1258 1281 return None
1259 1282
1260 1283 def load_module(self, name):
1261 1284 import imp
1262 1285 return imp.load_source('my_tmp', fullpath)
1263 1286
1264 1287 def get_code(self, fullname):
1265 1288 return compile(Path(fullpath).read_text(), "foo", "exec")
1266 1289
1267 1290 def is_package(self, __):
1268 1291 return False
1269 1292
1270 1293 sys.meta_path.insert(0, MyTempImporter())
1271 1294
1272 1295 with capture_output() as captured:
1273 1296 _ip.magic("run -m my_tmp")
1274 1297 _ip.run_cell("import my_tmp")
1275 1298
1276 1299 output = "Loaded my_tmp\nI just ran a script\nLoaded my_tmp\n"
1277 1300 nt.assert_equal(output, captured.stdout)
1278 1301
1279 1302 sys.meta_path.pop(0)
General Comments 0
You need to be logged in to leave comments. Login now