fix: format all code with black
and from now on should not deviate from that...
This commit is contained in:
parent
8a09fb1a3c
commit
4d0693862d
68 changed files with 6693 additions and 5659 deletions
|
|
@ -17,7 +17,7 @@ class TestUpgradeView(WebTestCase):
|
|||
return mod.UpgradeView(self.request)
|
||||
|
||||
def test_includeme(self):
|
||||
self.pyramid_config.include('wuttaweb.views.upgrades')
|
||||
self.pyramid_config.include("wuttaweb.views.upgrades")
|
||||
|
||||
def test_configure_grid(self):
|
||||
model = self.app.model
|
||||
|
|
@ -37,74 +37,79 @@ class TestUpgradeView(WebTestCase):
|
|||
self.assertIsNone(view.grid_row_class(upgrade, data, 1))
|
||||
|
||||
upgrade.status = enum.UpgradeStatus.EXECUTING
|
||||
self.assertEqual(view.grid_row_class(upgrade, data, 1), 'has-background-warning')
|
||||
self.assertEqual(
|
||||
view.grid_row_class(upgrade, data, 1), "has-background-warning"
|
||||
)
|
||||
|
||||
upgrade.status = enum.UpgradeStatus.SUCCESS
|
||||
self.assertIsNone(view.grid_row_class(upgrade, data, 1))
|
||||
|
||||
upgrade.status = enum.UpgradeStatus.FAILURE
|
||||
self.assertEqual(view.grid_row_class(upgrade, data, 1), 'has-background-warning')
|
||||
self.assertEqual(
|
||||
view.grid_row_class(upgrade, data, 1), "has-background-warning"
|
||||
)
|
||||
|
||||
def test_configure_form(self):
|
||||
self.pyramid_config.add_route('upgrades.download', '/upgrades/{uuid}/download')
|
||||
self.pyramid_config.add_route("upgrades.download", "/upgrades/{uuid}/download")
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
view = self.make_view()
|
||||
|
||||
# some fields exist when viewing
|
||||
with patch.object(view, 'viewing', new=True):
|
||||
with patch.object(view, "viewing", new=True):
|
||||
form = view.make_form(model_class=model.Upgrade, model_instance=upgrade)
|
||||
self.assertIn('created', form)
|
||||
self.assertIn("created", form)
|
||||
view.configure_form(form)
|
||||
self.assertIn('created', form)
|
||||
self.assertIn("created", form)
|
||||
|
||||
# but then are removed when creating
|
||||
with patch.object(view, 'creating', new=True):
|
||||
with patch.object(view, "creating", new=True):
|
||||
form = view.make_form(model_class=model.Upgrade)
|
||||
self.assertIn('created', form)
|
||||
self.assertIn("created", form)
|
||||
view.configure_form(form)
|
||||
self.assertNotIn('created', form)
|
||||
self.assertNotIn("created", form)
|
||||
|
||||
# test executed, stdout/stderr when viewing
|
||||
with patch.object(view, 'viewing', new=True):
|
||||
with patch.object(view, "viewing", new=True):
|
||||
|
||||
# executed is *not* shown by default
|
||||
form = view.make_form(model_class=model.Upgrade, model_instance=upgrade)
|
||||
self.assertIn('executed', form)
|
||||
self.assertIn("executed", form)
|
||||
view.configure_form(form)
|
||||
self.assertNotIn('executed', form)
|
||||
self.assertNotIn('stdout_file', form)
|
||||
self.assertNotIn('stderr_file', form)
|
||||
self.assertNotIn("executed", form)
|
||||
self.assertNotIn("stdout_file", form)
|
||||
self.assertNotIn("stderr_file", form)
|
||||
|
||||
# but it *is* shown if upgrade is executed
|
||||
upgrade.executed = datetime.datetime.now()
|
||||
upgrade.status = enum.UpgradeStatus.SUCCESS
|
||||
form = view.make_form(model_class=model.Upgrade, model_instance=upgrade)
|
||||
self.assertIn('executed', form)
|
||||
self.assertIn("executed", form)
|
||||
view.configure_form(form)
|
||||
self.assertIn('executed', form)
|
||||
self.assertIn('stdout_file', form)
|
||||
self.assertIn('stderr_file', form)
|
||||
self.assertIn("executed", form)
|
||||
self.assertIn("stdout_file", form)
|
||||
self.assertIn("stderr_file", form)
|
||||
|
||||
def test_objectify(self):
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
self.session.commit()
|
||||
view = self.make_view()
|
||||
|
||||
# user and status are auto-set when creating
|
||||
self.request.user = user
|
||||
self.request.method = 'POST'
|
||||
self.request.POST = {'description': "new one"}
|
||||
with patch.object(view, 'creating', new=True):
|
||||
self.request.method = "POST"
|
||||
self.request.POST = {"description": "new one"}
|
||||
with patch.object(view, "creating", new=True):
|
||||
form = view.make_model_form()
|
||||
self.assertTrue(form.validate())
|
||||
upgrade = view.objectify(form)
|
||||
|
|
@ -116,13 +121,14 @@ class TestUpgradeView(WebTestCase):
|
|||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
|
||||
appdir = self.mkdir('app')
|
||||
self.config.setdefault('wutta.appdir', appdir)
|
||||
appdir = self.mkdir("app")
|
||||
self.config.setdefault("wutta.appdir", appdir)
|
||||
self.assertEqual(self.app.get_appdir(), appdir)
|
||||
|
||||
user = model.User(username='barney')
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
user = model.User(username="barney")
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
|
|
@ -134,21 +140,24 @@ class TestUpgradeView(WebTestCase):
|
|||
self.assertIsNone(path)
|
||||
|
||||
# with filename
|
||||
path = view.download_path(upgrade, 'foo.txt')
|
||||
self.assertEqual(path, os.path.join(appdir, 'data', 'upgrades',
|
||||
uuid[:2], uuid[2:], 'foo.txt'))
|
||||
path = view.download_path(upgrade, "foo.txt")
|
||||
self.assertEqual(
|
||||
path,
|
||||
os.path.join(appdir, "data", "upgrades", uuid[:2], uuid[2:], "foo.txt"),
|
||||
)
|
||||
|
||||
def test_get_upgrade_filepath(self):
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
|
||||
appdir = self.mkdir('app')
|
||||
self.config.setdefault('wutta.appdir', appdir)
|
||||
appdir = self.mkdir("app")
|
||||
self.config.setdefault("wutta.appdir", appdir)
|
||||
self.assertEqual(self.app.get_appdir(), appdir)
|
||||
|
||||
user = model.User(username='barney')
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
user = model.User(username="barney")
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
|
|
@ -157,25 +166,29 @@ class TestUpgradeView(WebTestCase):
|
|||
|
||||
# no filename
|
||||
path = view.get_upgrade_filepath(upgrade)
|
||||
self.assertEqual(path, os.path.join(appdir, 'data', 'upgrades',
|
||||
uuid[:2], uuid[2:]))
|
||||
self.assertEqual(
|
||||
path, os.path.join(appdir, "data", "upgrades", uuid[:2], uuid[2:])
|
||||
)
|
||||
|
||||
# with filename
|
||||
path = view.get_upgrade_filepath(upgrade, 'foo.txt')
|
||||
self.assertEqual(path, os.path.join(appdir, 'data', 'upgrades',
|
||||
uuid[:2], uuid[2:], 'foo.txt'))
|
||||
path = view.get_upgrade_filepath(upgrade, "foo.txt")
|
||||
self.assertEqual(
|
||||
path,
|
||||
os.path.join(appdir, "data", "upgrades", uuid[:2], uuid[2:], "foo.txt"),
|
||||
)
|
||||
|
||||
def test_delete_instance(self):
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
|
||||
appdir = self.mkdir('app')
|
||||
self.config.setdefault('wutta.appdir', appdir)
|
||||
appdir = self.mkdir("app")
|
||||
self.config.setdefault("wutta.appdir", appdir)
|
||||
self.assertEqual(self.app.get_appdir(), appdir)
|
||||
|
||||
user = model.User(username='barney')
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
user = model.User(username="barney")
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
|
|
@ -183,19 +196,19 @@ class TestUpgradeView(WebTestCase):
|
|||
|
||||
# mock stdout/stderr files
|
||||
upgrade_dir = view.get_upgrade_filepath(upgrade)
|
||||
stdout = view.get_upgrade_filepath(upgrade, 'stdout.log')
|
||||
with open(stdout, 'w') as f:
|
||||
f.write('stdout')
|
||||
stderr = view.get_upgrade_filepath(upgrade, 'stderr.log')
|
||||
with open(stderr, 'w') as f:
|
||||
f.write('stderr')
|
||||
stdout = view.get_upgrade_filepath(upgrade, "stdout.log")
|
||||
with open(stdout, "w") as f:
|
||||
f.write("stdout")
|
||||
stderr = view.get_upgrade_filepath(upgrade, "stderr.log")
|
||||
with open(stderr, "w") as f:
|
||||
f.write("stderr")
|
||||
|
||||
# both upgrade and files are deleted
|
||||
self.assertTrue(os.path.exists(upgrade_dir))
|
||||
self.assertTrue(os.path.exists(stdout))
|
||||
self.assertTrue(os.path.exists(stderr))
|
||||
self.assertEqual(self.session.query(model.Upgrade).count(), 1)
|
||||
with patch.object(view, 'Session', return_value=self.session):
|
||||
with patch.object(view, "Session", return_value=self.session):
|
||||
view.delete_instance(upgrade)
|
||||
self.assertFalse(os.path.exists(upgrade_dir))
|
||||
self.assertFalse(os.path.exists(stdout))
|
||||
|
|
@ -206,13 +219,14 @@ class TestUpgradeView(WebTestCase):
|
|||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
|
||||
appdir = self.mkdir('app')
|
||||
self.config.setdefault('wutta.appdir', appdir)
|
||||
appdir = self.mkdir("app")
|
||||
self.config.setdefault("wutta.appdir", appdir)
|
||||
self.assertEqual(self.app.get_appdir(), appdir)
|
||||
|
||||
user = model.User(username='barney')
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
user = model.User(username="barney")
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
|
|
@ -224,139 +238,153 @@ class TestUpgradeView(WebTestCase):
|
|||
self.assertRaises(ConfigurationError, view.execute_instance, upgrade, user)
|
||||
|
||||
# script w/ success
|
||||
goodpy = self.write_file('good.py', """
|
||||
goodpy = self.write_file(
|
||||
"good.py",
|
||||
"""
|
||||
import sys
|
||||
sys.stdout.write('hello from good.py')
|
||||
sys.exit(0)
|
||||
""")
|
||||
self.app.save_setting(self.session, 'wutta.upgrades.command', f'{python} {goodpy}')
|
||||
""",
|
||||
)
|
||||
self.app.save_setting(
|
||||
self.session, "wutta.upgrades.command", f"{python} {goodpy}"
|
||||
)
|
||||
self.assertIsNone(upgrade.executed)
|
||||
self.assertIsNone(upgrade.executed_by)
|
||||
self.assertEqual(upgrade.status, enum.UpgradeStatus.PENDING)
|
||||
with patch.object(view, 'Session', return_value=self.session):
|
||||
with patch.object(self.config, 'usedb', new=True):
|
||||
with patch.object(view, "Session", return_value=self.session):
|
||||
with patch.object(self.config, "usedb", new=True):
|
||||
view.execute_instance(upgrade, user)
|
||||
self.assertIsNotNone(upgrade.executed)
|
||||
self.assertIs(upgrade.executed_by, user)
|
||||
self.assertEqual(upgrade.status, enum.UpgradeStatus.SUCCESS)
|
||||
with open(view.get_upgrade_filepath(upgrade, 'stdout.log')) as f:
|
||||
self.assertEqual(f.read(), 'hello from good.py')
|
||||
with open(view.get_upgrade_filepath(upgrade, 'stderr.log')) as f:
|
||||
self.assertEqual(f.read(), '')
|
||||
with open(view.get_upgrade_filepath(upgrade, "stdout.log")) as f:
|
||||
self.assertEqual(f.read(), "hello from good.py")
|
||||
with open(view.get_upgrade_filepath(upgrade, "stderr.log")) as f:
|
||||
self.assertEqual(f.read(), "")
|
||||
|
||||
# need a new record for next test
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
# script w/ failure
|
||||
badpy = self.write_file('bad.py', """
|
||||
badpy = self.write_file(
|
||||
"bad.py",
|
||||
"""
|
||||
import sys
|
||||
sys.stderr.write('hello from bad.py')
|
||||
sys.exit(42)
|
||||
""")
|
||||
self.app.save_setting(self.session, 'wutta.upgrades.command', f'{python} {badpy}')
|
||||
""",
|
||||
)
|
||||
self.app.save_setting(
|
||||
self.session, "wutta.upgrades.command", f"{python} {badpy}"
|
||||
)
|
||||
self.assertIsNone(upgrade.executed)
|
||||
self.assertIsNone(upgrade.executed_by)
|
||||
self.assertEqual(upgrade.status, enum.UpgradeStatus.PENDING)
|
||||
with patch.object(view, 'Session', return_value=self.session):
|
||||
with patch.object(self.config, 'usedb', new=True):
|
||||
with patch.object(view, "Session", return_value=self.session):
|
||||
with patch.object(self.config, "usedb", new=True):
|
||||
view.execute_instance(upgrade, user)
|
||||
self.assertIsNotNone(upgrade.executed)
|
||||
self.assertIs(upgrade.executed_by, user)
|
||||
self.assertEqual(upgrade.status, enum.UpgradeStatus.FAILURE)
|
||||
with open(view.get_upgrade_filepath(upgrade, 'stdout.log')) as f:
|
||||
self.assertEqual(f.read(), '')
|
||||
with open(view.get_upgrade_filepath(upgrade, 'stderr.log')) as f:
|
||||
self.assertEqual(f.read(), 'hello from bad.py')
|
||||
with open(view.get_upgrade_filepath(upgrade, "stdout.log")) as f:
|
||||
self.assertEqual(f.read(), "")
|
||||
with open(view.get_upgrade_filepath(upgrade, "stderr.log")) as f:
|
||||
self.assertEqual(f.read(), "hello from bad.py")
|
||||
|
||||
def test_execute_progress(self):
|
||||
model = self.app.model
|
||||
enum = self.app.enum
|
||||
view = self.make_view()
|
||||
|
||||
user = model.User(username='barney')
|
||||
user = model.User(username="barney")
|
||||
self.session.add(user)
|
||||
upgrade = model.Upgrade(description='test', created_by=user,
|
||||
status=enum.UpgradeStatus.PENDING)
|
||||
upgrade = model.Upgrade(
|
||||
description="test", created_by=user, status=enum.UpgradeStatus.PENDING
|
||||
)
|
||||
self.session.add(upgrade)
|
||||
self.session.commit()
|
||||
|
||||
stdout = self.write_file('stdout.log', 'hello 001\n')
|
||||
stdout = self.write_file("stdout.log", "hello 001\n")
|
||||
|
||||
self.request.matchdict = {'uuid': upgrade.uuid}
|
||||
with patch.multiple(mod.UpgradeView,
|
||||
Session=MagicMock(return_value=self.session),
|
||||
get_upgrade_filepath=MagicMock(return_value=stdout)):
|
||||
self.request.matchdict = {"uuid": upgrade.uuid}
|
||||
with patch.multiple(
|
||||
mod.UpgradeView,
|
||||
Session=MagicMock(return_value=self.session),
|
||||
get_upgrade_filepath=MagicMock(return_value=stdout),
|
||||
):
|
||||
|
||||
# nb. this is used to identify progress tracker
|
||||
self.request.session.id = 'mockid#1'
|
||||
self.request.session.id = "mockid#1"
|
||||
|
||||
# first call should get the full contents
|
||||
context = view.execute_progress()
|
||||
self.assertFalse(context.get('complete'))
|
||||
self.assertFalse(context.get('error'))
|
||||
self.assertFalse(context.get("complete"))
|
||||
self.assertFalse(context.get("error"))
|
||||
# nb. newline is converted to <br>
|
||||
self.assertEqual(context['stdout'], 'hello 001<br />')
|
||||
self.assertEqual(context["stdout"], "hello 001<br />")
|
||||
|
||||
# next call should get any new contents
|
||||
with open(stdout, 'a') as f:
|
||||
f.write('hello 002\n')
|
||||
with open(stdout, "a") as f:
|
||||
f.write("hello 002\n")
|
||||
context = view.execute_progress()
|
||||
self.assertFalse(context.get('complete'))
|
||||
self.assertFalse(context.get('error'))
|
||||
self.assertEqual(context['stdout'], 'hello 002<br />')
|
||||
self.assertFalse(context.get("complete"))
|
||||
self.assertFalse(context.get("error"))
|
||||
self.assertEqual(context["stdout"], "hello 002<br />")
|
||||
|
||||
# nb. switch to a different progress tracker
|
||||
self.request.session.id = 'mockid#2'
|
||||
self.request.session.id = "mockid#2"
|
||||
|
||||
# first call should get the full contents
|
||||
context = view.execute_progress()
|
||||
self.assertFalse(context.get('complete'))
|
||||
self.assertFalse(context.get('error'))
|
||||
self.assertEqual(context['stdout'], 'hello 001<br />hello 002<br />')
|
||||
self.assertFalse(context.get("complete"))
|
||||
self.assertFalse(context.get("error"))
|
||||
self.assertEqual(context["stdout"], "hello 001<br />hello 002<br />")
|
||||
|
||||
# mark progress complete
|
||||
session = get_progress_session(self.request, 'upgrades.execute')
|
||||
session = get_progress_session(self.request, "upgrades.execute")
|
||||
session.load()
|
||||
session['complete'] = True
|
||||
session['success_msg'] = 'yay!'
|
||||
session["complete"] = True
|
||||
session["success_msg"] = "yay!"
|
||||
session.save()
|
||||
|
||||
# next call should reflect that
|
||||
self.assertEqual(self.request.session.pop_flash(), [])
|
||||
context = view.execute_progress()
|
||||
self.assertTrue(context.get('complete'))
|
||||
self.assertFalse(context.get('error'))
|
||||
self.assertTrue(context.get("complete"))
|
||||
self.assertFalse(context.get("error"))
|
||||
# nb. this is missing b/c we already got all contents
|
||||
self.assertNotIn('stdout', context)
|
||||
self.assertEqual(self.request.session.pop_flash(), ['yay!'])
|
||||
self.assertNotIn("stdout", context)
|
||||
self.assertEqual(self.request.session.pop_flash(), ["yay!"])
|
||||
|
||||
# nb. switch to a different progress tracker
|
||||
self.request.session.id = 'mockid#3'
|
||||
self.request.session.id = "mockid#3"
|
||||
|
||||
# first call should get the full contents
|
||||
context = view.execute_progress()
|
||||
self.assertFalse(context.get('complete'))
|
||||
self.assertFalse(context.get('error'))
|
||||
self.assertEqual(context['stdout'], 'hello 001<br />hello 002<br />')
|
||||
self.assertFalse(context.get("complete"))
|
||||
self.assertFalse(context.get("error"))
|
||||
self.assertEqual(context["stdout"], "hello 001<br />hello 002<br />")
|
||||
|
||||
# mark progress error
|
||||
session = get_progress_session(self.request, 'upgrades.execute')
|
||||
session = get_progress_session(self.request, "upgrades.execute")
|
||||
session.load()
|
||||
session['error'] = True
|
||||
session['error_msg'] = 'omg!'
|
||||
session["error"] = True
|
||||
session["error_msg"] = "omg!"
|
||||
session.save()
|
||||
|
||||
# next call should reflect that
|
||||
self.assertEqual(self.request.session.pop_flash('error'), [])
|
||||
self.assertEqual(self.request.session.pop_flash("error"), [])
|
||||
context = view.execute_progress()
|
||||
self.assertFalse(context.get('complete'))
|
||||
self.assertTrue(context.get('error'))
|
||||
self.assertFalse(context.get("complete"))
|
||||
self.assertTrue(context.get("error"))
|
||||
# nb. this is missing b/c we already got all contents
|
||||
self.assertNotIn('stdout', context)
|
||||
self.assertEqual(self.request.session.pop_flash('error'), ['omg!'])
|
||||
self.assertNotIn("stdout", context)
|
||||
self.assertEqual(self.request.session.pop_flash("error"), ["omg!"])
|
||||
|
||||
def test_configure_get_simple_settings(self):
|
||||
# sanity/coverage check
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue