Skip to content

consolidate retry into retry wrapped lambda #353

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Mar 4, 2018
Merged

consolidate retry into retry wrapped lambda #353

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f8fbaad
consolidate retry into retry wrapped lambda
tony Mar 4, 2018
99b6550
add some docstring
tony Mar 4, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 36 additions & 45 deletions tests/test_workspacebuilder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,22 +9,42 @@

import kaptan
import pytest

from . import fixtures_dir
from libtmux import Window
from libtmux.common import has_gte_version
from libtmux.test import temp_session

from tmuxp import config, exc
from tmuxp._compat import text_type
from tmuxp.workspacebuilder import WorkspaceBuilder

from . import example_dir
from . import example_dir, fixtures_dir
from .fixtures._util import loadfixture


RETRY_TIMEOUT_SECONDS = int(os.getenv('RETRY_TIMEOUT_SECONDS', 8))


def retry(seconds=RETRY_TIMEOUT_SECONDS):
"""Retry a block of code until a time limit or ``break``.

.. code-block:: python

while retry():
p = w.attached_pane
p.server._update_panes()
if p.current_path == pane_path:
break


:param seconds: Seconds to retry, defaults to ``RETRY_TIMEOUT_SECONDS``,
which is configurable via environmental variables.
:type seconds: int
:rtype: void

:todo: Move to libtmux.test
"""
return (lambda: time.time() < time.time() + seconds)()


def test_split_windows(session):
yaml_config = loadfixture("workspacebuilder/two_pane.yaml")
s = session
Expand Down Expand Up @@ -110,15 +130,11 @@ def test_focus_pane_index(session):

pane_path = '/usr'

timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout

while True:
while retry():
p = w.attached_pane
p.server._update_panes()
if p.current_path == pane_path:
break
elif time.time() > timeout:
break

assert p.current_path == pane_path

Expand All @@ -131,14 +147,12 @@ def test_focus_pane_index(session):

p = None
pane_path = '/'
timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:

while retry():
p = window3.attached_pane
p.server._update_panes()
if p.current_path == pane_path:
break
elif time.time() > timeout:
break

assert p.current_path == pane_path

Expand Down Expand Up @@ -302,17 +316,14 @@ def test_window_options_after(session):

def assert_last_line(p, s):
correct = False
timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout

while True:
while retry():
pane_out = p.cmd('capture-pane', '-p', '-J').stdout
while not pane_out[-1].strip(): # delete trailing lines tmux 1.8
pane_out.pop()
if len(pane_out) > 1 and pane_out[-2].strip() == s:
correct = True
break
elif time.time() > timeout:
break

# Print output for easier debugging if assertion fails
if not correct:
Expand Down Expand Up @@ -351,13 +362,11 @@ def test_window_shell(session):
for w, wconf in builder.iter_create_windows(s):
if 'window_shell' in wconf:
assert wconf['window_shell'] == text_type('top')
timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:

while retry():
session.server._update_windows()
if w['window_name'] != 'top':
break
elif time.time() > timeout:
break

assert w.name != text_type('top')

Expand Down Expand Up @@ -403,38 +412,29 @@ def test_automatic_rename_option(session):
assert s.name != 'tmuxp'
w = s.windows[0]

timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
session.server._update_windows()
if w.name != 'sh':
break
elif time.time() > timeout:
break

assert w.name != 'sh'

pane_base_index = w.show_window_option('pane-base-index', g=True)
w.select_pane(pane_base_index)

timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
session.server._update_windows()
if w.name == 'sh':
break
elif time.time() > timeout:
break

assert w.name == text_type('sh')

w.select_pane('-D')

timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
session.server._update_windows()
if w['window_name'] != 'sh':
break
elif time.time() > timeout:
break

assert w.name != text_type('sh')

Expand Down Expand Up @@ -490,8 +490,7 @@ def test_start_directory(session, tmpdir):

for path, window in zip(dirs, session.windows):
for p in window.panes:
timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
p.server._update_panes()
pane_path = p.current_path
if pane_path is None:
Expand All @@ -505,8 +504,6 @@ def test_start_directory(session, tmpdir):
path in pane_path
)
break
elif time.time() > timeout:
break

# handle case with OS X adding /private/ to /tmp/ paths
assert result
Expand Down Expand Up @@ -558,8 +555,7 @@ def test_start_directory_relative(session, tmpdir):

for path, window in zip(dirs, session.windows):
for p in window.panes:
timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
p.server._update_panes()
# Handle case where directories resolve to /private/ in OSX
pane_path = p.current_path
Expand All @@ -574,8 +570,6 @@ def test_start_directory_relative(session, tmpdir):
path in pane_path
)
break
elif time.time() > timeout:
break

assert result

Expand Down Expand Up @@ -629,13 +623,10 @@ def test_pane_order(session):
# at 0 since python list.
pane_path = pane_paths[p_index - pane_base_index]

timeout = time.time() + RETRY_TIMEOUT_SECONDS # seconds timeout
while True:
while retry():
p.server._update_panes()
if p.current_path == pane_path:
break
elif time.time() > timeout:
break

assert p.current_path, pane_path

Expand Down